LangGraph入门：状态图、条件路由与Agent工具调用循环

最近连续有几个人跑来问我同一个问题LangGraph 和 LangChain 到底什么区别问得多了我发现大家不是不懂用法而是卡在了同一个点上——链式调用写多了之后根本想象不出图怎么组织 Agent。这篇就围绕标题里的三件事来写StateGraph 状态图、条件路由、Agent 的工具调用循环。理解这三块LangGraph 基本就算入门了剩下的 API 都是查文档就能解决的问题。这篇内容面向两类人一类是用过 LangChain 但被 Agent 循环绕晕的老手另一类是从没写过 LangGraph、想直接上手搭 Agent 的新人。我会把原理、代码、踩坑经验混在一起讲不铺垫废话直接拆。1. 为什么是图从顺序链到 LangGraph 的思维转变1.1 链式调用的天花板先说个实际场景。用 LangChain 写一个简单的客服 Agent最初版通常长这样result llm.invoke(user_input) if result.should_call_tool: tool_result call_tool(result.tool_calls) final llm.invoke([user_input] [result] [tool_result]) else: final result这段代码本身没问题但你会发现一旦业务流程复杂起来它就开始变形要加记忆得在外部维护数组要加重试机制得再包一层循环要让不同的用户问题走不同的处理器得写一堆 if-else。我见过不少项目最后变成了一坨几百行的顺序代码每个新需求都要往里面插一段逻辑改一个地方就要顺带检查全流程。这就是链式调用的天花板它隐含假设执行路径是线性的。但真实 Agent 的决策流程是分叉的、会回环的。模型输出一个工具调用你需要去执行工具执行完再把结果喂回去这本来就是循环循环里还可能根据不同的工具结果走不同的分支这又是条件跳转。用链去表达这些东西代码结构和业务逻辑根本不匹配。1.2 图模型的核心节点、边、状态LangGraph 换了一种表达方式把整个流程画成一张图。图里有三个基本概念。节点Node一个执行单元接收状态、处理逻辑、返回状态更新。在 Agent 里调用模型是一个节点执行工具是另一个节点。边Edge定义节点之间的跳转关系。普通边表示跑完 A 直接去 B条件边表示跑完 A 后根据某个结果决定去 B 还是去 C。状态State贯穿全程的共享数据容器。每个节点都能读它、更新它更新结果会叠加到状态上传给下一个节点。这个模型的妙处在于你把流程怎么走和每一步做什么拆开了。节点只关心自己那部分逻辑模型节点只负责生成回复工具节点只负责执行工具。至于下一步去哪由边来决定。这样改流程的时候你不用动节点内部代码只要重新连边就行。1.3 LangGraph 和 LangChain 到底差在哪很多人误以为 LangGraph 是 LangChain 的升级版其实它俩定位不一样。LangChain 侧重的是组件库它提供了模型封装、Prompt 模板、各种模型封装和工具接口解决的是怎么调用 LLM的问题。LangGraph 侧重的是编排层解决的是多个组件之间的执行流程怎么组织的问题。一个类比是LangChain 是零件仓库LangGraph 是流水线图纸。你当然可以直接从仓库拿零件手搓一个流水线但用图纸把工位、传送带、质检点画清楚后续维护和扩展都会轻松很多。实际项目中两者通常是配合使用的用 LangChain 的工具接口和模型封装用 LangGraph 把整个 Agent 流程串起来。2. StateGraph 状态设计Reducer、Schema 与共享内存2.1 用 TypedDict 声明状态第一步把全局变量定义清楚StateGraph 里的状态就是一个 TypedDict直接声明有哪些字段。入门时写复杂了容易懵最简单的状态长这样from typing import Annotated, TypedDict import operator class State(TypedDict): messages: Annotated[list, operator.add] count: int这里的messages是核心字段我会在 Agent 循环里反复往里追加消息。count是临时加的计数器用来演示普通字段的更新方式。声明状态用 TypedDict 而不是 Pydantic 模型是因为 TypedDict 足够轻量LangGraph 内部可以灵活处理它的合并逻辑。如果你偏好数据校验也可以换成 Pydantic 模型但入门阶段没必要。每次节点 return 的内容LangGraph 会把它当作对这个状态的一次更新然后根据字段对应的更新规则把新值合并进状态。如果字段没有声明合并规则默认是直接覆盖有了规则就按规则来。2.2 Reducer告诉 LangGraph 同一字段被多次更新时怎么合并Annotated[list, operator.add]是 LangGraph 里最容易卡住新人的写法。它的意思是当多个节点或多次执行都要更新messages字段时不做覆盖而是用operator.add把新列表追加到旧列表后面。为什么要这样做因为 Agent 循环里的消息列表天然是只增不减的。模型节点追加一条 AI 消息工具节点追加一条工具消息每次更新都是追加操作。如果你把messages声明成普通字段那么后一个节点更新的结果会把你上一轮的消息覆盖掉整个对话历史就丢了。自定义 Reducer 也不难就是一个签名(current_value, new_value) - final_value的函数def merge_lists(left: list, right: list) - list: return left right class State(TypedDict): messages: Annotated[list, merge_lists]我个人的习惯是所有只追加、不覆盖的字段都配上operator.add普通标量字段才留默认覆盖逻辑。这样能在一开始就避开大量并发更新导致的冲突问题——后面第 5 节我会专门讲这个坑。2.3 节点函数、add_edge 与编译运行的最小闭环节点就是一个普通函数输入 State输出一个 dictdict 的 key 必须是 State 声明的字段。from langgraph.graph import StateGraph, START, END def node_a(state: State): return {messages: [A 已执行], count: state[count] 1} def node_b(state: State): return {messages: [B 已执行], count: state[count] 1} builder StateGraph(State) builder.add_node(a, node_a) builder.add_node(b, node_b) builder.add_edge(START, a) builder.add_edge(a, b) builder.add_edge(b, END) graph builder.compile() result graph.invoke({messages: [], count: 0}) print(result[messages]) # [A 已执行, B 已执行]这里有两个细节值得注意。START和END是新版的写法早期文档里常见的是__start__和__end__。如果你在网上搜教程看到两种写法混着用别慌功能是一回事只是不同版本的命名调整。建议新项目一律用新写法。graph.compile()是必须的步骤。编译不只是做语法检查它会把整张图的结构固化下来生成内部执行计划。一些错误比如边指向不存在的节点会在编译阶段就暴露出来而不是等你运行到那一步才报错。2.4 从顺序执行到并行与分支的推演顺序图只是热身。StateGraph 真正厉害的地方在于节点之间的边可以组成任意形状。扇出A 同时连接 B 和 CA 执行完后 B 和 C 可以并行运行。汇聚B 和 C 都连接到 DD 要等 B、C 全部完成后才启动。自环节点 A 连接到它自己加上条件边就构成了循环。这里的关键词是并行。LangGraph 对并行分支的执行是异步调度、自动汇总的多个分支各自更新同一份状态各分支返回的字段更新会合并后再进入下一个节点。所以你在设计节点时不用自己写并发代码只要把图的结构搭对并行能力是天然的。但并行也带来约束如果有多个分支同时更新同一个字段那个字段必须配上 Reducer否则 LangGraph 不知道该听谁的会抛异常。我见过太多人在这里翻车——稀里糊涂写了个普通字段跑并行分支就报 Concurrent update 错误。解法就一句话凡是多个分支会碰的字段上 Reducer。3. 条件路由是怎么看图说话的3.1 路由函数的本质一个返回下一站的普通函数条件路由是 LangGraph 里最像逻辑判断的部分但它的原理很简单你写一个普通函数函数根据当前状态决定下一步去哪个节点然后返回值在图上被解释成一条边。from typing import Literal def route_function(state: State) - Literal[a, b]: if len(state[messages]) 5: return a return b builder.add_conditional_edges( current_node, route_function, {a: node_a, b: node_b}, )add_conditional_edges接收三个参数源节点名、路由函数、路径映射表。路由函数的返回值必须能在映射表里找到对应的目标节点。第三个参数的目的是解耦路由函数里可以不写真实的节点名而是返回一个抽象的路线标记映射表再把标记翻译成具体节点。这个设计很实用。假设你要替换节点实现只要改映射表不用动路由函数或者你想给同一个路由逻辑配不同的执行策略直接换张映射表就行。3.2 三个典型路由场景意图分流、校验重试、任务分类第一个场景是意图分流。用户消息进来先用分类节点判断走哪个处理流程def classify(state: State) - str: text state[query].lower() if 天气 in text: return weather if 计算 in text: return math return fallback builder.add_conditional_edges( classifier, classify, {weather: weather_node, math: math_node, fallback: fallback_node}, )第二个场景是校验重试。一个节点负责生成 SQL 语句另一个节点负责执行和校验如果校验失败就路由回生成节点重来一遍。这种当前节点的出边指向自己的写法就是重试循环的雏形。第三个场景是任务分类。从一组待办里识别出哪些该走人工审核、哪些自动处理然后分发给不同节点。核心都是用同一个路由函数根据状态内容直接跳到对应分支。3.3 条件路由踩坑清单条件路由我踩过不少坑列几条最典型的。路由函数返回的值如果在映射表里不存在运行时会抛异常所以返回类型和映射表的 key 要保持严格一致。我建议路由函数的返回类型直接声明成Literal[分支名1, 分支名2]这样类型检查器能在写代码时就帮你兜底。路由函数里读不到你想用的字段优先检查 state 的 key 名。TypedDict 不会在运行时强制 key 存在所以常常是字段名拼错了但编译不报错运行时报 KeyError。如果真的遇到先print(state.keys())看一遍真实字段。第三个坑是映射表值和实际节点名不一致。编译阶段通常会校验这一点但部分版本只在运行到条件边的时候才报错。遇到这种问题先检查节点名有没有拼错尤其注意大小写和下划线。4. Agent 工具调用循环让模型自己决定调用顺序4.1 工具调用循环的最小闭环Agent 的工具调用循环本质就是一个环形图模型节点生成回复如果回复里带着工具调用请求就进入工具节点去执行工具执行完把结果写回状态再回到模型节点让模型看到结果后继续决策。import operator from typing import Annotated, Literal, TypedDict, getattr from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode from langchain_core.tools import tool from langchain_openai import ChatOpenAI class State(TypedDict): messages: Annotated[list, operator.add] step_count: int tool def add(a: float, b: float) - float: Add two numbers together. return a b tools [add] tool_node ToolNode(tools) model ChatOpenAI(modelgpt-4o-mini) model_with_tools model.bind_tools(tools) def agent_node(state: State): response model_with_tools.invoke(state[messages]) return { messages: [response], step_count: state.get(step_count, 0) 1, } def should_continue(state: State) - Literal[continue, end]: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return continue return end builder StateGraph(State) builder.add_node(agent, agent_node) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_conditional_edges( agent, should_continue, {continue: tools, end: END}, ) builder.add_edge(tools, agent) graph builder.compile() initial_state {messages: [{role: user, content: 3 5 等于多少}], step_count: 0} result graph.invoke(initial_state) print(result[messages][-1].content)这段代码跑起来之后你观察到的执行顺序是agent → tools → agent然后模型判断没有更多工具调用输出最终答案走 END。工具调用的顺序完全由模型自己决定你只负责搭好回环结构。4.2 从 ReAct 模式理解 LangGraph 的循环结构很多 Agent 教程都会提 ReAct 模式它是一个经典的推理循环Thought思考该做什么→ Action调用工具→ Observation观察结果→ 回到 Thought。LangGraph 的工具调用循环本质上就是 ReAct 的工程化实现。区别在于ReAct 在论文里描述的是一个形式化流程落到代码里你需要自己去处理模型是否请求工具工具结果如何回传这些细节。LangGraph 把最麻烦的骨架给你搭好了ToolNode 负责执行工具和包装结果条件边负责判断是否继续循环。你只需要提供模型和工具列表。所以理解这个循环的关键是看消息列表的演进过程。每一轮AI 消息可能携带 tool_calls工具节点会把结果作为 ToolMessage 追加进去。模型每次调用时看到的不是单个结果而是完整的历史上下文——这也是为什么messages必须用 Reducer 追加而不是覆盖。4.3 循环终止的两个硬性保障循环结构最大的风险是不终止。模型在极端情况下可能反复请求同一个工具或者一直输出工具调用但永远不给出最终答案。这会导致运行时间无限拉长甚至触发 LangGraph 默认的递归上限。LangGraph 自带一个保护机制recursion_limit默认值是 25。超过这个步数运行会抛 GraphRecursionError。很多初学者第一次遇到这个报错都以为是 bug其实这是框架在替你踩刹车。更好的做法是主动设置最大迭代步数。常见方案是在条件路由里加一个计数器判断MAX_STEPS 5 def should_continue(state: State) - Literal[continue, end]: if state[step_count] MAX_STEPS: return end last_message state[messages][-1] if getattr(last_message, tool_calls, None): return continue return end注意step_count这个字段是普通 int没有 Reducer。因为环形图中每个节点在单次执行里只会被访问一次tools 走完才会到 agent不存在两个分支同时更新它默认覆盖逻辑完全够用。4.4 带记忆的工具调用循环从 stateless 到 stateful上面的例子是无状态的每次invoke都是全新上下文。如果想让 Agent 记住之前的对话可以在编译时挂一个 Checkpointer。Checkpointer 的作用是持久化图的执行状态并且支持按线程 ID 恢复。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph_with_memory builder.compile(checkpointermemory) result_1 graph_with_memory.invoke( {messages: [{role: user, content: 我的名字叫小明}], step_count: 0}, config{configurable: {thread_id: user-001}}, ) result_2 graph_with_memory.invoke( {messages: [{role: user, content: 我叫什么名字}], step_count: 0}, config{configurable: {thread_id: user-001}}, ) print(result_2[messages][-1].content) # 模型能答出来这里的核心变化是状态被保存下来下一次 invoke 会先加载历史状态再执行图。thread_id用来区分不同的会话同一个线程 ID 共享一段记忆。如果你用的是本地内存版本重启进程记忆就丢了生产环境建议换成持久化后端。这个方向涉及 Checkpointer 的选型入门阶段先用 MemorySaver 理解机制就够了。5. 实测中的问题记录与个人经验5.1 并行分支更新同一状态字段引发的冲突我第一次写多分支图时遇到过这样的报错Concurrent update to one or more channels detected, forcing a rewrite。我当时的状态里有两个分支同时往result这个普通字段里写内容LangGraph 看到同一个字段收到了两份互相冲突的新值无法决定覆盖顺序直接中止执行。这个报错教会我一件事普通字段不是不能存在而是要搞清楚哪些节点会写它。凡是可能被多个分支同时更新的字段一律加 Reducer。如果你确实想做多个并行结果汇总比如把多个工具响应收集到一个列表里再交给下一步那就必须用Annotated[list, operator.add]或自定义合并函数。顺带提醒一句并行分支的更新顺序是不确定的如果你的 Reducer 逻辑依赖顺序比如先到先得那结果可能不稳定。自定义 Reducer 尽量写成纯函数不依赖执行顺序。5.2 路由返回非法值报错之后怎么快速定位条件路由的报错通常长这样ValueError: Node xxx not found in graph或者 Invalid path xxx in conditional edge。第一反应别去翻框架源码先检查三件事。第一路由函数的返回值是不是原样传给了映射表。如果路由函数里做了取模、拼接、大小写转换之类的操作先打印一下最终返回值。第二映射表的 key 是否与返回值完全一致空格、大小写都算。第三映射表的值是否都指向已存在的节点名END 和 START 也要确认在当前版本里能直接用。我的做法是在路由函数里临时加一句 debugdef route_func(state): route a if state[count] 0 else b print(route:, route) return route看到打印值之后问题通常一秒就定位了。5.3 工具调用失败不等于循环结束工具执行失败是很常见的但不少初学者会在工具节点里让异常直接抛出导致整个图崩溃。正确的思路是把工具执行失败当作一个普通结果告诉模型让模型决定接下来怎么办。ToolNode 默认会捕获部分异常并把错误信息封装成 ToolMessage 回传给模型这是个很贴心的设计。如果你用自定义工具节点也建议自己包一层def safe_tool_node(state: State): try: return {messages: [execute_tool(state[messages][-1])]} except Exception as e: return {messages: [ToolMessage(contentf工具出错{e}, tool_call_id...)]}把异常信息塞回消息列表模型就有机会修正参数或换一个工具。这比直接崩溃要可靠得多也是 Agent 类项目稳定性的关键一环。5.4 调试和追踪把每一步状态摊开来看最后聊调试。LangGraph 的图结构容易让人产生错觉以为它是透明的其实节点多了之后你得主动追踪每一步的输入输出。最简单的调试方式是graph.invoke之后打印整个 state但这只能看到最终结果看不到中间过程。想逐步看用graph.streamfor chunk in graph.stream(initial_state): for node_name, update in chunk.items(): print(f节点 {node_name} 的更新, update)这样每一步的返回值都会单独打出来你在哪个节点出了什么问题一目了然。如果项目里开了 tracing 服务还能看到更完整的调用链和耗时数据但我个人经验是Debug 阶段先用stream打印就够解决了。最后分享一点个人体会。LangGraph 入门最大的障碍不是语法而是思维方式。链式编程让人习惯下一步是什么LangGraph 逼你思考这一步之后可能有哪些分支怎么判断去哪个分支状态如何流动和合并。等你把这套图式思维刻进脑子里再去看 Agent 的复杂循环、多 Agent 协作、人工介入机制都会清晰很多。我踩过的坑集中在状态合并和循环终止上希望上面这些记录能让你少走几步弯路。