LangGraph 编排：把 Agent 变成可恢复的状态机

RAG 与 Agent 里的 Agent 是一句 while 循环；Agent 架构通论讲的是形态与控制流。本页补上中间那层：当 Agent 要跑很久、要在中间等人审批、挂了要能接着跑、出错要能回到上一步重来时，循环就不够用了——需要把「步骤」和「状态」显式化，这就是图式编排框架 LangGraph 的位置。

一、为什么是图：while 循环哪里不够

一个裸循环能跑通 ReAct，但下面这些需求它都接不住：

需求	裸 while 循环	图式编排
进程崩了从断点继续	状态在内存，全丢	每步状态落盘（durable execution）
中途等人审批	阻塞线程或自己写队列	`interrupt()` 暂停，随时用 `Command` 恢复
回到某一步换个分支重跑	重跑整个任务	时间旅行：读历史状态、改状态、继续
并行分支再汇合	手写线程/协程与聚合	扇出节点 + 状态 reducer 聚合
每一步的状态可观测	自己打日志拼接	状态快照天然可序列化、可回放
长时间运行（小时/天级）	不现实	设计目标就是 long-running

核心变化：状态从「函数里的局部变量」变成「一等公民」。图 = 节点（函数）+ 边（流转）+ 状态（可持久化的数据结构）；
反过来，这些场景别上图：单轮问答、无状态的固定流水线、一次性的批处理脚本——直接写函数更清楚，图只会增加理解成本。

二、生态分层：别把 LangGraph 当全家桶用

按官方对自身产品线的划分，四层各管一段，混着谈容易选错：

层	定位	管什么
Deep Agents	Agent harness（脚手架）	规划、子智能体、文件系统工具、上下文管理，建在 LangGraph 之上
LangChain	Agent 框架	模型与工具的抽象和集成、预置的 Agent 循环
LangGraph	编排运行时	持久执行、流式、人机协同、持久化
LangSmith	平台	追踪（trace）、评测、提示词管理、部署

LangGraph 是底层（low-level）：它不抽象提示词，也不替你决定 Agent 架构，只提供编排能力；
LangGraph 可以脱离 LangChain 单独使用——文档里的示例常用 LangChain 组件是为了接模型与工具方便，不是依赖；
官方建议：刚起步或想要高层抽象时，先用 LangChain 的预置 Agent；等你确实需要精细控制每一步、需要持久化和人机协同，再下潜到 LangGraph；
追踪与评测统一走 LangSmith（与 Agent 评测与可观测的 Trace 设计对齐）。

三、核心概念：State / Node / Edge / Checkpointer

概念	是什么	关键细节
State	在节点间流动的数据结构	用 `TypedDict` 定义；字段的更新语义由 reducer 决定（如 `Annotated[list, operator.add]` 表示追加而非覆盖）
Node	一个普通函数	接收 state，返回状态增量；必须幂等（见第五节的重放语义）
Edge	节点间的连线	固定边 `add_edge`；条件边 `add_conditional_edges`（路由函数返回下一个节点名）
START / END	虚拟入口与出口	图的起点和终点
Checkpointer	状态持久化层	每个「步」结束后写快照，是持久执行与人机协同的地基

最小骨架（形态取自官方 Overview 示例，未在本地实跑，API 以官方文档为准）：

python

# pip install -U langgraph
from langgraph.graph import StateGraph, MessagesState, START, END

def mock_llm(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

graph = StateGraph(MessagesState)
graph.add_node(mock_llm)                 # 节点：函数名即节点名
graph.add_edge(START, "mock_llm")        # 入口
graph.add_edge("mock_llm", END)          # 出口
graph = graph.compile()

graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})

compile() 才是真正的分界线：编译前只是声明，编译时注入 checkpointer、store 与断点，运行时才有持久化与中断能力；
状态字段默认是覆盖语义，想要「消息不断追加」这类效果必须显式声明 reducer——这是新手最常踩的坑（消息历史被后一步整体覆盖掉）。

四、持久化：Checkpointer 与 Store 是两套东西

	Checkpointer	Store
存什么	图状态快照（graph state）	自定义键值数据
作用域	单个 thread	跨 thread
记忆类型	短期、线程内	长期、跨会话
用途	会话连续、人机协同、时间旅行、容错	用户偏好、事实、共享知识

用法：builder.compile(checkpointer=..., store=...)，调用时传 config={"configurable": {"thread_id": "..."}}；
thread_id 就是游标：复用同一个 id = 接着上次的断点；换一个新 id = 开一条全新的空状态线程；
后端选择：InMemorySaver / MemorySaver 仅开发用（进程重启即丢），SqliteSaver 适合本地文件持久化，PostgresSaver 用于生产（支持异步）；
四个官方点名的坑：
1. PostgresSaver 的 thread_id 有列长限制，控制在 255 字符内（用 UUID 或哈希）；
2. 内存型 saver 进程重启后全部丢失，别在生产用；
3. checkpoint 会无界增长，长会话要定期清理或设保留策略；
4. 子图有独立的 checkpoint 命名空间，父图不会立即看到子图的状态更新——跨边界共享的数据要走 Store。

五、人机协同：interrupt 与 Command

5.1 它是怎么暂停的

python

from langgraph.types import interrupt, Command

def approval_node(state: State):
    approved = interrupt("Do you approve this action?")   # 在此暂停
    return {"approved": approved}                          # 恢复后从节点开头重跑

interrupt() 通过抛出特殊异常让运行时暂停，checkpointer 保存状态，然后无限期等待；
恢复：再次调用图并传入 Command(resume=...)，这个值会成为 interrupt() 的返回值；必须用同一个 thread_id；
恢复时节点从开头重新执行，不是从 interrupt 那一行继续——这一条决定了下面所有纪律。

5.2 官方点名的四条规则

规则	原因
不要用裸 `try/except` 包住 `interrupt`	中断靠异常实现，裸 except 会把异常吞掉，中断失效
不要在同一节点内条件性跳过或循环调用 `interrupt`	恢复值按调用索引匹配，跳过或循环会导致错位；需要「反复追问直到合法」时用条件边回环，每次执行只调一次
不要给 `interrupt` 传不可序列化对象	函数、类实例等按 checkpointer 可能无法序列化；只传 JSON 友好的简单值或字典
`interrupt` 之前的副作用必须幂等	节点会重跑，之前的写库/追加操作会重复执行；把副作用挪到中断之后，或拆成独立节点

5.3 三种常用模式

审批（approve / reject）：节点里 interrupt 后按返回值路由——return Command(goto="proceed" if decision else "cancel")；
审阅并编辑状态：把当前内容放进 interrupt 载荷让人改，恢复值直接写回状态（纠正 LLM 输出、补漏信息）；
工具内中断：直接把 interrupt 写在工具函数里，任何人调用该工具都会先暂停待审，且恢复值可以覆盖入参再执行（例如改掉收件人再发信）。

并行分支同时中断：多个节点各自 interrupt 时，用 {interrupt_id: 恢复值} 的映射一次性恢复，确保每个回答配对到正确的中断；
静态断点（interrupt_before / interrupt_after）只用于调试（逐步执行、LangSmith Studio 里下断点），官方明确不建议拿它实现人机协同——那是动态 interrupt() 的职责。

5.4 驱动一个会中断的图

python

from langgraph.types import Command

stream_input = initial_input
while True:
    stream = graph.stream_events(stream_input, config=config, version="v3")
    for message in stream.messages:            # 逐 token 输出
        for token in message.text:
            display_streaming_content(token)
    if not stream.interrupted:
        final_state = stream.output
        break
    user_response = get_user_input(stream.interrupts[0].value)   # 拿中断载荷问人
    stream_input = Command(resume=user_response)                 # 带着回答恢复

stream.messages（模型输出流）、stream.values（每步状态快照）、stream.interrupted / stream.interrupts（是否暂停与载荷）三件套构成了交互式 Agent 的驱动循环；
只有 Command(resume=...) 是合法的输入形态。update / goto / graph 这些参数是给节点返回值用的，不要把它们当输入传给 invoke/stream 来续对话——多轮对话直接传普通输入字典。

六、常见图模式

模式	结构	适用
ReAct 循环	模型节点 ⇄ 工具节点，条件边按是否有 tool call 决定回环或结束	工具调用型 Agent 的默认形态
路由分发	分类节点 → 条件边 → 若干处理分支	意图明确、分支互斥（客服/工单分类）
Plan-execute	规划节点产出步骤 → 执行节点逐个跑 → 复盘节点决定是否重规划	长任务、需要中途调整
Map-reduce 扇出	一个节点分发 N 个子任务并行 → reducer 汇总	批量处理、多路检索后融合
Supervisor	主管节点调度多个专职子图	角色分工清晰的多智能体
Swarm / Handoff	智能体之间互相移交控制权	对话式交接、无法预先定死流程
子图嵌套	图作为节点被父图调用	复用与分层；注意状态隔离，跨边界数据走 Store

选型原则：能用固定边表达的就不要用条件边，能让模型决定的分支越少，行为越可预测、越容易测；
多智能体的代价是成本与调试复杂度（每个 handoff 都是一次模型调用，且轨迹变长），见 Agent 架构通论里对多智能体成本的提醒。

七、生产化要点

节点幂等是第一原则：持久执行与中断恢复都会重跑节点，任何「创建记录 / 追加日志 / 发通知」都要能安全重复执行（用 upsert、幂等键、或拆到独立节点）；
状态要可序列化且体积可控：状态里放大段检索结果会让 checkpoint 膨胀，只放必要的引用与摘要，正文留在外部存储（与 Agent 记忆与状态的上下文预算同理）；
可观测：接 LangSmith 记录每一步的状态转换与耗时；没有 trace 时，多步 Agent 的故障基本无法定位；
测试：把图当纯函数测（给定输入状态 → 断言状态增量），再测路由函数的分支覆盖，最后做端到端重放；
部署形态：长时间运行的有状态工作流对基础设施有特殊要求（状态存储、优雅重启），用托管 Agent Server 时持久化由服务端自动处理，自建则要自己保证 checkpointer 的高可用与备份；
版本与兼容：框架迭代快，升级前先看官方的向后兼容说明，并把关键图结构纳入回归测试。

八、什么时候不该用 LangGraph

情况	更合适的选择
单轮问答、固定顺序几步	直接写函数即可
只是想快速拼一个 ReAct	LangChain 的高层 Agent 抽象
需要规划/子智能体/文件系统等成套能力	Deep Agents 这类 harness
团队没有 Python 服务运维能力	低代码平台（见平台与框架选型）
流程由非技术同学维护	可视化编排平台，而不是代码里的图

九、踩坑清单

状态字段忘了声明 reducer：后一步的返回值把消息历史整体覆盖，表现为「模型忘了前面说过什么」；
compile() 时没配 checkpointer 就想要中断：interrupt 依赖持久化层，没 checkpointer 就无从保存与恢复；
恢复时换了 thread_id：等于开了新线程，状态是空的，看起来像「恢复没生效」；
用 MemorySaver 上生产：进程一重启所有会话与待审批任务全部消失；
thread_id 超长：Postgres 后端下列长限制会直接报错，用 UUID 或哈希；
裸 try/except 包住 interrupt：吞掉中断异常，暂停失效，节点直接往下跑；
在节点里循环调用 interrupt 做输入校验：每次恢复都从节点开头重放，循环体被指数级重复执行；改用条件边回环，每次只调一次；
interrupt 之前写库/追加：节点重跑导致重复记录，副作用必须幂等或后移；
把 Command(update=...) 当输入传给 invoke：续对话应传普通输入字典，Command 的 update/goto/graph 只用于节点返回值；
状态里塞大段内容：checkpoint 体积与延迟同步膨胀，只存引用与摘要；
checkpoint 从不清理：长会话场景下存储与延迟持续劣化，要有保留策略。

十、检查清单

[ ] 状态用 TypedDict 定义，需要累积的字段显式声明了 reducer
[ ] 节点是幂等的，副作用要么幂等要么放在独立节点
[ ] 生产环境使用 PostgresSaver（或同等持久后端），thread_id 长度受控
[ ] checkpoint 有清理/保留策略
[ ] 人机协同用动态 interrupt()，静态断点只用于调试
[ ] 同一节点内 interrupt 调用顺序稳定，未被 try/except 包裹
[ ] 传给 interrupt 的载荷是 JSON 可序列化的简单值
[ ] 驱动循环能正确处理 stream.interrupted 并用 Command(resume=...) 恢复
[ ] 接入了 LangSmith（或等价的 trace）覆盖每一步状态转换
[ ] 关键图结构与路由分支有回归测试

状态与参考

状态：已收录（2026-09-04）。本文的 API 形态、分层定位与中断规则均取自 LangChain 官方文档（LangGraph overview / Persistence / Interrupts），示例代码未在本地实跑，落地前请以对应版本的官方文档为准；
关联：Agent 架构通论（形态谱系与控制循环）、RAG 与 Agent（ReAct 最小循环实测）、Agent 记忆与状态（长短期记忆与状态持久化）、Agent 评测与可观测（Trace 与门禁）、Agent 安全与治理（审批闸门与动作分级）、Code Agent（写码循环）、GUI Agent（界面操作循环）、MCP（工具接入协议）、平台与框架选型；
提醒：LangGraph 1.0 之后生态分层（Deep Agents / LangChain / LangGraph / LangSmith）与流式接口（stream_events(version="v3")）都有过较大变化，网上大量教程仍停留在旧版 API，照抄容易踩坑。

想先理解「为什么要编排」回看 Agent 架构通论；想知道该不该上框架、还有哪些平台可选，见平台与框架选型。

LangGraph 编排：把 Agent 变成可恢复的状态机 ​

一、为什么是图：while 循环哪里不够 ​

二、生态分层：别把 LangGraph 当全家桶用 ​

三、核心概念：State / Node / Edge / Checkpointer ​

四、持久化：Checkpointer 与 Store 是两套东西 ​

五、人机协同：interrupt 与 Command ​

5.1 它是怎么暂停的 ​

5.2 官方点名的四条规则 ​

5.3 三种常用模式 ​

5.4 驱动一个会中断的图 ​

六、常见图模式 ​

七、生产化要点 ​

八、什么时候不该用 LangGraph ​

九、踩坑清单 ​

十、检查清单 ​

状态与参考 ​

LangGraph 编排：把 Agent 变成可恢复的状态机

一、为什么是图：while 循环哪里不够

二、生态分层：别把 LangGraph 当全家桶用

三、核心概念：State / Node / Edge / Checkpointer

四、持久化：Checkpointer 与 Store 是两套东西

五、人机协同：interrupt 与 Command

5.1 它是怎么暂停的

5.2 官方点名的四条规则

5.3 三种常用模式

5.4 驱动一个会中断的图

六、常见图模式

七、生产化要点

八、什么时候不该用 LangGraph

九、踩坑清单

十、检查清单

状态与参考