1. 为什么我转向了 LangGraph
做 Agent 开发的朋友应该都有一个感受:从零手写 Agent 循环,代码写到最后总是一团乱麻。最早我用 LangChain 的AgentExecutor,跑个简单的工具调用没问题,一旦涉及多步骤、条件分支、人工介入,或者想在中间插入记忆清理逻辑,改起来就非常痛苦。后来接触到 LangGraph,才意识到问题不在于 Agent 这个概念多难,而在于我们缺一个能把"状态流转"说清楚的东西。
LangGraph 是 LangChain 团队推出的一个低层编排框架,核心解决一个痛点:把 Agent 的每一步决策建模成一张图。图里有节点(Node),有边(Edge),有状态(State)。节点代表一次计算,边代表流转规则,状态则是整个流程的共享记忆。和 LangChain 的链式调用不同,LangGraph 支持循环,这意味着我们可以自然实现"LLM 决定调用哪个工具、调用完工具、把结果给 LLM 再看一次"这种 ReAct 循环,而不用靠什么while True之类的循环去硬凑。
这篇内容我主要围绕三个核心概念展开:StateGraph 的建模方式、条件路由的实现思路、以及 Agent 工具调用循环的完整落地。适合刚接触 LangGraph 的小白,也适合想从 LangChain 迁移过来的开发者。代码基于 Python 3.10+ 和 langgraph 0.2.x 版本,安装方式在下面会提到。
2. 先搞懂 LangGraph 的核心抽象
2.1 StateGraph 到底是什么
LangGraph 里的StateGraph本质上是一个状态机。你定义一张图,图里的节点会对状态做修改,边决定了状态如何在节点之间流转。状态是一个共享的对象(通常是 TypedDict 或 Pydantic 模型),所有节点都能读取和更新它。
我用一个生活化类比帮助理解:把整个 Agent 执行流程想象成一条流水线,State 就像工件上的托盘,每个工位(节点)读完托盘上的信息,往上面放点东西(修改状态),再传给下一个工位。关键是这个托盘只有一个,而且是全局共享的。LangGraph 里并没有"局部变量"这种说法,数据要么在 State 里,要么在节点函数的返回值里,后者也会自动合并进 State。
一个最基础的 StateGraph 定义可以是这样的:
from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END class AgentState(TypedDict): messages: list next_action: str def node_a(state: AgentState) -> dict: # 简单示例:把一条消息追加到消息列表 return {"messages": ["node_a 处理完成"]} def node_b(state: AgentState) -> dict: return {"messages": ["node_b 处理完成"]} graph = StateGraph(AgentState) graph.add_node("a", node_a) graph.add_node("b", node_b) graph.set_entry_point("a") graph.add_edge("a", "b") graph.add_edge("b", END) app = graph.compile() result = app.invoke({"messages": []}) print(result["messages"])这个例子里 State 是AgentState,两个节点函数各自返回一个新的字典片段,LangGraph 自动做合并。注意我用了messages: list,这里类型标注要配合 reducer 才追加,否则默认是覆盖。这个知识点很多人踩坑,后文专门讲。
2.2 LangChain 和 LangGraph 的区别
很多朋友搞不清 LangChain 和 LangGraph 的边界。简单说,LangChain 是一套给 LLM 应用提供组件的库——模型封装、Prompt 模板、输出解析、文档加载、向量存储、工具抽象都在这一层。它可以快速搭一个链式调用。但它的问题在于流程是线性定义的,分支和循环不够自然。
LangGraph 是 LangChain 生态里的编排层,它不关心你怎么调用模型、怎么解析输出,它只关心一件事:你的应用状态是怎么流动的。你可以把 LangChain 的工具、模型、解析器装进 LangGraph 的节点里,也可以完全不依赖 LangChain,直接用 LangGraph 做底层编排,只用 OpenAI SDK 或任何自定义代码。
所以从学习路径上看,LangGraph 并不要求你先把 LangChain 学全。你只需要了解 model、tool、prompt 这些基本组件就够了。实际上,LangGraph 的底层图执行引擎是独立的,你可以脱离 LangChain 单独跑一张简单图。理解了这一点,就不会再纠结"学哪个先"这种问题。
2.3 节点函数和边的关系
图里的节点函数有三个重要特征:
- 接收完整的 State 作为参数
- 返回一个字典,字典的 key 对应 State 里的字段,LangGraph 会根据 reducer 合并到原 State
- 不依赖全局变量做状态传递,每一次 invoke 都是独立运行
边则有普通边和条件边之分。普通边像"a 执行完直接去 b",条件是边则是一个路由函数,它接收 State,返回一个字符串(下一跳节点的名字)。条件边是实现条件路由的核心,后面单独展开。
3. 状态设计:这一步做不好后面全白搭
3.1 用 TypedDict 还是 Pydantic
Agent 的状态设计是整个图好不好维护的关键。刚开始用 LangGraph 时,很多人直接在 State 里塞了一堆杂七杂八的字段,后来发现调试时根本看不懂。我建议遵循几个原则:字段越少越好、字段职责单一、消息列表单独管理。
定义状态有两种方式:TypedDict和 Pydantic 模型。TypedDict 轻量,写起来快,适合状态结构简单的场景;Pydantic 支持字段校验和默认值,适合复杂项目。两者在节点函数里用起来差别不大,但如果你在节点里需要用到实体识别或其他验证逻辑,Pydantic 优势明显。
我常用的一个状态设计模板:
from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] current_tool: str tool_result: str finish: bool这里messages用了Annotated[list, add_messages],意思是消息字段不再是简单覆盖,而是通过add_messages这个 reducer 做追加。这是 LangGraph 官方定义消息列表的推荐方式,省得自己在每个节点里手动拼接 list。
3.2 Annotated 和 reducer 机制
Reducer 是 LangGraph 状态管理里最需要花时间理解的概念。默认情况下,节点返回的字段会覆盖 State 里已有的值。但只要字段类型写成了Annotated[list, add_messages],LangGraph 就会在合并时调用add_messages这个函数,而add_messages的逻辑是:新消息追加到旧消息列表尾部。这样你在每个节点里只需要返回追加的那一条消息,LangGraph 自动帮你维护完整消息历史。
自定义 reducer 也非常简单:
def merge_list(left: list, right: list) -> list: return left + right class CustomState(TypedDict): history: Annotated[list, merge_list]这个特性和 LangChain 的AgentExecutor有本质区别,LangChain 里你处理消息历史大部分时候是自己写逻辑,LangGraph 则把这件事变成了框架级的默认行为。
3.3 状态初始化与 Model 注入
状态设计完了,实际执行时还需要初始状态。app.invoke()传的字典就是初始状态。你可以在初始状态里注入用户输入、模型配置、工具列表等。有个小技巧:模型实例(比如ChatOpenAI)可以放在状态外面作为闭包变量,不需要塞进 State,因为 State 是要序列化的,塞一个模型对象进去容易出问题。
from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini", temperature=0) app = graph.compile() result = app.invoke({ "messages": [{"role": "user", "content": "帮我查一下明天的天气"}], })刚接触时容易犯的错是试图把 model 写进 State 里,一旦状态里有个不可序列化的对象,后面调试工具和断点重启都会变得非常棘手。保持 State 只放必要的数据,模型用闭包传递,这是经验之谈。
4. 条件路由的三种经典写法
4.1 什么是条件边
条件路由这个词听起来高大上,实际就是一个函数根据当前 State 决定走哪条边。你在add_condition_edges里给某个节点传入一个路由函数,函数返回值是下一个节点的名字。这个能力撑起了 Agent 的整个判断逻辑:LLM 说"要继续"就走循环边,说"结束了"就通往结束节点。
一个典型场景:LLM 调完工具后,需要判断是否还有下一步。如果工具结果里有"仍需继续"的标志,就走回 LLM 节点;否则走 finish 节点。这个判断函数拿到状态里的tool_result做分析,返回对应节点名。
def route_after_tool(state: AgentState) -> str: if state.get("finish"): return "end" return "agent"4.2 基于 LLM 决策的路由
更复杂一点的路由是让 LLM 做决策。比如有两类工具,一类用于查询数据库,一类用于调用外部 API,你可以让 LLM 输出一个 JSON 格式的意图,路由函数解析这个 JSON 决定下一跳。这本质上是在图内部实现了一个轻量级意图识别层。
需要注意的是,路由函数里尽量避免再调用模型或者做重量级 IO,因为它会执行得非常频繁。保持路由函数轻量,只做规则判断和字符串匹配,复杂计算放到节点函数里执行。
4.3 条件边的多分支处理
LangGraph 支持一个节点出度多个分支,每个分支是一个条件判断。举例:一个"意图分发"节点,有三个下游节点:"天气查询"、"时间查询"、"闲聊"。路由函数可以返回这三个节点名之一。这样一张图里就可以承载多种能力的 Agent 路由。
def classify_intent(state: AgentState) -> str: # 实际业务这里可以调分类模型 if "天气" in state["messages"][-1]["content"]: return "weather_node" return "chat_node" graph.add_conditional_edges( "intent_router", classify_intent, { "weather_node": "weather_node", "time_node": "time_node", "chat_node": "chat_node", } )这里第三参数是可选的映射字典。默认情况下路由函数返回的字符串直接就是节点名,不需要映射。但有的场景你要把语义化的分支名映射成节点名,这个参数就很方便。
5. Agent 工具调用循环:核心机制完全拆解
5.1 ReAct 循环到底在循环什么
Agent 工具调用循环,本质上是 ReAct(Reason + Act)模式的工程化实现。展开来说就是四步循环:
- 把当前的对话历史和工具描述喂给 LLM
- LLM 决定是否调用工具,如果调用,输出结构化指令(比如 function calling 格式)
- 程序执行对应工具,把工具结果作为一条新的消息追加到消息列表
- 回到第一步,让 LLM 看到工具结果后再决定下一步
这样一个循环,在 LangGraph 里就是两个节点之间互相指边:一个节点调用 LLM,一个节点执行工具。条件路由保证"需要工具时走工具节点,不需要工具时走结束节点"。如果拿流程图画出来,就是一个经典的带环流程图。
5.2 ToolNode 和工具的绑定方式
LangGraph 提供了内置的ToolNode,帮你省去手动执行工具和格式化结果的步骤。用法是:
from langgraph.prebuilt import ToolNode from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市当前的天气情况""" return f"{city} 今天的天气是晴朗,气温25摄氏度" tools = [get_weather] tool_node = ToolNode(tools) graph.add_node("agent", agent_node) graph.add_node("tools", tool_node) graph.add_conditional_edges("agent", should_continue, ["tools", END]) graph.add_edge("tools", "agent")ToolNode会自动读取 LLM 输出里的 tool_calls 指令,执行对应的 Python 函数,并把结果格式化成 ToolMessage 追加到消息列表。整个节点本身是预置的,不需要自己写工具分发逻辑。
有两个注意点:
- 工具函数必须写好 docstring,因为 LLM 靠这个了解工具用途
- 工具函数签名里参数名要带类型注解,LLM 靠这个生成正确入参
5.3 手写 Agent 节点和工具循环的条件判断
如果不想用太高层的封装,你完全可以自己写 Agent 节点。一个标准的 agent 节点大概长这样:
import json from langchain_core.messages import AIMessage def agent_node(state: AgentState) -> dict: response = model.invoke(state["messages"]) # 让模型看完整消息历史 return {"messages": [response]}注意这里直接返回了AIMessage对象,LangGraph 的add_messages会正确处理各种消息类型(HumanMessage、AIMessage、ToolMessage)。如果模型返回的AIMessage里有tool_calls字段,那说明模型希望调用工具,这时候条件路由应该走工具节点。
路由函数的判断逻辑很直接:
def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] # 如果 LLM 输出包含 tool_calls 字段,就继续调用工具 if hasattr(last_message, "tool_calls") and len(last_message.tool_calls) > 0: return "tools" return END很多教程里用after_model这个函数名,本质就是干这件事。
5.4 如何正确理解"工具调用循环"的结束条件
工具循环最怕的是无限循环。模型一直想调用工具,工具一直返回结果,状态永远走不到 END。LangGraph 没有内置的循环上限,所以必须自己做保护。办法有几种:
- 在 State 里加一个
turn_count字段,每次经过 agent 节点时加 1,在路由函数里判断超过阈值就强制结束 - 在工具节点内部捕获异常,如果工具执行失败就返回一个特殊消息,让模型基于这条消息决定是重试还是结束
- 在调用
app.invoke()时传入recursion_limit参数
result = app.invoke( {"messages": [{"role": "user", "content": "你好"}]}, config={"recursion_limit": 20} )recursion_limit是图执行的全局步数上限,超了会抛异常,可以当作兜底安全网。
6. 完整示例:一个带天气查询的问答 Agent
6.1 场景定义和工具准备
前面讲了很多理论,这里用一个完整代码把所有知识串起来。我们的目标是做一个简单的问答 Agent,它能从对话中识别出"查天气"这个意图,调用对应的工具,然后把结果返回给用户。
先准备工具:
from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市的实时天气信息。""" weather_map = { "北京": "晴,25°C,微风", "上海": "小雨,22°C,东南风3级", "广州": "雷阵雨,28°C,湿度80%", } return weather_map.get(city, f"暂未收录 {city} 的天气数据")这个工具故意做成字典查询,避免引入外部 API,方便大家直接复现。重点是看 LangGraph 怎么圈住一个真实场景并形成闭环。
6.2 构建带工具调用的 StateGraph
接下来把模型、图、路由串起来:
from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode from langchain_openai import ChatOpenAI from langchain_core.messages import AIMessage class AgentState(TypedDict): messages: Annotated[list, add_messages] model = ChatOpenAI(model="gpt-4o-mini", temperature=0) tools = [get_weather] model = model.bind_tools(tools) def agent_node(state: AgentState) -> dict: response = model.invoke(state["messages"]) return {"messages": [response]} def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and len(last_message.tool_calls) > 0: return "tools" return END graph = StateGraph(AgentState) graph.add_node("agent", agent_node) graph.add_node("tools", ToolNode(tools)) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue, ["tools", END]) graph.add_edge("tools", "agent") app = graph.compile()这一步是最核心的代码,只有十几个节点和三条边,但已经实现了一个完整的 ReAct Agent。
6.3 运行效果演示与分析
执行一次查询:
result = app.invoke({ "messages": [{"role": "user", "content": "北京今天天气怎么样?"}] }) print(result["messages"][-1].content)执行路径是这样的:入口是 agent 节点,LLM 看到用户的天气问题,输出一个带tool_calls的AIMessage,里面指明调用get_weather,参数是{"city": "北京"}。条件路由函数判断出有 tool_calls,走 tools 节点;ToolNode 执行工具,把"北京 今天的天气是晴朗,气温25摄氏度"封装成 ToolMessage 追加到消息列表。接着边把流转回 agent 节点,LLM 看到工具结果,生成面向用户的自然语言回答,这次不再有 tool_calls,路由函数放行到 END。
整个循环在两轮内就结束了。如果你手动打印每一步的状态,会看到消息列表从最开始的一问,逐步累计到一问、一 AI 调用指令、一个工具结果、最终回答这几条。
这就是 LangGraph 工具调用循环的完整机制。不用写任何循环代码,全靠图结构和条件边表达。
6.4 如果想加上限和容错
加上限的做法:
from typing import Annotated, TypedDict class AgentState(TypedDict): messages: Annotated[list, add_messages] turn_count: int def agent_node(state: AgentState) -> dict: response = model.invoke(state["messages"]) return {"messages": [response], "turn_count": state.get("turn_count", 0) + 1} def should_continue(state: AgentState) -> str: if state.get("turn_count", 0) >= 5: return END last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and len(last_message.tool_calls) > 0: return "tools" return END这里的turn_count字段是无 reducer 的普通字段,每次返回都会覆盖自身,所以拿来做计数非常方便。有了这个保护,就算模型抽风不断尝试调用工具,五轮之后也会强制结束,防止烧钱和死循环。
工具容错方面,可以在工具函数内部做 try-except,返回给人看的错误提示。因为工具返回的错误也会变成 ToolMessage 送回到模型手里,模型通常会根据错误信息修正调用方式,或者向用户解释失败原因。
7. 实操中的坑与排查清单
7.1 消息列表不追加、被覆盖的问题
很多人第一次用 LangGraph 都遇到过"为什么节点跑完 messages 只剩一条了"这种诡异现象。原因非常直接:State 字段没有声明 reducer。比如messages: list这样写,节点函数 return{"messages": [xxx]},LangGraph 会直接拿这个新 list 覆盖旧 list。解决办法就一个:改成messages: Annotated[list, add_messages]。
这个过程确实反直觉,因为 LangChain 的经验是"返回什么就替换什么"。到了 LangGraph,你要主动告诉她"这个字段需要累加"。理解了 reducer 就理解了这个框架的一半。
7.2 tool_calls 没有值,工具节点不执行
还有一种常见问题:Agent 节点返回的 AIMessage 里没有tool_calls,明明模型能理解工具描述,但就是不调用。排查方向有这几个:
- 确认
model.bind_tools(tools)已经执行,如果模型没有绑定工具,LLM 根本不知道有工具可用,自然不会有 tool_calls - 确认工具函数有完整的 docstring,且参数名本身语义化强,模糊的描述容易让 LLM 误判
- 检查工具名和参数描述是否和系统提示冲突,比如期望模型调用查天气的工具,却在 system prompt 里强调"直接回答",模型就会放弃调用
7.3 工具执行报错导致整个图崩溃
工具执行时抛出异常,默认会中断整个图的执行。如果你希望工具报错后 Agent 还能继续处理,有两种方案:给工具函数内部加 try-except,或者用 LangGraph 的异常处理配置。我一般选择前者,因为后者会丢弃 ToolMessage 的上下文。工具报错时的返回信息,本身也是模型推理的重要输入。
7.4 调试工具的使用
LangGraph 提供了丰富的调试能力。app.invoke(..., config={"debug": True})能打印每一步的节点执行、状态变换信息,这是排查"路由走了哪条边"最强力的工具。如果状态里塞了太多无关数据,调试输出会很难看,这也是我反复强调"State 字段要精简"的原因。
另外新版 langgraph 支持 LangSmith 集成,但本地开发先用 debug 模式就足够日常排查了。
7.5 图对象重用与并发安全
app.invoke()每次调用都是独立运行,不用担心状态污染。但如果你想在同一张图里并发跑多个不同会话,不要共用一个Resource里锁,直接并发调用invoke就行,LangGraph 会为每次调用创建独立的 state 实例。这一点比很多自己实现的循环逻辑要安全得多。
8. 从入门到进阶的后续学习路线
8.1 必读文档和官方示例
如果你看完这篇文章觉得 LangGraph 的理念是对的,下一步我建议直接啃官方文档的这几个页面:StateGraph 使用指南、条件边说明、预置 ToolNode 的用法,以及 langgraph-checkpoint 的介绍。官方示例仓库里有很多完整案例,比如多 Agent 协作、带人工审批的流程、带记忆的历史对话管理,每一个都能打开新的思路。
8.2 知识图谱式思考:把 LangGraph 当骨架
LangGraph 最值得借鉴的不只是 API,而是这种"状态图"的思维方式。以前做 Agent 时,脑子里全是一堆顺序执行的步骤;用 LangGraph 之后,我会先画数据流图,再想清楚每个节点的职责边界、状态字段的最小集合、路由函数要读取哪些信息。框架是死的,这种设计思路才是可以迁移到任何语言的。
8.3 和其他 Agent 框架的对比视角
市面上还有 AutoGPT、MetaGPT、CrewAI 等框架,各有侧重。LangGraph 的优势是灵活度和可定制性,它是"积木式"的,适合需要精确控制流程的复杂应用。CrewAI 更适合快速构建角色扮演式多 Agent,AutoGPT 则更适合自动化任务探索。但从工程角度看,LangGraph 的 StateGraph 设计对复杂应用的可测试性和可观测性是最好的。
8.4 实际项目中的应用建议
在真实项目里,我的建议是先在小场景验证可行性,再逐步加复杂度。先做单 Agent 单个工具调用,跑通以后再加条件分支,再加多工具分发,最后再考虑多 Agent 协作。如果一上来就想着做一个完美的多 Agent 系统,往往会被调试地狱劝退。
还要记住一点:LangGraph 不限制你的模型来源,OpenAI、Claude、通义千问、DeepSeek 只要支持 function calling 或 tool use,都能在这个框架里跑起来。这算是它又一个很实用的特性——不被某个模型厂商绑定。
最后分享一点实际体会
我用 LangGraph 写了几个生产级 Agent 之后,最大的感受是:图结构让"可解释性"变得非常具体。哪里出问题了,看一眼路由函数的判断条件和消息列表的最后几条就能定位,不需要像以前那样在循环里打十几行日志去猜状态。工具调用循环的本质不是让 LLM 更聪明,而是让整个执行过程变得可控、可观察、可干预。这个核心价值,用其他编排方案很难取代。
给新手的最终建议:先别急着上多 Agent,也不用把 LangChain 全家桶都学会。拿一个手头简单的工具调用需求,用 StateGraph 重写一遍,状态设计、条件路由、循环控制都会在这个过程中自然掌握。踩过几个坑之后,你对 Agent 的理解会比看十篇教程都深刻。