Pregel为何成为LangGraph编排的心脏

如果你正在使用 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 MemorySa ver() });

const result = await graph.invoke(input, config);

当你调用 invoke 的那一刻，幕后实际运转的核心引擎正是 Pregel。Pregel 承担着多项关键职责：决定每一步哪些节点应当被执行，让无依赖关系的节点实现并行运行，在每个步骤结束后自动保存快照（Checkpoint），以及在需要人类介入时精准暂停，再于后续从断点无缝恢复。

毫不夸张地说，掌握了 Pregel 的运行机制，就真正理解了 LangGraph 的底层逻辑。

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

2.1 Google Pregel：专为万亿级图而生

2010 年，Google 发表了划时代的论文 Pregel: A System for Large-Scale Graph Processing，提出了一种基于 BSP（整体同步并行模型）的大规模图计算框架。

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 TB
A[Phase 1: _prepareNextTasks
确定本轮要执行的任务] --> B[Phase 2: PregelRunner.tick
并发执行所有任务]
B --> C[Phase 3: _applyWrites
将写入应用到 Channel]
C --> D[Phase 4: _putCheckpoint
持久化快照]
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 LR
subgraph Pull 拉取调度
A1[Channel 被写入新版本] --> A2[triggerToNodes 反向索引]
A2 --> A3[版本检查通过]
A3 --> A4[_procInput 读取输入]
end
subgraph 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 MemorySa ver(),
    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. 列举历史 Checkpoint
const 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 的完整数据流：

sequenceDiagram
participant User
participant PE as Pregel Engine
participant PL as PregelLoop
participant PR as PregelRunner
participant CP as Checkpointer

User->>PE: invoke(input, config)
PE->>PL: initialize()
PL->>CP: getTuple(config)
CP-->>PL: CheckpointTuple
PL-->>PE: loop instance

rect rgb(240, 248, 255)
Note over PE,CP: Superstep N
PE->>PL: tick()
Note right of PL: first tick: _first()
otherwise: apply previous writes
PL->>PL: _prepareNextTasks()
Note right of PL: Pull + Push dual-track
alt tasks found
  PL-->>PE: return true
  PE->>PR: tick()
  PR->>PR: Task A, B, C in parallel
  Note right of PR: Promise.race + maxConcurrency
  PR-->>PE: all tasks settled
  PE->>PL: tick() next round
  PL->>PL: _applyWrites()
  Note right of PL: sort, versions_seen,
group by channel, update
  PL->>CP: putCheckpoint(step N)
  Note right of PL: step++
  opt interruptAfter hit
    PL-->>PE: throw GraphInterrupt
  end
else no tasks
  PL-->>PE: return false (done)
end
end
PE-->>User: final result

七、Pregel vs 传统 Agent 循环

八、设计启示录

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 核心运行时。