Pregel 为何会成为LangGraph编排的心脏

一、为什么你需要了解 Pregel

如果你用过 LangGraph 构建 Agent，你大概写过这样的代码：

const graph = new StateGraph(State).addNode("agent", agentNode).addNode("tools", toolNode).addEdge("__start__", "agent").addConditionalEdges("agent", route, { tools: "tools", end: "__end__" }).addEdge("tools", "agent").compile({ checkpointer: new MemorySaver() });const result = await graph.invoke(input, config);

调用 invoke 的那一刻，背后运转的正是 Pregel 执行引擎。它负责：

• 决定每一步哪些节点该执行
• 让互不依赖的节点并行跑起来
• 在每个步骤结束后自动保存快照（Checkpoint）
• 在需要人类介入时暂停，之后从断点精确恢复

理解 Pregel，就是理解 LangGraph 的「物理定律」。

二、一段跨越 16 年的技术演化

2.1 Google Pregel：为万亿级图而生

2010 年，Google 发表了论文 Pregel: A System for Large-Scale Graph Processing，提出了一种基于 BSP（Bulk Synchronous Parallel） 的大规模图计算模型。

BSP 的核心思想极其简洁——计算以离散的 Superstep（超级步） 推进，每个 Superstep 包含三个阶段：

┌──────────────────────────────────────────────┐│一个 BSP Superstep│├──────────────────────────────────────────────┤│ ① 本地计算每个节点独立运行 compute() ││ ↓││ ② 消息传递节点之间通过边交换消息││ ↓││ ③ 屏障同步全局等待，所有消息到达后才进入下一轮│└──────────────────────────────────────────────┘

关键约束：步骤②发送的消息，在步骤③同步完成之前，接收方看不到。这消除了数据竞争。

2.2 LangGraph 的关键改造

LangGraph 保留了 BSP 的屏障同步语义，但把「图计算」改造成了「Agent 编排」：

最核心的设计差异：Google Pregel 中节点通过 sendMessage(targetId, value) 直接定向发送；LangGraph 中节点写入 Channel，由 Channel 的语义决定值如何聚合——这是一种发布-订阅模型。

三、Superstep：Pregel 引擎的心跳

Pregel 的一切魔法都围绕 Superstep 展开。每一次 Superstep 经历五个精密的阶段：

graph TBA[Phase 1: _prepareNextTasks<br/>确定本轮要执行的任务] --> B[Phase 2: PregelRunner.tick<br/>并发执行所有任务]B --> C[Phase 3: _applyWrites<br/>将写入应用到 Channel]C --> D[Phase 4: _putCheckpoint<br/>持久化快照]D --> E[Phase 5: 中断检查 / 准备下一轮]E --> A

Phase 1：任务准备——Pull 与 Push 双轨调度

每个 Superstep 开始时，引擎需要确定本轮要执行哪些任务。任务来源有两条轨道：

Pull（拉取）：Channel 版本变化触发的常规节点。当一个 Channel 被上游写入新版本，订阅该 Channel 的节点会被自动触发。触发条件是一个简洁的数学不等式：

channel_versions[triggerChan] > versions_seen[nodeName][triggerChan]

左边是 Channel 的全局当前版本，右边是该节点上次看到的版本。前者大于后者，说明有新数据需要处理。

Push（推送）：上游节点通过 Send() API 主动分发的动态任务。每个 Send 包携带自定义输入数据，写入一个名为 TASKS 的特殊通道，在下一个 Superstep 被消费为并行任务。

graph LRsubgraph Pull 拉取调度A1[Channel 被写入新版本] --> A2[triggerToNodes 反向索引]A2 --> A3[版本检查通过]A3 --> A4[_procInput 读取输入]endsubgraph Push 推送调度B1[节点调用 Send] --> B2[写入 TASKS 通道]B2 --> B3[按 index 取出 Send 包]B3 --> B4[packet.args 作为输入]end

Pull 是静态可预测的（编译时就能确定触发关系），Push 是运行时动态决定的。两者取并集构成完整的任务集合。

Pull 调度还有一个精巧的优化——candidateNodes 函数利用 triggerToNodes 反向索引，只检查上一轮被修改的 Channel 关联的节点，避免了全量遍历：

Phase 2：并发执行——Promise.race + Barrier

所有准备好的任务被交给 PregelRunner 并发执行。核心机制是 Promise.race：

while (/* 还有任务 */) {// 启动尽可能多的任务（受 maxConcurrency 限制）for (; startedCount < tasks.length && activeCount < maxConcurrency; startedCount++) {executingTasksMap[task.id] = _runWithRetry(task, retryPolicy);}// 等待任意一个任务完成const settled = await Promise.race([...Object.values(executingTasksMap),barrier.wait,// 支持运行时动态注入新任务]);yield settled;}

这里有一个巧妙的设计——createPromiseBarrier。当某个任务在执行过程中通过 call() 或 Send() 动态创建子任务时，barrier 会被唤醒（barrier.next()），让 Promise.race 立即感知到新任务的加入，将其纳入并发执行池。

这意味着 Send() 产生的子任务可以在同一个 Superstep 内立即执行（通过 acceptPush 机制），而非等到下一个 Superstep，显著降低端到端延迟。

Phase 3：写入应用——Channel 是唯一的真相来源

这是整个 Superstep 最关键的步骤。所有并行任务的输出不会直接传给下一个节点，而是写入 Channel，由 Channel 的语义决定状态如何聚合：

// 按 path 排序保证确定性tasks.sort((a, b) => comparePaths(a.path, b.path));// 更新 versions_seen（标记节点已看到的 channel 版本）for (const task of tasks) {for (const chan of task.triggers) {checkpoint.versions_seen[task.name][chan] = checkpoint.channel_versions[chan];}}// 按 Channel 分组写入，调用 channel.update(vals)for (const [chan, vals] of Object.entries(pendingWritesByChannel)) {const updated = channels[chan].update(vals);// Channel 自行决定聚合策略if (updated) {checkpoint.channel_versions[chan] = getNextVersion(maxVersion);updatedChannels.add(chan);}}

核心洞察：_applyWrites 是 Channel-Driven 架构的枢纽——实现了生产者-消费者解耦。节点不知道谁会消费自己的输出，消费者也不知道输入来自哪个节点。

Phase 4 & 5：持久化与循环控制

每个 Superstep 结束后自动保存 Checkpoint，包含 Channel 值快照、版本号、已见版本和挂起写入。然后检查是否需要中断（interruptBefore / interruptAfter），或是否已无新任务可执行。

关键时序：_putCheckpoint 在 shouldInterrupt 之前执行——保证中断时断点已持久化到存储层。

四、Channel：并发冲突的终极解法

当多个并行节点同时试图修改共享状态时，传统方案是加锁。Pregel 采用了一条完全不同的路径——通过 五层防御体系 从根本上消除并发写入冲突。

第 1 层：写入缓冲

每个并行任务在执行期间不会直接修改 Channel。所有写入暂存在任务本地的 writes 数组中：

NodeA 执行中NodeB 执行中 NodeC 执行中 ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ taskA.writes │ │ taskB.writes │ │ taskC.writes │ │[x, 5]│ │[x, 10] │ │[y, "hello"]│ └──────────────┘ └──────────────┘ └──────────────┘ │││ └──────────┬───────────────┘──────────────────────────┘│ 全部完成后统一提交▼_applyWrites()

并行执行期间不存在共享可变状态，无需任何锁。

第 2 层：Superstep 屏障

只有当所有并行任务都完成后，_applyWrites 才会被调用，统一将写入应用到 Channel。这是对 BSP 屏障同步的忠实继承。

第 3 层：Channel 语义隔离（核心）

这是最精妙的设计。每种 Channel 类型定义了明确的冲突解决语义：

LastValue 的设计哲学特别值得一提：如果多个并行节点写入同一个 LastValue Channel，系统直接报错。这不是 bug，而是设计信号——告诉图设计者"你不应该让多个并行节点写同一个字段"。尽早暴露设计缺陷，而非静默产生不确定行为。

// LastValue 的 update 方法update(values: Value[]): boolean {if (values.length !== 1) {throw new InvalidUpdateError("LastValue can only receive one value per step.",{ lc_error_code: "INVALID_CONCURRENT_GRAPH_UPDATE" });}this.value = [values[values.length - 1]];return true;}

Reducer 的优雅：图设计者通过选择 Reducer 函数来显式定义合并策略。例如，多个并行节点写入 messages 字段时，Reducer 自动拼接所有消息；写入 totalTokens 时，Reducer 自动求和：

const State = Annotation.Root({// 消息列表：并行写入自动拼接messages: Annotation<BaseMessage[]>({ reducer: messagesReducer }),// Token 计数：并行写入自动累加totalTokens: Annotation<number>({ reducer: (a, b) => a + b }),// 当前步骤：只允许单写者（并行写入会报错）currentStep: Annotation<string>,});

第 4 层：确定性排序

即使有了 Channel 语义，写入的应用顺序也必须确定性。_applyWrites 按任务的 path（任务在图中的唯一地址标识）排序，保证 Reducer 合并顺序可重现：

排序前（Promise.race 完成顺序随机）:NodeB(writes=[counter,10]), NodeA(writes=[counter,5]), NodeC(writes=[counter,3])排序后（按 path 确定性排列）:NodeA < NodeB < NodeC→ Reducer 始终按 A→B→C 顺序合并：counter = 0 + 5 + 10 + 3 = 18

第 5 层：版本追踪

channel_versions 和 versions_seen 两套版本号确保节点触发的确定性——不会重复触发、不会遗漏触发、触发判定与执行顺序无关。

与传统并发模型的对比：

┌─────────────────┬────────────────┬────────────────┬──────────────────┐│ │ 锁（Mutex） │ Actor 模型│ Pregel Channel │├─────────────────┼────────────────┼────────────────┼──────────────────┤│ 冲突检测 │ 运行时死锁│ 消息顺序不定 │ 编译/运行时显式 ││ 冲突解决 │ 程序员手动处理 │ Actor 内部逻辑 │ Channel 语义定义 ││ 死锁风险 │ 高│ 无│ 无 ││ 与持久化的集成│ 需自行实现 │ 需自行实现 │ 原生 Checkpoint │└─────────────────┴────────────────┴────────────────┴──────────────────┘

五、Checkpoint + 断点续行：Agent 的"存档读档"

Agent 编排中有一个传统框架几乎无法优雅解决的问题：Agent 执行到一半需要人类审批，怎么办？

Pregel + Checkpoint 的组合提供了完整的断点续行能力，支撑四大场景：

5.1 三种暂停方式

interruptBefore / interruptAfter：编译期声明的中断点。

const graph = new StateGraph(State).addNode("agent", agent).addNode("human_review", humanReview).addNode("execute", execute).addEdge("__start__", "agent").addEdge("agent", "human_review").addEdge("human_review", "execute").compile({checkpointer: new MemorySaver(),interruptBefore: ["human_review"],// 执行 human_review 前暂停});

interrupt()：节点内部主动中断。

const humanReview = (state) => {const humanDecision = interrupt({question: `是否批准？最后消息: "${state.messages.at(-1)}"`,options: ["approved", "rejected"],});return { decision: humanDecision };};

interrupt() 底层有三路分支：

1. 已有 resume 值（恢复执行时）→ 直接返回
2. 有 nullResume 值（来自 Command 的全局恢复）→ 消费并返回
3. 无 resume 值 → 抛出 GraphInterrupt 中断执行

5.2 Command + resume：恢复执行

// 第一次调用：agent 执行 → human_review 中 interrupt() 触发中断const r1 = await graph.invoke({ messages: ["用户: 请帮我处理这个任务"] },{ configurable: { thread_id: "approval-001" } });// → { messages: [...], __interrupt__: [{ value: { question: "是否批准？..." } }] }// 人类审批后，恢复执行const r2 = await graph.invoke(new Command({ resume: "approved" }),{ configurable: { thread_id: "approval-001" } });// → { messages: [..., "方案 A 已执行！"], decision: "approved" }

5.3 shouldInterrupt 的防重复中断

恢复执行时有一个微妙的风险：shouldInterrupt 会不会再次触发中断？

答案在 versions_seen[INTERRUPT]——一个特殊的"中断游标"。每次 resume 时，它会被同步到最新的 channel_versions，使得 anyChannelUpdated 判断为 false，从而避免重复中断：

阶段 1：首次触发中断channel_versions = { messages: 2 }versions_seen[__interrupt__] = { messages: 1 }← 旧版本→ 2 > 1 → 触发中断 ✓阶段 2：恢复执行时（_first 中同步）versions_seen[__interrupt__] = { messages: 2 }← 同步到最新阶段 3：恢复后继续执行→ 中断游标已对齐，不会重复触发 ✓

5.4 时间旅行（Time Travel）

断点续行的进阶玩法——从历史某个 Checkpoint 恢复并创建分支：

// 1. 列举历史 Checkpointconst history = [];for await (const tuple of checkpointer.list(config)) {history.push(tuple);}// 2. 从 agent 完成后的状态分支，注入不同的决策await graph.updateState({ configurable: { thread_id: "approval-001", checkpoint_id: history[2].checkpoint.id } },{ decision: "rejected" },"human_review");// 3. 继续执行 → 走不同的路径const r3 = await graph.invoke(null, config);// → { messages: [..., "方案已取消。"] }

5.5 崩溃恢复

每个 Superstep 自动 Checkpoint，进程崩溃后只需重新 invoke(null, config)（同一 thread_id），引擎自动从最近的 Checkpoint 恢复，从未完成的节点继续执行。

三重保障：

1. _applyWrites 在 _putCheckpoint 之前执行 → Channel 状态一致
2. _putCheckpoint 在 shouldInterrupt 之前执行 → 中断前断点已持久化
3. putWrites 在任务完成时立即持久化 → 即使 Checkpoint 未保存，中间写入也不丢失

5.6 多次 interrupt 的逐步恢复

如果节点内有多个 interrupt()，resume 数组会逐步累积，按中断计数器索引消费：

const multiNode = (state) => {const intent = interrupt({ step: 1, question: "确认意图？" }); // 第 0 次const params = interrupt({ step: 2, question: "确认参数？" }); // 第 1 次return { messages: [`意图: ${intent}, 参数: ${params}`] };};

第 1 次调用：interrupt(idx=0) → 无 resume → 抛出中断第 2 次调用 resume="yes"：interrupt(idx=0) → 返回 "yes"；interrupt(idx=1) → 抛出中断第 3 次调用 resume="ok"：interrupt(idx=0) → 跳过（已恢复）；interrupt(idx=1) → 返回 "ok"；节点完成！

六、并行执行 → Checkpoint 全景时序

把以上所有机制串联起来，看一个包含 3 个并行节点的 Superstep 完整数据流：

sequenceDiagramparticipant Userparticipant PE as Pregel Engineparticipant PL as PregelLoopparticipant PR as PregelRunnerparticipant CP as CheckpointerUser->>PE: invoke(input, config)PE->>PL: initialize()PL->>CP: getTuple(config)CP-->>PL: CheckpointTuplePL-->>PE: loop instancerect rgb(240, 248, 255)Note over PE,CP: Superstep NPE->>PL: tick()Note right of PL: first tick: _first()<br/>otherwise: apply previous writesPL->>PL: _prepareNextTasks()Note right of PL: Pull + Push dual-trackalt tasks foundPL-->>PE: return truePE->>PR: tick()PR->>PR: Task A, B, C in parallelNote right of PR: Promise.race + maxConcurrencyPR-->>PE: all tasks settledPE->>PL: tick() next roundPL->>PL: _applyWrites()Note right of PL: sort, versions_seen,<br/>group by channel, updatePL->>CP: putCheckpoint(step N)Note right of PL: step++opt interruptAfter hitPL-->>PE: throw GraphInterruptendelse no tasksPL-->>PE: return false (done)endendPE-->>User: final result

七、Pregel vs 传统 Agent 循环

┌──────────────────┬───────────────────┬─────────────────────┐│ 特性│ 传统 Agent 循环│ LangGraph Pregel│├──────────────────┼───────────────────┼─────────────────────┤│ 执行模型│ 顺序调用链│ Superstep 并行││ 状态传递│ 函数返回值传递 │ Channel 解耦传递││ 状态聚合│ 手动合并│ Channel reducer 自动││ 持久化│ 需自行实现│ 每步自动 Checkpoint ││ 断点恢复│ 不支持 │ 原生支持 ││ 并行执行│ 需手动 Promise.all│ 自动并行无依赖节点 ││ 人机交互│ 不支持 │ interrupt 原生支持 ││ 时间旅行│ 不支持 │ updateState 回溯 ││ 动态路由│ if-else│ Send() + 条件边│└──────────────────┴───────────────────┴─────────────────────┘

八、设计启示录

8.1 "冲突"不是 bug，而是设计信号

Pregel 不试图隐式解决所有冲突。LastValue 的并发写入直接报错，告诉图设计者"你的状态 Schema 设计有问题"。这种尽早失败的哲学，远比静默产生不确定行为要负责。

8.2 Channel 选择 = 并发策略选择

图设计者在定义 State Schema 时，通过选择 Channel 类型（LastValue / Reducer / Topic）就已经确定了并发策略。这是一种声明式的并发控制，而非命令式。

8.3 BSP 模型的复兴

写入缓冲 + 屏障同步 = 无锁并发。这是对 1990 年代 BSP 并行计算模型的现代演绎——既有并行的性能优势，又有顺序执行的正确性保证。

8.4 Checkpoint 先行

整个执行流程中，Checkpoint 的持久化始终先于中断判断。这保证了一个不变量：任何中断点都有完整断点。无论是人类审批、进程崩溃还是时间旅行，都能从断点精确恢复。

九、关键源码索引

本文基于 LangGraph.js 源码分析，版本对应 @langchain/langgraph 核心运行时。