☰
LangGraph实战:StateGraph、条件路由与Agent工具循环详解
2026/10/1 7:52:56 网站建设 项目流程

LangGraph 这个工具我实际用了一年多,从最早的 0.0.x 版本一路跟到现在的稳定版,中间踩了无数坑,也帮团队落地了好几个生产级的 Agent 项目。很多朋友拿着 LangChain 的文档直接上手 LangGraph,结果被 StateGraph、条件路由、工具调用循环这几个概念绕得晕头转向。这篇文章我就把自己从入门到实战的完整经验梳理一遍,特别是 StateGraph 的状态设计、条件路由的几种典型模式,以及 Agent 工具调用循环的终止条件和防死循环策略,希望帮你少走弯路。

这篇文章适合谁看?你如果已经会写基本的 LLM 调用代码,但对 Agent 的编排架构还不熟悉;或者你用过 LangChain 的 Agent 但觉得它的控制力太弱、流程太黑盒;又或者面试被问到 LangChain 和 LangGraph 的区别时只能说出“LangGraph 是图”这种空话——那这篇文章就是为你准备的。

1. 为什么选 LangGraph:先搞清楚它和 LangChain 的区别

1.1 从 Chain 到 Graph:编排思路的一次升级

很多新手上手 LangGraph 之前,先用过 LangChain 的LCEL(LangChain Expression Language),里面最常见的概念就是Chain。Chain 的核心假设是:流程是线性的。你写一个prompt | model | parser,数据从前到后走一路,中间每个环节都是固定顺序的。但真实的 Agent 场景根本不是线性的——模型需要判断是否调用工具,工具返回结果后模型可能要再思考、再调用,这中间还有分支、有循环、有提前终止。

这里我打个比方。你把 Chain 想象成工厂流水线,产品从传送带起点走到终点,每个工位做固定的事情。但你做一个 Agent 的时候,需要的是一个“车间调度系统”——同一个工件到了某个工位,工人要判断:是返工、是进入下一个工位、还是直接出厂。流水线做不到这个,调度系统可以,LangGraph 就是那个调度系统。

LangGraph 的定位更准确地说是一个有状态、可编排、可细粒度控制的 Agent 运行时。它允许你把流程建模成一张图(Graph):图里有节点(Node)——代表具体执行的函数或步骤;有边(Edge)——代表节点间的流转方向。最关键的是,边可以是条件边,也就是模型执行完一个节点后,下一步走哪里由一段代码逻辑(甚至由 LLM 自行决定)来判断。

这个设计的价值在容错和生产级场景里体现得非常明显。比如你有一个多轮 Agent,第一轮模型决定调用天气 API,第二轮模型想调用日历 API,但工具返回了错误,你可能希望流程“回到”第一轮重新组织答案,而不是沿着固定链路一路跑下去。在 LangChain 里实现这种回退逻辑非常别扭,但在 LangGraph 里这就是一条普通的有向边而已。

1.2 StateGraph 的设计哲学:状态即一切

LangGraph 在图之上抽象了一个核心概念叫做StateGraph,我理解它就是把“图”和“状态”绑定在一起的一种图表驱动架构。状态(State)是整个图执行过程中共享的数据容器,每个节点执行完后可以把结果写入状态,下一个节点可以从状态里读取需要的数据。

这个设计继承了 Redux 一类前端状态管理库的思想。我刚开始接触的时候觉得有点小题大做,写个 Agent 而已,弄个全局变量不行吗?后来在生产环境里被教育了——Agent 执行过程中会产生大量的中间数据:用户的原始输入、模型的思考过程、工具调用的参数和结果、错误信息、执行历史……如果没有一个结构化、可追踪的状态容器,排起错来会非常痛苦。

StateGraph 的路由判断也完全是基于状态做的。你把模型上一次的输出、工具的返回结果都存进状态,然后写一个路由函数去读这些字段,决定下一步指向哪里。这样一来,整个 Agent 的执行链路是可视化的,任何一步的状态变化都可以记录下来,这对于调试和审计非常友好。

在实际项目中,我建议把 State 的设计放在搭建图结构之前。先想清楚你的 Agent 执行过程中需要哪些状态字段,每个字段是哪个节点写入的、哪个节点读取的,再开始写节点函数。状态是图的“血液”,血液流动的路径没想清楚,图搭得再漂亮也是空的。

2. StateGraph 基础:搭建你的第一个状态图

2.1 定义状态 Schema:给图里流动的数据一个“模具”

在 LangGraph 里定义状态,最常规的方式是继承TypedDict。这里我强烈建议你用typing.TypedDict而不是普通dict,因为 LangGraph 利用类型注解做了很多编译期检查。状态的定义决定了图上节点之间能传什么数据,相当于给数据流定了一个模板。

直接看一个最简单的例子:

from typing import TypedDict, Annotated from langgraph.graph import StateGraph class AgentState(TypedDict): messages: Annotated[list, operator.add] current_step: str

注意这里我用到了Annotated[list, operator.add]。LangGraph 支持为每个状态字段声明一个“归约器”(reducer),用来决定当多个节点往同一个字段写数据时,究竟是覆盖还是合并。operator.add表示增量追加——每个节点往messages里塞的数据会自动追加到已有列表后面,而不是把之前的列表整个覆盖掉。

这个设计真的很重要,我见过太多新手在这里踩坑。如果不指定 reducer,默认行为是后写的覆盖先写的,你辛辛苦苦在节点里往messages里 append 的内容,下一个节点一读取发现只有最后一条消息,怀疑人生。

再强调一个容易忽略的点:状态字段尽量按业务语义去拆分,不要把什么都塞进一个messages。比如你可以把tool_calls、tool_results、final_answer分开存放。这样 Node 的读取逻辑更清晰,条件路由的判断也更精准——你要检查“是否调用了工具”,去读tool_calls字段就好了,不需要从对话历史里翻找。

2.2 节点、边与编译执行:三个基础概念一次讲透

LangGraph 图里的节点就是一个普通的 Python 函数,函数签名是(state) -> dict。输入是当前完整状态,输出是一个字典,字典的 key 对应状态字段名,值是你想更新的内容。LangGraph 会自动将返回值合并进全局状态。

看一个最基础的双节点图:

def node_a(state: AgentState): print("进入节点 A") return {"current_step": "a_done"} def node_b(state: AgentState): print("进入节点 B") return {"current_step": "b_done"} builder = StateGraph(AgentState) builder.add_node("A", node_a) builder.add_node("B", node_b) builder.add_edge("A", "B") builder.set_entry_point("A") builder.set_finish_point("B") graph = builder.compile()

这里add_edge("A", "B")是普通边,表示 A 执行完必然走向 B。set_entry_point指定入口节点,set_finish_point指定终点节点。compile()返回一个可调用的对象,你只需要传入初始状态直接执行:

result = graph.invoke({"messages": [], "current_step": "start"})

这是最基本的流程,大部分教程都会讲到这里。但我想多说一句关于compile()的意义:LangGraph 在编译阶段会做大量的校验工作——检查节点是否真的存在、边的目标节点是否注册过、入口和终点是否合法。等编译通过,说明你图的拓扑结构从语法层面是正确的。这个机制在项目变复杂之后特别有用,再也不用担心图结构写错到了运行时才爆出来。

我知道有些朋友看到StateGraph加add_node加add_edge这种写法,会觉得“这不就是把函数调用包了一层壳吗,直接用 Python 写流程控制不是更简单?”——这个感受很真实,我也经历过。但是当你需要把流程的任意分支、中间状态和节点的执行过程可视化地展示出来,或者在运行中动态插入节点、动态修改路由逻辑时,这种显式的图定义就体现出优势了。图结构本身是可序列化的、可检视的,这在项目协作和运营层面价值很大。

3. 条件路由:让 Agent 自己决定下一步

3.1 add_conditional_edges 的原理:边也可以是“智能”的

条件路由是 LangGraph 区别于普通 workflow 工具的核心能力。普通边是固定的 A 到 B,条件边则是在 A 执行完之后,执行一个路由函数,根据函数返回值来决定接下来进入哪个节点。

看一个典型写法:

from typing import Literal def router_after_node_a(state: AgentState) -> Literal["B", "C"]: if state["current_step"] == "a_done": return "B" return "C" builder.add_conditional_edges( "A", router_after_node_a, { "B": "B", "C": "C" } )

路由函数接收当前完整状态,返回一个字符串,这个字符串是映射字典里的 key,对应到实际的目标节点。这里有个小细节:如果返回的字符串和目标节点名恰好一致,你可以省略映射字典的 value,LangGraph 会直接用返回值作为目标节点名。我第一次写的时候总是映射来映射去,后来发现直接返回节点名就行,简化了不少代码。

我特别不推荐把复杂逻辑写进路由函数里。比如很多人喜欢在路由函数里直接调 LLM 做意图判断,一个 HTTP 调用就花掉几秒钟。LangGraph 的add_conditional_edges第一个参数是源节点名,如果你需要路由函数里做 LLM 调用,请把它包裹成一个独立的节点,让节点的输出作为路由判断的依据,而不是在路由函数本身里做重计算。这样路由函数的执行非常快、几乎是纯内存操作,流程也更清晰——先有专门的节点做“决策”,再有轻量的路由函数做“分发”。

3.2 路由模式的几种典型场景:意图识别、结果校验、分支处理

条件路由的场景比想象的丰富得多,我总结三个最常用的模式,你在做 Agent 时大概率都会用到:

第一种是意图识别路由。用户输入进来,先由分类节点判断这句话属于“查天气”“查日历”还是“闲聊”,然后根据分类结果走不同的业务分支。这个模式对用户体验提升非常明显,可以把“能聊天”和“能办事”的能力解耦。

def intent_classifier(state: AgentState): # 假设这里已经调用 LLM 完成了意图分类 return {"intent": "weather"} def route_by_intent(state: AgentState) -> str: intent = state.get("intent", "chat") mapping = {"weather": "weather_tool_node", "calendar": "calendar_tool_node"} return mapping.get(intent, "chat_node")

第二种是结果校验路由。工具返回结果之后,你需要检查结果是否合法、是否包含关键字段,如果校验通过就进入生成答案的节点,不合格就重新调用工具或者进入人工处理的兜底节点。这种路由在生产环境里尤其重要,因为 LLM 调用工具时的参数合理性其实没有你想象的那么高。

第三种是循环回退路由。当某个节点的执行结果不满足预期,你可以让流程回到之前的任意一个节点重新执行一遍。这在 LangGraph 里写起来非常简单——条件分支的目标节点指向一个前面的节点就行。这种“回退”能力,在传统 Chain 里几乎没有可能优雅地实现。

条件路由本质上是一种显式的、可控制的 Agent 自主决策机制。它和让 LLM 直接自由发挥“下一步干什么”的区别是:路由的分支和去向是开发者预先设计好的,LLM 只能在这个预定义的候选分支里做选择,不能自己凭空创造流程。这在大规模工业化部署中是必要的——你不可能让一个模型来决定整个系统的流程边界,但你可以让它决定每条支路怎么走。

4. Agent 的工具调用循环:从“单次对话”到“自主完成”

4.1 工具调用循环的设计核心:把“调用工具”拆成两个节点

进入最核心的部分了,标题里说的“Agent 的工具调用循环”到底是什么?用一句话概括:通过一个图结构,让模型可以反复地决定调用工具、接收工具结果、再决定是否继续,直到它认为任务完成。

在 LangGraph 里实现这个循环,最核心的设计思路是把“模型调用工具”拆成两个独立节点。第一个节点叫agent_node,负责让模型思考并输出“是否要调用工具、调用哪些工具、参数是什么”;第二个节点叫tools_node(或者叫execute_tools),负责真正执行工具并返回结果。两个节点之间用条件边连接,形成循环。它不是一个节点内部做 while 循环,而是两个节点之间的“往返”在图上构成了环。

为什么要拆成两个节点?因为职责不同:agent_node是纯粹生成决策的,它不关心工具怎么实现;tools_node是纯粹执行的,它不关心模型下一步怎么想。这两个职责如果混在一个函数里,循环的终止条件和错误边界会非常难写。拆开之后,你可以单独给tools_node加超时、加重试、加记录日志,完全不影响模型决策。

下面是一个极简但完整的实现思路:

from langchain_openai import ChatOpenAI from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市的天气""" return f"{city} 今天是晴天" llm = ChatOpenAI(model="gpt-4o") llm_with_tools = llm.bind_tools([get_weather]) 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] tool_outputs = [] for tool_call in last_message.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] result = get_weather.invoke(tool_args) tool_outputs.append( {"role": "tool", "tool_call_id": tool_call["id"], "content": result} ) return {"messages": tool_outputs}

4.2 用条件边把两个节点“串成环”并设置退出条件

节点都写好了,接下来最关键的事情是把它们连成一个支持循环的图,并设置循环的终止条件。

def should_continue(state: AgentState) -> Literal["continue", "end"]: last_message = state["messages"][-1] if last_message.tool_calls: return "continue" return "end" builder = StateGraph(AgentState) builder.add_node("agent", agent_node) builder.add_node("tools", tools_node) builder.add_edge("tools", "agent") builder.add_conditional_edges( "agent", should_continue, {"continue": "tools", "end": "__end__"} ) builder.set_entry_point("agent") graph = builder.compile()

这个should_continue函数就是整个循环的“刹车阀”。它检查模型最后一轮的输出里是否包含工具调用请求,如果有就走到tools节点去执行工具;如果没有,说明模型认为答案已经完整,就走到特殊节点__end__,流程结束。

这里有一个我踩过很深的坑,必须提醒你:tools节点执行完工具返回结果之后,新的工具结果消息会被追加到messages状态里。此时agent节点再次运行时,它读取的是包含了工具结果的最新完整消息历史,模型基于这些结果进行下一轮推理。看起来是理所当然的,但如果你在agent_node里不小心只把最后一条用户消息传给模型,而不是把完整历史传过去,工具结果就“丢”了,模型会陷入一种“我说要调用工具、但看不到工具结果、于是又调用一次”的循环死锁。

我实际处理过的一个生产事故就是这么发生的:排查了半天,最后发现是团队里有人为了省 token 在agent_node里做了消息裁剪,把系统认为不重要的历史消息砍掉了,结果偏偏把 tool 消息误伤掉了。加了消息裁剪的白名单之后,问题才解决。所以提醒所有同学:在你对 LangGraph 和模型行为没有完全把握之前,不要在循环里剪裁消息。

还有一个实践要点是__end__这个特殊节点。它是 LangGraph 内置的终止标记,你不需要显示地定义它,直接在条件映射里指向"__end__"就行。很多新手不知道还有这么个东西,总想自己定义一个返回最终答案的节点,结果逻辑变得很绕。最终答案可以在agent节点的输出里直接给,只要模型不再请求调用工具,循环就自然走到__end__结束。

4.3 防死循环的三重保险:最大步数、超时、人工审核

工具调用循环一个比较头疼的问题是死循环——模型反复调用工具、工具反复返回结果、但结果始终不满足要求,或者模型就是停在某个状态里反复横跳。在真实项目里,这种问题非常频繁,我见过模型为了确认一个订单状态疯狂翻页查询二十多次的“事故现场”。

因此给循环上锁是必须的。LangGraph 官方提供了recursion_limit,它限制一次图调用里最多执行几个节点。超过限制会抛出异常,至少能让你在日志里看到“这轮执行异常地长”,而不是整个流程无声地卡住。

config = {"recursion_limit": 25} result = graph.invoke({"messages": initial_messages}, config=config)

除了recursion_limit,我自己还会在业务层面加两个保险。第一个是最大工具调用次数统计:在状态里加一个tool_attempts字段,每次tools_node执行完就加一,路由函数里如果检测到次数超过阈值(比如 5 次),就强制走一条“放弃工具,直接生成一个兜底答案”的路径。第二个是结果条件校验:如果工具返回的内容一直不满足业务校验规则(比如格式不对、关键字段为空),也应该提前退出循环进入人工处理流程,不能让它无限重试。

这些防护措施本质上是在回答一个问题:Agent 什么时候该“坚持”?什么时候该“认输”?把这个问题在路由设计阶段就想清楚,远比运行时报错了再救火要省心得多。我认为做 Agent 的核心理念是“给模型自由,但给流程边界”。LangGraph 的分支、条件路由、循环终止,就是约束模型自由度的护栏。

5. 踩坑记录与排查技巧

5.1 状态更新语义不一致:为什么有的字段是覆盖、有的是追加

我在前面提到过 reducer,但这里值得再深入讲一下,因为实际踩过的坑太典型了。举个例子,你在两个节点里都往同一个字段current_step写不同的值,如果这个字段没定义 reducer,那谁最后执行谁就覆盖前面的;但如果你用的是operator.add,那这个字段会被拼成一个列表。两种行为的语义完全不同,你如果用错,写路由函数的时候逻辑就是错的。

在真实的 Agent 项目里,messages字段用operator.add是最常见的,因为它天然是累积的。但是如果你把工具调用的参数也放进一个“累积”型字段里,那每一轮调用都会把上一次的参数追加在后面,推理和排查的成本会迅速上升。我的经验是:累计型字段宁可少用,也不要用多。只有真正需要全量历史的地方才用operator.add,其余字段尽量用覆盖型,让状态在任何时刻都保持简洁明确。

5.2 工具调用失败后的降级处理

工具调用失败是 Agent 场景里概率最高的错误之一,而且失败原因五花八门:工具超时、API 返回 500、参数校验不过、LLM 生成了不存在的工具名、工具返回了模型看不懂的格式……我在生产环境中见过所有你能想象到的失败路径。

一个比较实用的策略是,在tools_node里捕获所有异常,并把异常信息以一条tool角色的消息写回状态,而不是让异常直接打断图的执行。这样模型在下一轮能看到“你刚才调用工具失败了,原因是 xxx”,它可以自己决定是换个参数重试、还是换一个工具、还是直接放弃。

def tools_node(state: AgentState): last_message = state["messages"][-1] tool_outputs = [] for tool_call in last_message.tool_calls: try: result = execute_tool(tool_call) content = str(result) except Exception as e: content = f"工具调用失败: {type(e).__name__}: {e}" tool_outputs.append( {"role": "tool", "tool_call_id": tool_call["id"], "content": content} ) return {"messages": tool_outputs}

这个思路的好处在于,错误信息本身成为模型上下文的组成部分,模型可以“看见”错误并自行调整策略。当你把工具错误信息做成结构化、简洁、可提示的内容时,模型自己会学会归纳和判断,远比你在代码里做各种硬编码兜底要灵活。

5.3 调试工具链与方法:图的可视化与状态追踪

LangGraph 提供了一些很实用的调试手段。第一个建议是你把编译后的图输出成 ASCII 或者 Mermaid 图看看,眼过一遍整体结构:

print(graph.get_graph().draw_ascii())

注意:虽然 LangGraph 本身有 Mermaid 渲染功能,但在我这个版本里更推荐直接用draw_ascii()快速看结构,或者用在线可视化工具检查节点和边的关系。每次改动图结构后都输出一遍,确认边的走向跟你预期一致,尤其是条件边的分支是否都正确指向了目标节点。

第二个调试技巧是使用回调机制记录每个节点的输入输出。LangGraph 支持在编译时传入自定义的listener回调,或者简单粗暴地在每个节点函数开头打日志:

def agent_node(state: AgentState): print(f"[agent] messages数量={len(state['messages'])}") ...

在生产环境里,我会把每次节点执行的摘要(节点名、耗时、状态关键字段的变化)记录到结构化日志里,这样一次 Agent 执行就是一条完整的时间线,排查问题时把这条时间线翻出来,比对着代码猜要高效太多。

6. 我的实战经验与扩展建议

最后分享一些我在实际业务中积累的经验。LangGraph 最好的应用场景是那些流程比较复杂、分支比较多、对可观测性有要求的 Agent 任务——比如多步骤的办公助手、客服工单自动处理、数据分析 Agent。

在扩展方向上,有两个最值得关注的点。第一个是多 Agent 协作:LangGraph 底层支持在一个图里定义节点关系,因此你可以把多个 Agent 各自封装成一个子图作为大图的节点,让它们在同一个状态空间里协作。这个能力在处理复杂任务拆分时非常有价值,但一定要控制好 Agent 之间的通信边界,否则状态会变得极度混乱。第二个是持久化与 Checkpoint:LangGraph 提供了基于BaseStore的持久化机制,可以让 Agent 的状态在多次调用之间保持,这对于需要续聊和人工介入的长流程应用是个好特征的支撑。

我现在自己写 Agent 项目的习惯是:先用 LangGraph 快速搭建可运行的图,然后逐步把路由分支细化,加上循环防护、错误降级、结构化日志,最后再做持久化和人工审核。这个顺序比较稳妥,一上来就追求完美的架构反而容易过早陷入细节。

如果你是刚入门,我的建议是不要一上来就套复杂的框架,先用这个最简单的工具调用循环跑通一个真实场景——比如“查天气”“算费用”这种小工具——再慢慢往上加路由分支和多轮状态。亲手把一个循环跑通、看到模型调用工具、拿到结果、生成最终答案,你对 LangGraph 的理解会发生实质性的变化。

补充一个很容易被忽略的细节:官方文档和社区里给出的示例往往会省略python-dotenv配置环境变量、模型 API 连接超时处理等基础设施问题。你在本地跑示例时,建议先把这些基础环境问题解决掉,再开始研究图逻辑,否则很容易被一个 API 报错打断思路。

写到这里,我想起自己第一次跑通 Agent 工具循环时那种“通了”的感觉。LangGraph 的曲线确实需要一点耐心去适应,但一旦上手,你会对这种显式、可控的编排方式产生依赖。希望这篇文章能帮你也体会到这种流畅感。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询