拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

LangGraph 实战:StateGraph 与条件路由构建工具调用循环

LangGraph 实战:StateGraph 与条件路由构建工具调用循环

1. 从零理解 LangGraph 到底解决了什么问题

1.1 为什么单纯的链式调用不够用了

如果你用 LangChain 写过稍微复杂一点的东西,大概率会遇到一个尴尬的局面:业务逻辑一旦出现分支、循环、重试,整个 Chain 就变得极其难维护。比如一个客服机器人,用户问“帮我查下订单”,你得先判断意图,再决定是走查询订单的分支,还是走退换货的分支,查询失败还得重试,重试超过三次要转人工。这种带状态、带分支、带循环的流程,用传统的 Chain 写出来就是一堆 if-else 嵌套,调试的时候根本不知道卡在哪一步。

LangGraph 就是冲着这个痛点来的。它把整个流程抽象成一张有向图,节点是具体的处理函数,边是流转逻辑,而贯穿始终的是一个可持久化的State。你可以把它理解成给 LLM 应用装了一个“状态机引擎”,让 Agent 的每一步决策都有迹可循、可回放、可中断恢复。

我刚开始接触的时候也有疑问:这不就是把工作流引擎那套东西搬到 LLM 上吗?用下来发现确实有这个味道,但 LangGraph 针对 LLM 场景做了很多专门设计,比如支持流式输出、支持 human-in-the-loop 中断、支持 checkpoint 持久化,这些是通用工作流引擎给不了的。

1.2 StateGraph、条件路由、工具调用循环三者的关系

这三个概念其实是 LangGraph 构建 Agent 的三块基石,缺一不可。

StateGraph是骨架,它定义了整张图有哪些节点、状态怎么在节点之间传递。条件路由是神经,它决定了在当前状态下,下一步该走哪个节点——这是 Agent 具备“决策能力”的关键。工具调用循环是肌肉,它让 Agent 能够真正调用外部工具(搜索、计算、查数据库),拿到结果后回到模型继续推理,直到任务完成。

打个比方:StateGraph 是高速公路的路网,条件路由是每个路口的路牌和红绿灯,工具调用循环是车辆在路网里反复跑直到抵达目的地。少了任何一个,Agent 都跑不起来。

很多人学 LangGraph 卡住,不是因为 API 难,而是没想清楚这三者怎么配合。我见过不少初学者把工具调用逻辑硬塞进节点函数里,结果图结构一团乱。正确的做法是:让图结构去表达控制流,让节点函数只负责单一职责。这个原则后面会反复提到。

1.3 这篇文章适合谁看,能学到什么程度

如果你已经会写 Python,用过 LangChain 的基础组件(Prompt、LLM、Tool),但一想到要搭一个能自主决策的 Agent 就不知道从哪下手,那这篇内容就是给你准备的。我会从最基础的 StateGraph 定义讲起,一路讲到完整的工具调用循环,中间穿插条件路由的写法、状态设计的心法、以及我自己踩过的坑。

看完之后你应该能做到:独立设计一个带分支和循环的 Agent 图,知道状态字段该怎么定义,条件路由函数该怎么写才不会出 bug,工具调用循环怎么防止死循环。这些都是实际项目里天天要面对的问题,不是玩具 demo 级别的。

2. StateGraph 的核心机制与状态设计心法

2.1 StateGraph 的执行模型:节点、边、状态三者如何协作

LangGraph 的执行模型可以用一句话概括:状态在节点之间流动,边决定流动方向,节点负责修改状态。

具体来说,你首先定义一个 State 的结构(通常是一个 TypedDict),里面放所有需要在节点间共享的数据。然后创建 StateGraph 实例,把 State 类型传进去。接着用add_node注册节点,每个节点是一个函数,接收当前 State,返回一个字典表示要更新的字段。最后用add_edge和add_conditional_edges把这些节点连起来,编译成可执行图。

执行的时候,LangGraph 从入口节点开始,把初始 State 传进去,节点函数返回的更新会被合并到 State 里(默认是覆盖,用 Annotated 可以改成追加),然后根据边找到下一个节点,如此循环直到抵达 END。

这里有个容易忽略的点:节点函数返回的字典不需要包含所有字段,只需要包含你想更新的字段。LangGraph 会自动做合并。这个设计很关键,它让每个节点只关心自己负责的那部分状态,职责清晰。

2.2 状态字段怎么定义才不会踩坑

状态设计是 LangGraph 里最考验功力的地方。我总结了三条原则。

第一,消息列表用Annotated[list, add_messages]。这是官方推荐的做法,add_messages是一个 reducer,它会把新消息追加到列表里,而不是覆盖。如果你直接写messages: list,每次节点返回新消息就会把历史全冲掉,Agent 直接失忆。这个坑我踩过,调试了半天才发现是 reducer 没加。

第二,区分“输入态”和“中间态”。有些字段是用户一开始传进来的(比如 question),有些是流程中间产生的(比如 tool_calls、retry_count)。把它们都放在一个 State 里没问题,但心里要清楚哪些字段是只读的,哪些是可变的。

第三,控制状态膨胀。Agent 跑多轮之后,messages 会越来越长,token 消耗飙升。我的做法是在状态里加一个summary字段,当消息超过一定轮数时,用一个专门的节点把历史压缩成摘要,然后清空 messages。这样既保留了上下文,又控制了成本。

下面是一个我常用的状态定义模板:

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] question: str retry_count: int final_answer: str

2.3 节点函数的职责边界:单一职责是铁律

节点函数最容易犯的错误就是“什么都往里塞”。我见过有人在一个节点里既做意图识别,又做工具调用,还做结果格式化,最后这个函数两百多行,改一处崩三处。

正确的做法是一个节点只做一件事。意图识别一个节点,工具调用一个节点,结果汇总一个节点。这样图结构清晰,调试的时候一眼就能看出是哪一步出了问题,而且节点可以复用——比如工具调用节点,不管上游是哪个分支来的,逻辑都一样。

节点函数的签名也很简单,接收 state,返回 dict。返回的 dict 里放你要更新的字段。如果这个节点不需要更新任何状态(比如只是做日志),返回空字典{}也行。

提示:节点函数尽量保持纯函数风格,不要在里面做副作用操作(比如直接写数据库)。副作用应该封装成工具,通过工具调用节点来触发。这样整个图的行为是可预测的。

3. 条件路由:让 Agent 学会“看情况走哪条路”

3.1 条件路由的本质是一个返回字符串的函数

条件路由听起来高大上,其实本质特别简单:你写一个函数,接收当前 State,返回一个字符串,这个字符串是下一个节点的名字。LangGraph 根据这个返回值去查边表,找到对应的节点。

用add_conditional_edges注册的时候,需要传三个东西:源节点名、路由函数、以及一个映射表(把路由函数的返回值映射到节点名)。映射表这一步很多人会漏,其实它是可选的——如果路由函数直接返回节点名,映射表可以省略。但我建议还是写上,因为映射表让代码更清晰,路由函数返回语义化的标签(比如 "need_tool"、"done"),映射表负责翻译成实际节点名。

def route_after_llm(state: AgentState) -> str: last_message = state["messages"][-1] if last_message.tool_calls: return "call_tool" return "finish" graph.add_conditional_edges( "llm", route_after_llm, { "call_tool": "tool_node", "finish": END } )

这段代码的意思是:LLM 节点执行完之后,检查最后一条消息有没有工具调用请求。有就去工具节点,没有就结束。这就是最经典的工具调用循环的路由逻辑。

3.2 多分支路由的设计模式

实际项目里路由往往不止两个分支。比如一个客服 Agent,可能需要判断:是查询类问题、投诉类问题、还是闲聊。这时候路由函数就变成一个多路判断。

我的经验是,路由函数里只做判断,不做业务逻辑。判断依据全部来自 State 里已经计算好的字段。比如你可以在前面加一个“分类节点”,把用户意图写进 state["intent"],路由函数只读这个字段。这样路由函数极其简单,就是几个 if-elif,一眼能看懂。

如果分支特别多(超过五个),可以考虑用字典映射代替 if-elif 链:

INTENT_TO_NODE = { "query": "query_node", "complaint": "complaint_node", "chat": "chat_node", } def route_by_intent(state: AgentState) -> str: return INTENT_TO_NODE.get(state["intent"], "chat_node")

这种写法扩展性好,加新意图只需要改字典,不用动路由函数。

3.3 路由函数的常见陷阱与调试技巧

第一个陷阱是路由函数返回了不存在的节点名。LangGraph 编译的时候不会报错,运行时才炸,而且报错信息不一定直观。我的做法是在映射表里把所有可能的返回值都列出来,这样漏了哪个一眼能看出来。

第二个陷阱是状态字段还没被赋值就拿来判断。比如你在路由函数里读state["intent"],但分类节点因为某种原因没执行,就会 KeyError。解决办法是在 State 定义时给默认值,或者用state.get("intent", "chat")这种安全访问。

第三个陷阱是路由逻辑和节点逻辑耦合。有些人把判断逻辑写在节点里,节点返回不同的状态,然后路由函数根据状态判断。这本身没问题,但如果判断逻辑很复杂,建议还是抽到路由函数里,让节点保持纯粹。

调试路由的时候,我习惯在路由函数里加一行日志,打印当前状态和返回的路由结果。跑几次就能看清楚整个决策链路,比断点调试还快。

4. Agent 工具调用循环的完整实现

4.1 工具调用循环的标准结构

一个标准的工具调用循环包含三个核心部分:LLM 节点、工具执行节点、条件路由。流程是这样的:LLM 节点接收消息,调用模型,模型可能返回工具调用请求;路由判断有没有工具调用,有就去工具节点;工具节点执行工具,把结果作为 ToolMessage 追加到消息列表;然后回到 LLM 节点,模型看到工具结果继续推理,直到不再请求工具,路由走向 END。

这个循环的关键在于消息列表的累积。每一轮 LLM 的输出、工具的执行结果,都追加到 messages 里。模型每次都能看到完整的历史,所以它能基于工具返回的结果继续推理。这就是 ReAct 模式在 LangGraph 里的落地方式。

我建议把 LLM 节点和工具节点分开写,不要合并。合并的话,循环的边界就模糊了,而且没法在中间插入人工审核之类的环节。

4.2 工具节点的实现细节

工具节点的核心工作是:从最后一条 AIMessage 里取出 tool_calls,逐个执行,把结果包装成 ToolMessage 返回。

from langchain_core.messages import ToolMessage def tool_node(state: AgentState) -> dict: tool_calls = state["messages"][-1].tool_calls results = [] for call in tool_calls: tool = tools_by_name[call["name"]] output = tool.invoke(call["args"]) results.append( ToolMessage(content=str(output), tool_call_id=call["id"]) ) return {"messages": results}

这里有几个细节要注意。tool_call_id 必须和请求里的 id 对应,否则模型会报错说找不到对应的工具结果。content 必须是字符串,如果工具返回的是字典或列表,要序列化。异常要捕获,工具执行失败不能让整个图崩掉,应该把错误信息作为 ToolMessage 返回,让模型知道工具挂了,它可能会换个方式重试。

我一般会在工具节点里加一个 try-except,把异常信息包装成 ToolMessage。这样即使某个工具临时不可用,Agent 也能优雅降级,而不是直接报错退出。

4.3 防止死循环的三道防线

工具调用循环最大的风险就是死循环:模型一直请求工具,工具一直返回结果,模型不满意继续请求,无限循环下去。我一般设三道防线。

第一道是最大迭代次数。在 State 里加一个iteration字段,每次经过 LLM 节点就加一,路由函数里判断超过阈值就强制走向 END。阈值设多少看场景,我一般设 10 到 15。

第二道是工具调用去重。如果模型连续两次请求完全相同的工具和参数,说明它卡住了,这时候应该中断循环,返回当前结果。实现方式是在 State 里记录已调用的工具签名,路由函数里检查。

第三道是超时控制。整个图执行设置一个总超时,超过就中断。LangGraph 支持在编译时配置 recursion_limit,这个参数控制图的最大递归深度,设一个合理值能兜底。

graph = builder.compile() result = graph.invoke( {"messages": [user_msg]}, config={"recursion_limit": 25} )

这三道防线配合使用,基本能杜绝死循环。我实际项目里跑了几万次调用,没出现过无限循环的情况。

5. 实战:搭一个能查天气和算数的 Agent

5.1 完整代码结构与逐段解析

光讲理论没意思,我们直接搭一个能用的 Agent。功能很简单:用户问天气或者数学问题,Agent 自己决定调用哪个工具。

先定义工具:

from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市的天气""" fake_data = {"北京": "晴,25度", "上海": "多云,28度"} return fake_data.get(city, "暂无数据") @tool def calculate(expression: str) -> str: """计算数学表达式""" try: return str(eval(expression)) except Exception as e: return f"计算失败: {e}" tools = [get_weather, calculate] tools_by_name = {t.name: t for t in tools}

然后是状态定义和节点函数:

from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import ToolMessage from langchain_openai import ChatOpenAI class AgentState(TypedDict): messages: Annotated[list, add_messages] iteration: int llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools) def llm_node(state: AgentState) -> dict: response = llm.invoke(state["messages"]) return { "messages": [response], "iteration": state.get("iteration", 0) + 1 } def tool_node(state: AgentState) -> dict: tool_calls = state["messages"][-1].tool_calls results = [] for call in tool_calls: tool = tools_by_name[call["name"]] try: output = tool.invoke(call["args"]) except Exception as e: output = f"工具执行失败: {e}" results.append( ToolMessage(content=str(output), tool_call_id=call["id"]) ) return {"messages": results}

路由函数和图的组装:

def route_after_llm(state: AgentState) -> str: if state.get("iteration", 0) >= 10: return "finish" last = state["messages"][-1] if getattr(last, "tool_calls", None): return "call_tool" return "finish" builder = StateGraph(AgentState) builder.add_node("llm", llm_node) builder.add_node("tool", tool_node) builder.set_entry_point("llm") builder.add_conditional_edges( "llm", route_after_llm, {"call_tool": "tool", "finish": END} ) builder.add_edge("tool", "llm") graph = builder.compile()

跑一下:

result = graph.invoke({ "messages": [("user", "北京天气怎么样?顺便算一下 23 * 47")], "iteration": 0 }) print(result["messages"][-1].content)

5.2 关键参数的选择与计算过程

recursion_limit这个参数值得单独说一下。它控制的是图的最大递归步数,不是迭代轮数。一次完整的工具调用循环(LLM -> 工具 -> LLM)会消耗两步。所以如果你设 25,大概能支撑 12 轮工具调用。我一般按“最大轮数 × 2 + 缓冲”来算,比如想要 10 轮,就设 25。

iteration字段的阈值我设 10,是因为实际业务里超过 10 轮还没搞定的问题,基本是模型理解错了或者工具设计有问题,继续跑也是浪费 token。这时候中断返回,让用户重新描述问题,比硬撑更划算。

工具节点的异常处理我用了 try-except 包住 invoke,而不是包住整个循环。这样单个工具失败不影响其他工具的执行。如果一次请求里有多个工具调用,其中一个挂了,其他的还能正常返回。

5.3 运行结果分析与验证

跑上面那段代码,你会看到消息列表里依次出现:用户消息、带 tool_calls 的 AIMessage、两个 ToolMessage、最后的 AIMessage(包含最终答案)。整个流程完全符合预期。

验证 Agent 是否正常工作,我一般看三个点:工具是否被正确调用(看 tool_calls 的 name 和 args)、工具结果是否被正确回传(看 ToolMessage 的 content 和 tool_call_id)、最终答案是否基于工具结果(看最后一条 AIMessage 的内容)。

如果模型没有调用工具而是直接回答,说明 bind_tools 没生效或者模型不支持工具调用。如果工具结果回传后模型还在重复调用同一个工具,说明工具返回的内容模型看不懂,需要调整工具的返回格式。

6. 常见问题排查与避坑经验

6.1 状态更新不生效的排查思路

最常见的症状是:节点函数明明返回了新消息,但下一个节点读到的还是旧状态。九成的原因是reducer 没配对。messages 字段必须用Annotated[list, add_messages],如果你写成了普通 list,新消息会覆盖旧消息,看起来就像状态没更新。

另一个原因是节点函数返回的 key 和 State 里定义的 key 不一致。比如 State 里叫messages,你返回{"message": [...]},LangGraph 会忽略这个未知字段。这种错误不报错,特别隐蔽,建议返回前打印一下确认。

还有一种情况是在节点里直接修改了 state 对象,而不是返回新字典。LangGraph 依赖返回值来合并状态,直接改对象不会触发更新。记住:节点函数永远返回新字典,不要原地修改。

6.2 工具调用报错的典型场景

报错信息原因解决办法
tool_call_id not foundToolMessage 的 id 和请求不匹配用 call["id"] 赋值
tool not found工具名拼写错误或未注册检查 tools_by_name 的 key
content must be string工具返回了非字符串用 str() 或 json.dumps() 包装
model does not support tools模型不支持工具调用换支持 function calling 的模型

我遇到最多的是第一个,因为手动构造 ToolMessage 的时候容易忘记传 tool_call_id。用ToolMessage(content=..., tool_call_id=call["id"])这个模板就不会错。

6.3 性能优化的几个实操技巧

Agent 跑得慢,通常是三个原因:模型调用慢、工具执行慢、消息列表太长。

模型调用慢没法根治,但可以并行执行多个工具调用。如果模型一次请求了三个工具,用 asyncio.gather 并发跑,比串行快三倍。LangGraph 支持异步节点,把 tool_node 改成 async 函数就行。

工具执行慢的话,考虑加缓存。同样的工具参数短时间内重复调用,直接返回缓存结果。我在工具节点里加了一个简单的字典缓存,命中率还挺高的。

消息列表太长的话,定期压缩历史。前面提到的 summary 方案,或者只保留最近 N 轮对话,更早的丢弃。这个要权衡,丢太多模型会失去上下文,丢太少 token 成本高。我的经验是保留最近 6 到 8 轮比较合适。

注意:压缩历史的时候,工具调用的请求和结果要成对保留或成对丢弃,不能只留一个,否则模型会困惑。

7. 从能跑到好用:进阶优化方向

7.1 加入人工审核中断

有些场景下,Agent 调用工具之前需要人工确认,比如涉及支付、删除数据这类敏感操作。LangGraph 支持在节点执行前中断,等人工确认后再继续。

实现方式是在编译图的时候配置 interrupt_before 或 interrupt_after,指定在哪个节点前/后暂停。暂停后图的状态会被保存,人工审核通过后调用graph.update_state修改状态,再graph.invoke(None, config)继续执行。

这个功能在生产环境特别有用,它让 Agent 从“全自动”变成“半自动”,在关键节点上保留人的控制权。

7.2 持久化与断点恢复

LangGraph 的 checkpointer 机制可以把每一步的状态存到数据库(内存、SQLite、Postgres 都支持)。这样即使服务重启,Agent 也能从上次中断的地方继续跑。

配置方式是在 compile 的时候传 checkpointer:

from langgraph.checkpoint.memory import MemorySaver graph = builder.compile(checkpointer=MemorySaver()) result = graph.invoke( {"messages": [user_msg]}, config={"configurable": {"thread_id": "user-123"}} )

thread_id 是会话标识,同一个 thread_id 的多次调用会共享状态。这个机制让多轮对话变得非常简单,不用自己管理历史消息。

7.3 多 Agent 协作的图结构设计

单个 Agent 能力有限,复杂任务往往需要多个 Agent 协作。LangGraph 天然支持这种模式:把每个 Agent 做成一个子图,然后用一个主图来编排它们。

常见的模式是“主管 + 专家”:一个主管 Agent 负责拆解任务和分派,多个专家 Agent 各负责一个领域。主管根据任务类型路由到对应的专家,专家处理完把结果返回给主管,主管汇总后输出。

这种结构比单个大 Agent 更可控,每个专家可以有自己的工具集和提示词,职责清晰。缺点是图结构变复杂,调试成本上升。我的建议是先从单 Agent 做起,确实遇到瓶颈再拆多 Agent,不要一上来就搞复杂架构。

8. 我踩过的那些坑和最后想说的

说几个印象深刻的坑。有一次 Agent 死活不调用工具,排查半天发现是工具的 docstring 写得太模糊,模型不知道什么时候该用。后来把 docstring 改得具体一点,比如“查询指定城市的实时天气,参数是城市名”,立刻就正常了。工具的 docstring 就是给模型看的说明书,一定要写清楚。

还有一次状态莫名其妙丢失,最后发现是节点函数里用了state["messages"].append(...)这种原地修改,LangGraph 没检测到变化。改成返回新列表就好了。这个坑很隐蔽,因为代码逻辑上没错,但就是不符合 LangGraph 的更新机制。

最后一个建议:先用小模型跑通流程,再用大模型优化效果。开发阶段用 gpt-4o-mini 这种便宜快速的模型,把图结构和路由逻辑调对,最后再换成更强的模型。这样能省不少钱,调试也快。

LangGraph 的学习曲线确实比 LangChain 陡一点,但一旦理解了状态、节点、边这套模型,你会发现它能表达的东西比 Chain 多得多。工具调用循环只是最基础的应用,后面还有条件分支、并行执行、子图嵌套、人工中断等等玩法。把今天这套骨架吃透,剩下的都是在这个基础上做加法。

返回列表