尧图网页设计 
1. 从 LangChain 到 LangGraph为什么需要一张图如果你已经用过 LangChain 的AgentExecutor大概率经历过这种场景想让 Agent 先查资料、再判断要不要调用计算器、最后根据结果决定是继续查还是直接回答。用 Chain 写出来代码会变成一堆if/else嵌套调试的时候根本看不清执行到哪一步出了问题只能靠打日志猜。这不是你写代码的水平问题而是 Chain 这种线性结构本身就不适合表达带分支和循环的流程。LangGraph 要解决的就是这件事。它把 Agent 的执行过程建模成一张有向图节点Node代表一个具体的处理步骤边Edge代表步骤之间的流转关系而状态State则是在整张图上流动的数据。你可以把它理解成给 LLM 应用画了一张流程图只不过这张图是可以在运行时动态决定下一步走哪条边的。这里必须先厘清一个高频困惑LangGraph 和 LangChain 到底什么关系。简单说LangChain 提供的是组件层的东西——模型封装、Prompt 模板、工具定义、检索器、输出解析器这些在 LangGraph 里照样能用。LangGraph 则是在这些组件之上提供了一层编排层。它不替代 LangChain而是补上了 LangChain 在复杂控制流上的短板。所以你会看到 LangGraph 的代码里到处from langchain_core...这很正常两者是配合关系不是替代关系。那什么场景该上 LangGraph我的判断标准是三条第一流程里有条件分支下一步做什么取决于上一步的输出第二流程里有循环比如 Agent 反复调用工具直到拿到满意答案第三需要人工介入在某个节点暂停等确认再继续。这三条只要中了一条用 LangGraph 就比硬写 Chain 舒服得多。反过来如果就是一个输入→模型→输出的直线流程那用 LangChain 的 LCEL 就够了上 LangGraph 属于杀鸡用牛刀。下面这张表可以帮你快速做决策场景特征推荐方案原因单次模型调用、线性 Prompt 链LangChain LCEL结构简单无需图编排固定顺序的多步处理LangChain LCEL用管道符串联即可需要根据结果走不同分支LangGraph条件边天然支持Agent 反复调用工具LangGraph循环结构清晰可控需要人工审核后继续LangGraph支持中断与恢复多 Agent 协作LangGraph子图与状态共享机制成熟理解了这层定位后面的 StateGraph、条件路由、工具调用循环就都是水到渠成的事了。2. StateGraph 的核心状态怎么定义节点怎么读写StateGraph 是 LangGraph 里最基础的图类型几乎所有 Agent 都从它开始搭。它的核心思想是整张图共享一个状态对象每个节点读取状态、返回状态的更新部分。这个设计看起来简单但里面有几个坑不搞清楚写出来的 Agent 行为会很诡异。2.1 State 的 schema 设计与 reducer 机制状态的定义用TypedDict或者 Pydantic 模型都行我一般用TypedDict轻量且够用。一个典型的 Agent 状态长这样from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next_action: str tool_result: str这里最关键的是Annotated[list, add_messages]这个写法。add_messages是一个reducer 函数它决定了当多个节点都往messages里写东西时这些写入怎么合并。默认情况下LangGraph 对状态的更新是覆盖语义——节点返回什么状态里对应字段就变成什么。但对话历史显然不能覆盖你需要的是追加。add_messages做的就是这件事它会把新消息追加到列表末尾而且还会自动处理消息 ID 去重。我踩过的一个坑是一开始没加add_messages直接写messages: list结果 Agent 每轮对话都只记得最后一条消息前面全丢了。排查了半天才反应过来是 reducer 的问题。所以记住一条凡是需要累积的字段对话历史、工具调用记录、中间结果列表都要配 reducer凡是需要覆盖的字段当前步骤名、临时标志位用默认行为就行。reducer 也可以自己写。比如你想让某个字段做数值累加def add_numbers(left: int, right: int) - int: return left right class State(TypedDict): total: Annotated[int, add_numbers]这样每次节点返回{total: 1}状态里的total就会加 1 而不是被覆盖成 1。这个机制在做计数器、累积评分之类的场景很有用。2.2 节点函数的返回值语义节点就是一个普通函数签名是def node(state: AgentState) - dict。它接收完整状态返回一个字典表示要更新的字段。这里有个容易搞混的点返回的字典不需要包含所有字段只包含你要改的字段即可。LangGraph 会拿这个字典去和原状态做合并合并规则就是上面说的 reducer。def call_model(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]}这个节点只更新了messages其他字段原样保留。如果你返回了状态里不存在的字段LangGraph 会直接报错这点比某些框架严格但其实是好事能帮你早发现拼写错误。另一个细节节点函数可以是同步的也可以是异步的。如果你的 LLM 调用是ainvoke那节点就定义成async defLangGraph 会自动处理。但注意不要在同一个图里混用同步和异步节点虽然技术上可能跑得起来但事件循环的处理会变得很微妙我建议统一用一种。2.3 编译与入口点设置定义完节点和边之后要调用.compile()把图编译成一个可执行对象from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_edge(tools, agent) builder.add_conditional_edges(agent, should_continue, {continue: tools, end: END}) graph builder.compile()START和END是两个特殊节点分别代表图的入口和出口。add_edge(START, agent)表示图一启动就先执行agent节点。编译这一步会做一堆校验比如检查有没有孤立节点、条件边的目标是否都存在。我建议每次改完图结构都重新编译跑一遍让校验帮你抓低级错误比运行时才发现强。编译后的graph可以.invoke()也可以.stream()。调试阶段强烈建议用.stream()它能让你看到每个节点执行后的状态快照排查问题效率高很多。后面讲工具调用循环的时候我会具体演示怎么用。3. 条件路由让 Agent 自己决定下一步走向条件路由是 LangGraph 区别于普通 Chain 的灵魂。它的作用是在某个节点执行完之后根据当前状态动态决定走哪条边。对 Agent 来说最典型的应用就是模型输出里有没有工具调用请求有就去执行工具没有就直接结束。3.1 路由函数的写法与返回值约定路由函数接收状态返回一个字符串这个字符串会被映射到具体的下一个节点def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return continue return end然后在add_conditional_edges里把返回值映射到节点builder.add_conditional_edges( agent, should_continue, { continue: tools, end: END } )这里的映射字典是关键。路由函数返回的字符串必须在这个字典里有对应项否则运行时会报错。我习惯把映射字典写成变量方便复用和检查ROUTE_MAP {continue: tools, end: END}路由函数里可以做任意复杂的判断。比如你想让 Agent 在工具调用失败超过 3 次时强制结束就可以在状态里加个retry_count字段路由函数里判断这个计数def should_continue(state: AgentState) - str: if state.get(retry_count, 0) 3: return end last_message state[messages][-1] return continue if last_message.tool_calls else end这种带熔断的路由在实际项目里非常必要。我见过太多 Agent 因为工具一直报错、模型一直重试最后烧掉大量 token 还卡死的情况。加个重试上限成本可控体验也稳定。3.2 多分支路由与路径映射的坑条件路由不一定只有两条分支。比如一个客服 Agent可能需要根据用户意图走查订单退款转人工三条路def route_by_intent(state: AgentState) - str: intent state[intent] if intent query_order: return order_node elif intent refund: return refund_node else: return human_node builder.add_conditional_edges( classify, route_by_intent, { order_node: order_node, refund_node: refund_node, human_node: human_node } )这里有个坑映射字典的 key 必须和路由函数返回值完全一致包括大小写。我有一次路由函数返回Order_Node映射字典里写的是order_node结果运行时报KeyError但报错信息不太直观找了好一会儿。后来我养成了一个习惯把路由返回值定义成常量路由函数和映射字典都引用常量从根上杜绝拼写不一致。ROUTE_ORDER order_node ROUTE_REFUND refund_node ROUTE_HUMAN human_node def route_by_intent(state) - str: ... return ROUTE_ORDER builder.add_conditional_edges(classify, route_by_intent, { ROUTE_ORDER: order_node, ROUTE_REFUND: refund_node, ROUTE_HUMAN: human_node })3.3 用条件边实现规划-执行模式热词里提到的planning 模式 langgraph本质上就是用条件边实现的。思路是先让模型生成一个计划一串待执行的步骤然后一个执行节点按顺序执行每执行完一步就回到路由判断还有没有下一步有就继续执行没有就汇总输出。class PlanState(TypedDict): task: str plan: list[str] current_step: int results: Annotated[list, add_messages] def planner(state: PlanState) - dict: plan llm.invoke(f把任务拆成步骤{state[task]}) return {plan: plan.steps, current_step: 0} def executor(state: PlanState) - dict: step state[plan][state[current_step]] result llm.invoke(f执行这一步{step}) return { results: [result], current_step: state[current_step] 1 } def has_next_step(state: PlanState) - str: if state[current_step] len(state[plan]): return next return done这个模式的好处是执行过程完全可控你能清楚看到每一步在干什么而不是让模型在一个大循环里自由发挥。对于需要审计、需要复现的任务这种结构比纯 ReAct 靠谱得多。4. Agent 工具调用循环从绑定工具到循环终止工具调用循环是 Agent 的心脏。它的基本逻辑是模型决定调用工具 → 执行工具 → 把结果喂回模型 → 模型再决定下一步。这个循环什么时候停当模型不再请求工具、直接给出最终回答的时候。4.1 工具绑定与 ToolNode 的使用先把工具定义好用tool装饰器from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。 return f{city}今天晴25度 tool def calculate(expression: str) - str: 计算数学表达式。 return str(eval(expression))然后绑定到模型上tools [get_weather, calculate] llm_with_tools llm.bind_tools(tools)bind_tools会把工具的 schema 转换成模型能理解的格式模型在输出里就会带上tool_calls字段。接下来用 LangGraph 预置的ToolNode来执行工具from langgraph.prebuilt import ToolNode tool_node ToolNode(tools)ToolNode会自动读取最后一条消息里的tool_calls逐个执行然后把结果包装成ToolMessage追加到消息列表。这一步省了很多手写解析的功夫。但要注意工具函数的 docstring 非常重要模型就是靠它来判断该不该调用这个工具的。docstring 写得含糊模型就会乱调或者不调。我一般要求 docstring 里写清楚这个工具做什么、参数是什么含义、什么时候该用。4.2 循环的组装与终止条件把模型节点、工具节点、条件边拼起来就是一个完整的 Agent 循环builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_conditional_edges(agent, should_continue, { continue: tools, end: END }) builder.add_edge(tools, agent) graph builder.compile()注意builder.add_edge(tools, agent)这条边——工具执行完必须回到模型节点让模型看到工具结果再决定下一步。这就是循环的关键。如果忘了这条边工具执行完图就停了模型永远看不到结果。终止条件由should_continue控制模型输出里没有tool_calls就返回end图结束。这个判断逻辑简单但极其重要写错了要么死循环要么提前结束。4.3 用 stream 观察循环的每一步调试 Agent 循环stream是最好用的工具for chunk in graph.stream({messages: [(user, 北京天气怎么样顺便算一下 23*47)]}): for node_name, node_output in chunk.items(): print(f 节点: {node_name} ) print(node_output)你会看到类似这样的输出序列agent节点输出带tool_calls的消息 →tools节点输出工具结果 →agent节点输出最终回答。每一步的状态变化都清清楚楚。我排查模型为什么不调工具这类问题时就是靠这个输出看模型到底返回了什么。4.4 循环里最常见的三个坑第一个坑是工具调用参数格式错误。模型有时候会把参数写成字符串而不是对象或者漏掉必填参数。ToolNode会抛异常但如果你没做错误处理整个图就崩了。我的做法是在工具函数里加 try/except把错误信息作为正常结果返回给模型让模型自己纠正tool def calculate(expression: str) - str: 计算数学表达式。 try: return str(eval(expression)) except Exception as e: return f计算出错{e}请检查表达式格式第二个坑是无限循环。模型可能反复调用同一个工具每次都拿到相似结果但就是不给出最终答案。除了前面说的重试计数熔断还可以在路由函数里检查最近 N 条消息是否都是工具调用如果是就强制结束。第三个坑是消息历史膨胀。循环跑多了messages列表会越来越长每次调用模型都要把全部历史传进去token 消耗飞快。解决办法是加一个消息裁剪节点只保留最近 K 条消息或者用摘要的方式压缩早期历史。这个在长对话 Agent 里是必须做的优化。5. 把图跑起来一个完整可复现的 Agent 示例前面拆开讲了各个部件现在把它们组装成一个能直接跑的完整例子。这个 Agent 能查天气、能做计算会根据问题自动决定调哪个工具。5.1 完整代码与逐段说明from typing import Annotated, TypedDict from langchain_core.tools import tool from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode # 1. 定义工具 tool def get_weather(city: str) - str: 查询指定城市的天气情况。参数 city 是城市名。 weather_data {北京: 晴25度, 上海: 多云28度} return weather_data.get(city, f{city}的天气数据暂不可用) tool def calculate(expression: str) - str: 计算数学表达式参数 expression 是合法的 Python 算术表达式。 try: return str(eval(expression)) except Exception as e: return f计算失败{e} tools [get_weather, calculate] # 2. 定义状态 class AgentState(TypedDict): messages: Annotated[list, add_messages] # 3. 初始化模型并绑定工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools) # 4. 定义节点 def call_model(state: AgentState) - dict: response llm_with_tools.invoke(state[messages]) return {messages: [response]} tool_node ToolNode(tools) # 5. 定义路由 def should_continue(state: AgentState) - str: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return continue return end # 6. 组装图 builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_conditional_edges(agent, should_continue, { continue: tools, end: END }) builder.add_edge(tools, agent) graph builder.compile() # 7. 运行 result graph.invoke({ messages: [HumanMessage(content北京天气怎么样再帮我算下 128*32)] }) for msg in result[messages]: print(f[{msg.type}] {msg.content})跑下来你会看到消息序列用户消息 → AI 消息带两个 tool_calls→ 两条 ToolMessage → AI 最终回答。整个循环自动完成不需要你手动判断。5.2 从单轮到多轮状态如何跨轮保持上面的例子是单轮问答。要做多轮对话只需要把上一轮的messages传进下一轮state {messages: [HumanMessage(content北京天气怎么样)]} state graph.invoke(state) state[messages].append(HumanMessage(content那上海呢)) state graph.invoke(state)因为messages配了add_messagesreducer历史会自动累积。但要注意graph.invoke返回的是完整状态你直接拿它的messages继续用就行。如果用的是stream需要自己收集消息稍微麻烦一点。多轮场景下消息裁剪就变得重要了。我一般会在call_model节点里加一段逻辑如果len(state[messages]) 20就只保留系统消息加最近 10 条。这样既控制了 token又不会丢掉关键上下文。5.3 加一个人工确认节点有些工具调用是有副作用的比如发邮件、下单、删数据。这种操作最好加个人工确认。LangGraph 支持在节点里调用interrupt暂停图执行from langgraph.types import interrupt def confirm_node(state: AgentState) - dict: decision interrupt(即将执行敏感操作是否继续) return {messages: [HumanMessage(contentf用户决定{decision})]}图跑到这个节点会暂停把控制权交回给你。你确认后再用Command(resumeyes)恢复执行。这个机制在做需要审批的 Agent 时非常实用也是 LangGraph 相比手写循环的一大优势。6. 调试与优化让 Agent 稳定跑在生产环境Agent 能跑通和能上生产是两回事。下面这些是我在实际项目里踩出来的经验。6.1 常见报错与定位思路agent execution terminated due to error这类报错八成是工具执行抛异常了。定位方法是把stream打开看是哪个节点出的问题。如果是tools节点报错就去检查工具函数的参数和返回值如果是agent节点报错多半是模型调用的问题检查 API key、模型名、消息格式。还有一个高频问题是图编译报错比如节点不存在或边指向未定义的节点。这种一般是拼写错误仔细核对add_node和add_edge里的节点名。我建议节点名用常量定义避免手写字符串。6.2 控制 token 消耗的几个手段Agent 循环最烧钱的地方在于每轮都要把完整消息历史传给模型。三个优化手段一是消息裁剪只保留最近 K 条二是工具结果精简工具返回的内容不要塞一大堆无关信息只返回模型需要的关键字段三是给模型设置max_tokens防止它输出过长。另外temperature设成 0 对 Agent 场景很重要。Agent 需要的是稳定、可预测的决策不是创意发挥。温度高了模型可能这轮调工具、下轮不调行为不一致调试起来很痛苦。6.3 工具设计的经验法则工具不是越多越好。工具太多模型选择困难容易调错。我的经验是单个 Agent 的工具控制在 5 到 10 个超过就考虑拆成多个 Agent 或者用路由先分类。工具的描述要具体。比如查询订单这个工具docstring 里要写清楚根据订单号查询订单状态参数 order_id 是订单编号格式为纯数字。模型看到这些信息才知道什么时候该调、参数怎么填。工具的错误处理要友好。工具内部出错时不要直接抛异常而是返回一段描述性的错误信息让模型有机会自我纠正。这比整个图崩掉体验好得多。6.4 从单 Agent 到多 Agent 的演进路径当单个 Agent 的工具超过 10 个、或者任务类型差异很大时就该考虑多 Agent 了。LangGraph 支持子图你可以把每个 Agent 定义成一张子图然后用一张主图来编排它们。主图负责路由这个任务该交给哪个 Agent子图负责具体执行。多 Agent 的核心难点是状态共享和消息传递。子图之间的状态是隔离的需要通过主图的状态来传递。我一般会在主图状态里放一个shared_context字段各个子图读写这个字段来交换信息。这块展开讲能写一整篇这里先点到为止等你把单 Agent 玩熟了再往上走。最后分享一个我自己的习惯每次搭新 Agent先用stream跑几个典型 case把每个节点的输入输出都打印出来看一遍。确认流程符合预期了再去做优化和封装。跳过这一步直接上生产出问题的时候你会连从哪查起都不知道。 LangGraph 实战：从 StateGraph 到 Agent 工具调用循环
1. 从 LangChain 到 LangGraph为什么需要一张图如果你已经用过 LangChain 的AgentExecutor大概率经历过这种场景想让 Agent 先查资料、再判断要不要调用计算器、最后根据结果决定是继续查还是直接回答。用 Chain 写出来代码会变成一堆if/else嵌套调试的时候根本看不清执行到哪一步出了问题只能靠打日志猜。这不是你写代码的水平问题而是 Chain 这种线性结构本身就不适合表达带分支和循环的流程。LangGraph 要解决的就是这件事。它把 Agent 的执行过程建模成一张有向图节点Node代表一个具体的处理步骤边Edge代表步骤之间的流转关系而状态State则是在整张图上流动的数据。你可以把它理解成给 LLM 应用画了一张流程图只不过这张图是可以在运行时动态决定下一步走哪条边的。这里必须先厘清一个高频困惑LangGraph 和 LangChain 到底什么关系。简单说LangChain 提供的是组件层的东西——模型封装、Prompt 模板、工具定义、检索器、输出解析器这些在 LangGraph 里照样能用。LangGraph 则是在这些组件之上提供了一层编排层。它不替代 LangChain而是补上了 LangChain 在复杂控制流上的短板。所以你会看到 LangGraph 的代码里到处from langchain_core...这很正常两者是配合关系不是替代关系。那什么场景该上 LangGraph我的判断标准是三条第一流程里有条件分支下一步做什么取决于上一步的输出第二流程里有循环比如 Agent 反复调用工具直到拿到满意答案第三需要人工介入在某个节点暂停等确认再继续。这三条只要中了一条用 LangGraph 就比硬写 Chain 舒服得多。反过来如果就是一个输入→模型→输出的直线流程那用 LangChain 的 LCEL 就够了上 LangGraph 属于杀鸡用牛刀。下面这张表可以帮你快速做决策场景特征推荐方案原因单次模型调用、线性 Prompt 链LangChain LCEL结构简单无需图编排固定顺序的多步处理LangChain LCEL用管道符串联即可需要根据结果走不同分支LangGraph条件边天然支持Agent 反复调用工具LangGraph循环结构清晰可控需要人工审核后继续LangGraph支持中断与恢复多 Agent 协作LangGraph子图与状态共享机制成熟理解了这层定位后面的 StateGraph、条件路由、工具调用循环就都是水到渠成的事了。2. StateGraph 的核心状态怎么定义节点怎么读写StateGraph 是 LangGraph 里最基础的图类型几乎所有 Agent 都从它开始搭。它的核心思想是整张图共享一个状态对象每个节点读取状态、返回状态的更新部分。这个设计看起来简单但里面有几个坑不搞清楚写出来的 Agent 行为会很诡异。2.1 State 的 schema 设计与 reducer 机制状态的定义用TypedDict或者 Pydantic 模型都行我一般用TypedDict轻量且够用。一个典型的 Agent 状态长这样from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next_action: str tool_result: str这里最关键的是Annotated[list, add_messages]这个写法。add_messages是一个reducer 函数它决定了当多个节点都往messages里写东西时这些写入怎么合并。默认情况下LangGraph 对状态的更新是覆盖语义——节点返回什么状态里对应字段就变成什么。但对话历史显然不能覆盖你需要的是追加。add_messages做的就是这件事它会把新消息追加到列表末尾而且还会自动处理消息 ID 去重。我踩过的一个坑是一开始没加add_messages直接写messages: list结果 Agent 每轮对话都只记得最后一条消息前面全丢了。排查了半天才反应过来是 reducer 的问题。所以记住一条凡是需要累积的字段对话历史、工具调用记录、中间结果列表都要配 reducer凡是需要覆盖的字段当前步骤名、临时标志位用默认行为就行。reducer 也可以自己写。比如你想让某个字段做数值累加def add_numbers(left: int, right: int) - int: return left right class State(TypedDict): total: Annotated[int, add_numbers]这样每次节点返回{total: 1}状态里的total就会加 1 而不是被覆盖成 1。这个机制在做计数器、累积评分之类的场景很有用。2.2 节点函数的返回值语义节点就是一个普通函数签名是def node(state: AgentState) - dict。它接收完整状态返回一个字典表示要更新的字段。这里有个容易搞混的点返回的字典不需要包含所有字段只包含你要改的字段即可。LangGraph 会拿这个字典去和原状态做合并合并规则就是上面说的 reducer。def call_model(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response]}这个节点只更新了messages其他字段原样保留。如果你返回了状态里不存在的字段LangGraph 会直接报错这点比某些框架严格但其实是好事能帮你早发现拼写错误。另一个细节节点函数可以是同步的也可以是异步的。如果你的 LLM 调用是ainvoke那节点就定义成async defLangGraph 会自动处理。但注意不要在同一个图里混用同步和异步节点虽然技术上可能跑得起来但事件循环的处理会变得很微妙我建议统一用一种。2.3 编译与入口点设置定义完节点和边之后要调用.compile()把图编译成一个可执行对象from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_edge(tools, agent) builder.add_conditional_edges(agent, should_continue, {continue: tools, end: END}) graph builder.compile()START和END是两个特殊节点分别代表图的入口和出口。add_edge(START, agent)表示图一启动就先执行agent节点。编译这一步会做一堆校验比如检查有没有孤立节点、条件边的目标是否都存在。我建议每次改完图结构都重新编译跑一遍让校验帮你抓低级错误比运行时才发现强。编译后的graph可以.invoke()也可以.stream()。调试阶段强烈建议用.stream()它能让你看到每个节点执行后的状态快照排查问题效率高很多。后面讲工具调用循环的时候我会具体演示怎么用。3. 条件路由让 Agent 自己决定下一步走向条件路由是 LangGraph 区别于普通 Chain 的灵魂。它的作用是在某个节点执行完之后根据当前状态动态决定走哪条边。对 Agent 来说最典型的应用就是模型输出里有没有工具调用请求有就去执行工具没有就直接结束。3.1 路由函数的写法与返回值约定路由函数接收状态返回一个字符串这个字符串会被映射到具体的下一个节点def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return continue return end然后在add_conditional_edges里把返回值映射到节点builder.add_conditional_edges( agent, should_continue, { continue: tools, end: END } )这里的映射字典是关键。路由函数返回的字符串必须在这个字典里有对应项否则运行时会报错。我习惯把映射字典写成变量方便复用和检查ROUTE_MAP {continue: tools, end: END}路由函数里可以做任意复杂的判断。比如你想让 Agent 在工具调用失败超过 3 次时强制结束就可以在状态里加个retry_count字段路由函数里判断这个计数def should_continue(state: AgentState) - str: if state.get(retry_count, 0) 3: return end last_message state[messages][-1] return continue if last_message.tool_calls else end这种带熔断的路由在实际项目里非常必要。我见过太多 Agent 因为工具一直报错、模型一直重试最后烧掉大量 token 还卡死的情况。加个重试上限成本可控体验也稳定。3.2 多分支路由与路径映射的坑条件路由不一定只有两条分支。比如一个客服 Agent可能需要根据用户意图走查订单退款转人工三条路def route_by_intent(state: AgentState) - str: intent state[intent] if intent query_order: return order_node elif intent refund: return refund_node else: return human_node builder.add_conditional_edges( classify, route_by_intent, { order_node: order_node, refund_node: refund_node, human_node: human_node } )这里有个坑映射字典的 key 必须和路由函数返回值完全一致包括大小写。我有一次路由函数返回Order_Node映射字典里写的是order_node结果运行时报KeyError但报错信息不太直观找了好一会儿。后来我养成了一个习惯把路由返回值定义成常量路由函数和映射字典都引用常量从根上杜绝拼写不一致。ROUTE_ORDER order_node ROUTE_REFUND refund_node ROUTE_HUMAN human_node def route_by_intent(state) - str: ... return ROUTE_ORDER builder.add_conditional_edges(classify, route_by_intent, { ROUTE_ORDER: order_node, ROUTE_REFUND: refund_node, ROUTE_HUMAN: human_node })3.3 用条件边实现规划-执行模式热词里提到的planning 模式 langgraph本质上就是用条件边实现的。思路是先让模型生成一个计划一串待执行的步骤然后一个执行节点按顺序执行每执行完一步就回到路由判断还有没有下一步有就继续执行没有就汇总输出。class PlanState(TypedDict): task: str plan: list[str] current_step: int results: Annotated[list, add_messages] def planner(state: PlanState) - dict: plan llm.invoke(f把任务拆成步骤{state[task]}) return {plan: plan.steps, current_step: 0} def executor(state: PlanState) - dict: step state[plan][state[current_step]] result llm.invoke(f执行这一步{step}) return { results: [result], current_step: state[current_step] 1 } def has_next_step(state: PlanState) - str: if state[current_step] len(state[plan]): return next return done这个模式的好处是执行过程完全可控你能清楚看到每一步在干什么而不是让模型在一个大循环里自由发挥。对于需要审计、需要复现的任务这种结构比纯 ReAct 靠谱得多。4. Agent 工具调用循环从绑定工具到循环终止工具调用循环是 Agent 的心脏。它的基本逻辑是模型决定调用工具 → 执行工具 → 把结果喂回模型 → 模型再决定下一步。这个循环什么时候停当模型不再请求工具、直接给出最终回答的时候。4.1 工具绑定与 ToolNode 的使用先把工具定义好用tool装饰器from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的天气。 return f{city}今天晴25度 tool def calculate(expression: str) - str: 计算数学表达式。 return str(eval(expression))然后绑定到模型上tools [get_weather, calculate] llm_with_tools llm.bind_tools(tools)bind_tools会把工具的 schema 转换成模型能理解的格式模型在输出里就会带上tool_calls字段。接下来用 LangGraph 预置的ToolNode来执行工具from langgraph.prebuilt import ToolNode tool_node ToolNode(tools)ToolNode会自动读取最后一条消息里的tool_calls逐个执行然后把结果包装成ToolMessage追加到消息列表。这一步省了很多手写解析的功夫。但要注意工具函数的 docstring 非常重要模型就是靠它来判断该不该调用这个工具的。docstring 写得含糊模型就会乱调或者不调。我一般要求 docstring 里写清楚这个工具做什么、参数是什么含义、什么时候该用。4.2 循环的组装与终止条件把模型节点、工具节点、条件边拼起来就是一个完整的 Agent 循环builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_conditional_edges(agent, should_continue, { continue: tools, end: END }) builder.add_edge(tools, agent) graph builder.compile()注意builder.add_edge(tools, agent)这条边——工具执行完必须回到模型节点让模型看到工具结果再决定下一步。这就是循环的关键。如果忘了这条边工具执行完图就停了模型永远看不到结果。终止条件由should_continue控制模型输出里没有tool_calls就返回end图结束。这个判断逻辑简单但极其重要写错了要么死循环要么提前结束。4.3 用 stream 观察循环的每一步调试 Agent 循环stream是最好用的工具for chunk in graph.stream({messages: [(user, 北京天气怎么样顺便算一下 23*47)]}): for node_name, node_output in chunk.items(): print(f 节点: {node_name} ) print(node_output)你会看到类似这样的输出序列agent节点输出带tool_calls的消息 →tools节点输出工具结果 →agent节点输出最终回答。每一步的状态变化都清清楚楚。我排查模型为什么不调工具这类问题时就是靠这个输出看模型到底返回了什么。4.4 循环里最常见的三个坑第一个坑是工具调用参数格式错误。模型有时候会把参数写成字符串而不是对象或者漏掉必填参数。ToolNode会抛异常但如果你没做错误处理整个图就崩了。我的做法是在工具函数里加 try/except把错误信息作为正常结果返回给模型让模型自己纠正tool def calculate(expression: str) - str: 计算数学表达式。 try: return str(eval(expression)) except Exception as e: return f计算出错{e}请检查表达式格式第二个坑是无限循环。模型可能反复调用同一个工具每次都拿到相似结果但就是不给出最终答案。除了前面说的重试计数熔断还可以在路由函数里检查最近 N 条消息是否都是工具调用如果是就强制结束。第三个坑是消息历史膨胀。循环跑多了messages列表会越来越长每次调用模型都要把全部历史传进去token 消耗飞快。解决办法是加一个消息裁剪节点只保留最近 K 条消息或者用摘要的方式压缩早期历史。这个在长对话 Agent 里是必须做的优化。5. 把图跑起来一个完整可复现的 Agent 示例前面拆开讲了各个部件现在把它们组装成一个能直接跑的完整例子。这个 Agent 能查天气、能做计算会根据问题自动决定调哪个工具。5.1 完整代码与逐段说明from typing import Annotated, TypedDict from langchain_core.tools import tool from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode # 1. 定义工具 tool def get_weather(city: str) - str: 查询指定城市的天气情况。参数 city 是城市名。 weather_data {北京: 晴25度, 上海: 多云28度} return weather_data.get(city, f{city}的天气数据暂不可用) tool def calculate(expression: str) - str: 计算数学表达式参数 expression 是合法的 Python 算术表达式。 try: return str(eval(expression)) except Exception as e: return f计算失败{e} tools [get_weather, calculate] # 2. 定义状态 class AgentState(TypedDict): messages: Annotated[list, add_messages] # 3. 初始化模型并绑定工具 llm ChatOpenAI(modelgpt-4o-mini, temperature0) llm_with_tools llm.bind_tools(tools) # 4. 定义节点 def call_model(state: AgentState) - dict: response llm_with_tools.invoke(state[messages]) return {messages: [response]} tool_node ToolNode(tools) # 5. 定义路由 def should_continue(state: AgentState) - str: last_message state[messages][-1] if getattr(last_message, tool_calls, None): return continue return end # 6. 组装图 builder StateGraph(AgentState) builder.add_node(agent, call_model) builder.add_node(tools, tool_node) builder.add_edge(START, agent) builder.add_conditional_edges(agent, should_continue, { continue: tools, end: END }) builder.add_edge(tools, agent) graph builder.compile() # 7. 运行 result graph.invoke({ messages: [HumanMessage(content北京天气怎么样再帮我算下 128*32)] }) for msg in result[messages]: print(f[{msg.type}] {msg.content})跑下来你会看到消息序列用户消息 → AI 消息带两个 tool_calls→ 两条 ToolMessage → AI 最终回答。整个循环自动完成不需要你手动判断。5.2 从单轮到多轮状态如何跨轮保持上面的例子是单轮问答。要做多轮对话只需要把上一轮的messages传进下一轮state {messages: [HumanMessage(content北京天气怎么样)]} state graph.invoke(state) state[messages].append(HumanMessage(content那上海呢)) state graph.invoke(state)因为messages配了add_messagesreducer历史会自动累积。但要注意graph.invoke返回的是完整状态你直接拿它的messages继续用就行。如果用的是stream需要自己收集消息稍微麻烦一点。多轮场景下消息裁剪就变得重要了。我一般会在call_model节点里加一段逻辑如果len(state[messages]) 20就只保留系统消息加最近 10 条。这样既控制了 token又不会丢掉关键上下文。5.3 加一个人工确认节点有些工具调用是有副作用的比如发邮件、下单、删数据。这种操作最好加个人工确认。LangGraph 支持在节点里调用interrupt暂停图执行from langgraph.types import interrupt def confirm_node(state: AgentState) - dict: decision interrupt(即将执行敏感操作是否继续) return {messages: [HumanMessage(contentf用户决定{decision})]}图跑到这个节点会暂停把控制权交回给你。你确认后再用Command(resumeyes)恢复执行。这个机制在做需要审批的 Agent 时非常实用也是 LangGraph 相比手写循环的一大优势。6. 调试与优化让 Agent 稳定跑在生产环境Agent 能跑通和能上生产是两回事。下面这些是我在实际项目里踩出来的经验。6.1 常见报错与定位思路agent execution terminated due to error这类报错八成是工具执行抛异常了。定位方法是把stream打开看是哪个节点出的问题。如果是tools节点报错就去检查工具函数的参数和返回值如果是agent节点报错多半是模型调用的问题检查 API key、模型名、消息格式。还有一个高频问题是图编译报错比如节点不存在或边指向未定义的节点。这种一般是拼写错误仔细核对add_node和add_edge里的节点名。我建议节点名用常量定义避免手写字符串。6.2 控制 token 消耗的几个手段Agent 循环最烧钱的地方在于每轮都要把完整消息历史传给模型。三个优化手段一是消息裁剪只保留最近 K 条二是工具结果精简工具返回的内容不要塞一大堆无关信息只返回模型需要的关键字段三是给模型设置max_tokens防止它输出过长。另外temperature设成 0 对 Agent 场景很重要。Agent 需要的是稳定、可预测的决策不是创意发挥。温度高了模型可能这轮调工具、下轮不调行为不一致调试起来很痛苦。6.3 工具设计的经验法则工具不是越多越好。工具太多模型选择困难容易调错。我的经验是单个 Agent 的工具控制在 5 到 10 个超过就考虑拆成多个 Agent 或者用路由先分类。工具的描述要具体。比如查询订单这个工具docstring 里要写清楚根据订单号查询订单状态参数 order_id 是订单编号格式为纯数字。模型看到这些信息才知道什么时候该调、参数怎么填。工具的错误处理要友好。工具内部出错时不要直接抛异常而是返回一段描述性的错误信息让模型有机会自我纠正。这比整个图崩掉体验好得多。6.4 从单 Agent 到多 Agent 的演进路径当单个 Agent 的工具超过 10 个、或者任务类型差异很大时就该考虑多 Agent 了。LangGraph 支持子图你可以把每个 Agent 定义成一张子图然后用一张主图来编排它们。主图负责路由这个任务该交给哪个 Agent子图负责具体执行。多 Agent 的核心难点是状态共享和消息传递。子图之间的状态是隔离的需要通过主图的状态来传递。我一般会在主图状态里放一个shared_context字段各个子图读写这个字段来交换信息。这块展开讲能写一整篇这里先点到为止等你把单 Agent 玩熟了再往上走。最后分享一个我自己的习惯每次搭新 Agent先用stream跑几个典型 case把每个节点的输入输出都打印出来看一遍。确认流程符合预期了再去做优化和封装。跳过这一步直接上生产出问题的时候你会连从哪查起都不知道。