阶段 4 的核心其实就三个动词：Plan（备菜）→ Execute（炒菜）→ Update（结账）。今天讲头尾两个，都在 _algo.py 里：

• prepare_next_tasks(...)（_algo.py:392）= Plan：读当前 checkpoint 和 channels，算出"这一步该激活哪些节点"，把每个节点的输入读好、打包成一个个 task。
• apply_writes(...)（_algo.py:232）= Update：把这一步所有 task 攒的写入按 reducer 合并进 channel、给被写的 channel 升版本号，并返回"这一步到底更新了哪些 channel"。

先看整体结构。函数很长，但骨架就两段：先收 PUSH 任务（Send 动态派发），再收 PULL 任务（被 channel 触发的普通节点）。_algo.py:437-513：

# _algo.py:437
def prepare_next_tasks(checkpoint, pending_writes, processes, channels, ...) -> dict[str, ...]:
    tasks: list[PregelTask | PregelExecutableTask] = []
    # ① 消费 PUSH 队列：上一步 Send 出去的动态任务（Day 16 map-reduce）
    tasks_channel = cast(Topic[Send] | None, channels.get(TASKS))     # 442
    if tasks_channel and tasks_channel.is_available():
        for idx, _ in enumerate(tasks_channel.get()):                # 444
            if task := prepare_single_task((PUSH, idx), None, ...):   # 445
                tasks.append(task)
    # ② 计算 PULL 候选节点（见 L03 的优化）
    if updated_channels and trigger_to_nodes:                        # 475
        ...candidate_nodes = 只查被更新 channel 触发的那几个节点...
    elif not checkpoint["channel_versions"]:                         # 483
        candidate_nodes = ()                                         # 全新空图，没人可跑
    else:
        candidate_nodes = processes.keys()                           # 兜底：全图扫一遍
    # ③ 逐个候选节点尝试造 PULL 任务
    for name in candidate_nodes:                                     # 490
        if task := prepare_single_task((PULL, name), None, ...):     # 491
            tasks.append(task)
    return {t.id: t for t in tasks}                                  # 513 按 task_id 建字典

L02 里那段"算 candidate_nodes"是 Pregel 能扛大图的关键优化。放大看（_algo.py:468-486）：

# _algo.py:468 —— 注释原文：an optimization that allows which nodes will be active
if updated_channels and trigger_to_nodes:          # 475 上一步更新了哪些 channel + channel→节点反查表
    triggered_nodes: set[str] = set()
    for channel in updated_channels:               # 478 只遍历"变了的" channel
        if node_ids := trigger_to_nodes.get(channel):
            triggered_nodes.update(node_ids)        # 480 把订阅它的节点加进候选
    candidate_nodes: Iterable[str] = sorted(triggered_nodes)  # 482 排序保证确定性
elif not checkpoint["channel_versions"]:           # 483 从没写过任何 channel
    candidate_nodes = ()                            # 484 空图，没有候选
else:
    candidate_nodes = processes.keys()              # 486 兜底：老实扫全部节点

候选节点只是"可能要跑"，真正的判定在 prepare_single_task 的 PULL 分支里调 _triggers（_algo.py:597-612）：

# _algo.py:597
elif task_path[0] == PULL:
    name = cast(str, task_path[1])
    proc = processes[name]
    if _triggers(                                    # 606 真正判定"该不该跑"
        channels,
        checkpoint["channel_versions"],              # 608 每个 channel 现在第几版
        checkpoint["versions_seen"].get(name),       # 609 这个节点上次看到第几版
        checkpoint_null_version, proc,
    ):
        triggers = tuple(sorted(proc.triggers))      # 613 命中触发的通道，排序
        ...造 task...

核心比较就在 _triggers（_algo.py:1260）：

# _algo.py:1260
def _triggers(channels, versions, seen, null_version, proc) -> bool:
    if seen is None:                                 # 1267 从没跑过：只要订阅的通道有值就触发
        for chan in proc.triggers:
            if channels[chan].is_available():
                return True
    else:
        for chan in proc.triggers:                   # 1272 跑过：比版本号
            if channels[chan].is_available() and versions.get(
                chan, null_version
            ) > seen.get(chan, null_version):        # 1273 通道现版本 > 我已见版本 → 触发
                return True
    return False                                     # 1277 都没新版本 → 不跑

回到 L02 那段 PUSH 消费（_algo.py:442-466）。它处理的是"上一步某节点调了 Send"产生的动态任务：

# _algo.py:442
tasks_channel = cast(Topic[Send] | None, channels.get(TASKS))
if tasks_channel and tasks_channel.is_available():
    for idx, _ in enumerate(tasks_channel.get()):   # 444 TASKS 通道里每个 Send 一条
        if task := prepare_single_task(
            (PUSH, idx), None,                        # 446 用下标 idx 标识第几个 Send
            checkpoint=checkpoint, channels=channels, step=step, stop=stop,
            for_execution=for_execution, ...
        ):
            tasks.append(task)

换到结账端。执行器把本步任务跑完，after_tick 调 apply_writes（_algo.py:232）。前半段：排序、更新"已见版本"、按通道分组写入：

# _algo.py:232
def apply_writes(checkpoint, channels, tasks, get_next_version, trigger_to_nodes) -> set[str]:
    tasks = sorted(tasks, key=lambda t: task_path_str(t.path[:3]))   # 256 排序=确定性
    bump_step = any(t.triggers for t in tasks)                       # 259 有真任务才推进步

    # ① 更新每个节点的"已见版本"：它触发时看到的那些通道的当前版本
    for task in tasks:                                               # 262
        checkpoint["versions_seen"].setdefault(task.name, {}).update({
            chan: checkpoint["channel_versions"][chan]
            for chan in task.triggers
            if chan in checkpoint["channel_versions"]
        })

    # ② 把所有 task 的写入按通道分组
    pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)   # 295
    for task in tasks:
        for chan, val in task.writes:                               # 297
            if chan in (NO_WRITES, PUSH, RESUME, INTERRUPT, ...):   # 298 保留通道跳过
                pass
            elif chan in channels:
                pending_writes_by_channel[chan].append(val)         # 309 归到对应通道
            else:
                logger.warning(f"...wrote to unknown channel {chan}, ignoring it.")

apply_writes 后半段：真正把值 update 进通道、给被写通道升版本、并处理"步末通知"（_algo.py:315-345）：

# _algo.py:315
updated_channels: set[str] = set()
for chan, vals in pending_writes_by_channel.items():        # 317
    if channels[chan].update(vals) and next_version is not None:  # 319 交给通道 reducer 合并
        checkpoint["channel_versions"][chan] = next_version # 320 被写→升到新版本
        if channels[chan].is_available():
            updated_channels.add(chan)                      # 323 记入"本步更新集"

# 这步没被写的通道，通知它们"新的一步开始了"（某些通道靠这个清空瞬时值）
if bump_step:                                               # 326
    for chan in channels:
        if channels[chan].is_available() and chan not in updated_channels:
            if channels[chan].update(EMPTY_SEQ) and ...:    # 329 空更新
                ...

# 若本步更新的通道都触发不了任何节点 = 可能是最后一步，通知所有通道"收尾"
if bump_step and updated_channels.isdisjoint(trigger_to_nodes):  # 336
    for chan in channels:
        if channels[chan].finish() and next_version is not None:  # 338
            ...
return updated_channels                                     # 345 交回给 Day 20 的闭环

✋ 10 分钟动手

# 1. 读 apply_writes 全貌（合并 + 升版本 + 收尾）
sed -n '232,345p' libs/langgraph/langgraph/pregel/_algo.py

# 2. 读 prepare_next_tasks 骨架（PUSH + PULL + 增量优化）
sed -n '437,513p' libs/langgraph/langgraph/pregel/_algo.py

# 3. 读触发核心 _triggers（版本比较那一行）
sed -n '1260,1277p' libs/langgraph/langgraph/pregel/_algo.py

# 4. 打印看看版本号怎么涨（每步 counter 加 1）
python -c "
from langgraph.graph import StateGraph, START, END
from typing import Annotated; import operator
class S(dict): pass
g = StateGraph(dict)
g.add_node('a', lambda s: {'n': s.get('n',0)+1})
g.add_edge(START,'a'); g.add_edge('a', END)
app = g.compile()
for c in app.stream({'n':0}, stream_mode='debug'):
    print(c['type'], c.get('payload',{}).get('name',''))
"