链式编排擅长"一次输入、一条路径、一次输出"。但 Agent 一旦需要"想不通就再试一轮""查不到就去问人""同一批子任务并行跑"，链就塌了——它没有环路，也没有"停在中途等人"的语义。StateGraph 给的正是这三样：循环、条件分支、可中断的检查点。

---

1. 为什么是图不是链

需求	链	图
失败后重试直到满足条件	没有原生环路，得在外面包 while	条件边回指上游节点
按中间结果走不同路径	只能提前括号式分支	路由函数 + 条件边
中途需要人工批准	做不到	interrupt 暂停，恢复后从断点继续
失败重放	从头再跑一遍	checkpoint 断点续跑
一批子任务并行	手动并发、手动聚合	Send 分发，下游自动汇聚

判断标准很直接：只要你的流程图画出来有回头的箭头，就该用图。

---

2. 三要素

要素	是什么	写在哪
State	全图共享的数据结构	TypedDict 定义 + 节点返回值
Node	纯函数：读 state，返回增量	add_node("name", fn)
Edge	无条件转移或条件转移	add_edge / add_conditional_edges

最容易被写错的是 Node 的语义：节点不改状态，只返回"要合并进状态的增量"。合并规则由 reducer 决定，所以同一份代码既能表达"追加一条消息"，也能表达"覆盖当前步骤"。

---

3. State 与 reducer：合并语义写在这里

import operator
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # 追加 + 按 id 去重
    steps: Annotated[int, operator.add]       # 累加
    query: str                                # 默认覆盖
    evidence: Annotated[list, operator.add]   # 拼接

写法	合并语义	适合的数据
不加 Annotated	覆盖：后写的盖掉先写的	标量、当前阶段、最新结论
Annotated[list, add_messages]	追加并按消息 id 去重	对话历史、工具消息
Annotated[list, operator.add]	直接拼接	步骤记录、证据清单
Annotated[X, my_merge]	自定义 (left, right) -> new	计数器取最大、按 key 合并 dict

一个必须记住的报错：两个并行分支同时写同一个没有 reducer 的字段，会抛 InvalidUpdateError（并发更新）。解决方式只有两种——给字段加 reducer，或者让每个分支只写自己的字段。

---

4. 图骨架：循环就写在边上

from langgraph.graph import StateGraph, START, END

def planner(state: AgentState) -> dict:
    return {"messages": [("ai", "下一步计划：调用查询工具")], "steps": 1}

def call_tools(state: AgentState) -> dict:
    return {"messages": [("tool", run_tool(state["query"]))]}

def route(state: AgentState) -> str:
    if state["steps"] >= 6:
        return "end"
    return "tools" if needs_tool(state) else "end"

g = StateGraph(AgentState)
g.add_node("planner", planner)
g.add_node("tools", call_tools)
g.add_edge(START, "planner")
g.add_conditional_edges("planner", route, {"tools": "tools", "end": END})
g.add_edge("tools", "planner")          # 这条边构成循环
app = g.compile()

路由函数只做判定、返回已注册的标签，不要在里面调模型或写状态——它可能被调多次，且不在 checkpoint 的覆盖范围内。

---

5. compile() 与 checkpoint

from langgraph.checkpoint.memory import MemorySaver

app = g.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "user-42"}}

app.invoke({"query": "排查 502"}, cfg)
app.invoke({"messages": [("human", "继续")]}, cfg)   # 自动带上同一线程的历史

后端	包	存哪	适合
MemorySaver	内置	进程内存	开发、单测
SqliteSaver	langgraph-checkpoint-sqlite	单个 db 文件	单机小服务
PostgresSaver	langgraph-checkpoint-postgres	数据库	生产、多实例

thread_id 是会话主键：同一个 id 的多次 invoke 共享状态，换 id 等于开新会话。开发时随手用 MemorySaver 没问题，上线必须换掉——重启即丢的状态在排障时等于没有。

---

6. interrupt：人工审批怎么写

from langgraph.types import interrupt, Command

def risky_restart(state: AgentState) -> dict:
    decision = interrupt({"ask": "是否执行重启？", "target": state["query"]})
    if decision != "approve":
        return {"messages": [("ai", "操作已取消")]}
    return {"messages": [("ai", "已执行重启")]}

app.invoke({"query": "restart api"}, cfg)          # 停在 interrupt
app.invoke(Command(resume="approve"), cfg)         # 从断点继续

需求	做法	注意
暂停等人	interrupt(payload)	payload 以 __interrupt__ 出现在返回值里
恢复执行	Command(resume=value)	必须带同一个 thread_id
暂停期间改状态	app.update_state(cfg, {...})	改完再 resume，顺序别反
超时自动放行	外部定时器负责 resume	不要在 interrupt 前后 sleep

interrupt 之所以能这么写，是因为暂停点被存进了 checkpoint：人工审批不占进程，进程可以重启，恢复时读 checkpoint 即可。

---

7. Send：并行分发与汇聚

from langgraph.types import Send

def fanout(state: AgentState):
    return [Send("worker", {"query": q}) for q in state["shards"]]

g.add_conditional_edges("split", fanout, ["worker"])
g.add_edge("worker", "collect")     # 所有 worker 完成后才走 collect

Send 返回的是"要运行的任务列表"，每个任务带自己的输入。下游 collect 只有在全部 worker 结束后才执行，所以它天然是汇聚点——但结果必须用带 reducer 的字段（如 operator.add）接，否则并发写会冲突。

---

8. 子图复用

方式	写法	适用
编译后的图直接当节点	add_node("sub", sub_compiled)	子图与父图状态 schema 兼容
包一层适配节点	手动做入参转换 / 出参映射	子图有独立状态结构

子图是"复用一段稳定流程"的正确工具，比复制粘贴节点强得多。但要注意 checkpoint 的归属：子图默认共享父图的 config，跨线程复用子图时要显式传自己的 thread_id，否则两处状态会互相污染。

---

9. 状态爆炸的预防

症状	原因	处理
每轮 token 越来越多	messages 无限追加	加摘要节点，把早期消息压缩成一条
State 字段几十个	什么都往里塞	只留跨节点真正共享的字段
checkpoint 体积暴涨	大对象直接进状态	大对象放外部存储，状态只留 id
序列化报错	状态里有不可序列化对象	只放 dict / list / str / 数字
并行分支互相覆盖	缺 reducer	加 operator.add 或拆字段

原则一句话：状态是"节点之间要传的话"，不是"这次的运行时快照"。 任何能从别处重算出来的东西都不该进状态。

---

10. 与 CrewAI 的对照

维度	LangGraph	CrewAI
表达方式	节点 + 边，命令式	角色 + 任务 + 流程，声明式
控制流	写在代码里	由 Process 与模型决定
人工介入	interrupt 原生	需自己包一层
可回放	checkpoint + thread_id	重跑
适合	审批流、强状态机、需审计	角色分工清晰的多步生产

---

11. 排错

现象	原因	处理
InvalidUpdateError	并行分支写同一无 reducer 字段	加 reducer 或拆字段
图一进去就到 END	条件边标签与映射键不匹配	路由返回值必须等于映射的键
恢复后重新跑了一遍	thread_id 变了	用同一个 config
interrupt 之后的节点不执行	传的是新 init 而不是 Command(resume=)	恢复必须走 Command
内存一直涨	MemorySaver + 长会话	换数据库后端，或定期清线程
子图与父图状态互串	复用同一 thread_id	给子图独立线程命名空间

---

12. 边界

场景	建议	理由
直线式、无回头的流水线	用普通链式编排	图带来的是负担不是能力
需要分支、循环、重试	StateGraph	这正是它的主场
需要人工审批与审计	StateGraph + 数据库 checkpoint	暂停点必须落盘
一次性的脚本	不用	装框架的时间超过省下的时间
纯多角色内容生产	更强的角色抽象更合适	图的优势在控制流不在分工

---

13. 自测

1. 什么样的流程"必须"用图而不是链？给出两个判据。
2. 节点函数应该返回整个 state，还是只返回增量？为什么？
3. 并行分支写同一个没有 reducer 的字段会发生什么？
4. thread_id 在 checkpointer 里的角色是什么？
5. interrupt 暂停期间，进程能不能重启？为什么？
6. 你的 checkpoint 体积每天翻倍，最先怀疑哪个字段？