资讯动态

LangGraph 实战:用 StateGraph 编排带分支与循环的 Agent

发布时间:2026/9/29 20:33:39 来源:尧图企业网站定制
1. 从能跑通到能编排LangGraph 到底补了哪块短板很多人第一次接触 Agent 开发都是从 LangChain 的AgentExecutor起步的。写个提示词、挂两个工具、跑一个while循环看起来就智能了。但真把它放到稍微复杂一点的业务里问题立刻暴露多轮对话里状态怎么存工具调用失败要不要重试某一步该走 A 分支还是 B 分支这些在AgentExecutor里几乎全靠黑盒你只能祈祷它按你想的走。LangGraph 出现的意义就在这。它不是又一个更高级的 Agent 封装而是把 Agent 的执行过程从隐式循环变成显式图。你可以把每个节点当成一个函数把每条边当成一次跳转条件整个 Agent 的运行轨迹变成一张你能画出来、能调试、能断点续跑的图。这就是StateGraph的核心价值——状态在节点间流动路由由你定义循环由你控制。我自己的体会是如果你只是做一个问天气、查汇率的玩具 AgentLangChain 的AgentExecutor完全够用没必要上 LangGraph。但一旦涉及多步骤任务、条件分支、人工介入、失败重试、多 Agent 协作LangGraph 的图结构会让你少写大量胶水代码。它解决的不是能不能调用工具而是调用工具的过程能不能被精确编排。这篇文章我会按四个层次展开先把 StateGraph 的状态模型讲透再讲条件路由怎么写才不踩坑然后是 Agent 工具调用循环的完整实现最后聊聊实测中那些文档不会告诉你的细节。全程用可复现的代码Python 为主读完你应该能自己搭一个带分支和循环的 Agent。提示本文假设你已经会基本的 LangChain 用法Prompt、Tool、LLM 调用如果完全没接触过建议先跑通一个最简单的AgentExecutor再回来。2. StateGraph 的状态模型为什么共享字典是最容易翻车的地方2.1 State 不是普通字典而是带合并规则的通道刚上手 LangGraph 的人最容易犯的错就是把 State 当成一个普通dict觉得我往里面塞值节点读出来就行。但 LangGraph 的 State 本质是一组通道channel每个字段有自己的合并策略reducer。默认策略是覆盖——后一个节点返回的值直接盖掉前一个。这在单线程顺序执行时没问题但一旦涉及并行节点或者循环覆盖就会丢数据。举个真实场景你在做一个资料检索 Agent一个节点负责搜索一个节点负责总结还有一个节点负责把中间结果追加到历史里。如果你用默认覆盖策略历史列表每次都被新值替换最后只剩最后一条。正确做法是给这个字段指定operator.add作为 reducerfrom typing import Annotated, TypedDict import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 追加而非覆盖 current_step: str # 默认覆盖 retry_count: int # 默认覆盖Annotated[list, operator.add]这行的意思是任何节点返回的messages列表都会和现有列表做运算也就是追加。这是 LangGraph 里最常用、也最容易被忽略的一个细节。我见过太多人因为没加这个注解导致多轮对话历史莫名其妙丢失然后花半天时间 debug。2.2 状态字段的设计要够用但不臃肿状态里放什么直接决定了图的复杂度。我的经验是遵循三条放跨节点共享的数据比如消息历史、用户 ID、当前任务类型。只在单个节点内部用的临时变量不要塞进 State。放需要被路由判断的数据比如next_action、needs_human_review这种布尔或枚举字段条件边要靠它做决策。放需要持久化的数据如果你要用 checkpointer 做断点续跑State 里的字段就是会被存下来的东西别放不可序列化的对象比如数据库连接。一个常见的反模式是把整个 LLM 响应对象塞进 State。没必要存content和tool_calls就够了对象本身序列化麻烦还会让状态快照变得巨大。2.3 编译图之前先想清楚入口和终点StateGraph编译时必须指定entry_point也就是从哪个节点开始。终点则通过END常量标记。这里有个新手常踩的坑忘记给某条路径设置终点导致图跑到某个节点后没有出边运行时报错。我的习惯是画图之前先在纸上把节点和边列出来确认每个节点都有明确的下一步要么指向另一个节点要么指向END。from langgraph.graph import StateGraph, END builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, call_tools) builder.set_entry_point(agent) builder.add_conditional_edges(agent, should_continue, {tools: tools, end: END}) builder.add_edge(tools, agent) # 工具执行完回到 agent形成循环 graph builder.compile()这段代码就是最经典的 ReAct 循环骨架agent 决策 → 判断是否调用工具 → 调用工具 → 回到 agent。add_edge(tools, agent)这一条边就是循环的来源。理解了这个结构后面所有复杂编排都是在这个基础上加分支、加节点。3. 条件路由让 Agent 自己决定下一步去哪3.1 条件边的本质是一个返回字符串的函数add_conditional_edges的第二个参数是一个函数它接收当前 State返回一个字符串。这个字符串会和第三个参数的映射字典做匹配决定跳到哪个节点。就这么简单但威力很大。def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return tools return end这个函数是整个 Agent 的大脑开关。它不调用 LLM纯粹基于状态做判断所以执行极快、完全可控。我强烈建议把路由逻辑写得尽量简单、尽量显式不要在里面做复杂的 LLM 调用。路由函数越简单调试越容易。3.2 多分支路由用映射字典而不是 if-else 堆叠当你的 Agent 有多种走向时比如需要检索需要计算需要人工审核直接回答不要在路由函数里写一长串 if-else 然后返回不同字符串——虽然能跑但可读性差。更好的做法是让路由函数返回一个语义化的标签映射字典负责对应到节点def route_by_intent(state: AgentState) - str: intent state.get(intent, chat) return intent # 返回 search / calc / review / chat builder.add_conditional_edges( classify, route_by_intent, { search: search_node, calc: calc_node, review: human_review, chat: chat_node, } )这样意图分类节点只管输出intent字段路由映射集中在一处加新分支时只改字典不动逻辑。实测下来这种写法在分支超过三个时优势特别明显。3.3 循环的退出条件必须一定会被满足Agent 的工具调用循环最怕什么死循环。LLM 有时候会反复调用同一个工具或者路由函数因为状态没更新而一直返回tools。我一般会加两道保险在 State 里放一个retry_count或step_count每次进工具节点就 1路由函数里判断超过阈值就强制走END。在路由函数里检查最后一条消息如果连续 N 次都是同样的工具调用直接终止。def should_continue(state: AgentState) - str: if state.get(step_count, 0) 10: return end last state[messages][-1] return tools if last.tool_calls else end注意阈值不要设太小复杂任务可能需要 5-8 轮工具调用也不要设太大超过 15 轮基本说明 Agent 卡住了继续跑只是浪费 token。4. Agent 工具调用循环从能调到调得稳4.1 工具节点的标准写法与返回值陷阱工具节点负责真正执行tool_calls。标准写法是遍历最后一条消息里的所有 tool call逐个执行然后把结果包装成ToolMessage返回from langchain_core.messages import ToolMessage def call_tools(state: AgentState): last state[messages][-1] results [] for call in last.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, step_count: state.get(step_count, 0) 1}这里有两个关键点。第一tool_call_id必须和请求里的 id 对应否则下一轮 LLM 调用会因为消息不匹配而报错。第二工具异常一定要捕获并转成字符串返回不要让异常直接抛出中断整个图。Agent 的价值之一就是工具挂了还能自己想办法你把异常吞掉并告诉 LLM这个工具失败了它往往会换个方式重试。4.2 并行工具调用LangGraph 天然支持但要注意 reducer现代 LLM 经常一次返回多个 tool call。上面的for循环是串行执行的如果工具之间没有依赖可以并行。LangGraph 支持在节点里返回多个消息配合operator.addreducer 会自动合并。但要注意并行执行时每个工具的结果顺序可能和请求顺序不一致如果你的下游逻辑依赖顺序就得自己排序。我实测下来对于大多数场景串行执行工具反而更稳——因为工具之间经常有隐式依赖比如先查 ID 再查详情并行容易出问题。只有在工具完全独立比如同时查三个城市的天气时才考虑并行。4.3 消息裁剪别让历史无限膨胀Agent 跑十几轮之后messages列表会变得非常长每次调用 LLM 都带着全部历史token 消耗飙升还可能超出上下文窗口。我的做法是在 agent 节点调用 LLM 之前先做一次消息裁剪def trim_messages(messages, max_tokens4000): # 保留 system 消息 最近 N 条简单粗暴但有效 system [m for m in messages if m.type system] recent messages[-10:] return system recent更精细的做法是用trim_messages工具函数按 token 数裁剪但要注意别把tool_call和对应的ToolMessage拆散——拆散了 LLM 会报错。这是个很隐蔽的坑我踩过一次报错信息完全没提消息配对问题查了很久才发现。5. 实测中那些文档不会写的细节5.1 checkpointer 让断点续跑变成一行代码LangGraph 最让我惊喜的功能是 checkpointer。加上它之后整个图的状态会自动持久化你可以用同一个thread_id恢复执行from langgraph.checkpoint.memory import MemorySaver graph builder.compile(checkpointerMemorySaver()) config {configurable: {thread_id: user-123}} graph.invoke({messages: [HumanMessage(帮我查下订单)]}, config) # 下次用同样的 thread_id自动带上历史生产环境换成SqliteSaver或 Postgres 版本即可。这个功能对多轮对话、人工审核中断、长任务恢复都极其有用。我做过一个需要人工确认的审批 Agent用 checkpointer 实现暂停等审核 → 审核后继续代码量比手写状态机少了三分之二。5.2 流式输出要选对 stream_modeLangGraph 支持多种流式模式values输出每步完整状态updates只输出节点返回的增量messages专门流式输出 LLM token。做聊天界面时用messages模式体验最好做调试时用updates最清晰。别一股脑用values状态大的时候每次输出都很重。5.3 调试图结构先 print 再可视化LangGraph 可以生成图的可视化但在 notebook 里经常因为依赖问题画不出来。我的土办法是直接print(graph.get_graph().edges)把节点和边打印出来一眼就能看出哪里连错了。比折腾可视化工具快得多。5.4 常见报错与对应原因报错信息大概率原因解决方向InvalidUpdateErrorState 字段没配 reducer并行更新冲突给列表字段加operator.addtool_call_id不匹配ToolMessage 的 id 和请求对不上检查是否原样传递了call[id]图跑到某节点卡住该节点没有出边或路由没覆盖检查add_conditional_edges映射无限循环路由函数一直返回同一分支加step_count阈值消息配对错误裁剪时拆散了 tool_call 和 ToolMessage按对话轮次整体裁剪这张表是我自己踩坑攒出来的基本覆盖了 80% 的新手问题。遇到报错先对照这张表能省不少时间。6. 从单 Agent 到多 Agent图结构的自然延伸当你把单 Agent 的图跑顺之后多 Agent 协作其实就是把节点换成子图。LangGraph 支持把一个编译好的图作为另一个图的节点这样每个 Agent 有自己的状态和循环父图只负责调度。我做过一个研究员 写手 审核员的三 Agent 系统研究员和写手各自是独立的子图审核员是父图里的一个条件节点整体结构非常清晰。这里的关键经验是子图的 State 和父图的 State 要做显式映射不要指望它们自动共享。父图调用子图时通过输入输出 schema 转换把需要的字段传进去、把结果取出来。这个映射写清楚了多 Agent 系统就不会乱。另外多 Agent 不一定比单 Agent 好。我见过很多项目为了显得高级硬拆成多 Agent结果通信开销大、调试困难、效果还不如一个精心设计的单 Agent。我的建议是先用单 Agent 加条件路由解决实在解决不了再拆。拆分的信号通常是不同角色的提示词差异极大或者需要不同的工具集而不是任务步骤多。7. 我个人的几条实操建议第一先把图在纸上画出来再写代码。LangGraph 的代码结构和图结构是一一对应的图没想清楚代码一定乱。我现在的习惯是先用方框和箭头画一遍标好每个节点的输入输出再动手。第二路由函数保持纯函数。不要在里面调 LLM、不要读数据库、不要有副作用。它只做一件事看状态返回标签。这样它可测试、可预测、可复用。第三给每个节点加日志。LangGraph 的调试体验取决于你能看到什么。我在每个节点入口打印当前step_count和最后一条消息的类型出问题时一眼就能定位是哪一步跑偏了。第四别迷信框架该手写就手写。LangGraph 解决的是编排问题不是智能问题。如果你的 Agent 效果不好八成是提示词或工具设计的问题换框架救不了。我见过太多人把时间花在折腾框架上却不肯花半小时优化工具的描述文本——而工具描述恰恰是 LLM 决定调不调、怎么调的关键。第五从小图开始迭代。不要一上来就设计一个十几个节点的复杂图。先跑通agent tools两节点循环确认工具调用正常再加条件路由再加人工审核再加多 Agent。每加一层都验证一次比一次性写完再 debug 高效得多。这套东西我前后折腾了小半年从最开始被InvalidUpdateError折磨到后来能比较顺手地搭出带分支、循环、持久化的 Agent最大的感受就是LangGraph 的学习曲线不在 API而在用图的思维去拆解任务。API 半天就能看完思维方式的转变才是真正花时间的地方。等你习惯了状态在节点间流动这个心智模型回头看AgentExecutor那种黑盒循环会觉得它简单得有点可爱。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑