LangGraph 实战：StateGraph 与条件路由构建工具调用循环

1. 从零理解 LangGraph 到底解决了什么问题1.1 为什么单纯的链式调用不够用了如果你用 LangChain 写过稍微复杂一点的东西大概率会遇到一个尴尬的局面业务逻辑一旦出现分支、循环、重试整个 Chain 就变得极其难维护。比如一个客服机器人用户问“帮我查下订单”你得先判断意图再决定是走查询订单的分支还是走退换货的分支查询失败还得重试重试超过三次要转人工。这种带状态、带分支、带循环的流程用传统的 Chain 写出来就是一堆 if-else 嵌套调试的时候根本不知道卡在哪一步。LangGraph 就是冲着这个痛点来的。它把整个流程抽象成一张有向图节点是具体的处理函数边是流转逻辑而贯穿始终的是一个可持久化的State。你可以把它理解成给 LLM 应用装了一个“状态机引擎”让 Agent 的每一步决策都有迹可循、可回放、可中断恢复。我刚开始接触的时候也有疑问这不就是把工作流引擎那套东西搬到 LLM 上吗用下来发现确实有这个味道但 LangGraph 针对 LLM 场景做了很多专门设计比如支持流式输出、支持 human-in-the-loop 中断、支持 checkpoint 持久化这些是通用工作流引擎给不了的。1.2 StateGraph、条件路由、工具调用循环三者的关系这三个概念其实是 LangGraph 构建 Agent 的三块基石缺一不可。StateGraph是骨架它定义了整张图有哪些节点、状态怎么在节点之间传递。条件路由是神经它决定了在当前状态下下一步该走哪个节点——这是 Agent 具备“决策能力”的关键。工具调用循环是肌肉它让 Agent 能够真正调用外部工具搜索、计算、查数据库拿到结果后回到模型继续推理直到任务完成。打个比方StateGraph 是高速公路的路网条件路由是每个路口的路牌和红绿灯工具调用循环是车辆在路网里反复跑直到抵达目的地。少了任何一个Agent 都跑不起来。很多人学 LangGraph 卡住不是因为 API 难而是没想清楚这三者怎么配合。我见过不少初学者把工具调用逻辑硬塞进节点函数里结果图结构一团乱。正确的做法是让图结构去表达控制流让节点函数只负责单一职责。这个原则后面会反复提到。1.3 这篇文章适合谁看能学到什么程度如果你已经会写 Python用过 LangChain 的基础组件Prompt、LLM、Tool但一想到要搭一个能自主决策的 Agent 就不知道从哪下手那这篇内容就是给你准备的。我会从最基础的 StateGraph 定义讲起一路讲到完整的工具调用循环中间穿插条件路由的写法、状态设计的心法、以及我自己踩过的坑。看完之后你应该能做到独立设计一个带分支和循环的 Agent 图知道状态字段该怎么定义条件路由函数该怎么写才不会出 bug工具调用循环怎么防止死循环。这些都是实际项目里天天要面对的问题不是玩具 demo 级别的。2. StateGraph 的核心机制与状态设计心法2.1 StateGraph 的执行模型节点、边、状态三者如何协作LangGraph 的执行模型可以用一句话概括状态在节点之间流动边决定流动方向节点负责修改状态。具体来说你首先定义一个 State 的结构通常是一个 TypedDict里面放所有需要在节点间共享的数据。然后创建 StateGraph 实例把 State 类型传进去。接着用add_node注册节点每个节点是一个函数接收当前 State返回一个字典表示要更新的字段。最后用add_edge和add_conditional_edges把这些节点连起来编译成可执行图。执行的时候LangGraph 从入口节点开始把初始 State 传进去节点函数返回的更新会被合并到 State 里默认是覆盖用 Annotated 可以改成追加然后根据边找到下一个节点如此循环直到抵达 END。这里有个容易忽略的点节点函数返回的字典不需要包含所有字段只需要包含你想更新的字段。LangGraph 会自动做合并。这个设计很关键它让每个节点只关心自己负责的那部分状态职责清晰。2.2 状态字段怎么定义才不会踩坑状态设计是 LangGraph 里最考验功力的地方。我总结了三条原则。第一消息列表用Annotated[list, add_messages]。这是官方推荐的做法add_messages是一个 reducer它会把新消息追加到列表里而不是覆盖。如果你直接写messages: list每次节点返回新消息就会把历史全冲掉Agent 直接失忆。这个坑我踩过调试了半天才发现是 reducer 没加。第二区分“输入态”和“中间态”。有些字段是用户一开始传进来的比如 question有些是流程中间产生的比如 tool_calls、retry_count。把它们都放在一个 State 里没问题但心里要清楚哪些字段是只读的哪些是可变的。第三控制状态膨胀。Agent 跑多轮之后messages 会越来越长token 消耗飙升。我的做法是在状态里加一个summary字段当消息超过一定轮数时用一个专门的节点把历史压缩成摘要然后清空 messages。这样既保留了上下文又控制了成本。下面是一个我常用的状态定义模板from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] question: str retry_count: int final_answer: str2.3 节点函数的职责边界单一职责是铁律节点函数最容易犯的错误就是“什么都往里塞”。我见过有人在一个节点里既做意图识别又做工具调用还做结果格式化最后这个函数两百多行改一处崩三处。正确的做法是一个节点只做一件事。意图识别一个节点工具调用一个节点结果汇总一个节点。这样图结构清晰调试的时候一眼就能看出是哪一步出了问题而且节点可以复用——比如工具调用节点不管上游是哪个分支来的逻辑都一样。节点函数的签名也很简单接收 state返回 dict。返回的 dict 里放你要更新的字段。如果这个节点不需要更新任何状态比如只是做日志返回空字典{}也行。提示节点函数尽量保持纯函数风格不要在里面做副作用操作比如直接写数据库。副作用应该封装成工具通过工具调用节点来触发。这样整个图的行为是可预测的。3. 条件路由让 Agent 学会“看情况走哪条路”3.1 条件路由的本质是一个返回字符串的函数条件路由听起来高大上其实本质特别简单你写一个函数接收当前 State返回一个字符串这个字符串是下一个节点的名字。LangGraph 根据这个返回值去查边表找到对应的节点。用add_conditional_edges注册的时候需要传三个东西源节点名、路由函数、以及一个映射表把路由函数的返回值映射到节点名。映射表这一步很多人会漏其实它是可选的——如果路由函数直接返回节点名映射表可以省略。但我建议还是写上因为映射表让代码更清晰路由函数返回语义化的标签比如 need_tool、done映射表负责翻译成实际节点名。def route_after_llm(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return call_tool return finish graph.add_conditional_edges( llm, route_after_llm, { call_tool: tool_node, finish: END } )这段代码的意思是LLM 节点执行完之后检查最后一条消息有没有工具调用请求。有就去工具节点没有就结束。这就是最经典的工具调用循环的路由逻辑。3.2 多分支路由的设计模式实际项目里路由往往不止两个分支。比如一个客服 Agent可能需要判断是查询类问题、投诉类问题、还是闲聊。这时候路由函数就变成一个多路判断。我的经验是路由函数里只做判断不做业务逻辑。判断依据全部来自 State 里已经计算好的字段。比如你可以在前面加一个“分类节点”把用户意图写进 state[intent]路由函数只读这个字段。这样路由函数极其简单就是几个 if-elif一眼能看懂。如果分支特别多超过五个可以考虑用字典映射代替 if-elif 链INTENT_TO_NODE { query: query_node, complaint: complaint_node, chat: chat_node, } def route_by_intent(state: AgentState) - str: return INTENT_TO_NODE.get(state[intent], chat_node)这种写法扩展性好加新意图只需要改字典不用动路由函数。3.3 路由函数的常见陷阱与调试技巧第一个陷阱是路由函数返回了不存在的节点名。LangGraph 编译的时候不会报错运行时才炸而且报错信息不一定直观。我的做法是在映射表里把所有可能的返回值都列出来这样漏了哪个一眼能看出来。第二个陷阱是状态字段还没被赋值就拿来判断。比如你在路由函数里读state[intent]但分类节点因为某种原因没执行就会 KeyError。解决办法是在 State 定义时给默认值或者用state.get(intent, chat)这种安全访问。第三个陷阱是路由逻辑和节点逻辑耦合。有些人把判断逻辑写在节点里节点返回不同的状态然后路由函数根据状态判断。这本身没问题但如果判断逻辑很复杂建议还是抽到路由函数里让节点保持纯粹。调试路由的时候我习惯在路由函数里加一行日志打印当前状态和返回的路由结果。跑几次就能看清楚整个决策链路比断点调试还快。4. Agent 工具调用循环的完整实现4.1 工具调用循环的标准结构一个标准的工具调用循环包含三个核心部分LLM 节点、工具执行节点、条件路由。流程是这样的LLM 节点接收消息调用模型模型可能返回工具调用请求路由判断有没有工具调用有就去工具节点工具节点执行工具把结果作为 ToolMessage 追加到消息列表然后回到 LLM 节点模型看到工具结果继续推理直到不再请求工具路由走向 END。这个循环的关键在于消息列表的累积。每一轮 LLM 的输出、工具的执行结果都追加到 messages 里。模型每次都能看到完整的历史所以它能基于工具返回的结果继续推理。这就是 ReAct 模式在 LangGraph 里的落地方式。我建议把 LLM 节点和工具节点分开写不要合并。合并的话循环的边界就模糊了而且没法在中间插入人工审核之类的环节。4.2 工具节点的实现细节工具节点的核心工作是从最后一条 AIMessage 里取出 tool_calls逐个执行把结果包装成 ToolMessage 返回。from langchain_core.messages import ToolMessage def tool_node(state: AgentState) - dict: tool_calls state[messages][-1].tool_calls results [] for call in tool_calls: tool tools_by_name[call[name]] output tool.invoke(call[args]) results.append( ToolMessage(contentstr(output), tool_call_idcall[id]) ) return {messages: results}这里有几个细节要注意。tool_call_id 必须和请求里的 id 对应否则模型会报错说找不到对应的工具结果。content 必须是字符串如果工具返回的是字典或列表要序列化。异常要捕获工具执行失败不能让整个图崩掉应该把错误信息作为 ToolMessage 返回让模型知道工具挂了它可能会换个方式重试。我一般会在工具节点里加一个 try-except把异常信息包装成 ToolMessage。这样即使某个工具临时不可用Agent 也能优雅降级而不是直接报错退出。4.3 防止死循环的三道防线工具调用循环最大的风险就是死循环模型一直请求工具工具一直返回结果模型不满意继续请求无限循环下去。我一般设三道防线。第一道是最大迭代次数。在 State 里加一个iteration字段每次经过 LLM 节点就加一路由函数里判断超过阈值就强制走向 END。阈值设多少看场景我一般设 10 到 15。第二道是工具调用去重。如果模型连续两次请求完全相同的工具和参数说明它卡住了这时候应该中断循环返回当前结果。实现方式是在 State 里记录已调用的工具签名路由函数里检查。第三道是超时控制。整个图执行设置一个总超时超过就中断。LangGraph 支持在编译时配置 recursion_limit这个参数控制图的最大递归深度设一个合理值能兜底。graph builder.compile() result graph.invoke( {messages: [user_msg]}, config{recursion_limit: 25} )这三道防线配合使用基本能杜绝死循环。我实际项目里跑了几万次调用没出现过无限循环的情况。5. 实战搭一个能查天气和算数的 Agent5.1 完整代码结构与逐段解析光讲理论没意思我们直接搭一个能用的 Agent。功能很简单用户问天气或者数学问题Agent 自己决定调用哪个工具。先定义工具from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气 fake_data {北京: 晴25度, 上海: 多云28度} return fake_data.get(city, 暂无数据) tool def calculate(expression: str) - str: 计算数学表达式 try: return str(eval(expression)) except Exception as e: return f计算失败: {e} tools [get_weather, calculate] tools_by_name {t.name: t for t in tools}然后是状态定义和节点函数from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import ToolMessage from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[list, add_messages] iteration: int llm ChatOpenAI(modelgpt-4o-mini).bind_tools(tools) def llm_node(state: AgentState) - dict: response llm.invoke(state[messages]) return { messages: [response], iteration: state.get(iteration, 0) 1 } def tool_node(state: AgentState) - dict: tool_calls state[messages][-1].tool_calls results [] for call in tool_calls: tool tools_by_name[call[name]] try: output tool.invoke(call[args]) except Exception as e: output f工具执行失败: {e} results.append( ToolMessage(contentstr(output), tool_call_idcall[id]) ) return {messages: results}路由函数和图的组装def route_after_llm(state: AgentState) - str: if state.get(iteration, 0) 10: return finish last state[messages][-1] if getattr(last, tool_calls, None): return call_tool return finish builder StateGraph(AgentState) builder.add_node(llm, llm_node) builder.add_node(tool, tool_node) builder.set_entry_point(llm) builder.add_conditional_edges( llm, route_after_llm, {call_tool: tool, finish: END} ) builder.add_edge(tool, llm) graph builder.compile()跑一下result graph.invoke({ messages: [(user, 北京天气怎么样顺便算一下 23 * 47)], iteration: 0 }) print(result[messages][-1].content)5.2 关键参数的选择与计算过程recursion_limit这个参数值得单独说一下。它控制的是图的最大递归步数不是迭代轮数。一次完整的工具调用循环LLM - 工具 - LLM会消耗两步。所以如果你设 25大概能支撑 12 轮工具调用。我一般按“最大轮数 × 2 缓冲”来算比如想要 10 轮就设 25。iteration字段的阈值我设 10是因为实际业务里超过 10 轮还没搞定的问题基本是模型理解错了或者工具设计有问题继续跑也是浪费 token。这时候中断返回让用户重新描述问题比硬撑更划算。工具节点的异常处理我用了 try-except 包住 invoke而不是包住整个循环。这样单个工具失败不影响其他工具的执行。如果一次请求里有多个工具调用其中一个挂了其他的还能正常返回。5.3 运行结果分析与验证跑上面那段代码你会看到消息列表里依次出现用户消息、带 tool_calls 的 AIMessage、两个 ToolMessage、最后的 AIMessage包含最终答案。整个流程完全符合预期。验证 Agent 是否正常工作我一般看三个点工具是否被正确调用看 tool_calls 的 name 和 args、工具结果是否被正确回传看 ToolMessage 的 content 和 tool_call_id、最终答案是否基于工具结果看最后一条 AIMessage 的内容。如果模型没有调用工具而是直接回答说明 bind_tools 没生效或者模型不支持工具调用。如果工具结果回传后模型还在重复调用同一个工具说明工具返回的内容模型看不懂需要调整工具的返回格式。6. 常见问题排查与避坑经验6.1 状态更新不生效的排查思路最常见的症状是节点函数明明返回了新消息但下一个节点读到的还是旧状态。九成的原因是reducer 没配对。messages 字段必须用Annotated[list, add_messages]如果你写成了普通 list新消息会覆盖旧消息看起来就像状态没更新。另一个原因是节点函数返回的 key 和 State 里定义的 key 不一致。比如 State 里叫messages你返回{message: [...]}LangGraph 会忽略这个未知字段。这种错误不报错特别隐蔽建议返回前打印一下确认。还有一种情况是在节点里直接修改了 state 对象而不是返回新字典。LangGraph 依赖返回值来合并状态直接改对象不会触发更新。记住节点函数永远返回新字典不要原地修改。6.2 工具调用报错的典型场景报错信息原因解决办法tool_call_id not foundToolMessage 的 id 和请求不匹配用 call[id] 赋值tool not found工具名拼写错误或未注册检查 tools_by_name 的 keycontent must be string工具返回了非字符串用 str() 或 json.dumps() 包装model does not support tools模型不支持工具调用换支持 function calling 的模型我遇到最多的是第一个因为手动构造 ToolMessage 的时候容易忘记传 tool_call_id。用ToolMessage(content..., tool_call_idcall[id])这个模板就不会错。6.3 性能优化的几个实操技巧Agent 跑得慢通常是三个原因模型调用慢、工具执行慢、消息列表太长。模型调用慢没法根治但可以并行执行多个工具调用。如果模型一次请求了三个工具用 asyncio.gather 并发跑比串行快三倍。LangGraph 支持异步节点把 tool_node 改成 async 函数就行。工具执行慢的话考虑加缓存。同样的工具参数短时间内重复调用直接返回缓存结果。我在工具节点里加了一个简单的字典缓存命中率还挺高的。消息列表太长的话定期压缩历史。前面提到的 summary 方案或者只保留最近 N 轮对话更早的丢弃。这个要权衡丢太多模型会失去上下文丢太少 token 成本高。我的经验是保留最近 6 到 8 轮比较合适。注意压缩历史的时候工具调用的请求和结果要成对保留或成对丢弃不能只留一个否则模型会困惑。7. 从能跑到好用进阶优化方向7.1 加入人工审核中断有些场景下Agent 调用工具之前需要人工确认比如涉及支付、删除数据这类敏感操作。LangGraph 支持在节点执行前中断等人工确认后再继续。实现方式是在编译图的时候配置 interrupt_before 或 interrupt_after指定在哪个节点前/后暂停。暂停后图的状态会被保存人工审核通过后调用graph.update_state修改状态再graph.invoke(None, config)继续执行。这个功能在生产环境特别有用它让 Agent 从“全自动”变成“半自动”在关键节点上保留人的控制权。7.2 持久化与断点恢复LangGraph 的 checkpointer 机制可以把每一步的状态存到数据库内存、SQLite、Postgres 都支持。这样即使服务重启Agent 也能从上次中断的地方继续跑。配置方式是在 compile 的时候传 checkpointerfrom langgraph.checkpoint.memory import MemorySaver graph builder.compile(checkpointerMemorySaver()) result graph.invoke( {messages: [user_msg]}, config{configurable: {thread_id: user-123}} )thread_id 是会话标识同一个 thread_id 的多次调用会共享状态。这个机制让多轮对话变得非常简单不用自己管理历史消息。7.3 多 Agent 协作的图结构设计单个 Agent 能力有限复杂任务往往需要多个 Agent 协作。LangGraph 天然支持这种模式把每个 Agent 做成一个子图然后用一个主图来编排它们。常见的模式是“主管 专家”一个主管 Agent 负责拆解任务和分派多个专家 Agent 各负责一个领域。主管根据任务类型路由到对应的专家专家处理完把结果返回给主管主管汇总后输出。这种结构比单个大 Agent 更可控每个专家可以有自己的工具集和提示词职责清晰。缺点是图结构变复杂调试成本上升。我的建议是先从单 Agent 做起确实遇到瓶颈再拆多 Agent不要一上来就搞复杂架构。8. 我踩过的那些坑和最后想说的说几个印象深刻的坑。有一次 Agent 死活不调用工具排查半天发现是工具的 docstring 写得太模糊模型不知道什么时候该用。后来把 docstring 改得具体一点比如“查询指定城市的实时天气参数是城市名”立刻就正常了。工具的 docstring 就是给模型看的说明书一定要写清楚。还有一次状态莫名其妙丢失最后发现是节点函数里用了state[messages].append(...)这种原地修改LangGraph 没检测到变化。改成返回新列表就好了。这个坑很隐蔽因为代码逻辑上没错但就是不符合 LangGraph 的更新机制。最后一个建议先用小模型跑通流程再用大模型优化效果。开发阶段用 gpt-4o-mini 这种便宜快速的模型把图结构和路由逻辑调对最后再换成更强的模型。这样能省不少钱调试也快。LangGraph 的学习曲线确实比 LangChain 陡一点但一旦理解了状态、节点、边这套模型你会发现它能表达的东西比 Chain 多得多。工具调用循环只是最基础的应用后面还有条件分支、并行执行、子图嵌套、人工中断等等玩法。把今天这套骨架吃透剩下的都是在这个基础上做加法。