
先说一个我自己的体会刚接触 LangChain 里的 Agent 时总觉得那个AgentExecutor像个黑盒你给它一个任务它在模型、工具、记忆之间来回倒腾但你想在中间插入一次人工审核、想精确控制“什么时候必须停止调用工具”、想观察每一步状态是怎么变的都很别扭。后来 LangGraph 一出来我才算真正理解了 Agent 工具调用循环的底层逻辑。它可以把你脑子里的流程图变成可运行的代码状态显式存在节点负责干活条件路由决定下一步循环就是图上的一条路径。这篇文章就围绕 LangGraph 的StateGraph、条件路由和 Agent 工具调用循环展开用一个完整可跑的客服助手 Demo带你把这三件事一次性串起来。无论你是刚入门 AI Agent 开发、正在纠结 LangChain 和 LangGraph 区别还是想给现有项目加一个能自主调用工具的 Agent这篇都适用。我会把每一步的“为什么”也讲清楚而不只是贴代码。1. 为什么需要 LangGraph传统 Agent 的循环是“黑盒”1.1 Agent 的本质思考-行动-观察的循环任何一个能调用工具的 Agent本质都在跑同一个循环模型看到你的问题后决定“我要不要调用工具”如果要调就输出一个结构化指令比如tool_calls程序拿到指令后执行真实工具工具结果再作为一条消息还给模型模型看完结果继续判断直到它觉得信息够了给出最终回答。这个循环在 LangChain 最经典的AgentExecutor里其实也存在但它被封装得太死了。一般你只能拿到最终结果中间模型想了什么、调了哪个工具、结果是什么虽然后来也有中间步骤回调但控制权始终不在你手里。你很难在“模型刚输出工具调用、但还没执行工具”这个时机插入人工确认也很难在某个条件满足时直接中断循环。我记得有段时间做客服机器人客户要求所有调用“退单工具”的操作必须人工审批在 LangChain 老框架里实现这一条相当费劲。1.2 LangGraph 的把戏把流程画成一张图LangGraph 的解法很直接别把 Agent 当黑盒把它当作一张图。这张图里有四个核心概念状态State一份在节点之间传递的共享数据通常是TypedDict。节点Node一个普通的 Python 函数读当前状态返回一个状态更新。边Edge从一个节点到另一个节点的路径分普通边和条件边。条件边Conditional Edge节点执行完后调用一个路由函数由函数返回值决定下一步走向哪里。听起来抽象但其实就是流程图代码化。我画过一张特别朴素的图用户输入 → 模型节点 → 有没有工具调用有就去工具节点没有就输出 → 工具节点执行完回到模型节点。这整段逻辑LangGraph 能用不到五十行代码表达出来而且每一步状态都透明、可调试、可持久化。1.3 和 LangChain 到底是什么关系很多人在搜 LangChain 和 LangGraph 的区别这里我直接说结论两者不是替代关系而是层级关系。LangChain 是面向应用开发者的工具库提供了大量封装好的链、检索器、文档加载器LangGraph 是更底层的编排框架用于构建有状态、多分支、带循环的应用流程。实际项目中完全可以混用LangGraph 管流程LangChain 管模型调用、消息类型、工具绑定。甚至你不装 LangChain 也能用 LangGraph只要节点函数的输入输出符合约定即可。我的建议是别再纠结“选哪个”先学会用 LangGraph 把 Agent 循环搭出来你就知道哪个环节需要 LangChain 帮忙了。2. StateGraph 入门状态、节点和边一次只做一件事2.1 状态定义一切流程都是数据的流动在 StateGraph 里状态就是全部。它本质上是一个字典每个节点读它、改它、再传给下一个节点。最常见的设计是定义一个带messages字段的状态因为 Agent 循环需要把完整的对话历史一条条往后传。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]这里有个关键点add_messages。如果不加这个注解每次节点返回{messages: [新消息]}时会直接覆盖掉原来的消息列表。加上add_messages后LangGraph 会把新消息追加到历史后面。这是 LangGraph 状态机制里最容易被忽略的细节也是很多人写出来 Agent“没有记忆”的根源。你完全可以把State里的字段设计成user_input、intermediate_steps、final_answer等任意业务字段。只要记住节点返回的字典会被合并进总状态字段同名就按“默认覆盖或标注的合并规则”处理。2.2 节点普通 Python 函数不要写魔法节点函数是 LangGraph 里最接地气的部分。它接收当前状态做一件事然后返回一个状态更新。比如最简单的“调用模型”节点from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) def agent_node(state: AgentState): response llm.invoke(state[messages]) return {messages: [response]}函数返回的{messages: [response]}会走一遍add_messages合并逻辑把模型输出追加到对话历史里。这就是 LangGraph 对“节点”的全部要求输入一个 dict 结构的状态输出一个部分更新的 dict。我见过有人一开始把节点写成一等公民类、搞了一大堆抽象完全没必要。LangGraph 的设计就是让你用普通函数每个函数只负责一个动作图结构负责把动作串起来。2.3 图的构建一次性看清流程走向有了状态和节点构建图就三步建图、加节点、加边。from langgraph.graph import StateGraph, START, END graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_edge(START, agent) graph.add_edge(agent, END) app graph.compile()START是虚拟入口节点END是虚拟出口节点。graph.add_edge(START, agent)表示流程从agent节点开始graph.add_edge(agent, END)表示跑完就结束。compile()之后得到的app对象你只管调用它。调用方式也简单result app.invoke({messages: [{role: user, content: 你好}]})这个最简单的图当然没有实际价值但它把核心套路讲清楚了状态AgentState统一流转节点函数负责具体动作边决定流程方向。后面加条件路由、加工具节点都是在这个骨架上添砖加瓦。3. 条件路由让 Agent 自己决定“下一步该干什么”3.1 为什么需要路由Agent 不能永远走同一条路如果一张图只有固定的几条边那它就是一条流水线A 做完一定到 BB 做完一定到 C。但 Agent 的决策天然是分叉的——模型可能决定调用工具也可能决定直接回答。这个“可能”就是条件路由存在的意义。条件路由的思路很简单在某个节点执行完之后调用一个路由函数这个函数读取当前状态返回一个字符串LangGraph 根据这个字符串决定下一个节点是谁。def router(state: AgentState): last_message state[messages][-1] if last_message.tool_calls: return tools return END这里的关键是last_message.tool_calls。当你给模型绑定了工具后如果模型认为需要调用工具它返回的AIMessage里会有一个tool_calls列表里面包含工具名、参数和调用 ID。如果列表为空说明模型觉得不需要工具可以结束。注意路由函数的返回值不能随便写它必须是图里真实存在的节点名或者特殊值END。写错一个字符串运行时会立刻报错“节点不存在”这个特性其实很有用等于强制你把图结构理清楚。3.2 add_conditional_edges把路由函数挂到节点上路由函数写好了怎么通知 LangGraph用到add_conditional_edgesgraph.add_conditional_edges( agent, router, { tools: tools, END: END, } )这段代码的意思是当agent节点执行完后调用router(state)它的返回值是tools就去tools节点是END就结束。第三个参数是路径映射作用是把路由函数的返回值映射到实际节点名不写也可以LangGraph 会把返回值直接当节点名用。我建议初学者先写映射表。原因有两个一是防止路由函数里字符串写错映射表相当于一层校验二是后续如果节点改名只需要该映射表不用改路由函数内部逻辑。3.3 一个完整的岔路口什么时候走工具什么时候结束把前面几个部分拼起来你会得到这样一套逻辑用户问题进入agent节点。模型判断是否要调用工具输出AIMessage。条件路由检查最后一条消息。有tool_calls去tools节点没有走END。tools节点执行完无条件回到agent节点重复步骤 2。这里的“回到agent节点”在图上看起来是个环但这正体现了 LangGraph 的优雅之处循环不是一种特殊机制而是条件的自然结果。每次回到agent模型看的都是包含工具结果的最新状态因此它能继续推理直到不需要工具为止。吴恩达那门热门 Agent 教程里反复强调一个模式模型思考、调用工具、观察结果、再次思考。LangGraph 的条件路由就是用代码把这个模式落到了实处。我自己在面试候选人时也喜欢问“条件路由怎么终止循环”其实考的就是这个映射关系以及END的处理。4. Agent 的工具调用循环从模型思考到工具执行再到回填4.1 消息历史是整个循环的命脉前面说过AgentState里的messages是全流程的“公共记忆”。工具循环能不能跑对就看你能不能维护好这份记忆。一个典型的完整循环是这样的人类消息“北京今天多少度”模型消息包含一个 tool_call工具名get_weather参数{city: 北京}。工具消息工具返回的结果内容如“晴25°C”并携带一个tool_call_id指向刚才模型那条 tool_call。模型消息模型看到了工具结果后整合成最终回答“北京今天晴25°C”。如果你把循环中任意一类消息丢了模型就“失忆”。最典型的问题是不加add_messages工具结果把历史覆盖了第二次进入agent节点时模型根本看不到自己刚才调用了什么工具循环直接失控。另外提一句LangGraph 官方推荐用ToolMessage来封装工具结果并且一定要把它和原始tool_call_id关联上。很多模型服务商严格要求消息格式如果没带上正确的 ID模型甚至可能报错。4.2 tools 节点别在这里写死只有一个工具tools节点的职责是纯执行器遍历模型输出的tool_calls调用真实函数把结果转成ToolMessage。代码模式很固定tools_by_name { get_weather: get_weather, multiply: multiply, } def tools_node(state: AgentState): last_message state[messages][-1] outputs [] for tool_call in last_message.tool_calls: result tools_by_name[tool_call[name]](**tool_call[args]) outputs.append( ToolMessage( contentstr(result), tool_call_idtool_call[id], ) ) return {messages: outputs}几个注意点tool_call[args]是一个字典必须用**解包成关键字参数。一个AIMessage里可能同时有多个 tool_call所以要用循环处理。如果工具抛异常一定要在节点里捕获否则整张图会直接中断。我习惯把所有工具统一套一个 try-except结果返回错误文本让模型自己判断如何处理。这也算是 Agent 容错的重要一环。4.3 为什么 tools → agent 是无条件边在这个图里工具节点执行完一定会回到agent节点所以用的是普通边graph.add_edge(tools, agent)但agent → tools是条件边。这种设计不对称其实来自业务逻辑模型用完工具必须继续思考而“是否用工具”则取决于模型决策。理解这一点你就不会在agent和tools之间画两条无条件边导致流程变成死循环或者跳过模型直接执行工具。有时候我会在调试时打印一下每一步消息类型看到tools节点执行完立刻回到agent然后模型又输出新的tool_call这就是工具调用循环的正常运转。4.4 循环终止与递归上限既然图上存在环就必须考虑“跑不完怎么办”。LangGraph 有recursion_limit概念默认是 25意思是整张图最多执行 25 步。如果你的 Agent 陷入模型和工具来回绕圈超过上限后 LangGraph 会抛异常通常提示Recursion limit reached。实际开发中不要只依赖默认值。可以按任务复杂度显式设置result app.invoke( {messages: [{role: user, content: 北京天气怎么样同时算一下 12 * 8}]}, config{recursion_limit: 50} )我踩过的坑是把recursion_limit设大之后Agent 反而开始浪费时间反复尝试错误工具。所以限制不只是保护机制也是一种控制行为的手段。线上环境建议在关键节点加“最大工具调用轮数”的业务判断别把所有防护都压在框架上。5. 跑通一个真实案例带工具的智能客服助手5.1 完整代码天气查询 乘法计算下面是一个可以直接运行的完整 Demo。假设计算的是“客服助手”一个节点让模型决定是否用工具一个节点执行工具条件路由负责来回切换。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import AIMessage, HumanMessage, ToolMessage # ---------- 1. 定义状态 ---------- class AgentState(TypedDict): messages: Annotated[list, add_messages] # ---------- 2. 定义工具 ---------- def get_weather(city: str) - str: 查询一个城市的天气信息。 weather_map {北京: 晴25°C, 上海: 多云28°C, 广州: 雷阵雨30°C} return weather_map.get(city, f暂未收录 {city} 的天气数据) def multiply(a: float, b: float) - float: 计算两个数字相乘的结果。 return a * b tools [get_weather, multiply] tools_by_name {get_weather: get_weather, multiply: multiply} # ---------- 3. 初始化模型 ---------- llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools) # ---------- 4. 定义节点 ---------- 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] outputs [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] result tools_by_name[tool_name](**tool_args) outputs.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) return {messages: outputs} # ---------- 5. 定义条件路由 ---------- def router(state: AgentState): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return END # ---------- 6. 构建图 ---------- graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_edge(START, agent) graph.add_conditional_edges(agent, router, {tools: tools, END: END}) graph.add_edge(tools, agent) app graph.compile()5.2 运行与状态观察为了看清每一步的状态流转建议用stream而不是invoke用updates模式能看到每个节点返回了什么inputs {messages: [HumanMessage(content北京天气怎么样再帮我算一下 12 乘以 8。)]} for chunk in app.stream(inputs, config{recursion_limit: 50}, stream_modeupdates): for node_name, update in chunk.items(): print(f 当前节点: {node_name}) for msg in update.get(messages, []): print(f [{msg.type}] {msg.content}) print()你会看到类似下面的输出agent节点模型返回一个AIMessage其中有两个tool_calls一个调get_weather一个调multiply。tools节点依次执行两个工具返回两条ToolMessage。agent节点再次执行模型看到全部工具结果后输出最终AIMessage内容大概是“北京天气晴25°C12 乘以 8 等于 96”。路由判断最后一条消息没有tool_calls流程结束。这其实就还原了“智能客服助手”的核心逻辑。如果把这个 Demo 里的get_weather换成查订单、查库存、提交工单的 API再把模型换成你实际业务用的模型就差不多是生产环境里的一个初版 Agent 了。5.3 从 Demo 到生产的三个改动点实际项目不能直接照搬至少要改三处。第一工具必须有真实错误处理。你在 Demo 里直接调用工具函数但如果上游接口超时、参数校验失败LangGraph 的整条流会炸。生产实现里要在tools_node内部做异常兜底返回类似工具执行失败请求超时的ToolMessage让模型来决定是换个参数重试还是放弃。第二要加检索或知识库。客服助手里的大量答案不是靠模型幻觉生成的而是从企业知识库检索后回答。这属于 RAG 部分做法是在agent节点之前或内部加一个检索步骤把检索结果塞进消息上下文再交给模型。第三要加身份识别与持久化。每轮会话用thread_id区分LangGraph 带 checkpointer 机制可以把状态存起来这样用户退出重进后对话还能继续。这块后面细讲。6. LangGraph 实操心法踩坑清单、调试手段与进阶方向6.1 新手最容易踩的四个坑我在带着团队用 LangGraph 时总结出几个出现频率极高的错误这里逐个列出来都是真实经历。坑一State 里的 messages 字段没有加add_messages注解。结果每次模型输出都会把上一轮消息覆盖掉Agent 连续调用两次工具后彻底“失忆”。排查方法很简单打印state[messages]的长度看是否单调递增。这也解释了为什么状态定义是整个 LangGraph 的地基。坑二条件路由返回了不存在的节点名。路由函数像return tool但图里注册的节点叫tools一运行就报InvalidUpdateError或类似的节点缺失错误。我建议路由函数返回值用常量并且和add_node的名字保持同一处定义别散落两处。坑三工具调用出异常后没有捕获。模型生成的 tool_call 参数有时不合法比如传了负数给价格查询工具。工具函数直接抛异常整个图中断最终用户只看到“系统错误”。正确的做法是所有工具统一包一层异常处理异常文本作为工具结果返回给模型。坑四在循环里反复重试却不设置最大轮数。recursion_limit只是兜底如果业务要求最多调用三轮工具应该在路由函数或tools_node里数一下当前循环次数超过就强制END。否则模型可能反复调同一个失败工具浪费 token 也拖慢响应。6.2 调试不只靠 print要学会看状态快照LangGraph 天生适合调试因为所有中间状态都是可检查的。除了stream_modeupdates之外还有两个方法值得养成习惯。第一个是get_stateconfig {configurable: {thread_id: test-001}} current_state app.get_state(config) print(current_state.values[messages])这在挂上持久化之后特别有用你能看到某个线程中途的状态快照确认用户消息到底有没有进状态。第二个是利用 LangGraph 内置的可视化。调用app.get_graph().draw_mermaid_png()可以把图画出来。虽然平时我不会真去写这行代码但调试复杂多 Agent 流程时把图画出来一看哪个节点连错、哪条边缺失一目了然。最后还有一个我个人的调试习惯给关键节点加一层“日志包装”。比如agent节点里打印model response has X tool_callstools节点打印每个工具耗时。这些日志在线上排查“为什么 Agent 答非所问”时远比看用户反馈有用。6.3 从单 Agent 到多 AgentSupervisor 模式当你不再满足于单一 Agent想让“一个管家调度多个专家 Agent”时LangGraph 同样支持常见的就是 Supervisor主管模式。流程概括如下一个supervisor节点负责决定把任务派发给哪个子 Agent。多个子 Agent各自绑定不同工具和 Prompt。子 Agent 执行完路由回到supervisor由它决定是结束、还是继续派发给其他子 Agent。这种模式实现起来和单 Agent 工具循环的思路高度一致只是把“工具节点”换成了“子 Agent 节点”。LangGraph 里可以通过add_node直接挂一个编译好的子图实现图套图组合能力很强。我的经验是先不要一上来就搞多 Agent单 Agent 加工具都调试不顺多 Agent 只会放大问题。6.4 持久化、人工介入与生产落地生产级 Agent 还有三个关键词checkpointer、interrupt、断点续跑。checkpointer是 LangGraph 的功能组件可以把状态存到内存或数据库里。加上之后同一个thread_id的多次invoke会共享状态实现多轮对话记忆。interrupt允许图执行到某个节点时暂停等待外部输入。典型场景是人工审核模型要调用“退款”工具前图暂停等你确认之后再继续执行。结合 checkpointerLangGraph 天然支持“断点续跑”一次执行因为意外中断恢复后从前面的节点继续而不是从头开始。这些能力正是 LangGraph 区别于普通编排脚本的根本原因。你不再是像写函数一样“调一次算一次”而是真正在“编排一个有状态的长流程”。如果项目只是单次调用一家模型 API 回答一个问题完全不用上 LangGraph一旦涉及多轮工具调用、人工介入、状态持久化LangGraph 的价值就直接体现出来了。关于 LangGraph 入门我自己最大的感受是不要上来就啃文档里的所有概念先跑通一个最小工具循环再逐步加上路由、人工审批、记忆、多 Agent。工具调用循环看着复杂拆开看就是“模型节点、工具节点、条件路由”三个零件。把这三个零件理解透LangGraph 后面所有的进阶能力都只是在这上面加东西罢了。