LangGraph实战：StateGraph、条件路由与Agent工具循环详解

LangGraph 这个工具我实际用了一年多从最早的 0.0.x 版本一路跟到现在的稳定版中间踩了无数坑也帮团队落地了好几个生产级的 Agent 项目。很多朋友拿着 LangChain 的文档直接上手 LangGraph结果被 StateGraph、条件路由、工具调用循环这几个概念绕得晕头转向。这篇文章我就把自己从入门到实战的完整经验梳理一遍特别是 StateGraph 的状态设计、条件路由的几种典型模式以及 Agent 工具调用循环的终止条件和防死循环策略希望帮你少走弯路。这篇文章适合谁看你如果已经会写基本的 LLM 调用代码但对 Agent 的编排架构还不熟悉或者你用过 LangChain 的 Agent 但觉得它的控制力太弱、流程太黑盒又或者面试被问到 LangChain 和 LangGraph 的区别时只能说出“LangGraph 是图”这种空话——那这篇文章就是为你准备的。1. 为什么选 LangGraph先搞清楚它和 LangChain 的区别1.1 从 Chain 到 Graph编排思路的一次升级很多新手上手 LangGraph 之前先用过 LangChain 的LCELLangChain Expression Language里面最常见的概念就是Chain。Chain 的核心假设是流程是线性的。你写一个prompt | model | parser数据从前到后走一路中间每个环节都是固定顺序的。但真实的 Agent 场景根本不是线性的——模型需要判断是否调用工具工具返回结果后模型可能要再思考、再调用这中间还有分支、有循环、有提前终止。这里我打个比方。你把 Chain 想象成工厂流水线产品从传送带起点走到终点每个工位做固定的事情。但你做一个 Agent 的时候需要的是一个“车间调度系统”——同一个工件到了某个工位工人要判断是返工、是进入下一个工位、还是直接出厂。流水线做不到这个调度系统可以LangGraph 就是那个调度系统。LangGraph 的定位更准确地说是一个有状态、可编排、可细粒度控制的 Agent 运行时。它允许你把流程建模成一张图Graph图里有节点Node——代表具体执行的函数或步骤有边Edge——代表节点间的流转方向。最关键的是边可以是条件边也就是模型执行完一个节点后下一步走哪里由一段代码逻辑甚至由 LLM 自行决定来判断。这个设计的价值在容错和生产级场景里体现得非常明显。比如你有一个多轮 Agent第一轮模型决定调用天气 API第二轮模型想调用日历 API但工具返回了错误你可能希望流程“回到”第一轮重新组织答案而不是沿着固定链路一路跑下去。在 LangChain 里实现这种回退逻辑非常别扭但在 LangGraph 里这就是一条普通的有向边而已。1.2 StateGraph 的设计哲学状态即一切LangGraph 在图之上抽象了一个核心概念叫做StateGraph我理解它就是把“图”和“状态”绑定在一起的一种图表驱动架构。状态State是整个图执行过程中共享的数据容器每个节点执行完后可以把结果写入状态下一个节点可以从状态里读取需要的数据。这个设计继承了 Redux 一类前端状态管理库的思想。我刚开始接触的时候觉得有点小题大做写个 Agent 而已弄个全局变量不行吗后来在生产环境里被教育了——Agent 执行过程中会产生大量的中间数据用户的原始输入、模型的思考过程、工具调用的参数和结果、错误信息、执行历史……如果没有一个结构化、可追踪的状态容器排起错来会非常痛苦。StateGraph 的路由判断也完全是基于状态做的。你把模型上一次的输出、工具的返回结果都存进状态然后写一个路由函数去读这些字段决定下一步指向哪里。这样一来整个 Agent 的执行链路是可视化的任何一步的状态变化都可以记录下来这对于调试和审计非常友好。在实际项目中我建议把 State 的设计放在搭建图结构之前。先想清楚你的 Agent 执行过程中需要哪些状态字段每个字段是哪个节点写入的、哪个节点读取的再开始写节点函数。状态是图的“血液”血液流动的路径没想清楚图搭得再漂亮也是空的。2. StateGraph 基础搭建你的第一个状态图2.1 定义状态 Schema给图里流动的数据一个“模具”在 LangGraph 里定义状态最常规的方式是继承TypedDict。这里我强烈建议你用typing.TypedDict而不是普通dict因为 LangGraph 利用类型注解做了很多编译期检查。状态的定义决定了图上节点之间能传什么数据相当于给数据流定了一个模板。直接看一个最简单的例子from typing import TypedDict, Annotated from langgraph.graph import StateGraph class AgentState(TypedDict): messages: Annotated[list, operator.add] current_step: str注意这里我用到了Annotated[list, operator.add]。LangGraph 支持为每个状态字段声明一个“归约器”reducer用来决定当多个节点往同一个字段写数据时究竟是覆盖还是合并。operator.add表示增量追加——每个节点往messages里塞的数据会自动追加到已有列表后面而不是把之前的列表整个覆盖掉。这个设计真的很重要我见过太多新手在这里踩坑。如果不指定 reducer默认行为是后写的覆盖先写的你辛辛苦苦在节点里往messages里 append 的内容下一个节点一读取发现只有最后一条消息怀疑人生。再强调一个容易忽略的点状态字段尽量按业务语义去拆分不要把什么都塞进一个messages。比如你可以把tool_calls、tool_results、final_answer分开存放。这样 Node 的读取逻辑更清晰条件路由的判断也更精准——你要检查“是否调用了工具”去读tool_calls字段就好了不需要从对话历史里翻找。2.2 节点、边与编译执行三个基础概念一次讲透LangGraph 图里的节点就是一个普通的 Python 函数函数签名是(state) - dict。输入是当前完整状态输出是一个字典字典的 key 对应状态字段名值是你想更新的内容。LangGraph 会自动将返回值合并进全局状态。看一个最基础的双节点图def node_a(state: AgentState): print(进入节点 A) return {current_step: a_done} def node_b(state: AgentState): print(进入节点 B) return {current_step: b_done} builder StateGraph(AgentState) builder.add_node(A, node_a) builder.add_node(B, node_b) builder.add_edge(A, B) builder.set_entry_point(A) builder.set_finish_point(B) graph builder.compile()这里add_edge(A, B)是普通边表示 A 执行完必然走向 B。set_entry_point指定入口节点set_finish_point指定终点节点。compile()返回一个可调用的对象你只需要传入初始状态直接执行result graph.invoke({messages: [], current_step: start})这是最基本的流程大部分教程都会讲到这里。但我想多说一句关于compile()的意义LangGraph 在编译阶段会做大量的校验工作——检查节点是否真的存在、边的目标节点是否注册过、入口和终点是否合法。等编译通过说明你图的拓扑结构从语法层面是正确的。这个机制在项目变复杂之后特别有用再也不用担心图结构写错到了运行时才爆出来。我知道有些朋友看到StateGraph加add_node加add_edge这种写法会觉得“这不就是把函数调用包了一层壳吗直接用 Python 写流程控制不是更简单”——这个感受很真实我也经历过。但是当你需要把流程的任意分支、中间状态和节点的执行过程可视化地展示出来或者在运行中动态插入节点、动态修改路由逻辑时这种显式的图定义就体现出优势了。图结构本身是可序列化的、可检视的这在项目协作和运营层面价值很大。3. 条件路由让 Agent 自己决定下一步3.1 add_conditional_edges 的原理边也可以是“智能”的条件路由是 LangGraph 区别于普通 workflow 工具的核心能力。普通边是固定的 A 到 B条件边则是在 A 执行完之后执行一个路由函数根据函数返回值来决定接下来进入哪个节点。看一个典型写法from typing import Literal def router_after_node_a(state: AgentState) - Literal[B, C]: if state[current_step] a_done: return B return C builder.add_conditional_edges( A, router_after_node_a, { B: B, C: C } )路由函数接收当前完整状态返回一个字符串这个字符串是映射字典里的 key对应到实际的目标节点。这里有个小细节如果返回的字符串和目标节点名恰好一致你可以省略映射字典的 valueLangGraph 会直接用返回值作为目标节点名。我第一次写的时候总是映射来映射去后来发现直接返回节点名就行简化了不少代码。我特别不推荐把复杂逻辑写进路由函数里。比如很多人喜欢在路由函数里直接调 LLM 做意图判断一个 HTTP 调用就花掉几秒钟。LangGraph 的add_conditional_edges第一个参数是源节点名如果你需要路由函数里做 LLM 调用请把它包裹成一个独立的节点让节点的输出作为路由判断的依据而不是在路由函数本身里做重计算。这样路由函数的执行非常快、几乎是纯内存操作流程也更清晰——先有专门的节点做“决策”再有轻量的路由函数做“分发”。3.2 路由模式的几种典型场景意图识别、结果校验、分支处理条件路由的场景比想象的丰富得多我总结三个最常用的模式你在做 Agent 时大概率都会用到第一种是意图识别路由。用户输入进来先由分类节点判断这句话属于“查天气”“查日历”还是“闲聊”然后根据分类结果走不同的业务分支。这个模式对用户体验提升非常明显可以把“能聊天”和“能办事”的能力解耦。def intent_classifier(state: AgentState): # 假设这里已经调用 LLM 完成了意图分类 return {intent: weather} def route_by_intent(state: AgentState) - str: intent state.get(intent, chat) mapping {weather: weather_tool_node, calendar: calendar_tool_node} return mapping.get(intent, chat_node)第二种是结果校验路由。工具返回结果之后你需要检查结果是否合法、是否包含关键字段如果校验通过就进入生成答案的节点不合格就重新调用工具或者进入人工处理的兜底节点。这种路由在生产环境里尤其重要因为 LLM 调用工具时的参数合理性其实没有你想象的那么高。第三种是循环回退路由。当某个节点的执行结果不满足预期你可以让流程回到之前的任意一个节点重新执行一遍。这在 LangGraph 里写起来非常简单——条件分支的目标节点指向一个前面的节点就行。这种“回退”能力在传统 Chain 里几乎没有可能优雅地实现。条件路由本质上是一种显式的、可控制的 Agent 自主决策机制。它和让 LLM 直接自由发挥“下一步干什么”的区别是路由的分支和去向是开发者预先设计好的LLM 只能在这个预定义的候选分支里做选择不能自己凭空创造流程。这在大规模工业化部署中是必要的——你不可能让一个模型来决定整个系统的流程边界但你可以让它决定每条支路怎么走。4. Agent 的工具调用循环从“单次对话”到“自主完成”4.1 工具调用循环的设计核心把“调用工具”拆成两个节点进入最核心的部分了标题里说的“Agent 的工具调用循环”到底是什么用一句话概括通过一个图结构让模型可以反复地决定调用工具、接收工具结果、再决定是否继续直到它认为任务完成。在 LangGraph 里实现这个循环最核心的设计思路是把“模型调用工具”拆成两个独立节点。第一个节点叫agent_node负责让模型思考并输出“是否要调用工具、调用哪些工具、参数是什么”第二个节点叫tools_node或者叫execute_tools负责真正执行工具并返回结果。两个节点之间用条件边连接形成循环。它不是一个节点内部做 while 循环而是两个节点之间的“往返”在图上构成了环。为什么要拆成两个节点因为职责不同agent_node是纯粹生成决策的它不关心工具怎么实现tools_node是纯粹执行的它不关心模型下一步怎么想。这两个职责如果混在一个函数里循环的终止条件和错误边界会非常难写。拆开之后你可以单独给tools_node加超时、加重试、加记录日志完全不影响模型决策。下面是一个极简但完整的实现思路from langchain_openai import ChatOpenAI from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气 return f{city} 今天是晴天 llm ChatOpenAI(modelgpt-4o) llm_with_tools llm.bind_tools([get_weather]) def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} def tools_node(state: AgentState): last_message state[messages][-1] tool_outputs [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] result get_weather.invoke(tool_args) tool_outputs.append( {role: tool, tool_call_id: tool_call[id], content: result} ) return {messages: tool_outputs}4.2 用条件边把两个节点“串成环”并设置退出条件节点都写好了接下来最关键的事情是把它们连成一个支持循环的图并设置循环的终止条件。def should_continue(state: AgentState) - Literal[continue, end]: last_message state[messages][-1] if last_message.tool_calls: return continue return end builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, tools_node) builder.add_edge(tools, agent) builder.add_conditional_edges( agent, should_continue, {continue: tools, end: __end__} ) builder.set_entry_point(agent) graph builder.compile()这个should_continue函数就是整个循环的“刹车阀”。它检查模型最后一轮的输出里是否包含工具调用请求如果有就走到tools节点去执行工具如果没有说明模型认为答案已经完整就走到特殊节点__end__流程结束。这里有一个我踩过很深的坑必须提醒你tools节点执行完工具返回结果之后新的工具结果消息会被追加到messages状态里。此时agent节点再次运行时它读取的是包含了工具结果的最新完整消息历史模型基于这些结果进行下一轮推理。看起来是理所当然的但如果你在agent_node里不小心只把最后一条用户消息传给模型而不是把完整历史传过去工具结果就“丢”了模型会陷入一种“我说要调用工具、但看不到工具结果、于是又调用一次”的循环死锁。我实际处理过的一个生产事故就是这么发生的排查了半天最后发现是团队里有人为了省 token 在agent_node里做了消息裁剪把系统认为不重要的历史消息砍掉了结果偏偏把 tool 消息误伤掉了。加了消息裁剪的白名单之后问题才解决。所以提醒所有同学在你对 LangGraph 和模型行为没有完全把握之前不要在循环里剪裁消息。还有一个实践要点是__end__这个特殊节点。它是 LangGraph 内置的终止标记你不需要显示地定义它直接在条件映射里指向__end__就行。很多新手不知道还有这么个东西总想自己定义一个返回最终答案的节点结果逻辑变得很绕。最终答案可以在agent节点的输出里直接给只要模型不再请求调用工具循环就自然走到__end__结束。4.3 防死循环的三重保险最大步数、超时、人工审核工具调用循环一个比较头疼的问题是死循环——模型反复调用工具、工具反复返回结果、但结果始终不满足要求或者模型就是停在某个状态里反复横跳。在真实项目里这种问题非常频繁我见过模型为了确认一个订单状态疯狂翻页查询二十多次的“事故现场”。因此给循环上锁是必须的。LangGraph 官方提供了recursion_limit它限制一次图调用里最多执行几个节点。超过限制会抛出异常至少能让你在日志里看到“这轮执行异常地长”而不是整个流程无声地卡住。config {recursion_limit: 25} result graph.invoke({messages: initial_messages}, configconfig)除了recursion_limit我自己还会在业务层面加两个保险。第一个是最大工具调用次数统计在状态里加一个tool_attempts字段每次tools_node执行完就加一路由函数里如果检测到次数超过阈值比如 5 次就强制走一条“放弃工具直接生成一个兜底答案”的路径。第二个是结果条件校验如果工具返回的内容一直不满足业务校验规则比如格式不对、关键字段为空也应该提前退出循环进入人工处理流程不能让它无限重试。这些防护措施本质上是在回答一个问题Agent 什么时候该“坚持”什么时候该“认输”把这个问题在路由设计阶段就想清楚远比运行时报错了再救火要省心得多。我认为做 Agent 的核心理念是“给模型自由但给流程边界”。LangGraph 的分支、条件路由、循环终止就是约束模型自由度的护栏。5. 踩坑记录与排查技巧5.1 状态更新语义不一致为什么有的字段是覆盖、有的是追加我在前面提到过 reducer但这里值得再深入讲一下因为实际踩过的坑太典型了。举个例子你在两个节点里都往同一个字段current_step写不同的值如果这个字段没定义 reducer那谁最后执行谁就覆盖前面的但如果你用的是operator.add那这个字段会被拼成一个列表。两种行为的语义完全不同你如果用错写路由函数的时候逻辑就是错的。在真实的 Agent 项目里messages字段用operator.add是最常见的因为它天然是累积的。但是如果你把工具调用的参数也放进一个“累积”型字段里那每一轮调用都会把上一次的参数追加在后面推理和排查的成本会迅速上升。我的经验是累计型字段宁可少用也不要用多。只有真正需要全量历史的地方才用operator.add其余字段尽量用覆盖型让状态在任何时刻都保持简洁明确。5.2 工具调用失败后的降级处理工具调用失败是 Agent 场景里概率最高的错误之一而且失败原因五花八门工具超时、API 返回 500、参数校验不过、LLM 生成了不存在的工具名、工具返回了模型看不懂的格式……我在生产环境中见过所有你能想象到的失败路径。一个比较实用的策略是在tools_node里捕获所有异常并把异常信息以一条tool角色的消息写回状态而不是让异常直接打断图的执行。这样模型在下一轮能看到“你刚才调用工具失败了原因是 xxx”它可以自己决定是换个参数重试、还是换一个工具、还是直接放弃。def tools_node(state: AgentState): last_message state[messages][-1] tool_outputs [] for tool_call in last_message.tool_calls: try: result execute_tool(tool_call) content str(result) except Exception as e: content f工具调用失败: {type(e).__name__}: {e} tool_outputs.append( {role: tool, tool_call_id: tool_call[id], content: content} ) return {messages: tool_outputs}这个思路的好处在于错误信息本身成为模型上下文的组成部分模型可以“看见”错误并自行调整策略。当你把工具错误信息做成结构化、简洁、可提示的内容时模型自己会学会归纳和判断远比你在代码里做各种硬编码兜底要灵活。5.3 调试工具链与方法图的可视化与状态追踪LangGraph 提供了一些很实用的调试手段。第一个建议是你把编译后的图输出成 ASCII 或者 Mermaid 图看看眼过一遍整体结构print(graph.get_graph().draw_ascii())注意虽然 LangGraph 本身有 Mermaid 渲染功能但在我这个版本里更推荐直接用draw_ascii()快速看结构或者用在线可视化工具检查节点和边的关系。每次改动图结构后都输出一遍确认边的走向跟你预期一致尤其是条件边的分支是否都正确指向了目标节点。第二个调试技巧是使用回调机制记录每个节点的输入输出。LangGraph 支持在编译时传入自定义的listener回调或者简单粗暴地在每个节点函数开头打日志def agent_node(state: AgentState): print(f[agent] messages数量{len(state[messages])}) ...在生产环境里我会把每次节点执行的摘要节点名、耗时、状态关键字段的变化记录到结构化日志里这样一次 Agent 执行就是一条完整的时间线排查问题时把这条时间线翻出来比对着代码猜要高效太多。6. 我的实战经验与扩展建议最后分享一些我在实际业务中积累的经验。LangGraph 最好的应用场景是那些流程比较复杂、分支比较多、对可观测性有要求的 Agent 任务——比如多步骤的办公助手、客服工单自动处理、数据分析 Agent。在扩展方向上有两个最值得关注的点。第一个是多 Agent 协作LangGraph 底层支持在一个图里定义节点关系因此你可以把多个 Agent 各自封装成一个子图作为大图的节点让它们在同一个状态空间里协作。这个能力在处理复杂任务拆分时非常有价值但一定要控制好 Agent 之间的通信边界否则状态会变得极度混乱。第二个是持久化与 CheckpointLangGraph 提供了基于BaseStore的持久化机制可以让 Agent 的状态在多次调用之间保持这对于需要续聊和人工介入的长流程应用是个好特征的支撑。我现在自己写 Agent 项目的习惯是先用 LangGraph 快速搭建可运行的图然后逐步把路由分支细化加上循环防护、错误降级、结构化日志最后再做持久化和人工审核。这个顺序比较稳妥一上来就追求完美的架构反而容易过早陷入细节。如果你是刚入门我的建议是不要一上来就套复杂的框架先用这个最简单的工具调用循环跑通一个真实场景——比如“查天气”“算费用”这种小工具——再慢慢往上加路由分支和多轮状态。亲手把一个循环跑通、看到模型调用工具、拿到结果、生成最终答案你对 LangGraph 的理解会发生实质性的变化。补充一个很容易被忽略的细节官方文档和社区里给出的示例往往会省略python-dotenv配置环境变量、模型 API 连接超时处理等基础设施问题。你在本地跑示例时建议先把这些基础环境问题解决掉再开始研究图逻辑否则很容易被一个 API 报错打断思路。写到这里我想起自己第一次跑通 Agent 工具循环时那种“通了”的感觉。LangGraph 的曲线确实需要一点耐心去适应但一旦上手你会对这种显式、可控的编排方式产生依赖。希望这篇文章能帮你也体会到这种流畅感。