LangGraph 设计与实现

第7章任务调度与并行执行

作者杨艺韬 · 13,406 字 · 发布于 2026-04-05 · 更新于 2026-09-25

7.1 引言

上一章我们剖析了 Pregel 执行循环的宏观架构——tick()、after_tick() 和 BSP 超步模型。但在每个超步内部，还有一个同样复杂的世界：多个任务如何被并行调度？任务失败时如何重试？缓存如何避免重复计算？PUSH 任务和 PULL 任务在运行时有何不同？

本章将深入 LangGraph 的任务执行层，涉及以下核心组件：

PregelExecutableTask（types.py）—— 可执行任务的数据结构
PregelRunner（libs/langgraph/langgraph/pregel/_runner.py）—— 任务调度器，管理并行执行和结果收集
BackgroundExecutor / AsyncBackgroundExecutor（libs/langgraph/langgraph/pregel/_executor.py）—— 线程池和 asyncio 并行原语
run_with_retry / arun_with_retry（libs/langgraph/langgraph/pregel/_retry.py）—— 重试逻辑
缓存匹配机制 —— cache_policy 和 CacheKey 的协作

这些组件共同实现了一个高效的并行执行框架，在保证正确性的前提下最大化吞吐量。

本章要点

PregelExecutableTask 是任务执行的最小单元，包含输入、处理器、写入缓冲、配置等全部信息
PregelRunner 通过 FuturesDict 管理并发任务，支持"任一失败则全部停止"的语义
PULL 任务由 Channel 版本变更触发，输入从 Channel 读取；PUSH 任务由 Send API 创建，输入由调用者指定
BackgroundExecutor 使用线程池实现同步并行，AsyncBackgroundExecutor 使用 asyncio 任务实现异步并行
重试策略支持指数退避、抖动、最大重试次数，以及按异常类型匹配的多策略组合
缓存策略通过 CacheKey 关联节点身份和输入哈希，支持 TTL 过期

7.2 PregelExecutableTask：任务的全貌

PregelExecutableTask 定义在 libs/langgraph/langgraph/types.py:538（注意不是 pregel/types.py——后者在 1.1.6 里只是一层挂着 LangGraphDeprecatedSinceV10 的 re-export shim），是一个不可变的 dataclass：

@dataclass(frozen=True)
class PregelExecutableTask:
    name: str                          # 节点名称
    input: Any                         # 任务输入
    proc: Runnable                     # 可执行处理器（bound + writers 的组合）
    writes: deque[tuple[str, Any]]     # 写入缓冲区
    config: RunnableConfig             # 完整的运行配置
    triggers: Sequence[str]            # 触发此任务的 Channel 列表
    retry_policy: Sequence[RetryPolicy] # 重试策略
    cache_key: CacheKey | None         # 缓存键（如果启用了缓存）
    id: str                            # 全局唯一的任务 ID
    path: tuple[str | int | tuple, ...] # 任务路径（用于排序和标识）
    writers: Sequence[Runnable] = ()    # 写入器引用
    subgraphs: Sequence[PregelProtocol] = ()  # 子图引用

虽然标记为 frozen=True（不可变），但 writes 字段是一个 deque——它的引用不可变，但内容可变。这个设计使得任务执行过程中可以向 writes 追加数据，同时防止意外替换整个 writes 对象。

7.2.1 任务 ID 的生成

任务 ID 是通过确定性哈希函数生成的，确保同一个 Checkpoint 状态下，相同的任务总是获得相同的 ID：

# 对于 PULL 任务
task_id = task_id_func(
    checkpoint_id_bytes,    # Checkpoint ID 的字节表示
    checkpoint_ns,          # 命名空间（如 "parent:<task_id>|agent"）
    str(step),              # 步数
    name,                   # 节点名称
    PULL,                   # 任务类型
    *triggers,              # 触发 Channel
)

# 对于 PUSH 任务（Send API，_algo.py:930-937）
task_id = task_id_func(
    checkpoint_id_bytes,
    checkpoint_ns,
    str(step),
    packet.node,   # Send 的目标节点名
    PUSH,
    str(idx),      # Send 在 __pregel_tasks Topic 中的下标
)

# Functional API 的 Call 路径（_algo.py:775-783）才带父任务路径
task_id = task_id_func(
    checkpoint_id_bytes, checkpoint_ns, str(step), name, PUSH,
    task_path_str(task_path[1]), str(task_path[2]),
)

任务 ID 的哈希函数按 checkpoint 格式版本二选一（libs/langgraph/langgraph/pregel/_algo.py:528：task_id_func = _xxhash_str if checkpoint["v"] > 1 else _uuid5_str）——新格式（当前 Pregel 写出的是 v4）用 xxhash，v1 老 checkpoint 才回落到 uuid5。确定性的 ID 是 Checkpoint 恢复的关键——恢复后重新计算的任务 ID 与保存的 pending writes 中的 task ID 必须匹配，这样 _match_writes 才能正确地将已保存的写入结果关联到重建的任务。

7.2.2 proc 的构成

PregelExecutableTask.proc 取自 PregelNode.node（libs/langgraph/langgraph/pregel/_read.py:201），有写入器时就是一个 RunnableSeq，把用户逻辑和写入器串联：

flowchart LR
    INPUT[任务输入] --> BOUND[bound: 用户函数]
    BOUND --> W1[ChannelWrite: 状态更新 + 静态边路由]
    W1 --> W2[条件边的路由 Runnable（如有）]
    subgraph "proc = RunnableSeq(bound, *flat_writers)"
        BOUND
        W1
        W2
    end

执行 task.proc.invoke(task.input, task.config) 时：

首先调用用户函数，传入从 Channel 读取的状态
用户函数返回状态更新（如 {"count": 5}）
ChannelWrite 将更新转化为 Channel 写入元组，通过 CONFIG_KEY_SEND 发送。flat_writers 会把相邻的多个 ChannelWrite 合并成一个，所以静态边的路由写入（如 branch:to:b）和状态更新落在同一个 ChannelWrite 里
若节点后面接的是条件边，其后还有一个路由 Runnable，按路由函数的返回值把信号写入目标节点的触发 Channel

7.2.3 config 中注入的关键函数

每个任务的 config 中注入了几个关键回调，使得任务执行过程中能与 PregelLoop 交互：

config = patch_config(
    config,
    configurable={
        CONFIG_KEY_TASK_ID: task_id,
        CONFIG_KEY_SEND: writes.extend,     # 写入收集器
        CONFIG_KEY_READ: partial(            # 状态读取器
            local_read, scratchpad, channels, managed,
            PregelTaskWrites(path, name, writes, triggers),
        ),
        CONFIG_KEY_CHECKPOINTER: checkpointer,
        CONFIG_KEY_CHECKPOINT_NS: task_checkpoint_ns,
        CONFIG_KEY_SCRATCHPAD: scratchpad,
        CONFIG_KEY_RUNTIME: runtime,
    },
)

CONFIG_KEY_SEND：绑定到 writes.extend——当 ChannelWrite.do_write 被调用时，写入元组被追加到任务的 writes deque。deque.extend 是线程安全的。
CONFIG_KEY_READ：绑定到 local_read 函数——条件边通过此函数读取"应用了当前任务写入后"的状态快照。这确保条件判断基于最新状态。

7.3 PULL 任务 vs PUSH 任务

LangGraph 中有两种根本不同的任务触发方式：

flowchart TB
    subgraph "PULL 任务"
        direction TB
        EDGE["边/条件边写入\nbranch:to:agent"] --> CHAN_VER["Channel 版本更新"]
        CHAN_VER --> TRIGGERS["_triggers() 检测\n版本 > seen"]
        TRIGGERS --> PULL_PREP["prepare_single_task\npath = (PULL, name)"]
        PULL_PREP --> PULL_INPUT["输入 = Channel 读取结果\n读取 node.channels 指定的 Channel"]
        PULL_INPUT --> PULL_TASK["PregelExecutableTask"]
    end
    subgraph "PUSH 任务"
        direction TB
        SEND_API["节点返回 Send('tool', data)\n或条件边返回 Send"] --> TASKS_CH["__pregel_tasks Topic"]
        TASKS_CH --> PUSH_PREP["prepare_single_task\npath = (PUSH, idx)"]
        PUSH_PREP --> PUSH_INPUT["输入 = Send.arg\n由调用者直接指定"]
        PUSH_INPUT --> PUSH_TASK["PregelExecutableTask"]
    end
    style PULL_TASK fill:#c8e6c9
    style PUSH_TASK fill:#fff3e0

PULL 任务

PULL 任务是标准的 BSP 触发方式。在 prepare_single_task 中，对于 (PULL, name) 路径：

if task_path[0] == PULL:
    name = task_path[1]
    proc = processes[name]
    # 检查触发条件
    if _triggers(channels, checkpoint["channel_versions"],
                 checkpoint["versions_seen"].get(name),
                 null_version, proc):
        # 读取输入
        val = _proc_input(proc, managed, channels,
                          for_execution=True, ...)
        if val is MISSING:
            return  # Channel 为空，跳过
        # 创建任务
        return PregelExecutableTask(name, val, node, writes, ...)

PULL 任务的输入来自 Channel：_proc_input 根据 proc.channels 配置读取指定的 Channel 值，如果有 mapper 则进行类型转换。

PUSH 任务

PUSH 任务通过两种途径创建：

Send API（prepare_push_task_send）：当 __pregel_tasks Topic Channel 中有 Send 对象时
Functional API（prepare_push_task_functional）：当任务路径以 Call 对象结尾时

对于 Send API 的 PUSH 任务：

if task_path[0] == PUSH:
    # 获取 Send 对象
    send = tasks_channel.get()[task_path[1]]
    name = send.node
    val = send.arg  # 直接使用 Send 的参数作为输入
    proc = processes[name]
    # 创建任务（不检查 _triggers）
    return PregelExecutableTask(name, val, node, writes, ...)

关键区别：PUSH 任务不检查 _triggers——它们总是被执行。输入直接来自 Send.arg，而非从 Channel 读取。这使得同一个节点可以被多次调用，每次使用不同的输入。

7.4 PregelRunner：并行调度器

PregelRunner 定义在 libs/langgraph/langgraph/pregel/_runner.py 中，负责在每个超步中并行执行所有任务：

class PregelRunner:
    def __init__(self, *, submit, put_writes,
                 use_astream=False, node_finished=None):
        self.submit = submit          # 提交函数（弱引用）
        self.put_writes = put_writes  # 写入保存函数（弱引用）
        self.use_astream = use_astream
        self.node_finished = node_finished

7.4.1 同步 tick 的执行流程

def tick(self, tasks, *, reraise=True, timeout=None,
         retry_policy=None, get_waiter=None, schedule_task):
    tasks = tuple(tasks)
    futures = FuturesDict(
        callback=weakref.WeakMethod(self.commit),
        event=threading.Event(),
        future_type=concurrent.futures.Future,
    )
    # 让出控制权给调用者
    yield

    # 快速路径：单任务无超时
    if len(tasks) == 1 and timeout is None and get_waiter is None:
        t = tasks[0]
        try:
            run_with_retry(t, retry_policy, ...)
            self.commit(t, None)
        except Exception as exc:
            self.commit(t, exc)
            ...
        return

    # 调度所有任务到线程池
    for t in tasks:
        fut = self.submit()(
            run_with_retry, t, retry_policy, ...
        )
        futures[fut] = t

    # 等待任务完成，逐个处理
    while len(futures) > 0:
        done, inflight = concurrent.futures.wait(
            futures,
            return_when=concurrent.futures.FIRST_COMPLETED,
            timeout=...,
        )
        for fut in done:
            futures.pop(fut)
        if _should_stop_others(done):
            break
        yield  # 让出控制权给调用者处理流式输出

    # 等待所有回调完成
    futures.event.wait(timeout=...)
    yield

    # 检查异常
    _panic_or_proceed(futures.done, panic=reraise)

7.4.2 FuturesDict：智能的并发管理

FuturesDict 是一个自定义的 dict，它在 Future 完成时自动调用回调并管理计数器：

class FuturesDict(dict):
    event: threading.Event  # 所有任务完成时设置
    callback: weakref.ref   # commit 回调
    counter: int            # 活跃任务计数
    done: set              # 已完成的 Future 集合
    lock: threading.Lock    # 保护 counter / done 的互斥锁

    def __setitem__(self, key, value):
        super().__setitem__(key, value)
        if value is not None:
            with self.lock:          # counter 的自增不是线程安全的
                self.event.clear()
                self.counter += 1
            key.add_done_callback(partial(self.on_done, value))

    def on_done(self, task, fut):
        try:
            if cb := self.callback():
                cb(task, _exception(fut))
        finally:
            with self.lock:
                self.done.add(fut)
                self.counter -= 1
                if self.counter == 0 or _should_stop_others(self.done):
                    self.event.set()

sequenceDiagram
    participant Runner as PregelRunner
    participant FD as FuturesDict
    participant Pool as 线程池
    participant T1 as 任务 A
    participant T2 as 任务 B
    participant PL as PregelLoop
    Runner->>Pool: submit(run_with_retry, task_a)
    Runner->>FD: futures[fut_a] = task_a
    Runner->>Pool: submit(run_with_retry, task_b)
    Runner->>FD: futures[fut_b] = task_b
    Pool->>T1: 执行
    Pool->>T2: 执行
    T1-->>FD: on_done(task_a, result)
    FD->>Runner: commit(task_a, None)
    Runner->>PL: put_writes(task_a.id, task_a.writes)
    Note over Runner: yield - 让出控制权
    Note over PL: 处理流式输出
    T2-->>FD: on_done(task_b, result)
    FD->>Runner: commit(task_b, None)
    Runner->>PL: put_writes(task_b.id, task_b.writes)
    FD-->>FD: counter == 0, event.set()

7.4.3 单任务快速路径

当只有一个任务且没有超时时，PregelRunner 使用快速路径——直接在当前线程执行，避免线程池的开销：

if len(tasks) == 1 and timeout is None and get_waiter is None:
    t = tasks[0]
    try:
        run_with_retry(t, retry_policy, ...)
        self.commit(t, None)
    except Exception as exc:
        self.commit(t, exc)

这个优化在简单的线性图（每步只有一个节点执行）中显著减少了开销。

7.4.4 commit：结果提交

commit 方法根据任务结果的不同情况执行不同的处理（libs/langgraph/langgraph/pregel/_runner.py:425，略有简化）：

def commit(self, task, exception):
    if isinstance(exception, asyncio.CancelledError):
        # 被取消的任务：保存错误
        task.writes.append((ERROR, exception))
        self.put_writes()(task.id, task.writes)
    elif exception:
        if isinstance(exception, GraphInterrupt):
            # 中断：args[0] 非空时保存中断信息，连同本任务的 RESUME 写入
            if exception.args[0]:
                writes = [(INTERRUPT, exception.args[0])]
                if resumes := [w for w in task.writes if w[0] == RESUME]:
                    writes.extend(resumes)
                self.put_writes()(task.id, writes)
        elif isinstance(exception, GraphBubbleUp):
            # 冒泡异常：不保存，将在 _panic_or_proceed 中处理
            pass
        else:
            # 普通错误：保存错误信息
            task.writes.append((ERROR, exception))
            self.put_writes()(task.id, task.writes)
    else:
        # 成功：通知节点完成（带 TAG_HIDDEN 的任务除外），保存写入
        if self.node_finished and (
            task.config is None or TAG_HIDDEN not in task.config.get("tags", [])
        ):
            self.node_finished(task.name)
        if not task.writes:
            task.writes.append((NO_WRITES, None))
        self.put_writes()(task.id, task.writes)

注意 NO_WRITES 标记——即使任务没有产生任何写入，也会标记一个占位写入。这确保了 Checkpointer 知道该任务已经执行过，在恢复时不会重复执行。

7.4.5 错误传播与任务取消

_should_stop_others 函数检查是否有任务失败：

def _should_stop_others(done):
    for fut in done:
        if fut.cancelled():
            continue
        elif exc := fut.exception():
            if not isinstance(exc, GraphBubbleUp) and \
               fut not in SKIP_RERAISE_SET:
                return True
    return False

当一个任务失败时（非 GraphBubbleUp 异常），所有其他任务会被取消。GraphBubbleUp（包括 GraphInterrupt）是特殊的——它们不被视为错误，不会导致其他任务被取消。

_panic_or_proceed 在所有任务完成后进行最终检查：

def _panic_or_proceed(futs, *, timeout_exc_cls=TimeoutError, panic=True):
    done, inflight = set(), set()
    for fut in futs:                 # 按状态分桶，已取消的直接忽略
        if fut.cancelled():
            continue
        (done if fut.done() else inflight).add(fut)
    interrupts = []
    while done:
        fut = done.pop()
        if exc := _exception(fut):
            while inflight:          # 有任务失败：先取消所有仍在运行的任务
                inflight.pop().cancel()
            if panic:
                if isinstance(exc, GraphInterrupt):
                    interrupts.append(exc)
                elif fut not in SKIP_RERAISE_SET:
                    raise exc
    if interrupts:
        # 合并所有中断
        raise GraphInterrupt(
            tuple(i for exc in interrupts for i in exc.args[0])
        )
    if inflight:                     # 走到这里说明超时了
        while inflight:
            inflight.pop().cancel()
        raise timeout_exc_cls("Timed out")

多个任务的 GraphInterrupt 会被合并为一个，确保所有中断信息都被保留。

7.4.6 pregel/_algo.py 1258 行：任务规划器的真实源码账本

读 PregelRunner 解决的是"如何把一批任务跑起来"——但"这一批任务从哪来"是另一个同等重要的问题、答案藏在 libs/langgraph/langgraph/pregel/_algo.py。在本专栏基准 langgraph 1.1.6（commit c3ba590）上该文件 1258 行、是 pregel 目录第三大文件（仅次于 main.py 3718 和 _loop.py 1423）。它有两个主函数是本章理解任务调度必须对齐的外部接口——

prepare_next_tasks 函数签名（真实源码，libs/langgraph/langgraph/pregel/_algo.py:370-388）：

def prepare_next_tasks(
    checkpoint: Checkpoint,
    pending_writes: list[PendingWrite],
    processes: Mapping[str, PregelNode],
    channels: Mapping[str, BaseChannel],
    managed: ManagedValueMapping,
    config: RunnableConfig,
    step: int,
    stop: int,
    *,
    for_execution: bool,
    store: BaseStore | None = None,
    checkpointer: BaseCheckpointSaver | None = None,
    manager: None | ParentRunManager | AsyncParentRunManager = None,
    trigger_to_nodes: Mapping[str, Sequence[str]] | None = None,
    updated_channels: set[str] | None = None,
    retry_policy: Sequence[RetryPolicy] = (),
    cache_policy: CachePolicy | None = None,
) -> dict[str, PregelTask] | dict[str, PregelExecutableTask]:
    """Prepare the set of tasks that will make up the next Pregel step.
    ...
    """

签名里16 个参数每一个都有来路——

参数	来自	作用
`checkpoint`	`PregelLoop.checkpoint`	读取 channel_versions / versions_seen 决定 PULL 激活
`pending_writes`	`PregelLoop.checkpoint_pending_writes`	传给 `_scratchpad`，从中找出本任务（及全局）的 `RESUME` 恢复值；把已保存的写入对回任务的 `_match_writes` 由 `PregelLoop` 另行调用
`processes`	`compile` 阶段固化的 `node_name → PregelNode`	决定可调度节点集合
`channels`	`PregelLoop.channels`	PULL 任务读输入的数据源
`managed`	`PregelLoop.managed`	托管值（如 `IsLastStep`、`RemainingSteps`），读输入时按 scratchpad 现算
`config`	用户 invoke 时传入	注入各种 CONFIG_KEY_* 回调
`step` + `stop`	超步计数器 + 终止条件	进入任务 config 的 scratchpad
`for_execution`	`True` 出执行任务、`False` 出只读 PregelTask	两种调用方共用一套规划
`trigger_to_nodes`	`compile` 阶段算好的倒排索引	O(1) 查找"这个 channel 变了激活哪些节点"
`updated_channels`	上一步 `apply_writes` 返回值	增量激活、不遍历所有 channel
`retry_policy` / `cache_policy`	编译期配置	注入到生成的 PregelExecutableTask

trigger_to_nodes 和 updated_channels 这两个增量结构是规划器的性能关键——libs/langgraph/langgraph/pregel/_algo.py:453-464 把这个优化写得很直白：拿不到这两样东西时 candidate_nodes = processes.keys()，即每步都要把所有节点过一遍 prepare_single_task；拿得到时只遍历"上一步被改过的 channel 通过倒排索引点名的那些节点"。节点少时两者差别不明显；节点多、而每步只有少数 channel 变化时，后者省下的遍历量随图规模增长。这份集合还会经 create_checkpoint(..., updated_channels=...) 记进 checkpoint，恢复时由 _first 读回（libs/langgraph/langgraph/pregel/_loop.py:1199），让"增量规划"在中断恢复后也能接着用。

apply_writes 函数签名（真实源码，libs/langgraph/langgraph/pregel/_algo.py:218-224）：

def apply_writes(
    checkpoint: Checkpoint,
    channels: Mapping[str, BaseChannel],
    tasks: Iterable[WritesProtocol],
    get_next_version: GetNextVersion | None,
    trigger_to_nodes: Mapping[str, Sequence[str]],
) -> set[str]:
    """Apply writes from a set of tasks (usually the tasks from a Pregel step)
    to the checkpoint and channels, and return managed values writes to be
    applied externally.
    ...
    """

返回值 set[str] 就是下一次 prepare_next_tasks 的 updated_channels——两个函数首尾咬合形成闭环：prepare_next_tasks → run → apply_writes → updated_channels → prepare_next_tasks。这是 §7.4 的 Runner 为什么能完全不知道"下一步做什么"的原因——规划和执行被干净切分、规划算法全部锁在 _algo.py 里、Runner 只负责"把给我的任务跑完并 commit"。

源码文件分工表的对照（line 数据来自 wc -l）——

职责	文件	行	主要对外接口
规划"下一步跑什么"	`_algo.py`	1258	`prepare_next_tasks` / `apply_writes`
调度"当前这一批怎么跑"	`_runner.py`	768	`PregelRunner.tick` / `atick`
并发原语	`_executor.py`	223	`BackgroundExecutor` / `AsyncBackgroundExecutor`
单任务重试	`_retry.py`	318	`run_with_retry` / `arun_with_retry`
主循环串联上面四者	`_loop.py`	1423	`PregelLoop.tick` / `after_tick`

规划层（algo）不知道并行——执行层（runner）不知道如何生成任务——重试层（retry）不知道调度——执行器（executor）不知道任务语义。四层正交、通过数据结构（PregelExecutableTask、Checkpoint、RetryPolicy）而非继承关系耦合。这套分层是 LangGraph 自己的工程组织；从 Google 2010 年 Pregel 论文继承来的是 BSP 超步语义，而不是这种文件划分。

7.5 BackgroundExecutor：线程池并行

BackgroundExecutor 定义在 libs/langgraph/langgraph/pregel/_executor.py 中，是同步执行的后台任务管理器：

class BackgroundExecutor(AbstractContextManager):
    def __init__(self, config: RunnableConfig):
        self.stack = ExitStack()
        self.executor = self.stack.enter_context(
            get_executor_for_config(config)
        )
        self.tasks: dict[Future, tuple[bool, bool]] = {}

    def submit(self, fn, *args,
               __cancel_on_exit__=False,
               __reraise_on_exit__=True,
               __next_tick__=False,
               **kwargs) -> Future:
        ctx = copy_context()
        if __next_tick__:
            task = self.executor.submit(
                next_tick, ctx.run, fn, *args, **kwargs
            )
        else:
            task = self.executor.submit(ctx.run, fn, *args, **kwargs)
        self.tasks[task] = (__cancel_on_exit__, __reraise_on_exit__)
        task.add_done_callback(self.done)
        return task

7.5.1 关键参数

参数	含义	使用场景
`__cancel_on_exit__`	退出时是否取消任务	异步路径 `atick` 提交的节点任务与 Functional API 子任务都设为 True；同步线程池只能取消尚未开始的任务
`__reraise_on_exit__`	退出时是否重新抛出异常	PUSH 子任务设为 False（由父任务处理）
`__next_tick__`	是否在下一个"tick"执行	动态创建的子任务，确保当前步的写入先提交

7.5.2 next_tick 机制

__next_tick__ 参数触发 next_tick 函数：

def next_tick(fn, *args, **kwargs):
    time.sleep(0)  # 让出 CPU 时间片
    return fn(*args, **kwargs)

time.sleep(0) 看起来无意义，但它实际上让出了当前线程的执行权，允许线程池中的其他线程（特别是正在执行 commit 回调的线程）先跑。用意是让动态创建的子任务在本轮已有的写入提交、流式输出之后再开始执行（源码注释："starting a new task in the next tick ensures updates from this tick are committed/streamed first"）。

7.5.3 上下文管理器退出

# BackgroundExecutor.__exit__，libs/langgraph/langgraph/pregel/_executor.py:93
# （签名在源码里是多行带类型）
def __exit__(self, exc_type, exc_value, traceback):
    tasks = self.tasks.copy()
    # 取消标记为 cancel_on_exit 的任务
    for task, (cancel, _) in tasks.items():
        if cancel:
            task.cancel()
    # 等待所有任务完成
    if pending := {t for t in tasks if not t.done()}:
        concurrent.futures.wait(pending)
    # 关闭线程池
    self.stack.__exit__(exc_type, exc_value, traceback)
    # 重新抛出标记为 reraise 的任务异常
    if exc_type is None:
        for task, (_, reraise) in tasks.items():
            if not reraise:
                continue
            try:
                task.result()
            except concurrent.futures.CancelledError:
                pass

退出时的处理顺序很重要：先取消，再等待，再清理，最后处理异常。这确保了所有资源被正确释放。

7.6 AsyncBackgroundExecutor：异步并行

AsyncBackgroundExecutor 是异步版本，使用 asyncio 任务替代线程：

class AsyncBackgroundExecutor(AbstractAsyncContextManager):
    def __init__(self, config: RunnableConfig):
        self.tasks: dict[asyncio.Future, tuple[bool, bool]] = {}
        self.loop = asyncio.get_running_loop()
        if max_concurrency := config.get("max_concurrency"):
            self.semaphore = asyncio.Semaphore(max_concurrency)
        else:
            self.semaphore = None

    def submit(self, fn, *args, __cancel_on_exit__=False,
               __reraise_on_exit__=True, __next_tick__=False,
               **kwargs) -> asyncio.Future:
        coro = fn(*args, **kwargs)
        if self.semaphore:
            coro = gated(self.semaphore, coro)
        task = run_coroutine_threadsafe(
            coro, self.loop,
            name=__name__,
            context=copy_context(),  # 不支持 context 参数的环境下省略
            lazy=__next_tick__,
        )
        self.tasks[task] = (__cancel_on_exit__, __reraise_on_exit__)
        task.add_done_callback(self.done)
        return task

7.6.1 并发控制：Semaphore

flowchart TB
    subgraph "无并发限制"
        T1A[任务 A] --> EXEC1[立即执行]
        T1B[任务 B] --> EXEC1
        T1C[任务 C] --> EXEC1
    end
    subgraph "max_concurrency=2"
        T2A[任务 A] --> SEM["Semaphore(2)"]
        T2B[任务 B] --> SEM
        T2C[任务 C] --> SEM
        SEM -->|获取 permit| EXEC2A[执行 A]
        SEM -->|获取 permit| EXEC2B[执行 B]
        SEM -->|等待 permit| WAIT[任务 C 等待]
        EXEC2A -->|释放| SEM
        SEM -->|获取 permit| EXEC2C[执行 C]
    end

max_concurrency 通过 gated 协程实现：

async def gated(semaphore, coro):
    async with semaphore:
        return await coro

这提供了细粒度的并发控制——在 AI 工作流中，过多的并发可能导致 API 限流或内存溢出，max_concurrency 提供了安全阀。

7.6.2 同步 vs 异步对比

特性	BackgroundExecutor	AsyncBackgroundExecutor
并行原语	`ThreadPoolExecutor`	`asyncio.Task`
并发控制	线程池大小	`asyncio.Semaphore`
上下文传递	`copy_context()` + `ctx.run`	`copy_context()` 参数
取消机制	`Future.cancel()`（仅未启动）	`Task.cancel()`（在下一个 await 点生效）
next_tick	`time.sleep(0)`	`lazy=True` 延迟调度

异步版本的取消更加彻底——asyncio 任务可以在执行过程中被取消（通过 CancelledError），而线程池中的 Future 只能在未开始时被取消。

7.7 重试策略：run_with_retry

重试逻辑定义在 libs/langgraph/langgraph/pregel/_retry.py 中，run_with_retry 是同步版本：

def run_with_retry(task, retry_policy, configurable=None):
    retry_policy = task.retry_policy or retry_policy
    attempts = 0
    config = task.config
    if configurable is not None:
        config = patch_configurable(config, configurable)

    while True:
        try:
            # 清除上次尝试的写入
            task.writes.clear()
            # 执行任务
            return task.proc.invoke(task.input, config)
        except ParentCommand as exc:
            # Command 路由到当前图或父图
            ns = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
            cmd = exc.args[0]
            if cmd.graph in (ns, recast_checkpoint_ns(ns), task.name):
                for w in task.writers:
                    w.invoke(cmd, config)
                break
            elif cmd.graph == Command.PARENT:
                exc.args = (
                    replace(cmd, graph=_checkpoint_ns_for_parent_command(ns)),
                )
            raise
        except GraphBubbleUp:
            # 中断信号，直接向上传播
            raise
        except Exception as exc:
            if not retry_policy:
                raise

            # 查找匹配的重试策略
            matching_policy = None
            for policy in retry_policy:
                if _should_retry_on(policy, exc):
                    matching_policy = policy
                    break

            if not matching_policy:
                raise

            attempts += 1
            if attempts >= matching_policy.max_attempts:
                raise

            # 计算退避时间
            interval = matching_policy.initial_interval
            interval = min(
                matching_policy.max_interval,
                interval * (matching_policy.backoff_factor ** (attempts - 1)),
            )
            sleep_time = (
                interval + random.uniform(0, 1)
                if matching_policy.jitter else interval
            )
            time.sleep(sleep_time)

            # 标记为恢复模式
            config = patch_configurable(
                config, {CONFIG_KEY_RESUMING: True}
            )

7.7.1 RetryPolicy 数据结构

class RetryPolicy(NamedTuple):
    initial_interval: float = 0.5   # 首次重试等待时间（秒）
    backoff_factor: float = 2.0     # 退避倍数
    max_interval: float = 128.0     # 最大等待时间（秒）
    max_attempts: int = 3           # 最大尝试次数，含第一次（即最多再重试 2 次）
    jitter: bool = True             # 是否添加随机抖动
    retry_on: type[Exception] | Sequence[type[Exception]] | Callable = default_retry_on

7.7.2 退避算法可视化

flowchart TB
    subgraph "指数退避 + 抖动"
        A1["尝试 1: 执行失败"] --> W1["等待 0.5s + jitter"]
        W1 --> A2["尝试 2: 执行失败"]
        A2 --> W2["等待 1.0s + jitter"]
        W2 --> A3["尝试 3: 执行失败"]
        A3 --> FAIL["达到 max_attempts=3\n抛出异常"]
    end
    subgraph "退避时间计算"
        direction TB
        F["interval = initial_interval * backoff_factor^(attempts-1)"]
        F --> C["interval = min(interval, max_interval)"]
        C --> J{"jitter?"}
        J -->|是| JV["sleep = interval + random(0,1)"]
        J -->|否| NJ["sleep = interval"]
    end

7.7.3 多策略匹配

LangGraph 支持为同一个节点配置多个重试策略，按照异常类型进行匹配：

def _should_retry_on(retry_policy, exc):
    if isinstance(retry_policy.retry_on, Sequence):
        return isinstance(exc, tuple(retry_policy.retry_on))
    elif isinstance(retry_policy.retry_on, type):
        return isinstance(exc, retry_policy.retry_on)
    elif callable(retry_policy.retry_on):
        return retry_policy.retry_on(exc)

这允许精细的重试控制：

# 对 API 限流重试更多次，对网络错误快速重试
graph.add_node(
    "llm_call", llm_fn,
    retry_policy=[
        RetryPolicy(max_attempts=5, initial_interval=2.0,
                     retry_on=RateLimitError),
        RetryPolicy(max_attempts=3, initial_interval=0.1,
                     retry_on=ConnectionError),
    ]
)

7.7.4 重试中的状态管理

重试时有两个关键的状态操作：

task.writes.clear()：清除上次尝试的写入。这确保失败的写入不会被保留——只有成功执行的写入才会被提交。
CONFIG_KEY_RESUMING: True：标记为恢复模式。这通知子图"你正在被重试"，子图可以据此跳过已完成的步骤。

7.7.5 可观测性和分布式 runtime 的两个额外细节

上面的伪代码抓住了重试主干，但真实 libs/langgraph/langgraph/pregel/_retry.py 还有两处值得单独看的工程细节：

1. Python 3.11+ 的 exc.add_note 注入任务身份（libs/langgraph/langgraph/pregel/_retry.py:145-146）：

# _retry.py:30
SUPPORTS_EXC_NOTES = sys.version_info >= (3, 11)

# _retry.py:145 (在 except Exception 分支里)
if SUPPORTS_EXC_NOTES:
    exc.add_note(f"During task with name '{task.name}' and id '{task.id}'")

生产环境里一个链路几十个节点、每个节点可能多次重试——一个原始的 TimeoutError 或 ValidationError 在日志里显示的回溯栈没法直接指出"是哪个节点的哪次尝试出错"。add_note 是 PEP 678 给 Python 3.11 引入的 API——允许在异常对象上附加任意字符串注解、在 traceback 打印时自动显示。LangGraph 把任务名和 ID 注入进去、让运维能一眼看到：

Traceback (most recent call last):
  ...
httpx.TimeoutException: Request timed out after 30.0 seconds
During task with name 'llm_call_node' and id '7f3a1b2c-...'

这一行注解没改变异常类型、没破坏栈、也不影响 except 匹配——是纯增量的可观测性。SUPPORTS_EXC_NOTES 常量用 sys.version_info 在模块加载时判断、Python 3.10 及以下直接跳过这行，保持了向后兼容。logger.info(...) 的 retry 日志（libs/langgraph/langgraph/pregel/_retry.py:180-183）也带 exc_info=exc 让完整堆栈进监控。

2. _checkpoint_ns_for_parent_command——处理 namespace 里的纯数字段（libs/langgraph/langgraph/pregel/_retry.py:57-83）：

§7.7 的代码里，recast_checkpoint_ns(ns) 只用来判断 Command 是否指向当前图；Command.PARENT 分支计算父图 namespace 时调用的是专用辅助函数 _checkpoint_ns_for_parent_command。看它的注释和实现：

# _retry.py:57
def _checkpoint_ns_for_parent_command(ns: str) -> str:
    """Return the checkpoint namespace for the parent graph.

    The checkpoint namespace is a `|`-separated path. Each segment is usually
    of the form `name:task_id` (e.g. `parent_first:<uuid>|node:<uuid>`), but the
    runtime may also insert a purely-numeric segment (e.g. `|1`) to disambiguate
    concurrent tasks (e.g. `parent_first:<uuid>|1|node:<uuid>`).

    Numeric segments are not real path levels, so we drop them before computing
    the parent namespace.
    """
    parts = ns.split(NS_SEP)
    # Drop any trailing numeric selectors for the current frame
    while parts and parts[-1].isdigit():
        parts.pop()
    # Drop the current frame segment itself
    if parts:
        parts.pop()
    # Drop any trailing numeric selectors for the parent frame
    while parts and parts[-1].isdigit():
        parts.pop()
    return NS_SEP.join(parts)

这段代码在解决什么问题？同一个节点里先后调用多个子图时，要把它们的 namespace 区分开——PregelLoop 初始化时若父任务 scratchpad 的 subgraph_counter() 大于 0，就在 checkpoint_ns 后面追加一个纯数字段（libs/langgraph/langgraph/pregel/_loop.py:261-272）。在 pin 环境里实跑：节点 parent_first 里先后调用两个子图，第一个子图里节点看到的 namespace 是 parent_first:<id>|node:<id>，第二个则是 parent_first:<id>|1|node:<id>。这些数字段不是真正的图层级——计算父图 namespace 时得跳过它们。

所以parts.pop() 要做三轮：先剥掉当前帧的数字后缀、再剥掉当前帧本身、再剥掉父帧的数字后缀。recast_checkpoint_ns 做的是另一件事：去掉每段里的 task id、丢掉数字段，得到用于比较的"纯节点路径"（libs/langgraph/langgraph/_internal/_config.py:34），不能直接拿来当父图的 namespace。

这个函数来自 2026 年 2 月的修复提交 f178eb82（PR #6864）：此前的写法只处理末尾的数字段，第二次嵌套调用里发出的 Command.PARENT 会被算成形如 parent_first:<id>|1 的错误父图标识，路由不到父图节点。该提交附带的回归测试 libs/langgraph/tests/test_parent_command.py 就是上面这种"一个节点先后调用两个子图"的本地场景，在 pin 环境里实跑通过。

7.8 缓存策略

节点级别的缓存（CachePolicy + langgraph.cache 的 BaseCache）最早随 langgraph 0.4.4 发布，用于避免对确定性节点的重复计算；下面按本专栏基准 1.1.6 的实现讲。

开讲前先记一条静默失效：add_node(..., cache_policy=CachePolicy(...)) 只是往 PregelNode 上挂了个策略，真正开关缓存的是 compile(cache=...)（libs/langgraph/langgraph/graph/state.py:1042 的 cache 形参，:1160 处作为 cache=cache 传给 CompiledStateGraph 的构造）。只配 cache_policy 而不给 compile 传 cache，下面两处都会直接短路——match_cached_writes 在 self.cache is None 时返回空元组（libs/langgraph/langgraph/pregel/_loop.py:1096，异步版 :1292），put_writes 同样在 self.cache is None 时直接 return（:1121 / :1317）。没有任何警告或报错：cache_key 照算不误，图照跑，只是一次都不命中，你会以为缓存生效了而其实每次都在重算。编译期也不会帮你检查——cache_policy 在 graph/state.py 里只被透传，从未和 cache 交叉校验。

7.8.1 CachePolicy 和 CacheKey

@dataclass
class CachePolicy:
    key_func: Callable = default_cache_key  # 缓存键生成函数
    ttl: int | None = None                  # 过期时间（秒）

class CacheKey(NamedTuple):
    ns: tuple[str, ...]   # 命名空间（标识节点身份）
    key: str              # 缓存键（输入哈希）
    ttl: int | None       # 过期时间

缓存键由三部分组成：

命名空间：(CACHE_NS_WRITES, identifier(proc), name) —— 标识是哪个节点的哪个实现
键：xxh3_128_hexdigest(key_func(input)) —— 输入数据的哈希
TTL：可选的过期时间

7.8.2 缓存匹配流程

缓存匹配在两个地方发生：

1. 超步开始时（tick 之后）：

while loop.tick():
    for task in loop.match_cached_writes():
        loop.output_writes(task.id, task.writes, cached=True)
    # 只执行没有缓存命中的任务
    for _ in runner.tick(
        [t for t in loop.tasks.values() if not t.writes],
        ...
    ):

match_cached_writes 批量查询缓存：

# SyncPregelLoop.match_cached_writes，libs/langgraph/langgraph/pregel/_loop.py:1095
# （基类 PregelLoop 在 :575 只抛 NotImplementedError；异步版 amatch_cached_writes 在 :1291）
def match_cached_writes(self):
    if self.cache is None:
        return ()
    cached = {
        (t.cache_key.ns, t.cache_key.key): t
        for t in self.tasks.values()
        if t.cache_key and not t.writes  # 只查没有写入的任务
    }
    matched = []
    for key, values in self.cache.get(tuple(cached)).items():
        task = cached[key]
        task.writes.extend(values)  # 将缓存的写入填充到任务
        matched.append(task)
    return matched

2. 任务写入保存时（put_writes）：

def put_writes(self, task_id, writes):
    super().put_writes(task_id, writes)
    if not writes or self.cache is None or not hasattr(self, "tasks"):
        return
    task = self.tasks.get(task_id)
    if task is None or task.cache_key is None:
        return
    # 异步保存到缓存
    self.submit(
        self.cache.set,
        {(task.cache_key.ns, task.cache_key.key):
         (task.writes, task.cache_key.ttl)}
    )

flowchart TB
    START[超步开始] --> PREP[prepare_next_tasks\n创建任务，计算 cache_key]
    PREP --> MATCH[match_cached_writes\n批量查询缓存]
    MATCH --> HIT{缓存命中?}
    HIT -->|是| FILL[填充 task.writes\n发出 cached=True 事件]
    HIT -->|否| EXEC[runner.tick 执行任务]
    EXEC --> COMMIT[commit 提交结果]
    COMMIT --> SAVE[put_writes 保存写入]
    SAVE --> CACHE_SAVE[异步保存到缓存]
    FILL --> AFTER[after_tick 应用写入]
    CACHE_SAVE --> AFTER

7.8.3 缓存键的生成

默认的 default_cache_key 函数把位置参数和关键字参数一起 pickle 序列化：

# libs/langgraph/langgraph/_internal/_cache.py:26-31（省略 docstring）
def default_cache_key(*args: Any, **kwargs: Any) -> str | bytes:
    import pickle

    # protocol 5 strikes a good balance between speed and size
    return pickle.dumps(
        (_freeze(args), _freeze(kwargs)), protocol=5, fix_imports=False
    )

_freeze 先把不可哈希的结构转成可哈希、可确定性序列化的形式：字典按键排序（避免键顺序不同的字典算出不同的键），列表转成元组，带 tobytes() 的数组类对象转成字节。

在 prepare_single_task 中，缓存键的完整计算过程：

if cache_policy:
    args_key = cache_policy.key_func(val)  # 用户定义或默认的键函数
    cache_key = CacheKey(
        (CACHE_NS_WRITES, identifier(proc) or "__dynamic__", name),
        xxh3_128_hexdigest(
            args_key.encode() if isinstance(args_key, str) else args_key
        ),
        cache_policy.ttl,
    )

identifier(proc) 返回处理器的唯一标识（通常基于函数的模块和名称），确保不同实现的同名节点不会共享缓存。

7.9 动态任务创建：accept_push

在 Functional API 中，任务可以在执行过程中动态创建子任务。这通过 PregelLoop.accept_push 方法实现：

# PregelLoop.accept_push，libs/langgraph/langgraph/pregel/_loop.py:422
# （:1110 是 SyncPregelLoop 的同名覆写）
def accept_push(self, task, write_idx, call=None):
    """接受一个来自正在执行的任务的 PUSH 请求。"""
    pushed = prepare_single_task(
        (PUSH, task.path, write_idx, task.id, call),
        None,
        checkpoint=self.checkpoint,
        ...
    )
    if pushed:
        # 发出调试事件
        self._emit("tasks", map_debug_tasks, [pushed])
        # 保存新任务
        self.tasks[pushed.id] = pushed
        # 匹配已有写入
        if not self.is_replaying:
            self._match_writes({pushed.id: pushed})
        return pushed

accept_push 作为 schedule_task 回调传递给 PregelRunner.tick，在任务执行过程中被调用：

# libs/langgraph/langgraph/pregel/_runner.py:533 起 _call 函数（节选，省略了"子任务已在运行"与 ERROR 写入两个分支）
if next_task := schedule_task(task(), scratchpad.call_counter(), call):
    if next_task.writes:
        # 已经有结果（从缓存或 Checkpoint 恢复），直接返回
        fut = concurrent.futures.Future()
        ret = next((v for c, v in next_task.writes if c == RETURN), MISSING)
        fut.set_result(ret)
    else:
        # 调度新任务到线程池
        fut = submit()(
            run_with_retry, next_task, retry_policy, ...,
            __next_tick__=True,  # 确保写入先提交
        )
        SKIP_RERAISE_SET.add(fut)
        futures()[fut] = next_task

动态创建的子任务使用 __next_tick__=True，确保父任务的当前写入先被提交到 PregelLoop，然后子任务才开始执行。这维护了写入的因果顺序。

7.10 异常处理的分层设计

LangGraph 的异常处理分为多个层次：

flowchart TB
    subgraph "异常分层"
        direction TB
        L1["GraphBubbleUp（基类）"]
        L2["GraphInterrupt\n(中断信号)"]
        L3["ParentCommand\n(跨图命令)"]
        L4["普通 Exception\n(可重试)"]
        L5["CancelledError\n(被取消)"]
        L1 --- L2
        L1 --- L3
    end
    subgraph "处理策略"
        direction TB
        S1["GraphBubbleUp: 不重试，不停止其他任务，向上传播"]
        S2["GraphInterrupt: 保存中断信息，合并后向上传播"]
        S3["ParentCommand: 路由到目标图，或继续冒泡"]
        S4["Exception: 匹配重试策略，超限后保存错误并停止其他任务"]
        S5["CancelledError: 保存错误到 Checkpoint"]
    end
    L1 --> S1
    L2 --> S2
    L3 --> S3
    L4 --> S4
    L5 --> S5

run_with_retry 中的异常处理

# libs/langgraph/langgraph/pregel/_retry.py:122 起（异步版在 :232 起）
while True:
    try:
        task.writes.clear()
        return task.proc.invoke(task.input, config)
    except ParentCommand as exc:
        # 检查 Command 的目标图
        if cmd.graph in (ns, recast_checkpoint_ns(ns), task.name):
            # 当前图处理
            for w in task.writers:
                w.invoke(cmd, config)
            break
        elif cmd.graph == Command.PARENT:
            # 设置父图命名空间后冒泡（见 §7.7.5）
            exc.args = (
                replace(cmd, graph=_checkpoint_ns_for_parent_command(ns)),
            )
        raise
    except GraphBubbleUp:
        raise  # 直接向上传播
    except Exception as exc:
        if not retry_policy:
            raise
        # 匹配并执行重试策略...

这种分层设计确保了：

控制流信号（Interrupt、Command）不会被重试逻辑意外捕获
普通异常可以根据策略重试
异常信息会被保存到 Checkpoint，支持调试和恢复

7.11 执行过程中的写入流

让我们追踪一次完整的写入流，从节点函数返回到 Channel 更新：

sequenceDiagram
    participant Node as 用户函数
    participant CW as ChannelWrite
    participant Writes as task.writes (deque)
    participant Runner as PregelRunner
    participant PL as PregelLoop
    participant Chan as Channels
    Note over Node: return {"count": 5}
    Node->>CW: invoke({"count": 5})
    Note over CW: _get_updates 提取元组
    CW->>CW: do_write(config, writes)
    Note over CW: config[CONF][CONFIG_KEY_SEND]
    Note over CW: 静态边的路由条目已由 flat_writers 并入同一个 ChannelWrite
    CW->>Writes: extend([("count", 5), ("branch:to:next", None)])
    Note over Runner: task.proc.invoke 完成
    Runner->>Runner: commit(task, None)
    Runner->>PL: put_writes(task.id, task.writes)
    Note over PL: 保存到 checkpoint_pending_writes
    Note over PL: 异步保存到 Checkpointer
    Note over PL: after_tick()
    PL->>Chan: apply_writes(checkpoint, channels, tasks, ...)
    Note over Chan: channels["count"].update([5])
    Note over Chan: channels["branch:to:next"].update([None])

关键观察：写入在任务执行过程中被收集到 task.writes 中，但直到 after_tick() 调用 apply_writes 时才真正应用到 Channel。这就是 BSP 模型的"步内隔离"——同一超步中的不同任务看不到彼此的写入。

7.12 设计决策分析

为什么使用弱引用（weakref）？

PregelRunner 中的 submit 和 put_writes 都使用弱引用：

class PregelRunner:
    def __init__(self, *, submit, put_writes, ...):
        self.submit = submit          # weakref.WeakMethod
        self.put_writes = put_writes  # weakref.WeakMethod

PregelRunner 只经弱引用持有 PregelLoop 的方法，自身不延长 PregelLoop 的生命周期；tick 里注入的 commit 回调（weakref.WeakMethod(self.commit)）、_call 拿到的 futures、父任务引用也都是弱引用。这样 PregelLoop 退出上下文管理器后，相关对象不会因为这些互相引用而滞留。

为什么 writes 使用 deque 而非 list？

源码在注入 CONFIG_KEY_SEND 处特意注明 # deque.extend is thread-safe（libs/langgraph/langgraph/pregel/_algo.py:701）。在并行执行场景中，写入可能来自不同线程，选 deque 就是看中这一点。

为什么 PUSH 子任务使用 SKIP_RERAISE_SET？

当任务 A 动态创建子任务 B 时，B 的异常应该由 A 处理（通过返回的 Future），而不是由 PregelRunner 的 _panic_or_proceed 处理。SKIP_RERAISE_SET 记录了哪些 Future 的异常应该被跳过：

# _runner.py
fut = submit()(run_with_retry, next_task, ...)
SKIP_RERAISE_SET.add(fut)  # 标记：异常由父任务处理
futures()[fut] = next_task

这实现了"异常沿调用链传播"的语义，而非"所有异常在顶层汇合"的语义。

为什么重试时清除 writes 而不是创建新 deque？

task.writes.clear()  # 而不是 task = replace(task, writes=deque())

因为 PregelExecutableTask 是 frozen=True 的 dataclass，不能替换字段。同时，config 中注入的 CONFIG_KEY_SEND 引用的是原始 deque 的 extend 方法——如果创建新 deque，写入会丢失到旧对象中。clear() 在不改变引用的前提下清空内容。

7.12.1 `langgraph/pregel/` 22 文件 11392 行的真实骨架

读到这里你已经接触了 runner、executor、retry、algo 四个核心模块——把它们放在整个 pregel/ 目录的尺度看——

文件	行	本章涉及	角色
`main.py`	3718	-	`Pregel` 主类——invoke、stream、astream 的用户入口（`Pregel` 本身没有 `compile`，编译在 `StateGraph.compile`）
`_loop.py`	1423	间接	`PregelLoop` 超步主循环——驱动 runner、处理 checkpoint、管理 channel 版本
`_algo.py`	1258	部分	任务规划——`prepare_next_tasks` / `apply_writes` / channel version 递增算法
`remote.py`	1191	-	RemoteGraph——跨进程调用其他 LangGraph Graph 的 client
`_runner.py`	768	核心	本章 §7.4 主角——`FuturesDict` + `PregelRunner.tick/atick`
`_retry.py`	318	§7.7	`run_with_retry / arun_with_retry` + 指数退避实现
`_draw.py`	294	-	`draw_graph`：为 `get_graph` 构建图结构，供可视化
`protocol.py`	288	部分	`PregelProtocol` 抽象——Pregel 和 RemoteGraph 的公共超类
`debug.py`	279	-	`map_debug_tasks` / `map_debug_checkpoint` 等 debug 事件映射
`_read.py`	277	间接	`ChannelRead` 与 `PregelNode` 定义
`_call.py`	269	-	Functional API 的底层支撑——`call()`、`get_runnable_for_task`、`identifier`（`@task` / `@entrypoint` 本身在 `libs/langgraph/langgraph/func/__init__.py`）
`_messages.py`	250	-	`StreamMessagesHandler`——实现 `stream_mode="messages"` 的回调处理器
`_executor.py`	223	§7.5-7.6	`BackgroundExecutor` + `AsyncBackgroundExecutor`
`_utils.py`	218	间接	小工具函数
`_write.py`	192	§7.11	`ChannelWrite` 以及写入路径
`_io.py`	174	-	通道读取与输入/输出映射（`read_channels`、`map_input`、`map_output_*`）
`_validate.py` / `_checkpoint.py` / `types.py` / `__init__.py` / `_log.py` / `_config.py`	≤ 120	-	校验 / 公共类型 / 薄壳

三条值得记住的观察——

main.py 3718 行占 32.6%——单一 Pregel 类把 invoke / stream / astream / get_graph / get_state / update_state / get_state_history 等 public API 塞在一起——这是 LangGraph "一个类是主入口" 的设计取舍——对立面是 LangChain 把 LCEL 切成 Runnable / RunnableLambda / RunnableParallel 等十几个类
_algo.py 1258 行是第三大——任务规划（看哪些 channel 有新值、对应激活哪些 node、生成 PULL 任务列表）的纯算法层独立成文件——这让 PregelLoop 可以不关心"下一步做什么"、只负责"怎么做"
remote.py 1191 行接近 _algo.py——RemoteGraph 是调用实现 LangGraph Server API 的远程部署的客户端，它的代码量与本章主讲的 runner + executor + retry 三个文件之和（1309 行）相当

7.13 PregelRunner 源码的几个容易误读之处

读 _runner.py 的 768 行、有几个地方第一次读容易误解——集中澄清能省去后续大量试错。以下行号以本专栏基准 langgraph 1.1.6（commit c3ba590）为准——

误读 1：`yield` 不是协程——是"让调用方插入流式消费"

PregelRunner.tick 的签名返回 Iterator[None]——方法体里散布着 yield 语句：

# _runner.py: PregelRunner.tick 内部
yield  # 1: 调度前先让出一次（源码注释：give control back to the caller）
# ...
while len(futures) > 0:
    done, inflight = concurrent.futures.wait(...)
    # 处理完成的 futures
    if _should_stop_others(done):
        break
    yield  # 2: 每 wait 一批完成后让出
# ...
yield  # 3: 最后的收尾让出

这三个 yield 不是 asyncio 的"让出事件循环"——它们是 Python 生成器的 yield、唯一作用是让调用方取回控制权。调用方是 Pregel.stream，它这样消费：

# libs/langgraph/langgraph/pregel/main.py:2733 起，Pregel.stream 内部（节选）
while loop.tick():
    for task in loop.match_cached_writes():
        loop.output_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,
        get_waiter=get_waiter,
        schedule_task=loop.accept_push,
    ):
        # 每一次 yield 后，在这里把已产生的流式输出交给用户
        yield from _output(stream_mode, print_mode, subgraphs, stream.get, ...)
    loop.after_tick()

不是协程 = 不需要 async 即可实现"每完成一个 task 就输出一次"——这是 LangGraph 流式输出能在同步 API 里工作的关键设计。async 版本 atick 才用真正的 await、但机械上结构一致。新人最容易犯的错是把 Iterator[None] 理解成"异步"、结果 debug 时把断点打在错误位置。

误读 2：`SKIP_RERAISE_SET` 是模块级 `WeakSet`——不是实例属性

源码第 66-68 行定义——

SKIP_RERAISE_SET: weakref.WeakSet[concurrent.futures.Future | asyncio.Future] = (
    weakref.WeakSet()
)

这是模块级全局变量、所有 PregelRunner 实例共享一个 WeakSet。为什么能这样？因为 Future 对象全局唯一——A Runner 的 fut_1 不可能和 B Runner 的 fut_2 ID 冲突。WeakSet 的关键在于——当 Future 被垃圾回收、自动从 Set 里消失、不需要显式清理。这避免了内存泄漏——长时间运行的 LangGraph 服务如果用普通 set、每次 PUSH 子任务都往里塞 Future、又不清理，集合会一直增长。读源码时容易盯着实例属性找不到它。

误读 3：`FuturesDict` 继承 dict——不是 Mapping

第 71 行——

class FuturesDict(Generic[F, E], dict[F, PregelExecutableTask | None]):
    event: E
    callback: weakref.ref[...]
    counter: int
    done: set[F]
    lock: threading.Lock

它是 dict 的子类、有真实的 []、pop()、len() 能力——不是抽象接口。这带来两个真实后果——

futures[fut] = t 触发 __setitem__ 重载、在 super 调用后加计数器递增和 add_done_callback
futures.pop(fut) 是 dict 原生行为、不触发计数器递减——计数器递减在 on_done 回调里发生、独立于 pop

这个"set 时加计数、回调时减计数"的非对称设计很微妙——如果想自己改 Runner、不理解这个会把计数器搞乱、整个并发语义崩坏。线程安全靠 lock: threading.Lock 守护——event.wait() 的唤醒靠 on_done 里 event.set()。lock 只包住对 counter、done 与 event 的修改，on_done 里的 commit 回调本身在锁外执行。

误读 4：单任务快速路径并非总能走到

§7.4.3 的快速路径要同时满足三个条件：本步只有一个任务、没有超时（timeout）、没有 waiter（libs/langgraph/langgraph/pregel/_runner.py:164），其中 timeout 来自图的步超时配置。另外，图的 stream 方法在 stream_eager、subgraphs=True、或 stream_mode 含 "messages" / "custom" 时都会创建 waiter（libs/langgraph/langgraph/pregel/main.py:2706-2711），这时即便一步只有一个节点，也照样提交到线程池执行。在 pin 环境里实跑同一个单节点图：invoke 时节点运行在 MainThread，stream(..., stream_mode="custom") 时则运行在 ThreadPoolExecutor-… 线程里。

快速路径也不改变异常与上下文的语义：两条路径抛出的异常都用同一组 EXCLUDED_FRAME_FNAMES 裁剪 traceback（libs/langgraph/langgraph/pregel/_runner.py:56）；线程池路径里的上下文变量由 BackgroundExecutor.submit 的 copy_context() + ctx.run 传递（§7.5）。所以把它理解为"省掉一次线程切换"的优化即可。

7.14 super-step vs DAG / 协程 / Actor——四种并发调度模式对照

LangGraph 的"super-step（超步）"来自 Google 2010 年 Pregel paper 的 BSP（Bulk Synchronous Parallel）模型——它不是常见的并发范式。把它和当代工程里最主流的三种调度模式对照、能更清楚看到 LangGraph 的取舍。

7.14.1 模式四象限

flowchart LR
    subgraph "super-step (BSP)"
        direction TB
        BSPS["步 N：所有 task 并行执行"] --> BSPA["after_tick：原子 apply_writes"]
        BSPA --> BSPN["步 N+1：按新 channel 版本激活 task"]
    end
    subgraph "DAG (Airflow / Prefect)"
        direction TB
        DAGA["task A 完成"] --> DAGB["依赖 A 的 B 立即可跑"]
        DAGB --> DAGC["一跑到完、无重复访问同一节点"]
    end
    subgraph "协程 pipeline (asyncio / Go channel)"
        direction TB
        CO1["producer →"] --> CO2["stage1 →"]
        CO2 --> CO3["stage2 →"]
        CO3 --> CO4["consumer"]
    end
    subgraph "Actor (Akka / Erlang / Ray)"
        direction TB
        ACT1["actor A ←→ mailbox"]
        ACT2["actor B ←→ mailbox"]
        ACT1 -.-> ACT2
        ACT2 -.-> ACT1
    end

7.14.2 语义对比表

维度	super-step (LangGraph)	DAG (Airflow)	协程 (asyncio)	Actor (Ray)
同一节点可多次激活	是（多步循环）	否（一次 run）	流式、不谈"节点"身份	是（actor 接收多消息）
状态模型	共享 Channel + BSP 隔离	每 task 独立输入输出	隐式在队列里	每 actor 私有可变状态
并行粒度	步内全部 task 并行	依赖满足即并行	stage 间 pipeline 并行	无限制、actor 间异步
步间屏障	是（apply_writes 原子）	无全局屏障	无	无
失败恢复	Checkpoint → 超步粒度重放	task 粒度重试	无原生支持	无原生支持、需 snapshot
中断 / 人类介入	一等公民（GraphInterrupt）	需手搭	需手搭	需手搭
循环支持	一等公民（Channel 版本递增）	困难（DAG 定义不含环）	需手搭	困难（消息循环易死锁）
典型场景	中小规模的 agent 图	批量 task 的 ETL	生产者/消费者流	大规模 actor 集群
调度开销	中（每步屏障）	低（纯拓扑）	低	低
观测粒度	每 task 的 writes + checkpoint	每 task 日志	弱	actor 消息追踪

7.14.3 为什么 LangGraph 选择 super-step？

核心原因有三——

1. Agent 的天然循环属性——LLM agent 的执行是"思考 → 调工具 → 看结果 → 再思考"的循环、同一个 agent 节点会被反复激活多次。DAG 模型不支持循环（或者要展开成重复 task、观测困难）；Actor 模型支持但状态跟踪麻烦（每个 actor 实例各自的 state 要对账）。super-step 的 Channel 版本机制天然适配——Channel 值变了就再激活一次、同一个 agent 节点名在历史上有 N 次 PregelExecutableTask 实例、都被 checkpoint 完整记录。

2. 人类介入的确定性语义——用户在第 5 步触发 interrupt、系统应当知道"这一步没跑完、下次从这一步开始"。BSP 的"步"是天然的 checkpoint 粒度——每步之间原子 apply_writes、之间不保存半成品状态——所以恢复逻辑极其简单（从最近 checkpoint 重放一步）。DAG 如果中途中断、已跑过但未产出的 task 难以幂等恢复；协程被中断基本等于丢数据。

3. 写入隔离带来正确性保证——一步内的 N 个并行 task 看到的 Channel 值是这步开始时的快照、不会读到"同步被另一 task 刚写的值"。这避免了一类最难调试的 bug：并发 task 互相看到对方"半成品"的更新、导致不可复现的行为。代价是性能——没有 write-through——但对 Agent 场景（延迟由 LLM 主导、不是调度主导），这个代价几乎免费。

代价面也真实存在——

步间屏障限制了吞吐——纯数据流场景（ETL、推理流水）中 super-step 明显不如 DAG 或协程流水
step 计数器隐含上限——默认 recursion_limit 是 DEFAULT_RECURSION_LIMIT（1.1.6 里为 10007，见第 6 章 §6.10）、agent 循环超过这个数就抛 GraphRecursionError——要收紧就在 config 里显式传小值
观测的维度高——同一节点多步激活让日志量线性放大（Channel 历史、多个 task_id 对应同一 name）

LangGraph 不是"最快的"调度框架、它是"最适合 agent 循环 + 人类介入 + checkpoint 恢复"的调度框架——理解这三个约束才能理解为什么它选择 BSP 而非更常见的 DAG 或协程。拿它当纯数据流 ETL 引擎用会显得偏重——它不是为那个问题设计的。

7.15 小结

本章深入分析了 LangGraph 的任务调度和并行执行机制。核心要点回顾：

PregelExecutableTask 是不可变的任务容器，其 proc 串联了用户逻辑和写入器，writes deque 作为线程安全的写入缓冲区
PULL vs PUSH：PULL 任务由 Channel 版本变更触发，输入从 Channel 读取；PUSH 任务由 Send API 创建，输入由调用者指定，不检查版本
PregelRunner 通过 FuturesDict 管理并发任务，支持单任务快速路径、超时控制、"任一失败则全部停止"语义
BackgroundExecutor 使用线程池 + copy_context 实现同步并行；AsyncBackgroundExecutor 使用 asyncio + Semaphore 实现异步并行和并发控制
重试策略支持指数退避、随机抖动、最大次数限制，以及按异常类型匹配的多策略组合。重试时清除旧写入并标记恢复模式
缓存策略通过节点身份 + 输入哈希生成 CacheKey，在超步开始时批量查询，命中后直接填充 writes 跳过执行
写入流遵循 BSP 模型：执行期间收集到 deque，commit 时保存到 Checkpointer，after_tick 时统一应用到 Channel

物理层面：整个 pregel/ 目录 22 文件 11392 行——本章主讲的 _runner.py（768）+ _executor.py（223）+ _retry.py（318）合计 1309 行、占 11.5%——是理解 LangGraph 运行时的最小必读子集。

7.16 与本专栏其他章节的串联——任务调度如何穿插全专栏

任务调度不是孤立的主题、它串联了本专栏几乎所有其他章节。把 §7 放在全专栏坐标系里、能更清楚每一节外接了谁——

7.16.1 向上依赖——state / channels / checkpoint

本章内容	依赖章节	关联点
PULL 任务从 Channel 读输入	§3 StateGraph + §4 Channels	`_proc_input` 通过 `proc.channels` 映射读取 Channel；reducer 是 Channel 的核心能力
`task.writes` 在 `apply_writes` 里汇入 Channel	§4 Channels	写入的合并规则来自 Channel 类型（LastValue / Topic / BinaryOperatorAggregate 等）
任务 ID 由 checkpoint_id + step + name 生成	§8 Checkpoint	`_match_writes` 靠确定性 ID 把恢复的任务对回已持久化的 writes
`CONFIG_KEY_RESUMING` 通知重试	§8 Checkpoint + §11 Subgraphs	子图据此跳过已完成步骤
`CachePolicy` 存储的 `(writes, ttl)`	§8 Checkpoint	共用 put_writes 基础设施、但写的是 cache namespace 不是 checkpoint namespace

7.16.2 向下支持——interrupt / command / send / streaming / multi-agent

本章内容	支撑章节	关联点
`GraphBubbleUp` 不触发重试	§9 Interrupt	`GraphInterrupt` 继承 `GraphBubbleUp`，runner 识别后合并各任务的中断再抛出
`ParentCommand` 的跨图路由	§10 Command	`recast_checkpoint_ns` / `_checkpoint_ns_for_parent_command` 专门为 Command 服务
PUSH 任务的 `Send.arg` 输入机制	§12 Send	除 Functional API 外，PUSH 任务都来自 Send
`yield` 让出控制权给 PregelLoop	§13 Streaming	Runner 每完成一个 task 就 yield 一次、PregelLoop 趁机把 writes 流式输出
`max_concurrency` Semaphore	§16 Prebuilt	`ToolNode` 同步路径复用 config 里的 `max_concurrency` 限制并行调工具（§16 讲过异步路径不读它）
`schedule_task` 回调的 PUSH 子任务	（Functional API 本专栏无专章，见本章 §7.9）	Functional API 的 `@task` 调用经 `CONFIG_KEY_CALL` 回到所在图的 `PregelLoop.accept_push`（§7.9），由该图自己的 Runner 调度

7.16.3 API 索引——本章出现过的符号速查

用本章时作为速查表：把所有本章介绍的模块、类、函数列成一张表、方便回头查阅时按 Ctrl+F 精准定位。

符号	类型	来源文件	本章小节
`PregelExecutableTask`	frozen dataclass	`types.py`	§7.2
`PregelRunner`	类	`libs/langgraph/langgraph/pregel/_runner.py`	§7.4
`PregelRunner.tick`	同步方法	`_runner.py`	§7.4.1
`PregelRunner.atick`	异步方法	`_runner.py`	§7.6
`PregelRunner.commit`	方法	`_runner.py`	§7.4.4
`FuturesDict`	dict 子类	`libs/langgraph/langgraph/pregel/_runner.py:71`	§7.4.2
`_should_stop_others`	自由函数	`libs/langgraph/langgraph/pregel/_runner.py:462`	§7.4.5
`_panic_or_proceed`	自由函数	`_runner.py`	§7.4.5
`SKIP_RERAISE_SET`	模块级 WeakSet	`libs/langgraph/langgraph/pregel/_runner.py:66`	§7.13
`BackgroundExecutor`	上下文管理器	`libs/langgraph/langgraph/pregel/_executor.py`	§7.5
`AsyncBackgroundExecutor`	异步上下文管理器	`_executor.py`	§7.6
`next_tick`	辅助函数	`_executor.py`	§7.5.2
`gated`	协程包装器	`_executor.py`	§7.6.1
`run_with_retry`	同步函数	`libs/langgraph/langgraph/pregel/_retry.py`	§7.7
`arun_with_retry`	异步函数	`_retry.py`	§7.7
`_should_retry_on`	辅助函数	`_retry.py`	§7.7.3
`_checkpoint_ns_for_parent_command`	辅助函数	`libs/langgraph/langgraph/pregel/_retry.py:57`	§7.7.5
`SUPPORTS_EXC_NOTES`	模块级常量	`libs/langgraph/langgraph/pregel/_retry.py:30`	§7.7.5
`RetryPolicy`	NamedTuple	`types.py`	§7.7.1
`CachePolicy`	dataclass	`types.py`	§7.8.1
`CacheKey`	NamedTuple	`types.py`	§7.8.1
`default_cache_key`	自由函数	`libs/langgraph/langgraph/_internal/_cache.py`	§7.8.3
`prepare_next_tasks`	规划器入口	`libs/langgraph/langgraph/pregel/_algo.py`	§7.4.6
`apply_writes`	写入应用函数	`_algo.py`	§7.4.6
`prepare_single_task`	单任务准备	`_algo.py`	§7.3
`PregelLoop.accept_push`	方法	`libs/langgraph/langgraph/pregel/_loop.py`	§7.9
`PregelLoop.match_cached_writes`	方法	`_loop.py`	§7.8.2
`GraphBubbleUp`	异常基类	`errors.py`	§7.10
`GraphInterrupt`	异常子类	`errors.py`	§7.10
`ParentCommand`	异常	`errors.py`	§7.10
`CONFIG_KEY_SEND` / `CONFIG_KEY_READ` / `CONFIG_KEY_RESUMING` / `CONFIG_KEY_TASK_ID`	配置键常量	`libs/langgraph/langgraph/_internal/_constants.py`	§7.2.3 / §7.7.4
`NO_WRITES` / `INTERRUPT` / `ERROR` / `RETURN`	写入标签常量	`libs/langgraph/langgraph/_internal/_constants.py`	§7.4.4 / §7.9

这个表可以当索引用——任何时候读 LangGraph 源码碰到本表中符号、可以直接回这里找所在章节。表中所有 行号/函数名 以本专栏基准 langgraph 1.1.6（commit c3ba590）为准、后续版本如有调整、以官方仓库 git blame 为准。

7.16.4 一个串联的心智图

flowchart TB
    subgraph "编译期 (§3 §4 §5)"
        SG[StateGraph] --> CMP[compile]
        CH[Channels + reducer] --> CMP
        CMP --> NODE["PregelNode 字典\ntrigger_to_nodes 倒排索引"]
    end
    subgraph "运行期核心 (§6 §7)"
        NODE --> LOOP["PregelLoop.tick\n(§6)"]
        LOOP --> ALGO["prepare_next_tasks\n(§7.4.6)"]
        ALGO --> TASKS[PregelExecutableTask 列表]
        TASKS --> CACHE["match_cached_writes\n(§7.8)"]
        CACHE --> RUNNER["PregelRunner.tick\n(§7.4)"]
        RUNNER --> EXEC["BackgroundExecutor\n(§7.5-7.6)"]
        EXEC --> RETRY["run_with_retry\n(§7.7)"]
        RETRY --> COMMIT["put_writes"]
        COMMIT --> APPLY["apply_writes\n→ updated_channels"]
        APPLY --> LOOP
    end
    subgraph "控制流 (§9 §10 §12)"
        RETRY -.->|"GraphInterrupt"| INT["§9 Interrupt\n合并中断"]
        RETRY -.->|"ParentCommand"| CMD["§10 Command\n跨图路由"]
        COMMIT -.->|"Send write"| SEND["§12 Send\nPUSH 新任务"]
        SEND -.-> ALGO
    end
    subgraph "持久化 (§8)"
        COMMIT -.-> CKPT["Checkpointer"]
        LOOP -.-> CKPT
        CKPT -.-> RESUME["恢复时 _match_writes"]
        RESUME -.-> ALGO
    end
    subgraph "输出 (§13)"
        RUNNER -.->|"yield"| STREAM["§13 Streaming\nstream_mode"]
        COMMIT -.-> STREAM
    end

这张图要传达的最重要一条信息是——任务调度（§7）不是一个孤立的"模块"。节点执行期间产生的用户可观察行为（节点内的 interrupt、节点返回的 Command、节点里发出的 stream 输出）都要经过 PregelRunner.tick → run_with_retry → commit 这条路径，配了 checkpointer 时再经 put_writes 落盘。所以本章是后续 §8-§13 各"运行期"章节的共同底座。

读完本章后建议的学习顺序：→ §8 Checkpoint（深化 put_writes 的持久化路径）→ §9 Interrupt（深化 GraphBubbleUp 的分支）→ §12 Send（深化 PUSH 任务的来源）→ §13 Streaming（深化 yield 的消费端）→ §11 Subgraphs / §17 Multi-agent（看整套机制在跨图和多 Agent 场景的组合）。这条路径按"由近及远"的耦合度排序。

7.16.5 本章与工程实践的三条直接收益

最后把抽象理论落到三条你做 LangGraph 项目时立即能用的工程结论上——

结论 1：线性图请勿多开线程。如果你的图是"开始 → A → B → C → 结束"这种无分叉的线性结构、每步只有一个节点激活、PregelRunner 的单任务快速路径不会占用线程池（前提是没设 step_timeout，也没用会创建 waiter 的流式模式，见 §7.13 误读 4）——给线程池配大了也没用、线程池大小只影响有分叉的步。配置 max_concurrency 对这种图也无效。

结论 2：重试策略要匹配失败类型。retry_policy 支持按异常类型配多个策略——LLM API 的 RateLimitError 给 5 次、首次退避 2 秒；网络瞬时抖动的 ConnectionError 给 3 次、首次退避 0.1 秒——不要一股脑全配一个 max_attempts=3。生产环境里这个细节直接决定"该重试的重试够、不该重试的早点失败"——底层是利用了 _should_retry_on 的多策略顺序扫描逻辑（§7.7.3：第一个匹配的策略生效）。

结论 3：Command.PARENT 没路由到预期父图时——先看 _checkpoint_ns_for_parent_command。§7.7.5 讲过、同一个节点里先后调用多个子图时，第二次起的 namespace 会带数字段（|1、|2），计算父图 namespace 需要跳过这些段；1.1.6 已包含对应修复（PR #6864）。遇到"Command.PARENT 没路由到预期图"的 bug、优先打印 checkpoint_ns 看它的字符串格式。