先说一个我自己的体会：刚接触 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(model="gpt-4o-mini", temperature=0) 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 做完一定到 B，B 做完一定到 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_edges：
graph.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( content=str(result), tool_call_id=tool_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(model="gpt-4o-mini", temperature=0) 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(content=str(result), tool_call_id=tool_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_mode="updates"): 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°C；12 乘以 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_mode="updates"之外，还有两个方法值得养成习惯。
第一个是get_state：
config = {"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_calls，tools节点打印每个工具耗时。这些日志在线上排查“为什么 Agent 答非所问”时，远比看用户反馈有用。
6.3 从单 Agent 到多 Agent：Supervisor 模式
当你不再满足于单一 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允许图执行到某个节点时暂停，等待外部输入。典型场景是人工审核：模型要调用“退款”工具前，图暂停，等你确认，之后再继续执行。- 结合 checkpointer，LangGraph 天然支持“断点续跑”：一次执行因为意外中断，恢复后从前面的节点继续，而不是从头开始。
这些能力正是 LangGraph 区别于普通编排脚本的根本原因。你不再是像写函数一样“调一次算一次”，而是真正在“编排一个有状态的长流程”。如果项目只是单次调用一家模型 API 回答一个问题，完全不用上 LangGraph；一旦涉及多轮工具调用、人工介入、状态持久化，LangGraph 的价值就直接体现出来了。
关于 LangGraph 入门，我自己最大的感受是：不要上来就啃文档里的所有概念，先跑通一个最小工具循环，再逐步加上路由、人工审批、记忆、多 Agent。工具调用循环看着复杂，拆开看就是“模型节点、工具节点、条件路由”三个零件。把这三个零件理解透，LangGraph 后面所有的进阶能力都只是在这上面加东西罢了。
