LangGraph 核心机制 — 图引擎是怎么运转的
这篇讲什么：LangGraph 底层四个关键机制——Pregel 执行模型、State 与 Reducer、Checkpointer 持久化、Interrupt 人机协作。 读完你能回答：图是「怎么跑起来」的？并行节点同时写状态怎么办？崩溃后凭什么能续跑？人工审批怎么插进自动流程？
机制一：Pregel 执行模型 —— 图按「拍子」运转
没有它会怎样：图里有循环、有并行分支，「先执行谁、后执行谁、什么时候算一轮结束」如果没有一个明确的执行模型，每个框架使用者都要自己发明一套调度规则——并发 bug 和死循环会层出不穷。
方案：LangGraph 的运行时叫 Pregel（致敬 Google 同名论文），采用 BSP（整体同步并行） 模型，把执行切成一拍一拍的 super-step（超级步）：
- 每个 super-step 内，所有「收到新消息」的节点并行执行
- 节点执行时只读输入、只写输出，彼此不直接调用
- 本步所有节点都跑完（全局屏障），写入生效，进入下一拍
- 没有任何节点被激活时，图结束
两个推论很关键：
- 节点间靠消息传递，不靠函数调用——节点在 super-step N 写入 channel，订阅它的节点在 super-step N+1 被「收件箱」触发。这就是循环边能安全存在的原因：回边不会栈溢出，只是「再排进下一拍」
- 屏障即存档点——super-step 之间的全局屏障是天然的一致性时刻，Checkpointer（机制三）就挂在这里
一句话记住：Pregel = 打拍子跑图：一拍内并行、拍间同步、靠收件箱触发下一拍。
机制二：State 与 Reducer —— 并行写入怎么不打架
没有它会怎样：super-step 内多个节点并行跑，如果两个节点同时返回 {"messages": [...]}，State 里到底听谁的？直接覆盖会丢数据，直接追加又无法表达「修改某条消息」。
方案：给 State 的每个字段声明一个 Reducer（合并函数）。节点返回增量，框架用 reducer 把增量合并进旧值：
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # 有 reducer：合并
user_id: str # 无 reducer：覆盖（默认）- 无 reducer 的字段：后写覆盖先写（适合状态标记类字段）
- 同一 super-step 内并行写同一字段：由各字段的 reducer 合并出确定结果；字段没声明 reducer 又发生并行写，框架直接报
InvalidUpdateError——宁可报错，不猜语义 add_messages的精妙设计：默认追加（消息历史只增不减），但如果新消息与已有消息 id 相同则原地更新——这一条同时支持了「追加对话」和「修改/重试某条消息」两种场景
实测：两个并行节点分别返回 {"vals": ["x"]} 和 {"vals": ["y"]}，字段声明 Annotated[list, operator.add]，结果是 ['x', 'y']——合并行为完全确定。
一句话记住：节点只交增量，合并看 reducer；add_messages = 追加为主、同 id 更新。
机制三：Checkpointer —— 崩溃续跑与多轮会话的根基
没有它会怎样：Agent 跑到第 10 轮崩了，所有上下文在内存里灰飞烟灭，只能从头再来；用户关掉页面明天再聊，Agent 完全失忆。没有持久化，Agent 只能是「一次性脚本」。
方案：在每个 super-step 边界（机制一的全局屏障处）把完整 State 落盘，称为 checkpoint。每次调用时带上 thread_id，同一个 thread 的历史 checkpoint 串成一条时间线：
from langgraph.checkpoint.memory import InMemorySaver
graph = builder.compile(checkpointer=InMemorySaver()) # 编译时挂上存档器
config = {"configurable": {"thread_id": "user-123"}} # 调用时带上会话 id
graph.invoke({"messages": [("user", "你好")]}, config) # 第一轮
graph.invoke({"messages": [("user", "接着聊")]}, config) # 自动恢复历史，记得"你好"这套设计带来的能力远超「防丢数据」：
| 能力 | 怎么来的 |
|---|---|
| 崩溃恢复 | 从最后一个 checkpoint 重放即可续跑 |
| 多轮会话记忆 | 同一 thread_id 自动加载历史 State |
| 时间旅行 | checkpoint 是一条历史链，可回到任意一步分叉重跑 |
| 人机协作 | interrupt 暂停的本质 = 在 checkpoint 上挂起等待（机制四） |
生产环境换存储后端即可，接口不变：InMemorySaver（开发）→ SqliteSaver（langgraph-checkpoint-sqlite 包）→ PostgresSaver（langgraph-checkpoint-postgres 包）。
是不是有 event-sourcing 的味道？对比 pi 拆解 · 会话持久化 的「JSONL 追加 + 重放重建」——不同语言生态，同一个思想：状态可重放，系统就敢崩。
一句话记住：每拍边界存一次档，thread_id 串起一条时间线；能重放，就敢崩。
机制四：Interrupt —— 把「人工审批」插进自动流程
没有它会怎样：Agent 要执行危险操作（删数据、发邮件、下单），你想让它「先问我一声」。没有框架支持，你得在节点里手写轮询、外部状态标记、人工回调……审批逻辑和业务逻辑缠成一团。
方案：在节点里调用一个函数 interrupt()——图当场暂停并把 State 存档（复用机制三的 checkpoint），等外部用 Command(resume=...) 送入人类的决定，图从暂停处接着跑：
from langgraph.types import interrupt, Command
def approval_node(state):
# 首次执行到这里：图暂停，payload 透传给调用方
answer = interrupt({"question": "确认删除这 3 个文件吗？"})
# 恢复后：interrupt() 的返回值就是人类传入的 resume 值
return {"approved": answer == "yes"}config = {"configurable": {"thread_id": "t-1"}}
result = graph.invoke(input, config) # 跑到 interrupt 停下，result 含 __interrupt__
# ……把问题展示给用户，拿到答复……
graph.invoke(Command(resume="yes"), config) # 从暂停处继续三个实现细节，面试常追问：
- 暂停不是挂起线程——是「抛异常 + 落盘 checkpoint」，进程可以退出，恢复时重新加载
- 恢复后节点从头重跑——
interrupt()之前的代码会再执行一遍，所以 interrupt 要放在节点开头、之前不要有副作用 - 必须有 checkpointer——没有存档就无从恢复，
compile(checkpointer=...)是 interrupt 的前提
另有静态断点
interrupt_before/interrupt_after（编译或调用时指定在哪些节点前/后暂停），定位是调试与环外审批，与动态interrupt()并存。
一句话记住：interrupt = 抛异常存档等命令，resume = 从 checkpoint 原地复活。
速记卡
| 机制 | 一句话 | 考点 |
|---|---|---|
| Pregel | 打拍子跑图：拍内并行、拍间同步、收件箱触发 | super-step 是什么、循环为什么不爆栈 |
| Reducer | 节点交增量，合并看 reducer | 并行写冲突怎么办、add_messages 的 id 语义 |
| Checkpointer | 每拍存档，thread_id 串时间线 | 断点续跑原理、生产用什么后端 |
| Interrupt | 暂停=存档，恢复=Command(resume=) | 为什么恢复后节点重跑、为什么需要 checkpointer |
总纲一句：LangGraph = Pregel 打拍子 + Reducer 管合并 + Checkpoint 敢崩溃 + Interrupt 能等人。