【LangGraph基础】Pregel 执行内幕：一次超步的 Plan / Execute / Update 三阶段

源码基准：langgraph 1.2.16。

---

三大步骤

• Plan 只计算“这一拍要跑哪些任务”，不执行节点，也不修改状态。
• Execute 并行执行节点，结果只写入 checkpoint_pending_writes，不直接修改 channel。
• Update 统一把 writes apply 到 channel，得到新的 updated_channels，然后保存 checkpoint。
• 第 N 拍产生的 writes，在第 N+1 拍才可见；条件边的 route() 在第 N 拍执行，目标节点在第 N+1 拍才被唤醒。
• 三阶段的两个关键边界是 **put_writes()**（Execute → Update）和 **apply_writes()**（Update 真正改状态）。

---

1. 为什么是三大步骤？

Pregel 采用 Bulk Synchronous Parallel（BSP） 模型。一次图执行由若干 超步（superstep） 组成，每个超步内：

1. 同一批任务可以并行执行；
2. 节点之间不能看到彼此在当前超步的写入；
3. 所有任务结束后，统一应用写入，再进入下一超步。

这带来一个反直觉但非常重要的现象：

节点 A 写入了 state，并不代表同一超步内的节点 B 能读到这个值。 这些写入要先经过 Update 阶段进入 channel，下一拍的 Plan 才会看到它们。

因此，LangGraph 的执行引擎把一次超步拆成三个职责清晰的阶段：

flowchart LR
  subgraph PLAN["Plan · 计算任务"]
    direction TB
    P0["checkpoint<br/>updated_channels<br/>trigger_to_nodes"] --> P1["PregelLoop.tick()"]
    P1 --> P2["prepare_next_tasks()"]
    P2 --> P3["tasks: PUSH + PULL"]
  end

  subgraph EXEC["Execute · 并行执行"]
    direction TB
    E1["PregelRunner.tick()"] --> E2["run_with_retry()"]
    E2 --> E3["task.proc.invoke()"]
    E3 --> E4["commit() → put_writes()"]
    E4 --> E5["checkpoint_pending_writes"]
  end

  subgraph UPDATE["Update · 应用写入"]
    direction TB
    U1["after_tick()"] --> U2["apply_writes()"]
    U2 --> U3["channels.update()"]
    U3 --> U4["updated_channels"]
    U4 --> U5["_put_checkpoint()"]
  end

  P3 --> E1
  E5 --> U1
  U4 -.->|下一超步 N+1| P1

  classDef plan fill:#eff6ff,stroke:#2563eb,color:#1e3a8a;
  classDef exec fill:#fffbeb,stroke:#d97706,color:#92400e;
  classDef upd fill:#ecfdf5,stroke:#059669,color:#065f46;
  class P0,P1,P2,P3 plan;
  class E1,E2,E3,E4,E5 exec;
  class U1,U2,U3,U4,U5 upd;

三阶段的分工可以压缩成一张表：

阶段	输入	输出	是否修改状态
Plan	checkpoint、channels、上一拍的 updated_channels	本拍任务集合	否
Execute	任务集合	pending writes	否（只写 buffer）
Update	pending writes	新 channels、updated_channels、checkpoint	是

---

2. 入口与输入超步：从 invoke() 到 input checkpoint

一次同步执行的入口链路大致是：

Pregel.invoke()
  → Pregel.stream()
    → SyncPregelLoop.__enter__()
      → checkpointer.get_tuple()          # 取最新 checkpoint
      → channels_from_checkpoint()        # 恢复 channels
      → PregelLoop._first()               # 处理输入 / resume / time-travel
        → map_input()                     # 输入 → (channel, value)
        → apply_writes()                  # 先应用输入
        → _put_checkpoint("input")        # 保存 input checkpoint
    → while loop.tick():                  # 进入超步循环

对应源码：

位置	作用
pregel/main.py:3924	Pregel.invoke()
pregel/main.py:2748	Pregel.stream()，创建 loop 和 runner
pregel/_loop.py:1805	SyncPregelLoop.__enter__()
checkpoint/memory/__init__.py:239	InMemorySaver.get_tuple()
pregel/_checkpoint.py:458	channels_from_checkpoint()
pregel/_loop.py:915	PregelLoop._first()
pregel/_io.py:81	map_input()
pregel/_loop.py:1179	_put_checkpoint({"source": "input"})

核心循环在 Pregel.stream() 中，结构非常清晰：

# langgraph/pregel/main.py:2990-3079（简化）
with SyncPregelLoop(input, config=config, ...) as loop:
    runner = PregelRunner(
        submit=loop.submit,
        put_writes=loop.put_writes,
        ...
    )

    while loop.tick():
        for task in loop.match_cached_writes():
            loop.put_writes(task.id, task.writes, cached=True)

        for _ in runner.tick(
            [t for t in loop.tasks.values() if not t.writes],
            timeout=self.step_timeout,
            schedule_task=loop.accept_push,
        ):
            ...  # 流式输出

        loop.after_tick()

        if durability_ == "sync":
            loop._put_checkpoint_fut.result()

注意：输入本身也是一次 apply_writes()。 _first() 会把输入映射成 writes，先应用到 channel，再保存 source="input" 的 checkpoint。 所以第一次 tick() 看到的 updated_channels，已经包含了输入带来的通道变化。

---

3. Plan：这一拍要跑哪些任务？

Plan 阶段的核心是 PregelLoop.tick()。

# langgraph/pregel/_loop.py:650（简化）
def tick(self) -> bool:
    if self.step > self.stop:
        self.status = "out_of_steps"
        return False

    self.tasks = prepare_next_tasks(
        self.checkpoint,
        self.checkpoint_pending_writes,
        self.nodes,
        self.channels,
        ...
        updated_channels=self.updated_channels,
        trigger_to_nodes=self.trigger_to_nodes,
    )

    if not self.tasks:
        self.status = "done"
        return False

    ...
    return True

它做四件事：

1. 检查递归/步数上限；
2. 调用 prepare_next_tasks() 生成本拍任务；
3. 没有任务则 status="done"，结束循环；
4. 处理 resume 恢复、interrupt_before 等边界，然后返回 True。

3.1 prepare_next_tasks() 如何选出节点？

prepare_next_tasks()（pregel/_algo.py:391）会合并两类任务：

• PUSH：Send、call() 等显式派发的任务；
• PULL：被 channel 更新触发的普通节点。

PULL 的优化非常关键：它不会遍历所有节点，而是用上一拍的 updated_channels 去查 trigger_to_nodes：

# langgraph/pregel/_algo.py:474（简化）
if updated_channels and trigger_to_nodes:
    triggered_nodes = set()
    for channel in updated_channels:
        if node_ids := trigger_to_nodes.get(channel):
            triggered_nodes.update(node_ids)
    candidate_nodes = sorted(triggered_nodes)

也就是说：

updated_channels + trigger_to_nodes
        ↓
被唤醒的节点集合
        ↓
prepare_single_task() → PregelExecutableTask

flowchart TD
  A["prepare_next_tasks()"] --> B["消费 TASKS 通道（若有 Send / call）"]
  B --> B1["prepare_single_task(PUSH, ...)"]
  B1 --> C["updated_channels × trigger_to_nodes"]
  C --> D["候选节点集合（排序保证确定性）"]
  D --> E["prepare_single_task(PULL, name)"]
  E --> F["合并为 {task_id: PregelExecutableTask}"]
  B1 --> F

相关的关键函数：

函数	位置	作用
prepare_next_tasks()	pregel/_algo.py:391	合并 PUSH / PULL 任务
prepare_single_task()	pregel/_algo.py:523	生成单个任务，计算 task id
prepare_push_task_send()	pregel/_algo.py:937	处理 Send
prepare_push_task_functional()	pregel/_algo.py:799	处理 call() 等函数式任务
_xxhash_str()	pregel/_algo.py:1403	任务 id 生成（checkpoint v2）
_emit("tasks", ...)	pregel/_loop.py:728	发出 tasks 调试事件

3.2 条件边为什么能在一个超步内触发多个节点？

如果条件边的 path 返回列表：

def route(state):
    return ["weather", "news"]

那么在同一次节点执行中，条件边 writer 会同时写入 branch:to:weather 和 branch:to:news。 Update 阶段这两个 channel 都进入 updated_channels，下一拍 Plan 通过 trigger_to_nodes 同时唤醒两个节点，于是它们在同一超步内并行执行。

---

4. Execute：并发执行，只写 buffer

Execute 阶段由 PregelRunner 负责。同步版本是 tick()，异步版本是 atick()。

4.1 调度：单任务快路径 vs 多任务并发

PregelRunner.tick()（pregel/_runner.py:177）会：

1. 先 yield，把控制权交回调用方，便于流式输出；
2. 单任务且无 timeout / waiter 时，当前线程直接执行 run_with_retry()；
3. 多任务时通过 self.submit() 提交到线程池（sync）或事件循环（async）；
4. 用 FuturesDict 管理 Future，并通过 concurrent.futures.wait(FIRST_COMPLETED) 或 asyncio.wait(FIRST_COMPLETED) 等待任务完成；
5. 每完成一个任务就 yield，让调用方输出流式块。

PregelRunner.tick(tasks)
  ├─ 单任务快路径 → run_with_retry(task)
  └─ 多任务 → submit(run_with_retry, task)
       ├─ BackgroundExecutor.submit()       # sync
       └─ AsyncBackgroundExecutor.submit()  # async

4.2 节点执行：task.proc.invoke()

run_with_retry()（pregel/_retry.py:573）负责重试和超时控制，最终调用：

# langgraph/pregel/_retry.py:617
return task.proc.invoke(task.input, config)

task.proc 是一个 RunnableSeq，它按顺序执行：

1. 用户节点函数；
2. 状态 writer（把返回值写入 state channel）；
3. 条件边 writer（BranchSpec._route() → _finish() → ChannelWrite.do_write()）。

flowchart LR
  A["节点函数"] --> B["状态 writer<br/>ChannelWrite"]
  B --> C{"有条件边？"}
  C -->|是| D["BranchSpec._route()"]
  D --> E["_finish()<br/>ends 映射"]
  E --> F["ChannelWrite.do_write()<br/>写 branch:to:target"]
  C -->|否| G["结束"]

对应源码：

函数 / 类	位置
PregelNode.invoke()	pregel/_read.py:261
BranchSpec._route()	graph/_branch.py:146
BranchSpec._finish()	graph/_branch.py:192
ChannelWrite.do_write()	pregel/_write.py:106

4.3 动态任务：call() / Send

节点执行期间，还可以通过 call()、Command、Send 产生新的 PUSH 任务：

_call() / _acall()                  _runner.py:707 / 796
  → schedule_task()
    → PregelLoop.accept_push()      _loop.py:1738 / 1994
      → prepare_single_task(PUSH)   _loop.py:609
        → 新任务插入当前 runner 的 Future 集合

新任务可能在当前超步内继续执行，也可能被安排到下一拍，具体取决于 __next_tick__ 等标志。

4.4 写回：commit() → put_writes()

任务完成时，FuturesDict 的回调会调用 commit()（pregel/_runner.py:575）：

• 成功：把 writes 交给 put_writes()；
• 异常：追加 ERROR，必要时路由到 error_handler；
• 中断：追加 INTERRUPT；
• 取消：追加 ERROR。

put_writes()（pregel/_loop.py:461 / sync 覆盖 pregel/_loop.py:1783）做的是：

1. 去重特殊 channel 的写入；
2. ensure_message_ids()；
3. 把 writes 追加到 checkpoint_pending_writes；
4. 异步调用 checkpointer_put_writes()，提前持久化 pending writes；
5. 发出 writes 流式事件、更新 cache。

关键边界：put_writes() 只写 pending buffer，不会修改 channel。 节点之间之所以互相不可见，就是因为大家都在往 buffer 里写，而不是直接改 state。

sequenceDiagram
  autonumber
  participant Caller as Pregel.stream
  participant PLoop as PregelLoop
  participant Runner as PregelRunner
  participant Task as task.proc.invoke
  participant CP as checkpointer

  Caller->>PLoop: loop.tick()
  PLoop-->>Caller: True / False
  Caller->>PLoop: loop.match_cached_writes()
  Caller->>Runner: runner.tick(tasks)
  Runner->>Task: run_with_retry(task)
  Task-->>Runner: writes
  Runner->>PLoop: commit() → put_writes()
  PLoop->>CP: checkpointer_put_writes()（后台）
  Caller->>PLoop: loop.after_tick()
  PLoop->>PLoop: apply_writes() → updated_channels
  PLoop->>CP: _put_checkpoint() → put()
  Caller->>PLoop: 下一拍 loop.tick()

4.5 重试、超时与错误兜底

这些能力都发生在 Execute 阶段：

• RetryPolicy：由 run_with_retry() / arun_with_retry() 驱动；
• run_timeout / idle_timeout：由 _start_timed_attempt()、_arun_with_timeout() 控制；
• error_handler：重试耗尽后，由 commit() / schedule_error_handler() 路由到兜底节点。

对应源码：

能力	位置
重试判定 _should_retry_on()	pregel/_retry.py:834
超时启动 _start_timed_attempt()	pregel/_retry.py:343
异步超时 _arun_with_timeout()	pregel/_retry.py:422
异常路由 schedule_error_handler()	pregel/_loop.py:640

---

5. Update：统一应用写入并保存 checkpoint

Update 阶段从 PregelLoop.after_tick()（pregel/_loop.py:737）开始。

5.1 apply_writes()：状态真正改变的地方

apply_writes()（pregel/_algo.py:233）的主要步骤：

1. 按 task path 排序，保证 apply 顺序确定；
2. 更新 versions_seen；
3. 对任务读取过的 channel 调用 consume()；
4. 按 channel 分组 writes；
5. 调用 channels[chan].update(vals)；
6. 对未更新的 channel 发送 EMPTY_SEQ；
7. 如果没有可触发节点，调用 channels[chan].finish()；
8. 返回 updated_channels。

flowchart TD
  A["after_tick(): 收集本拍 writes"] --> B["按 task path 排序"]
  B --> C["更新 versions_seen"]
  C --> D["consume() 被读取的通道"]
  D --> E["按 channel 分组 writes"]
  E --> F["channels[chan].update(vals)"]
  F --> G["未更新通道 update(EMPTY_SEQ)"]
  G --> H{"updated_channels 与 trigger_to_nodes 有交集？"}
  H -->|否| I["channel.finish()"]
  H -->|是| J["保留 updated_channels"]
  I --> K["返回 updated_channels"]
  J --> K
  K --> L["_put_checkpoint() → create_checkpoint()"]
  L --> M["checkpointer.put()"]

对应源码：

步骤	位置
收集 writes	pregel/_loop.py:739
apply_writes()	pregel/_algo.py:233
versions_seen 更新	pregel/_algo.py:262
consume()	pregel/_algo.py:290
channel.update(vals)	pregel/_algo.py:318
EMPTY_SEQ	pregel/_algo.py:328
channel.finish()	pregel/_algo.py:337

5.2 values 输出与 buffer 清理

apply_writes() 返回后，after_tick() 会：

• 如果 output channel 被更新，发出 values 流；
• 清空 checkpoint_pending_writes；
• 重置 is_replaying；
• 调用 _put_checkpoint({"source": "loop"})。

5.3 checkpoint 保存链路

_put_checkpoint()                       _loop.py:1231
  → create_checkpoint()                 _checkpoint.py:268
  → submit(_checkpointer_put_after_previous)
      → checkpointer.put()              checkpoint/memory/__init__.py:430

_checkpointer_put_after_previous()（pregel/_loop.py:1706 / async 1962）会等待上一个 put 和 delta writes 完成，保证 checkpoint 按顺序保存。

保存默认是异步的；只有 durability="sync" 时，主循环才会在拍末等待：

# langgraph/pregel/main.py:3079
if durability_ == "sync":
    loop._put_checkpoint_fut.result()

5.4 interrupt_after

Update 完成后，after_tick() 会检查 interrupt_after：

# langgraph/pregel/_loop.py:805（简化）
if self.interrupt_after and should_interrupt(
    self.checkpoint, self.interrupt_after, self.tasks.values()
):
    self.status = "interrupt_after"
    raise GraphInterrupt()

同时清理 CONFIG_KEY_RESUMING，避免下一拍误判为 resume。

---

6. 循环与退出

一个超步结束后，控制流并不是简单的“继续下一拍”，而是回到 while loop.tick()，由下一次 tick() 重新判断：

flowchart TD
  A["while loop.tick()"] --> B{"tasks 为空？"}
  B -->|是| C["status = done<br/>return False"]
  C --> D["退出循环 → _suppress_interrupt()"]
  D --> E["最终 checkpoint + output"]
  B -->|否| F["match_cached_writes()"]
  F --> G["过滤已有 writes 的任务"]
  G --> H["PregelRunner.tick()"]
  H --> I["after_tick() → apply_writes()"]
  I --> J["_put_checkpoint()"]
  J --> K{"durability == sync？"}
  K -->|是| L["等待 put 完成"]
  K -->|否| A
  L --> A
  I -.->|updated_channels 驱动下一拍| A

退出路径：

• tick() 返回 False，status="done"；
• 上下文退出时 _suppress_interrupt()（pregel/_loop.py:1494）保存最终 checkpoint、pending writes，并读取最终 output；
• 如果步数耗尽，status="out_of_steps"，外部会抛 GraphRecursionError。

---

7. 条件边：执行在 N 拍，生效在 N+1 拍

这是理解 Pregel 最关键、也最容易踩坑的一点。

sequenceDiagram
  autonumber
  participant N as Execute（第 N 拍）
  participant U as Update（第 N 拍）
  participant P as Plan（第 N+1 拍）

  N->>N: route() 返回 "weather"
  N->>N: BranchSpec._finish() 写 branch:to:weather
  Note over N: 只进入 pending buffer，channel 未变
  N->>U: commit() → put_writes()
  U->>U: apply_writes()<br/>branch:to:weather 进入 updated_channels
  U->>P: _put_checkpoint() → 下一拍
  P->>P: trigger_to_nodes["branch:to:weather"]<br/>→ 生成 weather 任务

对应代码路径（可以通过 pregel/conditional_edge.py 的 traced_finish() 完整观察）：

task.proc.invoke()
  → RunnableSeq
    → BranchSpec._route()          graph/_branch.py:146
      → self.path.invoke(value)    # 你的 route() 在这里被调用
      → _finish()                  graph/_branch.py:192
        → ChannelWrite.do_write()  pregel/_write.py:106
          → 写 branch:to:weather

下一拍 Plan：
  updated_channels × trigger_to_nodes
    → trigger_to_nodes["branch:to:weather"]
      → prepare_single_task(PULL, "weather")

所以：

route() 的决策发生在第 N 拍的 Execute； route() 的结果生效发生在第 N 拍的 Update； 目标节点的实际执行发生在第 N+1 拍的 Execute。

理解这一点，就不会再问“为什么我在 route 之后立刻读 state 读不到”了。

---

8. 一个可运行的例子：planner → (weather ∥ news) → summarize

以本仓库的 pregel/debug_pregel.py 为例：

g = StateGraph(AgentState)
g.add_node("planner", planner)
g.add_node("weather", call_weather)
g.add_node("news", call_news)
g.add_node("summarize", summarize)

g.add_edge(START, "planner")
g.add_conditional_edges("planner", route, {
    "weather": "weather",
    "news": "news",
})
g.add_edge("weather", "summarize")
g.add_edge("news", "summarize")
g.add_edge("summarize", END)

其中：

def route(state):
    return ["weather", "news"]  # 返回列表 → 同一超步并行触发两个节点

执行时间线：

flowchart LR
  S0["输入超步<br/>messages=[user], token_count=1"] --> S1["第 1 拍<br/>planner"]
  S1 --> S2["第 2 拍<br/>weather ∥ news"]
  S2 --> S3["第 3 拍<br/>summarize"]
  S3 --> S4["第 4 拍<br/>无任务 → done"]
  S4 --> S5["最终 output"]

各拍的关键写入：

超步	计划任务	节点 writes	Update 后
输入超步	—	输入 writes	messages、token_count=1 进入 channels，保存 input checkpoint
第 1 拍	planner	plan、token_count=100、branch:to:weather、branch:to:news	token_count=101；两个 branch 进入 updated_channels
第 2 拍	weather、news	weather=sunny、token_count=50；news=market up、token_count=50	token_count=201；weather、news、branch:to:summarize 更新
第 3 拍	summarize	messages=[ai, "..."]	messages 更新；没有新的可触发节点
第 4 拍	无	—	status="done"，退出并保存最终 checkpoint

token_count 使用了 operator.add reducer，所以最终值是：

1 + 100 + 50 + 50 = 201

与 pregel/compiled_graph.py 的输出一致。

---

9. 源码定位表

阶段	关键函数	位置	作用
入口	Pregel.invoke()	pregel/main.py:3924	公开入口
入口	Pregel.stream()	pregel/main.py:2748	创建 loop / runner，驱动超步循环
初始化	SyncPregelLoop.__enter__()	pregel/_loop.py:1805	读取 checkpoint、恢复 channels
初始化	PregelLoop._first()	pregel/_loop.py:915	输入 / resume / time-travel
初始化	map_input()	pregel/_io.py:81	输入映射为 writes
Plan	PregelLoop.tick()	pregel/_loop.py:650	本拍 Plan 主体
Plan	prepare_next_tasks()	pregel/_algo.py:391	合并 PUSH / PULL 任务
Plan	prepare_single_task()	pregel/_algo.py:523	生成任务、计算 task id
Plan	trigger_to_nodes 反查	pregel/_algo.py:474	根据更新通道找候选节点
Plan	_reapply_writes_to_succeeded_nodes()	pregel/_loop.py:827	resume 时恢复已完成 writes
Plan	match_cached_writes()	pregel/_loop.py:1723	缓存命中直接复用 writes
Execute	PregelRunner.tick() / atick()	pregel/_runner.py:177 / 361	调度并发任务
Execute	run_with_retry() / arun_with_retry()	pregel/_retry.py:573 / 685	重试与超时
Execute	task.proc.invoke()	pregel/_retry.py:617	执行节点 RunnableSeq
Execute	PregelNode.invoke()	pregel/_read.py:261	调用节点函数
Execute	BranchSpec._route/_finish()	graph/_branch.py:146 / 192	条件边路由
Execute	ChannelWrite.do_write()	pregel/_write.py:106	写 branch:to:*
Execute	_call() / _acall()	pregel/_runner.py:707 / 796	动态任务
Execute	accept_push()	pregel/_loop.py:1738 / 1994	接收 PUSH 任务
Execute	commit()	pregel/_runner.py:575	Future 完成回调
Execute	put_writes()	pregel/_loop.py:461 / 1783	写 pending buffer
Execute	_panic_or_proceed()	pregel/_runner.py:657	收敛异常 / 超时
Update	after_tick()	pregel/_loop.py:737	本拍 Update 主体
Update	apply_writes()	pregel/_algo.py:233	writes → channels
Update	_put_checkpoint()	pregel/_loop.py:1231	生成并保存 checkpoint
Update	create_checkpoint()	pregel/_checkpoint.py:268	构造 checkpoint
Update	checkpointer.put()	checkpoint/memory/__init__.py:430	持久化
退出	_suppress_interrupt()	pregel/_loop.py:1494	最终 checkpoint / pending writes
退出	_put_pending_writes()	pregel/_loop.py:559	退出时补存 pending writes

---

完整的、可交互的流程图：

Pregel执行流程演示

包含：

• 三泳道执行图：Plan / Execute / Update，27 个关键节点，全部标注源码位置；
• 核心步骤路程图：10 个里程碑，覆盖 P1–P9、E1–E8、U1–U10；
• 点击核心步骤里的 P1 / E3 / U9 等编号，可直接定位并高亮上方泳道节点；
• 支持 30%–200% 缩放、适应宽度、Ctrl / ⌘ + 滚轮 缩放。

---

10. 最后的记忆清单

1. Plan 只读，不执行，不改状态。
2. Execute 只写 pending buffer，不直接改 channel。
3. Update 才 apply writes，产生 updated_channels 和 checkpoint。
4. N 拍的写入在 N+1 拍才可见；条件边执行与生效跨一拍。
5. put_writes() 和 apply_writes() 是两个最重要的阶段边界。

把这五条记住，再回头看 PregelLoop.tick()、PregelRunner.tick() 和 PregelLoop.after_tick()，整个执行过程就会非常清晰。