这是「近五年 AI 技术演进」系列深挖的第六篇。前五篇把地基铺完了全景、RAG、Function Calling、Tool、MCP。这一篇正式进入编排层。当模型会说话、会查资料、会表达调用意图、工具也能被标准地发现之后剩下的问题只有一个反复调工具直到干完这个循环用什么写。一句话定位用图结构编排 Agent 循环的框架。全文约 5500 字含一段可以直接跑的 ReAct Agent 代码。一、链的尽头是一堵墙先说清楚链到底是个什么东西。LangChain 里的 Chain以及后来的 LCEL本质上是一条有向无环的调用管道A 的输出喂给 BB 的输出喂给 Cprompt | model | parser就是最短的一条。写起来确实优雅一行竖线串起整个流程像 Unix 的管道。但有向无环这四个字已经把它能做的事说完了。它能往前走不能回头也不能自己决定去哪。平时写业务代码这个限制不明显。做文本摘要、做翻译、做格式转换——流程是固定的人是提前知道每一步该干嘛的链完全够用。一旦开始做 Agent三件事会轮流撞上来第一件失败要重试。模型输出了一段不合法的 JSON你想让它带着错误信息再试一次。在链里第 3 步出错你只能把整条链从头再跑一遍——因为链里没有从第 3 步退回到第 2 步这条路。于是你在外面套一个for循环套着套着发现自己在手写一个状态机。第二件走哪条路要看模型说了什么。用户问帮我查下订单先判断这是查询、投诉还是退款三种情况走三套流程。链里的RunnableBranch能在编译时定下几条支路但判断条件写在代码里。而 Agent 的判断是运行时从模型输出里长出来的——模型说我还需要再查一次物流这件事写代码的人事先不知道。第三件也是致命的做几步才知道要不要继续。查天气 → 发现要查的其实是另一个城市 → 再查一次 → 发现要对比两个城市 → 再查一次。循环几次取决于中间结果。这不是分支分支是有限条路选一条这是回环同一段逻辑可以走 N 遍。三件事凑齐你就明白了链把下一步做什么在写代码时就定死了而 Agent 的下一步是在运行时才长出来的。更麻烦的还在后面。这些补丁一旦叠上去代码会开始变形for retry in range(3)套着if needs_more:套着try/except中间还夹着几个session_state变量。三个月后你回来看会发现这段代码已经没人能说清它到底会走几条路——它不再是一个流程而是一堆流程的残骸。真正让团队决定换框架的往往不是跑不通而是看不懂。二、从链到图把 while 和 if 搬进数据结构手写 Agent 的人最后都会写出类似这样一段代码while True: resp model.invoke(messages) if not resp.tool_calls: break messages.append(resp) for call in resp.tool_calls: messages.append(run_tool(call))就这么几行能跑。问题不在能不能跑在于这个循环是一段藏起来的控制流你没法在中间停下、没法存档、没法在出错的那一步单独重试、也没法在监控里看到现在走到第几步了。它是一段黑盒。LangGraph 做的事说白了只有一句把这个while和里面的if从代码里搬出来变成一张可以检查、可以存档、可以中断的数据结构。搬出来之后原来的东西各自有了名字循环体里的变量messages→State循环体里干活的函数model.invoke、run_tool→Nodeif not resp.tool_calls: break→条件边循环回到开头 →回边循环的退出 →END于是链变成图。差别不只是画法好看控制流一旦变成数据它就能被序列化、被持久化、被人在半路改掉。这才是 LangGraph 真正的价值不是图是图可以被停下来。三、三个核心概念LangGraph 的 API 表面上有十几个方法但真正需要理解的只有三个名词。剩下的都是这三个的排列组合。3.1 State整张图共享的一份数据State 是一个TypedDict也可以用 dataclass 或 Pydantic声明这张图运行时有哪些字段。from typing import Annotated from typing_extensions import TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] steps: int关键在那个Annotated[list, add_messages]。默认的合并策略是覆盖。节点返回{steps: 3}框架就把 State 里的steps改成 3。这对标量没问题但对列表是灾难模型节点返回一条新消息如果按覆盖处理前面所有对话历史就没了。所以列表字段必须挂一个reducer——一个告诉框架两个值怎么合并的函数。add_messages就是官方给的 reducer追加、去重按消息 id、并按 id 更新已有消息。没有它你的多轮对话在第二轮就会失忆。这里有个容易忽略的推论并行分支必须用 reducer。如果你让一个节点同时扇出到三个节点Send或一条边连多条边三个节点会在同一轮里各自返回增量然后统一合并。没有 reducer 的字段三个结果会互相覆盖最后谁赢取决于执行顺序——这种 bug 极难复现。还有一个认知要先摆正State 不是全局变量它是每一步之后的快照。每个 super-step 结束时框架把当前 State 存一份如果配了 checkpointer。节点拿到的永远是上一轮结束时的只读视图它只能通过返回值表达我要改什么。这个约束让整个流程可回放——这是后面 Checkpointer 和人在回路的全部前提。设计 State 有个很实用的经验字段不是越多越好而是必须跨节点传递的才放。临时变量留在节点内部就好不要什么都往 State 里塞。原因很直接——State 每个 super-step 都会被序列化一次字段越多checkpoint 越大回放越慢而且一旦某个字段只在一个节点里用你以后想改它就得先确认没有别的节点在读。能塞进消息里的就别开新字段因为messages已经有 reducer、有标准结构、还能直接被模型看见是最省事的那块共享内存。真正值得单独开字段的通常是流程控制用的计数器和需要人看的中间产物比如steps、draft、approval_status。3.2 Node一个函数一次动作节点就是一个普通函数签名(state) - dict。返回的 dict 是增量不是完整的 State。def call_model(state: AgentState) - dict: resp model_with_tools.invoke(state[messages]) return {messages: [resp]} # 只说要改 messages框架负责把{messages: [resp]}用add_messages合并进去steps之类的其他字段原样保留。节点里放什么框架不管一次模型调用、一次数据库查询、一段 Python 计算、甚至另一个编译好的图叫子图。它只要求两件事不要改传进来的 state。你要返回新值不要state[x] 1。改了也不会报错但会让回放和并行出问题。尽量幂等。同一个节点可能被重跑——尤其是中断恢复的时候后面会讲。发邮件、扣款这类副作用要么放在能确认只执行一次的位置要么自己加幂等键。节点名会出现在追踪、可视化、以及get_state()的返回值里。起个能看懂的名字调试时省下的时间远超你想的。3.3 Edge普通边管顺序条件边管决策边决定这个节点跑完下一个是谁。普通边是无条件跳转add_edge(tools, agent)tools 跑完一定回 agent。它负责把节点串成流水线。条件边是 Agent 的灵魂def should_continue(state) - str: last state[messages][-1] return tools if getattr(last, tool_calls, None) else end builder.add_conditional_edges( agent, should_continue, # 路由函数读 state返回一个 key {tools: tools, end: END}, # 映射表key → 下一个节点 )它做三件事跑一个路由函数、拿到一个字符串、拿这个字符串去映射表里查下一个节点。为什么说它是灵魂因为这一个函数就是Agent 什么时候停的全部定义。模型不再是生成一段文本它是生成一个决定流程走向的信号。你把判断逻辑写在路由函数里流程的拓扑就从静态变成了动态——同一张图这次跑 2 轮下次跑 7 轮。顺便说清三件常被误解的事第一边不搬运数据。很多人第一反应是边是管道数据从这条边流过去。不是。数据永远只走 State边只负责路由。理解这一点你才不会在边上找怎么传参。第二条件边可以指回上游。这就是回边就是循环。add_conditional_edges(agent, ..., {tools: tools, end: END})加上add_edge(tools, agent)一个 ReAct 循环就成立了。第三循环必须有刹车。LangGraph 用recursion_limit限制单次调用的 super-step 数1.0.6 之后默认 1000。一轮 ReAct 大约消耗 2 步模型 1 步 工具 1 步所以 1000 差不多等于 500 轮。听起来够用但模型要是卡在两个工具之间来回撞它会在给出结论前先把额度烧光。生产里显式设置recursion_limit并且捕获GraphRecursionError别让它变成一个没人处理的 500。四、写一个真正能跑的 ReAct Agent理论讲完动手。下面这段 90 行左右pip install langgraph langchain-openai之后填上 API Key 就能跑。版本提醒LangGraph 1.0 在 2025 年 10 月发布并承诺到 2.0 之前不破坏兼容。网上大量 0.x 教程里的create_react_agent已被高层封装create_agent取代MemorySaver也改了名叫InMemorySaver旧名仍是别名。下面的代码按 1.x 写手搓 StateGraph跨版本最稳。# react_agent.py —— 一个完整可运行的 ReAct AgentLangGraph 1.x # 依赖pip install langgraph langchain-openai from typing import Annotated, Literal from typing_extensions import TypedDict from langchain_core.messages import ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import START, END, StateGraph from langgraph.graph.message import add_messages # ---------- 1. 工具普通函数加个装饰器签名自动变 JSON Schema ---------- tool def get_weather(city: str) - str: 查询指定城市的当前天气。city 用中文城市名例如「杭州」。 db {杭州: 多云 22℃东南风 2 级, 北京: 晴 18℃微风} return db.get(city, f没有 {city} 的天气数据) tool def query_orders(status: str) - str: 按状态查询订单。status 只能是 pending 或 shipped。 db {pending: [SO-1024 ¥2680, SO-1025 ¥899], shipped: [SO-1026 ¥15600]} rows db.get(status, []) return f共 {len(rows)} 条 .join(rows) if rows else f没有 {status} 的订单 TOOLS [get_weather, query_orders] TOOL_MAP {t.name: t for t in TOOLS} # 名字 → 工具执行时按名查找 # ---------- 2. State整张图共享的一份数据 ---------- class AgentState(TypedDict): messages: Annotated[list, add_messages] # reducer追加不是覆盖 steps: int # 自己记轮数方便兜底 # ---------- 3. 模型节点想一步 ---------- model ChatOpenAI(modelgpt-4o-mini, temperature0) model_with_tools model.bind_tools(TOOLS) # 把工具清单交给模型 def call_model(state: AgentState) - dict: 把历史消息喂给模型它要么说人话要么给出 tool_calls。 resp model_with_tools.invoke(state[messages]) return {messages: [resp], steps: state.get(steps, 0) 1} # ---------- 4. 工具节点做一步 ---------- def call_tools(state: AgentState) - dict: 执行最后一条 AIMessage 里的所有 tool_calls一条结果一条 ToolMessage。 last state[messages][-1] results [] for call in last.tool_calls: # 一次可能要求调多个工具 fn TOOL_MAP.get(call[name]) try: out fn.invoke(call[args]) if fn else f未知工具{call[name]} except Exception as e: # 失败也回填让模型自己纠错 out f工具执行失败{e} results.append(ToolMessage(contentstr(out), tool_call_idcall[id])) return {messages: results} # ---------- 5. 条件边Agent 的灵魂 ---------- def should_continue(state: AgentState) - Literal[tools, end]: 只看最后一条消息带 tool_calls 就去执行不带就结束。 return tools if getattr(state[messages][-1], tool_calls, None) else end # ---------- 6. 组装成图 ---------- builder StateGraph(AgentState) builder.add_node(agent, call_model) # 节点一次模型调用 builder.add_node(tools, call_tools) # 节点一次工具执行 builder.add_edge(START, agent) # 入口 builder.add_conditional_edges( # 条件边运行时分流 agent, should_continue, {tools: tools, end: END}, ) builder.add_edge(tools, agent) # 回边这就是循环 app builder.compile(checkpointerInMemorySaver()) # 带检查点编译 # ---------- 7. 调用 ---------- if __name__ __main__: config {configurable: {thread_id: demo-1}, recursion_limit: 12} out app.invoke( {messages: [(user, 杭州现在天气怎么样顺便看下还有多少待发货订单。)], steps: 0}, config, ) for m in out[messages]: calls getattr(m, tool_calls, None) print(f[{m.type}] {str(m.content)[:60]}, f→ {[c[name] for c in calls]} if calls else )跑起来会发生什么第一次进agent模型看到两个工具决定同时调get_weather(杭州)和query_orders(pending)返回两条 tool_calls。条件边看到 tool_calls路由到tools。tools逐个执行产出两条ToolMessage。回边把控制权交回agent。第二次进agent模型手上已经有天气和订单直接输出一段人话没有 tool_calls。条件边路由到END。总共4 个 super-step。这段代码里每一部分在干什么部分代码作用Statemessagesadd_messages唯一的共享数据只增不改模型节点call_model读全部历史产出一条 AIMessage工具节点call_tools读 tool_calls执行产 ToolMessage条件边should_continue决定再来一轮还是收工回边add_edge(tools, agent)把直线掰成环编译compile(checkpointer...)产出可 invoke 的app注意一个细节recursion_limit12。这不是随便写的。一轮循环 2 步加上首尾12 步大约允许 5 轮工具调用——足够正常任务又能让失控的 Agent 早点停下来。宁可让它早死也不要让它悄悄花掉两百次模型调用。调试这张图别只会看最终输出。app.stream(input, config, stream_modeupdates)会每跑完一个节点就吐一次增量你能清楚看到第几轮、哪个节点、往 State 里写了什么app.get_state(config)则能随时拿到当前快照和next字段——next告诉你如果现在继续下一个要跑谁。线上排查 Agent 为什么绕圈靠的就是这两句。如果跑起来发现它一直在agent和tools之间来回八成是两件事之一工具返回了空字符串模型以为没查到就一直重试或者工具描述写得让模型误判比如query_orders的说明里没写清status只能取哪两个值。这两个坑第三篇和第四篇里都踩过换到图里也一样会踩——图能管住流程管不住工具的描述质量。五、Checkpointer让 Agent 可以被暂停前面提了好几次存一份。现在说清楚这是什么。进程会死。用户会关页面。审批要等人。Agent 跑一次可能十几分钟。如果状态只在内存里上面任何一件事发生这次执行就永久丢了——而且是从第一步重来。Checkpointer 就是给每一步拍快照。每个 super-step 结束后框架把当前的 State、下一步该去哪个节点、以及一些元数据写进存储。这份快照叫 checkpoint一串 checkpoint 挂在同一个thread_id下面。app builder.compile(checkpointerInMemorySaver()) config {configurable: {thread_id: user-42}} app.invoke({messages: [(user, 帮我查下订单)]}, config) # 第一次 app.invoke({messages: [(user, 那发货了吗)]}, config) # 第二次自动带上历史两个 API换来三样东西断点续跑。进程崩了用同一个thread_id再invoke一次它从最近一个 checkpoint 接着跑。你不用写任何恢复逻辑——恢复逻辑是框架送的。多轮对话记忆。很多人第一次用 LangGraph 会问记忆怎么存。答案是历史就是 checkpoint 链。同一个 thread 下的 State 会累积你什么都不用做。这也是为什么 1.0 之后不再需要单独的 memory 组件。时间旅行。app.get_state_history(config)能拿到这条 thread 的所有快照每个带checkpoint_id。你可以把checkpoint_id塞进 config从任意一个历史点重新跑——调试时极其好用模型在第三步做了个蠢决定你不用从头复现直接回到第三步换句话再跑。存储怎么选InMemorySaver只在内存里进程一退全没只适合 notebook 和单测。SqliteSaver适合本地开发。生产用PostgresSaverpip install langgraph-checkpoint-postgres第一次用记得调setup()建表另外thread_id长度要控制在 255 字符以内。还有一句要提醒checkpoint 里存的是完整对话和工具返回结果等于把你所有的 prompt、用户输入、业务数据都落了一份盘。做数据合规的时候这张表要跟业务主表一个待遇。最后一个细节做多 Agent 时很关键子图不会自己开一个 thread。你把一张编译好的图当成节点塞进父图它默认跟父图共享同一个 checkpointer 和同一个thread_id——也就是说父子两层的 checkpoint 会写进同一条链里。想让子图独立存档、独立重跑要么在compile()时给子图单独配 saver要么用namespace把它隔离出来。这件事不搞清楚debug 的时候你会看到一条长得离谱的历史分不清哪一步是谁跑的。六、人在回路在不可逆的那一步交回方向盘Checkpointer 最实用的一个副产品是让停下来等人变成一件自然的事。早期做法是在compile()时写interrupt_before[tools]——跑之前就声明到 tools 之前停一下。问题是它太硬每次都在同一个地方停停几次也不知道停的时候也拿不到上下文。等于在流水线上焊了个闸门。现在的做法是在节点内部主动中断from langgraph.types import interrupt, Command def approve_and_send(state: AgentState) - dict: draft state[draft] # 执行到这里直接停住把这个问题抛给外面的人 decision interrupt({action: send_email, to: state[to], body: draft}) if not decision.get(approved): return {status: rejected} send_email(state[to], draft) # 副作用放在 interrupt 之后 return {status: sent}外面这样接config {configurable: {thread_id: mail-1}} app.invoke({messages: [(user, 给客户发一封延期说明)]}, config) # 跑到 interrupt 停住 # ……人可以走开进程可以退出第二天回来…… app.invoke(Command(resume{approved: True}), config) # 带着答案继续interrupt()会把问题抛出去并让整个图暂停同时把状态落盘。线程此时不占任何资源——它不需要挂着一个进程等也不需要开着长连接。等人和等定时器在这套模型里是同一件事这是设计上最聪明的一处。四个坑一个比一个重要① 恢复时节点是从头重跑的。interrupt()恢复后不是从下一行继续而是把整个节点函数再执行一遍只是这次interrupt()直接返回你传进来的值。所以interrupt()上面的代码会跑两次。副作用必须放在它下面或者自己加幂等键。这条是最容易出事的一条有人在interrupt()上面写了扣款恢复时扣了两次。② 必须有 checkpointer。没有它暂停那一刻状态就丢了interrupt()直接抛异常。③ 别用 try/except 包住interrupt()。它靠异常机制中断流程你一个宽泛的except Exception会把中断吞掉变成跑一半报错。④ 恢复值就是interrupt()的返回值。不要再额外存一份待审批记录在别处——两份状态一定会不同步。七、什么时候该用什么时候别用这是最该讲清楚的一节因为 LangGraph 现在是人人都说该用的状态而大部分 Agent 根本不需要它。Anthropic 在《Building Effective Agents》里的建议是对的从最简单的方案开始只有当它明显不够用时才加一层。按这个思路问自己四个问题① 一步就能答完吗那就直接调模型。连链都不用。② 需要模型反复调工具循环吗不需要一条链或者几个if就够。③ 中途要停下来等人、或者跨进程续跑吗需要这就是 LangGraph 的主场。Checkpointer interrupt()这一套手写至少一个月而且很难写对。④ 需要复杂分支、并行扇出、子图嵌套吗需要图的表达力确实比代码直观改起来也安全。四个都答否——就用一段while循环自己写。你比框架更懂自己的场景多引入一层依赖只会多一层调试成本。一张表把链和图摆在一起维度链Chain / LCEL图LangGraph表达能力有向无环线性或静态分支有向可循环任意拓扑循环不支持只能在外面套while原生支持回边分支编译期定死靠RunnableBranch运行时决定条件边读 State状态管理值在管道里流动无共享状态显式 State reducer可持久化中断与恢复不支持Checkpointer interrupt()可观测性调用树、回调每一步都是命名节点可追踪、可回放适用场景单轮、固定流程、数据处理管道多步决策、要人等、长任务、多 Agent 协作这张表最右边那列其实也在提醒一件事图不是链的替代品是链的下一层。一个真实系统里两者经常同时存在——入口做意图识别用的是链识别完之后按意图进不同的图图里某个节点内部又调了一条链来做格式清洗。别把上 LangGraph理解成把所有链都改写成图那是纯粹的自找麻烦。反过来也允许自己从图上退下来。上线三个月发现某个图的循环从来没超过一轮、条件边永远走同一个分支那就说明这个 Agent 其实是条链拆掉它省下的维护成本比省下的代码行数多得多。架构该跟着真实流量走不是跟着趋势走。八、框架是语法模式是文法最后一个容易混的问题LangGraph 和 Orchestrator 是什么关系它们不在同一个维度上所以不是二选一。LangGraph 是框架——它规定你的代码长什么样StateGraph、add_node、add_conditional_edges、compile。换掉它这些行要重写。Orchestrator 是模式——它规定谁负责决定、谁负责干活一个主脑拆任务、分派给若干 Worker、汇总结果。模式里没有一个import它住在你的架构图和分工里。换掉框架这个模式一点没变。所以正确的用法是叠着用Orchestrator 模式里的每一个 Worker都可以是 LangGraph 里的一个节点或者一整张子图subgraph。主脑本身也是一张图——它的条件边负责派给谁子图负责这个 worker 怎么干。顺便把模式这个词说得再具体一点。常见的编排模式其实就这么几种而且每一种在图上的形状都不一样Orchestrator是一个主节点扇出到若干子节点再汇总形状像扇子Router是一个条件边分流到不同处理节点形状像岔路口Evaluator-Optimizer是生成节点 评审节点之间来回转的一条回边形状是个环本质就是一个 ReAct 循环的变体。你把这些模式认出来之后会发现LangGraph 的 API 只是给这几种形状提供了画法选型时该想的是我这个业务属于哪种形状而不是我该不该用 LangGraph。顺带说清一个高频困惑官方的create_agent和手搓StateGraph是什么关系create_agent就是官方替你搓好的那张图模型节点 工具节点 条件边 回边外加一层 middleware 机制让你在固定位置挂钩子。它适合标准 Agent 循环你想改拓扑的时候再往下走一层。两者不是新旧关系是封装层次的关系。九、写在最后回到最开始那个while循环。它没有错。它只是把控制流藏进了代码里——藏进去之后你没法暂停它、没法存档、没法在监控里看见它、也没法让一个人在半路插手。LangGraph 做的事是把这段控制流摊开画在纸上状态是什么、有哪几步、什么条件下走哪条路、什么时候停。摊开之后图就变成了一份可以被工具读、被人改、被系统存档的资产。代价是你得比写while多花半天时间画清楚这张图——而这半天通常会在第一次线上事故里赚回来。放回系列里看Prompt是怎么和模型说话RAG是它的外挂知识Function Calling是它表达意图的语法Tool是它的手脚MCP是让手脚插得上任何机器的接口而LangGraph 是让这只手脚能动起来、能反复动、还能被人叫停的那套编排。再往后多个 Agent 之间怎么协作A2A、能力怎么打包分发Skills、运行时怎么把这些全串起来Harness是后面几篇的事。如果只记一句话链回答按顺序做什么图回答什么时候该回头。Agent 和流水线的全部差别就在这个回头上。