LangGraph核心模型与实战：从条件路由到并行分支的Agent状态机设计

先回答一个很多人问过的问题：LangChain 学了很久，示例能跑，但一遇到“答案不满意就重试”“检索完判断要不要调用工具”“多个任务并行执行再汇总”这种真实 Agent 需求，代码就变成了一堆散落的 if-else 和全局变量，维护成本直线上升。

LangGraph 解决的正是这个问题：它把 Agent 的流程变成一张有向图，把中间数据变成显式的 State，让开发者重新拿回控制权。

本文会从零讲清楚 LangGraph 的核心模型，并给出一套可运行的实战样例，覆盖条件路由、循环检测、State 修改、并行分支、子图和状态持久化。读完你能自己搭建一个带重试机制和并行任务的 Agent 图，也知道该去哪里排查问题。文章配套的代码尽量保持最小化，方便你复制后直接运行验证。

1. 为什么 LangGraph 值得专门花时间学

先看一个真实场景。你有一个 RAG 问答功能：用户输入问题，系统检索知识库，大模型生成答案。用普通 Python 写，大概是“检索函数 + 生成函数 + 主流程”。看起来不难，但加上以下需求后就开始失控：

• 答案质量不合格时，要改写问题重新检索，最多重试 3 次。
• 某些问题需要调用外部工具（查天气、查订单、查库存），调完之后还要再生成一次。
• 用户连续追问时，要保留历史消息，避免模型“失忆”。
• 多个检索源要并行执行，全部完成后再汇总。

如果用普通代码管理，你需要自己维护循环条件、状态传递、超时控制、重试上限，稍不注意就出现无限循环或状态覆盖。很多项目最后变成“能用，但不敢改”。

LangGraph 的价值在于：它把“流程控制”从业务代码里剥离出来，交给图执行引擎。节点函数只关心“输入 State 是什么，输出什么更新”，至于下一步走到哪、是否循环、是否并行、是否中断，由图的边和配置决定。

适合学 LangGraph 的人，通常已经具备以下特征之一：

• 用 LangChain 做过 Demo，但对 Chain 的线性执行模型不满意。
• 正在开发客服、RAG 重写、工具调用型 Agent，需要复杂的控制流。
• 希望把 Agent 流程做得可测试、可恢复、可观测。

不适合的情况也有：如果只是调一次大模型接口返回结果，不需要图；如果业务流程非常简单且不会增长，上 LangGraph 是过度设计。

2. LangGraph 核心概念：一张图、四样东西

LangGraph 的学习曲线不在于 API 数量，而在于思维方式的转变。它的核心抽象可以浓缩成四个概念。

2.1 State：图运行过程中的数据中枢

State 是一个TypedDict，定义了图在运行期间需要维护的全部数据。所有节点都能读取当前 State，节点返回的更新会被合并回 State。

例如一个问答 Agent 的 State 可能是这样的：

from typing import TypedDict class QAState(TypedDict): question: str answer: str retry_count: int

设计 State 时有一个原则：只放节点之间需要共享的数据，能推导出来的字段不要存。比如answer_length这种可以由answer算出来的字段，就不应该进入 State。

2.2 Node：图中的一个处理单元

Node 就是一个普通的 Python 函数或可调用对象。它接收当前 State，返回一个 dict，表示要更新的字段。

def generate(state: QAState) -> dict: return {"answer": "这是生成的答案"}

这里有一个新手最容易踩的坑：不要在节点函数里直接修改传入的 state 对象，而是返回需要更新的字段字典。LangGraph 的推荐做法是返回一个 partial state，由框架负责合并。

2.3 Edge：节点之间的流转关系

Edge 定义节点之间的连接。最基本的边是无条件边：A 执行完之后，一定去 B。

graph.add_edge("node_a", "node_b")

边的类型有两种：

• add_edge：无条件跳转。
• add_conditional_edges：条件路由，根据 State 动态决定下一步。

2.4 Conditional Edge：真正的控制流入口

条件路由是 LangGraph 区别于普通链式框架的关键。它允许一个节点执行完后，由路由函数决定跳转到哪个节点。

graph.add_conditional_edges( "evaluate", router, # 路由函数，接收 state，返回节点名 { "continue": "generate", "end": END, } )

路由函数是一个纯函数：输入 State，输出一个字符串 key，LangGraph 根据 key 在映射表里找到目标节点。这个设计让“判断”和“执行”分离，逻辑非常清晰。

2.5 LangGraph 和 LangChain 的关系

很多人在搜“LangGraph 和 LangChain 的区别”。准确地说，它们解决的不是同一层的问题。

LangGraph 并不替代 LangChain，它们可以配合使用。你完全可以在 LangGraph 的节点里调用 LangChain 的 Retriever、Tool、ChatModel。LangGraph 管“流程怎么走”，LangChain 管“每一步用什么组件”。

3. 环境准备与最小示例

实操之前，先准备好环境。本文使用的是 Python 生态，代码以 LangGraph 的核心 API 为主。为了避免版本绑定问题，我建议你按照实际操作时的官方安装方式为准。

pip install langgraph

如果要在节点里使用 LangChain 的模型消息类型，额外安装：

pip install langchain-core

环境要求：建议 Python 3.9 及以上，具体版本以你安装的 LangGraph 依赖要求为准。

下面是最小示例：两个节点，从 START 进入 node_a，再到 node_b，最后到 END。

# graph_minimal.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class GraphState(TypedDict): text: str def node_a(state: GraphState) -> dict: return {"text": state["text"] + " -> A"} def node_b(state: GraphState) -> dict: return {"text": state["text"] + " -> B"} builder = StateGraph(GraphState) builder.add_node("node_a", node_a) builder.add_node("node_b", node_b) builder.add_edge(START, "node_a") builder.add_edge("node_a", "node_b") builder.add_edge("node_b", END) app = builder.compile() result = app.invoke({"text": "start"}) print(result)

运行：

python graph_minimal.py

预期输出：

{'text': 'start -> A -> B'}

这个示例虽然简单，但已经把最核心的执行逻辑讲清楚了：图接收初始 State，按边执行节点，每个节点返回的 dict 更新 State，最终输出完整的 State。

如果运行失败，优先检查两点：安装的 langgraph 是否成功；Python 版本是否满足依赖要求。

4. 条件路由与循环检测：做一个带重试的 QA Agent

条件路由最典型的应用场景就是“判断结果是否合格，决定下一步动作”。下面用一个带重试机制的 QA Agent 来演示。

业务逻辑是：

1. 生成答案。
2. 评估答案质量。
3. 如果答案质量不合格且还没超过最大重试次数，回到生成节点重新生成。
4. 超过次数后无条件结束。

代码如下：

# graph_conditional.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class QAState(TypedDict): question: str answer: str retry_count: int route: str def generate(state: QAState) -> dict: current_try = state["retry_count"] + 1 return { "answer": f"第 {current_try} 次生成的答案", "retry_count": current_try, } def evaluate(state: QAState) -> dict: # 模拟质量评估：超过最大次数，或答案包含关键词，则结束 if state["retry_count"] >= 3: return {"route": "end"} if "可用" in state["answer"]: return {"route": "end"} return {"route": "regenerate"} def router(state: QAState) -> str: return state["route"] builder = StateGraph(QAState) builder.add_node("generate", generate) builder.add_node("evaluate", evaluate) builder.add_edge(START, "generate") builder.add_edge("generate", "evaluate") builder.add_conditional_edges( "evaluate", router, { "regenerate": "generate", "end": END, } ) app = builder.compile() result = app.invoke({ "question": "LangGraph 是什么", "answer": "", "retry_count": 0, "route": "", }) print(result)

这里的关键在于add_conditional_edges的用法：evaluate 节点执行完后，不会直接走固定的下一条边，而是调用 router 函数，router 返回regenerate就回到 generate，返回end就结束。

运行这个示例，你会看到 retry_count 从 1 增长到 3，最终停在 END。它证明了 LangGraph 天然支持循环，不需要你用 while 自己控制。

循环检测有两个层面。第一个层面是业务层面：在 State 中维护计数，由节点函数判断是否继续。第二个层面是框架层面：LangGraph 有recursion_limit配置，防止因逻辑 bug 导致无限循环。

result = app.invoke( {"question": "LangGraph 是什么", "answer": "", "retry_count": 0, "route": ""}, config={"recursion_limit": 10}, )

当图的执行步数超过recursion_limit时，框架会抛出异常。生产环境建议两个层面都设置：业务逻辑里做循环上限判断，invoke 时也设置一个大于预期最大步数的 limit 作为兜底。

5. 在节点函数中正确修改 State 状态值

这是 LangGraph 新手问得最多的问题之一：节点函数里到底怎么改 State？

很多人的第一反应是：

def my_node(state): state["answer"] = "xxx" # 不推荐

LangGraph 的推荐做法是：节点函数不直接原地修改 State 对象，而是返回一个 dict，包含想要更新的字段。LangGraph 会把返回值合并进全局 State。

def my_node(state): return {"answer": "xxx"}

为什么这样设计？因为 LangGraph 需要知道 State 的变更历史，才能支持断点恢复、时间旅行、并行分支合并等能力。如果所有节点都原地改同一个字典，框架很难追踪“谁改了什么”。

但如果字段是列表、消息序列这类需要“追加”而不是“覆盖”的数据，直接返回新值会把旧值整个替换掉。解决办法是使用 reducer。

下面是一个消息追加的示例：

# graph_state_update.py from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage class ChatState(TypedDict): messages: Annotated[list, add_messages] status: str def chat_node(state: ChatState) -> dict: user_text = state["messages"][-1].content return { "messages": [AIMessage(content=f"收到：{user_text}")], "status": "processed", } builder = StateGraph(ChatState) builder.add_node("chat", chat_node) builder.add_edge(START, "chat") builder.add_edge("chat", END) app = builder.compile() result = app.invoke({ "messages": [HumanMessage(content="你好 LangGraph")], "status": "pending", }) print(result)

输出结果中，messages 会同时包含 HumanMessage 和 AIMessage，而不是把 HumanMessage 覆盖掉。这是因为messages字段的类型是Annotated[list, add_messages]，LangGraph 发现该字段有 reducer，就会用 reducer 合并新旧值。

不同字段的更新行为对比：

如果你需要自定义合并逻辑，只要在类型标注里写Annotated[list, my_reducer]即可。reducer 是一个接收两个列表并返回一个新列表的函数。

6. 并行分支与子图：复杂 Agent 的组装方式

真实 Agent 经常需要同时做多件事。比如用户输入一个问题后，系统要并行检索知识库、查询历史会话、调用天气工具，等所有结果都返回后再汇总。

LangGraph 支持两种并行方式：静态并行和动态并行。

6.1 静态并行：fan-out 与 fan-in

在一个 StateGraph 中，多个节点可以同时从一个起始节点流出，然后汇聚到一个节点。下面是一个并行检索的示例。

# graph_parallel.py from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END def merge_list(left: list, right: list) -> list: return left + right class ParallelState(TypedDict): question: str results: Annotated[list, merge_list] def search_kb(state: ParallelState) -> dict: return {"results": [f"知识库结果：{state['question']}"]} def search_history(state: ParallelState) -> dict: return {"results": [f"历史会话结果：{state['question']}"]} def aggregate(state: ParallelState) -> dict: return {"results": [f"汇总：{state['results']}"]} builder = StateGraph(ParallelState) builder.add_node("search_kb", search_kb) builder.add_node("search_history", search_history) builder.add_node("aggregate", aggregate) builder.add_edge(START, "search_kb") builder.add_edge(START, "search_history") builder.add_edge("search_kb", "aggregate") builder.add_edge("search_history", "aggregate") builder.add_edge("aggregate", END) app = builder.compile() result = app.invoke({"question": "LangGraph 并行", "results": []}) print(result)

这段代码的关键在于results字段使用了merge_listreducer。两个并行节点都会向results追加内容，如果没有 reducer，后返回的节点会覆盖先返回的节点，aggregate 只能看到一个来源的数据。

并行节点更新同一个 State 字段时，必须通过 reducer 定义合并规则，这是并行写法里最重要的注意事项。

6.2 动态并行：使用 Send API

如果并行任务的数量是动态的，比如用户上传了 5 个文件，每个文件需要一个处理节点，这时静态写边就不够了。LangGraph 提供了SendAPI，用于在条件路由中动态创建多个并行任务。

from langgraph.types import Send def continue_to_process(state): return [ Send("process_one", {"item": item}) for item in state["items"] ]

然后在图中使用条件边：

builder.add_conditional_edges( "distribute", continue_to_process, ["process_one"], )

Send的第一个参数是目标节点名，第二个参数是传给该节点的 state。这样就能把一个分发节点拆成 N 个并行任务。具体写法在不同版本中可能略有差异，建议以官方文档为准。

6.3 子图：把一张图塞进另一个节点

当一个图变得复杂时，你应该把它拆成多个子图。LangGraph 支持把一张已编译的图作为另一个图的节点来使用，这很适合模块化设计。

inner_builder = StateGraph(InnerState) # ... 定义 inner 图的节点和边 ... inner_app = inner_builder.compile() outer_builder = StateGraph(OuterState) outer_builder.add_node("inner", inner_app)

这种做法相当于把一个完整的子流程封装成一个节点。外层图只看到“有一个节点叫 inner”，内部逻辑完全隔离。注意子图的输入输出字段需要和外层 State 对齐，如果字段名不一致，可以在子图入口和出口加转换节点。

7. 状态持久化与断点恢复

Agent 应用一旦进入生产环境，“执行到一半进程挂了怎么办”“用户的会话状态怎么保存”就是必须面对的问题。LangGraph 的 Checkpointer 机制解决的就是这个问题。

引入 checkpointer 后，LangGraph 会在每一步执行后保存 State 快照。配合thread_id，可以区分不同会话。

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() app = builder.compile(checkpointer=memory) config = {"configurable": {"thread_id": "thread-1"}} result1 = app.invoke({"question": "你好"}, config=config) result2 = app.invoke({"question": "继续刚才的话题"}, config=config)

有了 checkpointer，同一个thread_id的多次 invoke 之间，State 是共享的。这意味着多轮对话不再需要自己手动拼接历史消息，checkpoint 会帮你维护。

生产环境不建议使用MemorySaver，因为数据只存在内存里，进程重启就丢了。更稳妥的做法是使用基于 SQLite 或 Postgres 的持久化 saver，并将存储会话数据的数据库单独备份管理。具体 saver 的导入路径和配置方式会随版本变化，接入前务必查阅官方文档。

需要注意的安全边界：checkpoint 里可能包含用户敏感消息。如果用于生产，存储端要做访问控制，遵循最小权限原则，不要把所有用户的 thread 放在一个无鉴权的存储里。

8. 常见问题与排查思路

排错时有一个通用顺序：先看异常发生在“图执行层”还是“节点内部”。如果是图层面，错误信息通常会提到节点名和边；如果是节点内部，通常能看到业务代码的堆栈。给每个节点函数加一行入口日志，是定位问题最便宜的手段。

9. 工程落地与最佳实践

9.1 设计 State 时保持克制

State 是图的数据协议，字段越少越容易维护。每新增一个字段前，问自己：这个字段真的需要跨节点共享吗？能否由其他字段推导出来？

9.2 节点函数尽量写成纯函数

节点函数只依赖传入的 State，不读写全局变量，不直接操作外部状态。这样每个节点都可以单独测试，也方便在失败时重放 State。

9.3 条件路由函数保持轻量

路由函数只负责根据 State 返回节点名，不要在里面做重 IO、调模型、查数据库。路由函数执行频率高，一旦变慢会影响整个图的调度。

9.4 永远做循环防护

即使业务上认为不会无限循环，也要设置recursion_limit，并在 State 里维护计数。线上 Agent 的异常往往不是模型报错，而是流程失控。

9.5 密钥与安全边界

模型 API Key、数据库密码等敏感信息放在环境变量或密钥管理服务中，不要提交到代码库。节点函数如果访问外部系统，要考虑超时、限流和失败降级。Agent 工具调用还可能涉及用户数据权限，工具层要做鉴权，不能因为“流程跑通了”就跳过权限校验。

9.6 测试策略

对条件路由函数做单元测试是最划算的：给定 State 输入，断言返回的节点名正确。对完整图做集成测试时，用 mock 模型固定返回内容，避免测试结果不稳定。涉及数据库或外部服务的变更，先在测试环境验证，再走发布流程。

9.7 不要过度设计

如果一个流程只需要三步线性调用，直接写函数调用更清晰。Graph 的价值在复杂度出现以后才显现。一张图如果超过十几个节点，建议拆成多个子图，否则排查问题时会很难受。

9.8 关注版本兼容性

LangGraph 迭代速度不慢，API 偶尔会有调整。学习时以官方文档为准，不要照搬过时的博客代码。搜索“LangGraph 是否有 Rust 版本”这类问题时也要注意：实际开发中 Python 生态最为完整，除非你有明确的跨语言需求，否则优先走 Python 路线，把核心的状态机设计思想学扎实。

10. 总结与后续学习方向

本文的核心判断是：LangGraph 不值得死记 API，真正需要理解的是“State 定义数据、Node 处理状态、Edge 控制流转、Conditional Edge 实现决策”这套状态机模型。把最小示例跑通之后，再逐步加入条件路由、循环检测、并行分支、子图和 checkpointer，就是一条很平滑的学习路径。

如果你今天只做一件事，建议把第 4 节的条件路由示例亲自跑一遍，并试着把“重试判断”改造为自己的业务逻辑。跑通了之后，再去看官方文档里的高级主题，你会发现很多概念已经自然理解了。

下一步可以从这几个方向继续深入：

• 用 LangGraph 重写一个已有的 LangChain 线性 Demo，从链式调用改造成图结构。
• 给图加上 checkpointer，体验多轮对话和历史恢复。
• 实现一个人工审核节点：Agent 生成结果后暂停，人工确认后再继续执行。
• 把多个子图组合成一个完整的上层应用。

关于网上各种“2026 全套教程”的说法，有一点可以放心：LangGraph 的核心模型短期内不会发生颠覆性变化。把 Graph、State、Node、Edge 这四个概念吃透，再多的版本更新对你来说都只是增量知识。