1. 为什么手写 Agent 循环迟早会让你崩溃
如果你已经用 LangChain 或原生 API 手写过一版 Agent 循环,大概率经历过这样的场景:一开始代码只有几十行,一个while循环里塞着模型调用、工具解析、结果回填,跑得挺顺。等到要加多轮记忆、加人工确认、加条件分支、加并行工具调用的时候,那个循环就变成了一坨谁都不敢动的意大利面。我自己最早写的一个客服 Agent,从 80 行长到 600 行只用了两周,最后连我自己都要画图才能理清哪条路径会走到哪个if。
这就是"篇三"要聊的核心问题:把 Agent 循环的控制权,从你手里交给 LangGraph。关键词里出现的 LangGraph、StateGraph、create_agent,本质上都是在回答同一个问题——当 Agent 的流程不再是"调模型→执行工具→再调模型"这种线性结构,而是带环、带分支、带状态持久化的图结构时,你该怎么组织代码。
先说清楚这篇适合谁看。如果你完全没接触过 Agent,建议先补一下 LangChain 的基础调用和工具(Tool)定义,否则直接上 LangGraph 会有点懵。如果你已经手写过循环、被状态管理折磨过,或者正在纠结"LangGraph 和 LangChain 到底啥区别""create_agent 和 StateGraph 该用哪个",那这篇就是写给你的。我会从手写循环的真实痛点讲起,拆解 LangGraph 的图模型到底解决了什么,再给出从零迁移的完整实操,最后聊聊那些文档里不会写的坑。
需要提前说明的是,LangGraph 的 API 迭代很快,本文基于当前主流版本的写法,核心概念(节点、边、状态、检查点)是稳定的,但具体函数签名请以你安装的版本为准。我踩过的坑里有一半是因为照着半年前的教程写,结果 API 已经变了。
2. 手写循环的三个致命伤,以及 LangGraph 的图模型怎么治
2.1 状态散落在闭包和全局变量里,根本没法调试
手写循环最典型的样子是这样的:你在函数里维护一个messages列表,工具调用的结果 append 进去,模型的回复也 append 进去,中间可能还有几个临时变量记录"当前走到第几步""是否已经调用过某个工具"。这些状态要么是局部变量,要么挂在闭包上,要么干脆用全局变量。
问题在于,当流程变复杂,你根本不知道某一轮循环进来的时候,状态到底是什么。想加个断点看看?可以,但你得在几十个地方打断点。想复现一个线上 bug?抱歉,状态没持久化,复现不了。我遇到过一个诡异的问题:Agent 偶尔会重复调用同一个工具三次,排查了半天才发现是某次工具返回的格式和预期不一致,导致解析失败后走了重试分支,而重试分支又没清理临时状态。
LangGraph 的第一个核心设计就是把状态显式化。它用StateGraph定义一个状态结构(通常是一个 TypedDict 或 Pydantic 模型),所有节点读写这个共享状态,节点之间通过边连接。状态不再是藏在闭包里的黑盒,而是一个你能打印、能序列化、能持久化的对象。
from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] step_count: int pending_tool: str | None这里add_messages是个 reducer,告诉 LangGraph 当多个节点都想往messages里写东西时,用追加而不是覆盖的方式合并。这个细节很关键,手写循环里你手动 append,图模型里你得声明合并规则,否则后写的节点会把前面的消息冲掉。我第一次用的时候没加Annotated,结果工具返回的消息全丢了,debug 了半小时才反应过来。
2.2 循环和分支用 if-else 硬编码,改一条路径要动全身
手写循环里,控制流就是while加一堆if。想加一个"如果模型连续两次调用同一个工具就强制结束"的逻辑?你得在循环里插一段判断。想加一个"人工确认后再执行敏感工具"的环节?你得在工具执行前插一个阻塞等待。每加一个需求,循环体就膨胀一圈,而且这些逻辑互相耦合,改 A 可能影响 B。
LangGraph 把控制流抽象成节点(Node)和边(Edge)。节点是一个函数,接收状态、返回状态更新;边定义节点之间的跳转关系。普通边是固定的 A→B,条件边(conditional edge)则根据状态决定下一步走哪。
from langgraph.graph import StateGraph, START, END def call_model(state: AgentState): response = model.invoke(state["messages"]) return {"messages": [response], "step_count": state["step_count"] + 1} def should_continue(state: AgentState): last = state["messages"][-1] if last.tool_calls: return "tools" return END 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, {"tools": "tools", END: END}) builder.add_edge("tools", "agent") graph = builder.compile()这段代码画出来的图是:START → agent →(有工具调用就去 tools,否则结束)→ tools → agent,形成一个环。这个环就是 Agent 的核心循环,但它不再是藏在while里的隐式结构,而是图上一眼能看出来的显式结构。想加人工确认?在 tools 前面插一个节点,加一条条件边就行,不用动 agent 节点本身。
2.3 没有持久化,中断恢复和多轮对话全靠自己造轮子
手写循环的第三个坑是持久化。多轮对话你要自己存 messages,中断恢复你要自己序列化整个状态,想做个"从某一步重新执行"的功能几乎不可能。我见过有团队为了支持"用户中途关掉页面再回来继续",硬生生写了一套状态快照机制,代码量比 Agent 本身还大。
LangGraph 内置了Checkpointer(检查点)机制。编译图的时候传入一个 checkpointer,每次节点执行后状态都会自动保存。配合thread_id,你可以实现多轮对话的记忆、中断后恢复、甚至时间旅行(回到历史某个检查点重新执行)。
from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() graph = builder.compile(checkpointer=memory) config = {"configurable": {"thread_id": "user-123"}} graph.invoke({"messages": [("user", "帮我查下订单")]}, config) # 下次同一个 thread_id 调用,会自动带上历史状态 graph.invoke({"messages": [("user", "那再帮我退了吧")]}, config)生产环境一般用SqliteSaver或 Postgres 的 checkpointer,MemorySaver 只适合本地调试。这里有个容易忽略的点:checkpointer 保存的是整个状态,如果你的状态里塞了大对象(比如原始文件内容),存储会迅速膨胀。我建议状态里只放必要的消息和元数据,大内容存外部再引用。
3. StateGraph 和 create_agent 到底该选哪个
3.1 create_agent 是"开箱即用",但它的边界你得清楚
LangGraph 提供了create_agent(早期叫create_react_agent)这个高阶 API,一行代码就能得到一个标准的 ReAct 风格 Agent。
from langgraph.prebuilt import create_agent agent = create_agent(model, tools=[search, calculator]) result = agent.invoke({"messages": [("user", "北京今天天气怎么样")]})它内部帮你搭好了"模型→工具→模型"的循环图,你不需要自己定义状态、节点、边。对于标准的"给模型一堆工具让它自己决定调用"的场景,这玩意儿确实省事。关键词里create_agent热度高,就是因为很多人想快速跑通一个 Agent。
但create_agent的边界也很明显:它的状态结构是固定的(主要是 messages),控制流是固定的(ReAct 循环)。你想在中间插一个人工审核节点?想根据工具类型走不同的分支?想加一个"先规划再执行"的两阶段流程?这些它都不直接支持,你得退回到StateGraph自己搭。
我的经验是:原型阶段用 create_agent 快速验证,一旦流程有定制需求就立刻切到 StateGraph。不要试图在 create_agent 外面包一层来 hack 它的行为,那样写出来的代码比直接用 StateGraph 还难维护。
3.2 StateGraph 的灵活性来自"状态即契约"
StateGraph的核心思想是:状态是节点之间的契约。每个节点声明它读什么、写什么,图引擎负责按 reducer 合并。这种设计带来的灵活性是 create_agent 给不了的。
举个实际例子。我做过一个内容审核 Agent,流程是:先让模型判断内容类型,如果是敏感类型就走人工审核节点,否则走自动处理节点,处理完再让模型生成回复。用 StateGraph 表达:
class ReviewState(TypedDict): content: str category: str needs_human: bool result: str def classify(state: ReviewState): cat = classifier.invoke(state["content"]) return {"category": cat, "needs_human": cat in SENSITIVE_TYPES} def route(state: ReviewState): return "human" if state["needs_human"] else "auto" builder = StateGraph(ReviewState) builder.add_node("classify", classify) builder.add_node("human", human_review) builder.add_node("auto", auto_process) builder.add_node("respond", generate_response) builder.add_edge(START, "classify") builder.add_conditional_edges("classify", route, {"human": "human", "auto": "auto"}) builder.add_edge("human", "respond") builder.add_edge("auto", "respond") builder.add_edge("respond", END)这种"分类→分支→汇合"的结构,用 create_agent 根本表达不了。而且因为状态是显式的,人工审核节点可以只读content和category,写result,职责非常清晰。
3.3 一张表看清两者的取舍
| 维度 | create_agent | StateGraph |
|---|---|---|
| 上手成本 | 极低,一行代码 | 中等,需定义状态和节点 |
| 状态结构 | 固定(messages 为主) | 完全自定义 |
| 控制流 | 固定 ReAct 循环 | 任意图结构 |
| 人工介入 | 需额外包装 | 原生支持(插节点即可) |
| 多阶段流程 | 不支持 | 支持 |
| 适用场景 | 标准工具调用 Agent | 复杂编排、多分支、多阶段 |
| 调试难度 | 低(结构简单) | 中(需理解图) |
我的建议很直接:如果你在纠结选哪个,先问自己"我的流程里有没有非线性的控制需求"。有,就 StateGraph;没有,create_agent 够用。别为了用框架而用框架,也别为了省事而把复杂流程硬塞进 create_agent。
4. 从手写循环迁移到 LangGraph 的完整实操
4.1 第一步:把循环里的状态抽出来定义清楚
迁移的第一件事不是写代码,是把你手写循环里所有的状态变量列出来。拿一张纸,把messages、step_count、current_tool、retry_count、user_context这些全写下来,然后判断哪些是真正需要在节点间传递的,哪些是节点内部的临时变量。
临时变量不要放进状态。比如某个节点内部算了一个中间结果,用完就扔,那就别污染状态。状态里只放跨节点需要共享的东西。这个判断标准能帮你避免状态膨胀。
定义状态时,注意 reducer 的选择。messages用add_messages,计数器类的字段如果多个节点都可能更新,要么用operator.add,要么明确只有一个节点写它。我见过有人给所有字段都加了add,结果step_count被加了好几次,逻辑全乱了。
4.2 第二步:把循环体拆成节点,一个职责一个节点
手写循环里的一段逻辑,往往对应图里的一个节点。拆分的粒度原则是:一个节点只做一件事,且这件事是原子的。比如"调用模型"是一个节点,"执行工具"是一个节点,"判断是否继续"是一个条件边而不是节点。
拆分的时候有个常见误区:把太多逻辑塞进一个节点。我见过有人把"调用模型+解析工具调用+执行工具"全塞一个节点里,那跟手写循环没区别,图只是个摆设。正确的做法是让每个节点短小、可测试、可复用。
工具执行节点通常用 LangGraph 预置的ToolNode,它会自动处理工具调用的解析和并行执行。如果你有自定义的工具执行逻辑(比如加权限校验),可以自己写节点,但建议先看看ToolNode能不能满足,别重复造轮子。
4.3 第三步:用条件边替代 if-else,把分支画出来
手写循环里的if判断,在图里就是条件边。条件边接收一个函数,函数返回一个字符串,图引擎根据这个字符串决定走哪条边。
def route_after_agent(state: AgentState) -> str: last_msg = state["messages"][-1] if not last_msg.tool_calls: return "end" if state["step_count"] > MAX_STEPS: return "force_end" return "tools" builder.add_conditional_edges( "agent", route_after_agent, {"tools": "tools", "end": END, "force_end": END} )这里force_end和end都指向 END,但语义不同,方便你在日志里区分是正常结束还是超步数强制结束。这种"多个条件指向同一目标但语义分离"的技巧,在排查问题时特别有用。
4.4 第四步:接上 checkpointer,把持久化交给框架
迁移的最后一步是持久化。手写循环里你可能自己存了对话历史,现在把它交给 checkpointer。编译图的时候传入 checkpointer,调用时传thread_id。
from langgraph.checkpoint.sqlite import SqliteSaver with SqliteSaver.from_conn_string("agent.db") as checkpointer: graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "session-001"}} graph.invoke({"messages": [("user", "你好")]}, config)有个细节要注意:checkpointer 保存的是每个节点执行后的状态快照,所以你的状态必须可序列化。如果你在状态里放了不可序列化的对象(比如数据库连接、文件句柄),会直接报错。这也是为什么我前面强调状态要精简。
迁移完成后,你会发现代码结构完全变了:不再有一个巨大的循环函数,而是一堆小节点加一张图。可读性、可测试性、可扩展性都是质的提升。代价是你得理解图模型,前期有学习成本,但一旦跨过去,后面加需求就是加节点加边的事。
5. 那些文档不会告诉你的实战坑
5.1 状态 reducer 写错,消息莫名其妙丢失
这是新手最容易踩的坑。messages字段如果不加Annotated[list, add_messages],默认行为是覆盖。也就是说,后执行的节点返回的 messages 会把前面的全冲掉。表现就是:Agent 好像失忆了,工具返回的结果模型看不到。
我第一次遇到的时候以为是模型的问题,换了模型还是一样,最后打印状态才发现 messages 只剩最后一条。加上add_messages就好了。类似的,任何需要"累积"的字段都要显式声明 reducer,别指望默认行为。
5.2 条件边返回了未声明的目标,直接抛异常
条件边的映射字典必须包含函数可能返回的所有字符串。如果你返回了一个字典里没有的 key,LangGraph 会直接报错。这个设计是好的(强制你穷举所有分支),但调试时如果没注意,会一脸懵。
我的习惯是:条件边函数里用常量而不是裸字符串,映射字典也用同样的常量,这样拼写错误在 IDE 里就能发现。
TOOLS = "tools" END_NODE = "end" def route(state): return TOOLS if state["messages"][-1].tool_calls else END_NODE builder.add_conditional_edges("agent", route, {TOOLS: "tools", END_NODE: END})5.3 无限循环:Agent 反复调用同一个工具
即使有了图,Agent 还是可能陷入死循环——模型反复调用同一个工具,工具反复返回同样的结果。这不是 LangGraph 的锅,是 Agent 本身的经典问题。解决办法是在状态里加计数器,在条件边里判断。
def route_after_agent(state): if state["step_count"] >= MAX_STEPS: return "force_end" # 检测重复工具调用 recent = state["messages"][-4:] tool_names = [m.name for m in recent if hasattr(m, "name")] if len(tool_names) >= 3 and len(set(tool_names)) == 1: return "force_end" return "tools" if state["messages"][-1].tool_calls else "end"这个"检测最近 N 条消息里是否重复调用同一工具"的逻辑,是我踩了好几次坑之后加的。阈值设多少要看你的场景,3 次是个比较稳的起点。
5.4 流式输出和图的执行顺序对不上
LangGraph 支持流式输出,但流出来的事件顺序可能和你预期的不一样。因为图是并行执行可能的,多个节点的事件可能交错。如果你要做"打字机效果",得用stream_mode="messages"专门过滤模型 token 事件,而不是用默认的values模式。
for event in graph.stream(inputs, config, stream_mode="messages"): message_chunk, metadata = event if message_chunk.content: print(message_chunk.content, end="")这个坑我在做前端对接时踩过,默认模式下拿到的是状态快照,不是 token 流,前端根本没法做逐字显示。
5.5 检查点存储膨胀,跑几天数据库就爆了
用 SqliteSaver 或 Postgres checkpointer 时,每次节点执行都会写一条记录。如果你的 Agent 调用频繁,几天下来表就几百万行。解决办法有两个:一是定期清理旧 thread 的检查点,二是用checkpoint_id做保留策略,只保留最近 N 个。
另外,如果你的状态里 messages 很长(比如带了大段文档),每条检查点都会存一份完整 messages,存储放大很严重。这种情况建议把大内容外置,状态里只存引用 ID。
6. 关于 LangGraph 和 LangChain 的关系,以及学习路径
6.1 别再问"LangGraph 和 LangChain 啥区别"了
关键词里langgraph和langchain的区别热度很高,说明很多人卡在这个概念上。一句话说清楚:LangChain 是组件库,LangGraph 是编排框架。
LangChain 提供的是模型封装、工具定义、提示词模板、检索器这些"零件"。LangGraph 提供的是把这些零件组装成有状态、有分支、有循环的"流程图"的能力。你可以只用 LangChain 不用 LangGraph(手写循环就是),也可以只用 LangGraph 不用 LangChain(自己封装模型调用也行),但两者配合最顺。
create_agent是 LangGraph 里的高阶封装,它内部用了 LangChain 的工具抽象。所以三者不是替代关系,是层次关系。
6.2 学习路径:先跑通,再拆解,最后定制
我给的建议路径是这样的。第一步,用create_agent跑通一个最简单的工具调用 Agent,感受一下整体流程。第二步,用StateGraph手动搭一个等价的 ReAct 循环,理解节点、边、状态、reducer 这些概念。第三步,在第二步的基础上加一个自定义节点(比如人工审核),体会图模型的扩展性。第四步,接上 checkpointer,实现多轮对话和中断恢复。
这个路径走下来,大概两三天能摸熟。别一上来就啃官方文档的所有 API,会淹死在细节里。先跑通最小闭环,遇到问题再查文档,效率高得多。
6.3 面试里会被问到的几个点
关键词里有langchain和langgraph面试题,我顺带说几个高频问题。一是"LangGraph 的状态 reducer 是干嘛的",答清楚合并策略和add_messages就行。二是"条件边和普通边的区别",核心是条件边根据状态动态决定下一步。三是"checkpointer 的作用和实现原理",答持久化、thread_id、时间旅行这几点。四是"create_agent 和 StateGraph 怎么选",就是我前面那张表的逻辑。
面试官真正想听的不是你会不会背 API,而是你理不理解为什么要有图模型。能讲清楚手写循环的痛点,以及图模型怎么解决这些痛点,基本就稳了。
7. 我个人的一点使用体会
从手写循环迁到 LangGraph,最大的感受不是"代码变少了",而是"思路变清晰了"。手写循环的时候,我脑子里装的是执行顺序;用图之后,我脑子里装的是状态流转。这个转变一开始不适应,但一旦转过来,加需求就变成了"加个节点、连条边"这么简单的事。
当然 LangGraph 也不是银弹。它的抽象层有学习成本,简单场景用它反而啰嗦。我现在的判断标准是:如果流程能用一张图清楚地画出来,就用 LangGraph;如果画不出来或者图很简单,就手写。别为了追新而用框架,工具是服务于问题的。
最后分享一个小技巧:调试复杂图的时候,把graph.get_graph().draw_mermaid()的输出贴到支持 Mermaid 的编辑器里,能直接看到图的结构。虽然本文不让用 Mermaid 图表,但这个自检手段在开发时真的能帮你快速发现"边连错了"这种低级问题。图对不对,画出来一看便知。