LangGraph 入门：为什么 Agent 需要状态机

在构建多步骤 AI Agent 时，一个常见的挑战是流程控制：如何让 Agent 在执行过程中，能够根据中间结果做出不同的决策，而不是完全依赖 LLM 自行判断下一步做什么。

状态机：一种管理程序"当前所处阶段"的设计方式——程序在任意时刻都处于某个明确的状态（如"搜索中"、"生成答案"），并根据条件在不同状态之间切换。

LangChain 的 ReAct Agent 是处理简单任务的常见方案，但随着任务复杂度上升，它的局限会逐渐显现。LangGraph 正是为解决这个问题而设计的。

图：左侧 ReAct Agent 的流程完全由 LLM 自己决定，开发者无法干预；右侧 LangGraph 由开发者定义图结构，LLM 只在决策节点参与。

为什么需要状态机：从一个失控的 Agent 说起

假设你在构建一个研究 Agent，用户请求是"帮我研究 2026 年 AI 框架的发展趋势"。

用 LangChain ReAct Agent 会发生什么

Agent 开始搜索"2026 AI 框架趋势"
搜索返回 20 篇文章，Agent 决定逐篇阅读
读到第 5 篇时，发现提到了"multi-agent 系统"，Agent 决定深入研究这个方向
Agent 去搜索 multi-agent，又找到 15 篇文章
循环继续……30 分钟后，Agent 还在搜索，没有任何输出

这个问题有个名字：目标漂移（Goal Drift）。Agent 的注意力被中间步骤的细节吸引，偏离了原始目标。

ReAct Agent 的根本问题在于：流程控制完全由 LLM 自己决定。LLM 什么时候停、走哪条路、用哪个工具，开发者无法从外部干预。

LangGraph 如何解决这个问题

LangGraph 引入状态机，把执行流程从 LLM 的脑子里搬到开发者定义的图结构上：

code

研究 Agent 的图结构：
START → 分解任务 → 并行搜索(最多3个关键词) → 汇总结果 → 评估质量
         ↑                                                    ↓
         └──────────────────如果质量不够────────────────────┘
                                     ↓ 如果质量够了
                              生成报告 → END

这张图明确规定了：最多搜索 3 个关键词（不能无限扩展），质量评估后决定是重搜还是生成（条件分支），流程有明确的终止条件。LLM 只负责在"评估质量"节点判断"够不够好"，不负责决定"下一步做什么"。

LangChain Agent 的局限

LangChain 的 ReAct Agent 工作模式是：LLM 判断要用什么工具 → 调用工具 → 把结果喂回 LLM → LLM 再判断 → 循环直到 LLM 说"我完成了"。

这个模式有几个根本性的弱点：

流程控制完全由 LLM 决定：LLM 什么时候停、走哪条路、用哪个工具，开发者无法从外部干预。不同的输入、不同的随机性，执行路径可能完全不同。

状态是隐式的：Agent 的执行状态（当前步骤、已完成的工具调用、中间结果）全部隐藏在框架内部。开发者无法直接访问这些状态，出错时只能靠日志猜测哪一步出了问题。

无法精确实现条件逻辑："先做 A，再根据 A 的结果决定做 B 还是 C"这种条件逻辑，在 ReAct Agent 里只能靠 Prompt 描述，能不能执行得到取决于 LLM 的理解。

循环难以控制：Agent 可能重复调用同一个工具，或者在某个分支上无限循环。设置最大步数是个粗糙的保护，不能精细控制。

归根结底，ReAct Agent 适合简单的问答任务，不适合需要精确流程控制的复杂业务场景。

LangGraph 是什么

LangGraph 是 LangChain 团队开发的框架，核心思路是：把 Agent 的执行流程建模成有向图（图：像流程图一样，节点之间的连线有方向，规定了执行顺序）。

每个节点（Node）是一个处理步骤，边（Edge）决定节点之间的执行顺序。整个 Agent 的行为变成了一张开发者定义的图，流程完全可控。

开发者不再依赖 LLM 自己决定下一步做什么，而是通过定义图的结构来控制流程，LLM 只负责在它该做决策的节点里做决策。

这个思路和 Java 里的工作流引擎（比如 Activiti、Flowable，这类工具专门用于定义和执行业务流程，如"收到申请→审批→通知"这样的多步骤流转）非常相似，但更轻量，专门为 LLM 应用场景设计。

LangGraph 状态机工作原理——State 共享数据结构、图中的节点与条件边、以及执行流程三者协同

四个核心概念

图：State、Node、Edge、Conditional Edge 四个核心概念的关系——State 是共享数据，Node 读写 State，Edge 连接节点，Conditional Edge 根据 State 动态路由。

State（状态）

State 是贯穿整个图的共享数据结构。所有节点都从 State 里读数据，也把处理结果写回 State。

用 Python 的 TypedDict（Python 内置的类型工具，用于定义字典的结构和字段类型，相当于给字典加上"字段说明书"）来定义：

python

from typing import TypedDict, List

class AgentState(TypedDict):
    user_input: str           # 用户输入
    search_results: List[str] # 搜索结果
    quality_score: float      # 质量评分
    final_answer: str         # 最终答案
    step_count: int           # 执行步数，用于防止死循环

State 的设计是整个图的核心。每个节点不需要知道其他节点的存在，只需要从 State 读取它需要的字段，写入它产出的字段。这种解耦方式让每个节点都可以独立测试。

State 的设计原则：

只在 State 里存储节点之间需要共享的信息，不需要共享的就放在节点函数的局部变量里
字段名要清晰，让人一眼就能看出这个字段的用途
对于会被并发节点写入的字段，需要使用 Reducer（稍后介绍）

Node（节点）

节点是一个普通的 Python 函数，接收当前 State，返回需要更新的字段（以字典形式返回，LangGraph 会自动合并到 State 里）。

python

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def search_node(state: AgentState) -> dict:
    """执行搜索的节点"""
    query = state["user_input"]
    # 实际项目里调用真实的搜索 API（如 Tavily）
    results = [
        f"搜索结果1：关于 '{query}' 的最新信息...",
        f"搜索结果2：'{query}' 的历史背景...",
    ]
    print(f"[搜索节点] 搜索：{query}，找到 {len(results)} 条结果")
    return {
        "search_results": results,
        "step_count": state.get("step_count", 0) + 1
    }

def evaluate_node(state: AgentState) -> dict:
    """评估搜索结果质量的节点"""
    results = state["search_results"]
    # 简化示例：基于结果数量判断质量
    # 实际项目里让 LLM 评估相关性
    score = min(len(results) / 3.0, 1.0)  # 至少需要 3 条结果才算质量达标
    print(f"[评估节点] 质量分数：{score:.2f}")
    return {"quality_score": score}

def answer_node(state: AgentState) -> dict:
    """生成答案的节点"""
    search_results = state["search_results"]
    user_input = state["user_input"]

    combined = "\n".join(search_results)
    response = llm.invoke(f"基于以下搜索结果，回答问题：{user_input}\n\n搜索结果：\n{combined}")

    print(f"[答案节点] 生成最终答案")
    return {"final_answer": response.content}

节点函数只更新它负责的字段，不需要返回完整的 State。LangGraph 会把返回的字典和现有 State 合并。

Edge（边）

边定义了节点之间的执行顺序。最简单的是普通边，固定从节点 A 到节点 B：

python

graph.add_edge("search", "evaluate")  # 搜索完成后，进入评估

还有条件边（Conditional Edge），根据 State 的值动态决定下一步：

python

def should_continue(state: AgentState) -> str:
    """路由函数：根据质量分数决定下一步"""
    if state.get("quality_score", 0) >= 0.7:
        return "answer"   # 质量够了，生成答案
    elif state.get("step_count", 0) >= 3:
        return "answer"   # 超过 3 次搜索，强制结束
    else:
        return "search"   # 质量不够，重新搜索

graph.add_conditional_edges(
    "evaluate",      # 从评估节点出发
    should_continue, # 路由函数决定走哪条路
    {
        "search": "search",  # 返回 "search" 则去搜索节点
        "answer": "answer",  # 返回 "answer" 则去答案节点
    }
)

StateGraph（状态图）

StateGraph 是把节点和边组合在一起的容器，是整个图的"画板"。

python

from langgraph.graph import StateGraph, START, END

graph = StateGraph(AgentState)  # 指定 State 类型
graph.add_node("search", search_node)
graph.add_node("evaluate", evaluate_node)
graph.add_node("answer", answer_node)

graph.add_edge(START, "search")          # 入口：从搜索开始
graph.add_edge("search", "evaluate")     # 搜索后进入评估
graph.add_conditional_edges("evaluate", should_continue, {
    "search": "search",
    "answer": "answer",
})
graph.add_edge("answer", END)            # 答案节点后结束

app = graph.compile()  # 编译成可执行对象

一个完整的示例：带循环的研究 Agent

以下示例演示一个会循环搜索、直到质量达标才停止的研究 Agent：

LangGraph 执行流程

图：LangGraph 的典型执行流程——START 进入后经路由节点判断，分叉为工具调用或直接回答两条路径，最终汇聚到生成答案节点，右侧标注每步 State 的变化。

python

from typing import TypedDict, List
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)


# 1. 定义 State
class ResearchState(TypedDict):
    query: str
    search_results: List[str]
    quality_score: float
    search_count: int      # 搜索次数
    final_answer: str


# 2. 定义节点函数
def search_node(state: ResearchState) -> dict:
    """搜索节点：执行搜索并追加结果"""
    query = state["query"]
    count = state.get("search_count", 0)

    # 模拟搜索（实际项目替换为 Tavily API）
    new_results = [
        f"第{count+1}次搜索结果1：关于 '{query}' 的信息...",
        f"第{count+1}次搜索结果2：'{query}' 的详细说明...",
    ]

    # 追加到已有结果（不是覆盖）
    existing_results = state.get("search_results", [])
    updated_results = existing_results + new_results

    print(f"[搜索] 第 {count+1} 次搜索，累计 {len(updated_results)} 条结果")

    return {
        "search_results": updated_results,
        "search_count": count + 1,
    }


def evaluate_node(state: ResearchState) -> dict:
    """评估节点：判断搜索结果是否足够"""
    results = state.get("search_results", [])
    # 简化：5 条以上认为质量达标
    # 实际项目里让 LLM 评估相关性和完整性
    score = min(len(results) / 5.0, 1.0)
    print(f"[评估] 当前 {len(results)} 条结果，质量分数：{score:.2f}")
    return {"quality_score": score}


def answer_node(state: ResearchState) -> dict:
    """答案节点：基于搜索结果生成最终报告"""
    results = state.get("search_results", [])
    query = state["query"]

    context = "\n".join(results[:8])  # 最多使用 8 条结果
    response = llm.invoke(
        f"基于以下研究资料，回答问题：{query}\n\n资料：\n{context}"
    )

    print(f"[答案] 基于 {len(results)} 条研究资料，生成了最终答案")
    return {"final_answer": response.content}


# 3. 定义路由函数
def route_after_evaluate(state: ResearchState) -> str:
    """路由：质量达标或搜索次数超限就生成答案，否则继续搜索"""
    quality = state.get("quality_score", 0)
    count = state.get("search_count", 0)

    if quality >= 0.8:
        print(f"[路由] 质量达标（{quality:.2f}），生成答案")
        return "answer"
    elif count >= 3:
        print(f"[路由] 已搜索 {count} 次，强制生成答案")
        return "answer"
    else:
        print(f"[路由] 质量不够（{quality:.2f}），继续搜索")
        return "search"


# 4. 构建图
def build_research_graph():
    graph = StateGraph(ResearchState)

    # 添加节点
    graph.add_node("search", search_node)
    graph.add_node("evaluate", evaluate_node)
    graph.add_node("answer", answer_node)

    # 添加边
    graph.add_edge(START, "search")
    graph.add_edge("search", "evaluate")
    graph.add_conditional_edges(
        "evaluate",
        route_after_evaluate,
        {
            "search": "search",   # 继续搜索（形成循环！）
            "answer": "answer",
        }
    )
    graph.add_edge("answer", END)

    return graph.compile()


# 5. 运行
if __name__ == "__main__":
    app = build_research_graph()

    result = app.invoke({
        "query": "2026年LangGraph的主要特性更新",
        "search_results": [],
        "search_count": 0,
        "quality_score": 0.0,
        "final_answer": "",
    })

    print("\n=== 最终答案 ===")
    print(result["final_answer"])
    print(f"\n总搜索次数：{result['search_count']}")

运行输出示例：

code

[搜索] 第 1 次搜索，累计 2 条结果
[评估] 当前 2 条结果，质量分数：0.40
[路由] 质量不够（0.40），继续搜索
[搜索] 第 2 次搜索，累计 4 条结果
[评估] 当前 4 条结果，质量分数：0.80
[路由] 质量达标（0.80），生成答案
[答案] 基于 4 条研究资料，生成了最终答案

=== 最终答案 ===
（LLM 生成的研究报告内容）

总搜索次数：2

这里有一个关键点：图中存在循环。evaluate 节点可能把流程路由回 search 节点，形成"搜索 → 评估 → 再搜索 → 再评估"的循环。这在 LCEL 线性链中是无法表达的，但在 LangGraph 的图结构中是自然的。

图的结构可视化

上面这个示例对应的图结构：

这张图清晰地表达了整个 Agent 的执行逻辑。任何人看到这张图，都能理解：Agent 会先搜索，然后评估，如果不够好就再搜索，直到够好或超过次数限制才生成答案。

关键区别：这张图是开发者定义的，不是 LLM 在运行时决定的。执行路径的可能性被明确约束在这张图里。

和 LangChain ReAct Agent 的对比

维度	LangChain ReAct Agent	LangGraph
流程控制	LLM 自己决定	开发者定义图结构
循环控制	容易失控，依赖 LLM 判断终止	通过计数器和条件边精确控制
条件逻辑	依赖 LLM 判断，不稳定	用条件边精确定义
可调试性	很难，黑盒	每个节点独立可测试
人工介入	不支持	原生支持 interrupt
状态可见性	隐式，开发者看不到	显式 TypedDict，完全可见
并行执行	不支持	Send API 支持
持久化	不支持	Checkpointer 支持
适合场景	简单问答，几步以内	复杂多步骤任务，生产级 Agent

ReAct Agent 并不是坏的选择，对于简单的"查个信息、算个数字"类型的任务，它足够用，代码量也少。但一旦任务复杂起来，一旦需要可靠性，LangGraph 就是更合适的工具。

对 Java 开发者的类比

State 就是一个数据对象（类似 Java 的 VO/DTO），Node 就是一个 Service 方法，Edge 就是方法之间的调用关系。整个 LangGraph 图就是一个 Service 层的业务流程，只不过 LangGraph 把这个结构形式化了，原生支持暂停、恢复、并行执行这些特性。

如果用过 Spring Batch 的 Step 和 Flow，会觉得 LangGraph 非常眼熟：

Spring Batch 的 Step = LangGraph 的 Node
Spring Batch 的 Flow = LangGraph 的 StateGraph
Spring Batch 的 Jobexecution context = LangGraph 的 State
Spring Batch 的 Conditional Flow = LangGraph 的 Conditional Edge

核心思路完全一样：把复杂流程拆成可管理的步骤，每个步骤独立运行，步骤之间通过状态传递数据。

小结

LangGraph 不是 LangChain 的替代品，而是在 LangChain 之上专门解决"复杂 Agent 流程控制"这个问题的框架。

需要精确控制流程的场景——多步骤任务、条件分支、循环逻辑、人工审核——LangGraph 是目前最成熟的选择。

如果任务简单，直接用 LangChain 或者手写几十行 Python 就够了。一旦任务复杂起来，一旦遇到"这个 Agent 为什么总是绕圈子"、"这个 Agent 崩溃了怎么从断点恢复"这类问题，就是引入 LangGraph 的时候了。

下一篇会讲 Conditional Edge 的详细用法，实现"根据搜索结果质量决定是否继续搜索"这种真正有用的循环 Agent。