LangGraph 1.2.16 · Pregel / BSP

Pregel 执行过程 · Plan / Execute / Update 泳道图

一个超步（superstep）的完整生命周期。重点看三件事：Plan 只计算任务、 Execute 只把结果写进 pending buffer、Update 才真正 apply 到 channel 并生成 checkpoint。 图中每个节点都标注了 langgraph 源码位置，以及本仓库 pregel/ 示例中的对应观察点。

源码基准 langgraph 1.2.16 路径前缀 .venv/lib/python3.10/site-packages/langgraph/ 示例 pregel/debug_pregel.py 同步 / 异步双实现均标注
01 · PLAN

计算本拍要跑什么

读取 checkpoint、channels、上一拍的 updated_channels，通过 trigger_to_nodes 反查被唤醒的节点，生成 PregelExecutableTask。不执行节点，不改状态。

PregelLoop.tick()
02 · EXECUTE

并行执行并收集写入

线程池或事件循环并发执行任务。节点返回值只写入 checkpoint_pending_writes，同一超步内节点之间互不可见。

PregelRunner.tick()
03 · UPDATE

统一应用写入

把本拍所有 writes 排序后 apply 到 channel，得到新的 updated_channels，保存 checkpoint，然后进入下一拍。

PregelLoop.after_tick()
Plan 泳道 Execute 泳道 Update 泳道 同超步顺序 跨超步循环 syncasync双实现 file:line源码位置（悬停看完整路径）
LANE 01
Plan
计算任务
只读 checkpoint / channels
产出 tasks
不改 state
P1入口

Pregel.invoke()

公开入口；内部转调 stream()，由流式生成器驱动整个超步循环。

pregel/main.py:3924 → pregel/main.py:2748
P2初始化

SyncPregelLoop.__enter__()

读取该 thread 最新 checkpoint，反序列化并恢复 channels；建立后续拍次的起点。

pregel/_loop.py:1805
  • get_tuple()checkpoint/memory/__init__.py:239
  • channels_from_checkpoint()pregel/_checkpoint.py:458
  • AsyncPregelLoop.__aenter__()pregel/_loop.py:2061
P3输入超步

PregelLoop._first()

区分新输入 / resume / time-travel；把输入映射成 channel writes，并保存 input checkpoint。

pregel/_loop.py:915
  • map_input()pregel/_io.py:81
  • prepare_next_tasks()（丢弃未完成任务）pregel/_loop.py:1113
  • apply_writes()（应用输入）pregel/_loop.py:1134
  • _put_checkpoint("input")pregel/_loop.py:1179
P4超步循环

while loop.tick():

BSP 主循环。第 N 拍的 channel 更新只在第 N+1 拍可见；循环由 checkpoint 与 writes 驱动。

pregel/main.py:3055
  • match_cached_writes()pregel/main.py:3056
  • runner.tick(tasks, timeout=step_timeout)pregel/main.py:3058
  • loop.after_tick()pregel/main.py:3075
P5Plan 主体

PregelLoop.tick()

检查步数上限；调用任务准备函数；无任务则 status="done" 并结束循环。

pregel/_loop.py:650
  • if self.step > self.stoppregel/_loop.py:658
  • prepare_next_tasks()pregel/_loop.py:663
  • if not self.tasks: status="done"pregel/_loop.py:704
  • _emit("tasks", map_debug_tasks)pregel/_loop.py:728
P6核心

prepare_next_tasks()

生成本拍任务集合：PUSH（Send / call）与 PULL（被 channel 触发）两类任务合并。

pregel/_algo.py:391 ← pregel/_loop.py:663
  • updated_channels × trigger_to_nodespregel/_algo.py:474
  • prepare_single_task()pregel/_algo.py:523
  • prepare_push_task_send()pregel/_algo.py:937
  • prepare_push_task_functional()pregel/_algo.py:799
  • _xxhash_str() 生成 task idpregel/_algo.py:1403
P7恢复 / 中断

恢复 writes · 路由 error_handler

resume 时把已完成任务的 writes 归还给内存 task；失败任务按映射转交给 error_handler，而不是重跑原节点。

pregel/_loop.py:827 + pregel/_loop.py:840
  • should_interrupt()pregel/_algo.py:156
  • interrupt_before → GraphInterruptpregel/_loop.py:714
  • schedule_error_handler()pregel/_loop.py:640
P8缓存 / 调试

match_cached_writes()

缓存命中的任务直接写回上次结果，不再执行节点；同时发出 tasks 调试事件。

pregel/_loop.py:1723 / pregel/_loop.py:1979
  • put_writes(..., cached=True)pregel/_loop.py:1783
  • map_debug_tasks()pregel/debug.py:44
P9交接

过滤已写任务 → Execute

已有 writes 的任务（缓存命中 / resume 恢复）跳过执行；剩余任务交给 runner。

pregel/main.py:3059
LANE 02
Execute
并行执行
节点并发执行
结果只进 pending buffer
不改 channel
E1调度

PregelRunner.tick()

提交本拍任务；先 yield 把控制权交回调用方，便于流式输出；再等待 Future。

pregel/_runner.py:177 / pregel/_runner.py:361
  • FuturesDict 管理并发 Futurepregel/_runner.py:76
  • self.submit()(run_with_retry, ...)pregel/_runner.py:261
  • BackgroundExecutor.submit()pregel/_executor.py:54
  • AsyncBackgroundExecutor.submit()pregel/_executor.py:142
E2重试 / 超时

run_with_retry()

单任务无超时可走当前线程快路径；否则提交到线程池 / 事件循环。失败按 RetryPolicy 指数退避。

pregel/_retry.py:573 / pregel/_retry.py:685
  • 单任务快路径pregel/_runner.py:204
  • _should_retry_on()pregel/_retry.py:834
  • _start_timed_attempt()pregel/_retry.py:343
  • _arun_with_timeout()pregel/_retry.py:422
E3节点执行

task.proc.invoke()

执行节点的 RunnableSeq：用户节点函数 → 状态 writer → 条件边 writer。节点函数返回值在这里产生。

pregel/_retry.py:617 / pregel/_retry.py:737
  • PregelNode.invoke()pregel/_read.py:261
  • BranchSpec._route() 条件边入口graph/_branch.py:146
  • BranchSpec._finish() ends 映射graph/_branch.py:192
  • ChannelWrite.do_write() 写 branch:to:*pregel/_write.py:106
E4可选 · 节点内

call() / Send 动态任务

节点执行期间可通过 call()、Command、Send 产生 PUSH 任务；新任务插入 runner 的 Future 集合。

pregel/_runner.py:707 / pregel/_runner.py:796
  • accept_push()pregel/_loop.py:1738
  • aaccept_push()pregel/_loop.py:1994
  • prepare_single_task((PUSH, ...))pregel/_loop.py:609
  • _emit("tasks", ...)pregel/_loop.py:631
E5完成回调

commit()

Future 完成回调。成功、异常、中断统一转成 writes 并交给 loop；异常可路由到 error_handler。

pregel/_runner.py:575
  • 成功 → put_writes(task.id, writes)pregel/_loop.py:1783
  • 异常 → 追加 ERRORpregel/_runner.py:598
  • 中断 → 追加 INTERRUPTpregel/_runner.py:589
E6阶段边界

PregelLoop.put_writes()

关键边界：只把 writes 追加到 checkpoint_pending_writes，不碰 channel；同时后台落盘 pending writes 并发流事件。

pregel/_loop.py:461 / pregel/_loop.py:1783
  • checkpoint_pending_writes.extend()pregel/_loop.py:514
  • checkpointer_put_writes() 后台pregel/_loop.py:530
  • output_writes() 流式事件pregel/_loop.py:1593
  • cache.setpregel/_loop.py:1793
E7收敛

_panic_or_proceed()

等待所有 Future；任一任务失败则取消其余任务并向上抛；全部成功则本拍执行结束。

pregel/_runner.py:657
  • _should_stop_others()pregel/_runner.py:618
  • concurrent.futures.wait()pregel/_runner.py:284
  • asyncio.wait(FIRST_COMPLETED)pregel/_runner.py:483
E8流式输出

_output()

调用方在每个任务完成时 yield 流式块；执行与输出解耦，慢任务不会阻塞已完成任务的结果。

pregel/main.py:4374 ← pregel/main.py:3064
LANE 03
Update
应用写入
apply writes → channels
更新 updated_channels
保存 checkpoint
U1Update 主体

PregelLoop.after_tick()

本拍执行结束。收集所有 task 的 writes，准备统一应用到 channel。

pregel/_loop.py:737
  • writes = [w for t in tasks for w in t.writes]pregel/_loop.py:739
  • apply_writes()pregel/_loop.py:761
U2状态在这里改变

apply_writes()

按 task path 排序后 apply；返回 updated_channels，它决定下一拍唤醒谁。

pregel/_algo.py:233
  • 更新 versions_seenpregel/_algo.py:262
  • channels[chan].consume()pregel/_algo.py:290
  • channels[chan].update(vals)pregel/_algo.py:318
  • EMPTY_SEQ 通知未更新通道pregel/_algo.py:328
  • channels[chan].finish()pregel/_algo.py:337
U3输出

_emit("values")

若 output channel 被更新，则按 map_output_values 产生 values 流。

pregel/_loop.py:774 → pregel/_io.py:100
U4清缓冲

清空 pending writes

writes 已并入 channel；清空 buffer，并结束 replay 状态。

pregel/_loop.py:799
  • checkpoint_pending_writes.clear()pregel/_loop.py:799
  • is_replaying = Falsepregel/_loop.py:801
U5持久化

_put_checkpoint()

生成新 checkpoint，并异步提交保存；下一拍不必等待落盘（durability 可调）。

pregel/_loop.py:803 → pregel/_loop.py:1231
  • create_checkpoint()pregel/_checkpoint.py:268
  • get_new_channel_versions()pregel/_utils.py:27
  • submit(_checkpointer_put_after_previous)pregel/_loop.py:1360
U6落盘

checkpointer.put()

串行化 checkpoint 保存：等待上一个 put 和 delta writes 完成，保证存档顺序。

pregel/_loop.py:1706 / pregel/_loop.py:1962
  • InMemorySaver.put()checkpoint/memory/__init__.py:430
  • BaseCheckpointSaver.put()checkpoint/base/__init__.py:301
U7中断

interrupt_after 检查

更新完成后检查 after 断点；清理 resume 标记，准备下一拍或退出。

pregel/_loop.py:805 – pregel/_loop.py:811
  • raise GraphInterruptpregel/_loop.py:809
  • pop(CONFIG_KEY_RESUMING)pregel/_loop.py:811
U8同步落盘

durability 同步等待

durability="sync" 时等 checkpoint 真正落盘再进入下一拍；async / exit 不等待。

pregel/main.py:3079
  • loop._put_checkpoint_fut.result()pregel/main.py:3079
  • await loop._put_checkpoint_fut（async）pregel/main.py:3551
U9循环判定

updated_channels → 下一拍？

把 updated_channels 交给下一拍 Plan：有可触发节点则继续；否则 tick() 返回 False，进入退出路径。

pregel/_loop.py:761 → pregel/main.py:3055
  • updated_channels × trigger_to_nodespregel/_algo.py:474
U10结束

最终 checkpoint / 输出

循环结束后上下文退出：补存最终 checkpoint 与 pending writes，读取最终 output；步数耗尽则抛 GraphRecursionError。

pregel/_loop.py:1494 → pregel/main.py:3093
  • _put_checkpoint(final)pregel/_loop.py:1510
  • _put_pending_writes()pregel/_loop.py:559
  • read_channels() 最终输出pregel/_read.py
可拖动底部横条或使用上方缩放控件查看完整超步 · 虚线为跨超步循环
CORE JOURNEY · 10 MILESTONES

一个超步的核心步骤

把上方 27 个细节点压缩成 10 个可执行里程碑。每张卡片给出关键调用链， 并列出它覆盖的泳道节点；点击编号可定位并高亮上方对应节点。

TIP点击 P1 / E3 / U9 定位节点

先算清楚，再执行

只读 checkpoint / channels，产出本拍任务；不改 state。

01
PLAN · 输入 入口

恢复 checkpoint 与输入

读取该 thread 的最新 checkpoint，恢复 channels；新输入 / resume 被映射成 channel writes，并保存 input checkpoint。

get_tuple → channels_from_checkpoint → _first → map_input
泳道映射
02
PLAN · 计算 核心

计算本拍任务

用上一拍的 updated_channels × trigger_to_nodes 反查被唤醒节点，合并 PUSH（Send / call）与 PULL（边触发）任务。

tick → prepare_next_tasks → prepare_single_task
泳道映射
03
PLAN · 复用 检查

缓存、恢复与中断检查

已完成任务复用 writes；失败任务路由到 error_handler；检查 interrupt_before，并过滤已有 writes 的任务。

_reapply_writes → match_cached_writes → should_interrupt
泳道映射

并发执行，只写 buffer

节点结果先落 pending writes，同一超步内互不可见。

04
EXECUTE · 调度 并发

调度本拍任务

把任务交给 PregelRunner；单任务走当前线程快路径，多任务提交到线程池（sync）或事件循环（async）。

PregelRunner.tick → run_with_retry / submit
泳道映射
05
EXECUTE · 执行 节点

执行节点与 writer

运行节点 RunnableSeq：节点函数 → 状态 writer → 条件边 writer；节点内还可以派发 call() / Send 动态任务。

task.proc.invoke → _call / accept_push
泳道映射
06
EXECUTE · 写回 阶段边界

写回 pending buffer

Future 完成回调 commit；成功、异常、中断统一转成 writes，追加到 checkpoint_pending_writes，不直接改 channel。

commit → PregelLoop.put_writes
泳道映射
07
EXECUTE · 收敛 输出

收敛与流式输出

等待所有 Future；任一任务失败则取消其余任务。每完成一个任务，调用方就 yield 一个流式块。

_panic_or_proceed → _output
泳道映射

统一应用，推进下一拍

writes 在这里才真正进入 channels，并生成新 checkpoint。

08
UPDATE · 应用 核心

apply writes 到 channels

把所有 writes 排序后统一 apply；更新 versions_seen / consume / finish，返回 updated_channels，并产生 values 输出。

after_tick → apply_writes → _emit(values)
泳道映射
09
UPDATE · 持久化 checkpoint

保存 checkpoint

清空 pending buffer，创建新 checkpoint 并异步落盘；durability="sync" 时等待保存完成。

_put_checkpoint → create_checkpoint → checkpointer.put
泳道映射
10
UPDATE · 循环 下一拍

下一拍或退出

updated_channels 交给下一拍 Plan；没有可触发节点则退出，保存最终 checkpoint 并读取 output。

updated_channels → while loop.tick() / _suppress_interrupt
泳道映射
↺
下一超步 N+1：Update 产生的 updated_channels 回到 Plan 的 while loop.tick()， 再经 trigger_to_nodes 唤醒下一批节点。这正是条件边“执行在 N 拍、生效在 N+1 拍”的原因。
→

关键不变量

INVARIANT 01

同一超步内 channel 不可变

Execute 阶段的节点写的都是 checkpoint_pending_writes，不会直接改 channel。N 拍的写入只在 N+1 拍可见。

INVARIANT 02

条件边“执行”与“生效”跨一拍

route() 在 Execute 执行并写 branch:to:target；Update 才让它进入 updated_channels；下一拍 Plan 才唤醒 target。

INVARIANT 03

Plan 的输入不是“当前 state”

Plan 依据的是 checkpoint + 上一拍的 updated_channels + trigger_to_nodes，而不是实时读取节点刚写完的值。

INVARIANT 04

put_writes 是阶段边界

Execute 的结果先进入 pending buffer；只有 after_tick → apply_writes 才把 buffer 合并进 channels 并保存 checkpoint。

示例 ↔ 源码对照

本仓库示例 观察到的阶段 源码落点
pregel/debug_pregel.py:129
traced_tick()		 Plan：本拍触发通道、计划执行的任务、task id PregelLoop.tick() _loop.py:650 → prepare_next_tasks() _algo.py:391
pregel/debug_pregel.py:151
traced_runner_tick()		 Execute：任务提交、并行执行、各任务 writes PregelRunner.tick() _runner.py:177；run_with_retry() _retry.py:573
pregel/debug_pregel.py:99
traced_after_tick()		 Update：channel 值、updated_channels、checkpoint PregelLoop.after_tick() _loop.py:737 → apply_writes() _algo.py:233
pregel/conditional_edge.py:119
traced_finish()		 Execute 写 branch:to:* → Update 生效 → 下一拍 Plan 唤醒 BranchSpec._route/_finish() graph/_branch.py:146/192；ChannelWrite.do_write() _write.py:106
pregel/loop_concurrency_demo.py SyncPregelLoop 线程池 vs AsyncPregelLoop 事件循环 PregelRunner.tick/atick _runner.py:177/361；BackgroundExecutor.submit _executor.py:54/142
pregel/timeout_demo.py run_timeout / idle_timeout / RetryPolicy / error_handler _start_timed_attempt _retry.py:343；_arun_with_timeout _retry.py:422；_should_retry_on _retry.py:834
pregel/compiled_graph.py channels / trigger_to_nodes / 并行分支如何决定下一拍任务 prepare_next_tasks() _algo.py:391/474；apply_writes() _algo.py:233
pregel/compile_internals.py compile 产物：PregelNode / channels / writers，是运行时三阶段的输入 PregelNode.invoke() _read.py:261；_first() _loop.py:915