Pregel アルゴリズム層:中断・書き戻し・调度

源码版本1.2.9

役割

_algo.py は PregelLoop と PregelRunner の「数学カーネル」です。loop は「何周走るか」を、runner は「どう並列実行するか」を担い、algo は「各ステップのセマンティクス」を担います: いつ中断すべきか、ノードの書き戻した writes をどうチャネル (channel) へマージするか、次ステップでどのノードを触発するか。状態も IO も持たない——すべて純関数で、入力は checkpoint + チャネル + tasks、出力は新しい channel バージョン表 / updated_channels 集合 / 新しい task 辞書です。

三つのコア関数は PregelLoop.tick + after_tick の三段の呼び出しに対応します: should_interrupt(should_interrupt:155)は tick 中段で呼ばれ、実行前に GraphInterrupt を投げるか判定します; apply_writes(apply_writes:232)は after_tick 冒頭で呼ばれ、本ステップの全 task の writes をチャネルへマージし updated_channels を返します; prepare_next_tasks(prepare_next_tasks:392)は tick 冒頭で呼ばれ、現在の checkpoint から本ステップで実行する task を算出します。

言い換えると、この三関数が「Pregel の BSP ループで各ステップが何を見て、何をし、収尾後にチャネルがどうなるか」を決めます。これらを理解すれば Pregel の「状態変化規則」を理解したことになります。

設計動機

なぜアルゴリズムロジックを純関数に分けるのか?

テスト可能性:should_interrupt / apply_writes / prepare_next_tasks はすべて純関数で、入力は checkpoint dict + channel mapping、出力はデータ構造です。Pregel の IO、submit、checkpointer から離れて単体テストできます——これが libs/langgraph/tests/unit/ の大量テストの形です。
同期/非同期で共用:PregelLoop には SyncPregelLoop と AsyncPregelLoop の二つのサブクラスがありますが、algo 層は完全 side-effect なしで両サブクラスが同じロジックを共用します(tick calls prepare_next_tasks:612)。さもなくば同期/非同期でアルゴリズムを二度書きし、バグも二箇所で直すことになります。
updated_channels 最適化:apply_writes が updated_channels 集合を返し、prepare_next_tasks はこれを hint として受け取り、「全ノードを走査して trigger が発火したか見る」をスキップします(updated_channels hint:475-486)。大規模グラフではこのステップが O(N×M) の trigger 走査を O(updated×triggered) に下げ、最も重要な性能最適化です。
should_interrupt は versions_seen で重複除去:INTERRUPT チャネルの versions_seen(versions_seen INTERRUPT:163)は前回の interrupt 時に見たチャネルバージョン番号を記録します。「前回の interrupt 以降に新たなチャネル更新があった」場合のみ interrupt を考慮します——interrupt を触発後に resume してまた interrupt する無限ループを避けます。
apply_writes 末尾の finish 信号:bump_step and updated_channels.isdisjoint(trigger_to_nodes) のとき channels[chan].finish() を呼び(finish:336-342)、全チャネルに「これが最後のスーパーステップ」信号を送ります。恒久チャネル(LastValue など)はこの機会に自身を unavailable とし、prepare_next_tasks が以降どのノードも触発しないようにします——これがグラフの「自然死」です。

主要ファイル

should_interrupt:155-185 — グラフが中断すべきかチェック。まず versions_seen[INTERRUPT] で新更新があるか見て、次に触発された task が interrupt_nodes リストに入るか見る。
any_updates_since_prev_interrupt:161-168 — チャネルバージョン番号で「前回の interrupt 以降に更新があるか」を比較。interrupt 重複除去の核心。
interrupt_nodes filter:170-184 — interrupt_nodes == "*" は非 hidden task すべてを意味し、それ以外は task.name in interrupt_nodes でフィルタ。
apply_writes:232-345 — 本ステップの task の writes をチャネルへマージし、updated_channels 集合を返す。
bump_step:256-259 — いずれかの task が triggers を持てば bump_step=True。このステップがチャネルを消費し step カウンタを進めることを意味。
update seen versions:262-269 — 各 task が読んだ trigger チャネルのバージョン番号を versions_seen に記録。次回の _triggers 判定で「前回実行以降に新更新があるか」になる。
consume channels:284-292 — channels[chan].consume() を呼び、本ステップで読んだチャネルを消費済みにする。PULL チャネルの「一度読んだらクリア」セマンティクスの実装。
apply writes to channels:315-323 — channels[chan].update(vals) で writes を本当にチャネルへマージ。is_available() のチャネルのみ updated_channels に入る。
bump_step notify:326-333 — bump_step=True のとき本ステップで更新されなかった可用チャネルも update(EMPTY_SEQ) し、バージョン番号を進める——これらを購読しても本ステップで触発されなかったノードが次回 tick で誤って触発されないようにする。
finish signal:336-342 — 本ステップで更新触発されたノードがないとき、全チャネルに finish() 信号を送り、恒久チャネルが unavailable になりグラフが自然終了。
prepare_next_tasks:392-513 — 次ステップで走らせる task を算出。PUSH(Send fan-out 由来)と PULL(エッジ触発由来)がここから出る。
updated_channels optimization:475-486 — 前ステップの updated_channels + trigger_to_nodes で触発されたノード集合を逆引きし、全表走査をスキップ。
prepare_single_task:524 — 単一 task の構築。PUSH は prepare_push_task_functional / prepare_push_task_send、PULL は _triggers 判定 + _proc_input で入力取得。
_triggers:1260-1277 — ノードの trigger チャネルに「未読の新バージョンがあるか」を判定。PULL 调度の中核条件。
local_read:188-224 — ノード関数がチャネル値を読む入口。fresh=True で「自分が書いたばかりでまだチャネルにマージされていない」ローカルビューを読む、条件エッジ (conditional edge) の即書き即読みセマンティクスに対応。

データフロー

should_interrupt は三つの中で最も単純で、ロジックは「新更新あり + 現 task がリストに命中」です:

python

def should_interrupt(
    checkpoint: Checkpoint,
    interrupt_nodes: All | Sequence[str],
    tasks: Iterable[PregelExecutableTask],
) -> list[PregelExecutableTask]:
    """Check if the graph should be interrupted based on current state."""
    version_type = type(next(iter(checkpoint["channel_versions"].values()), None))
    null_version = version_type()  # type: ignore[misc]
    seen = checkpoint["versions_seen"].get(INTERRUPT, {})
    # interrupt if any channel has been updated since last interrupt
    any_updates_since_prev_interrupt = any(
        version > seen.get(chan, null_version)  # type: ignore[operator]
        for chan, version in checkpoint["channel_versions"].items()
    )
    # and any triggered node is in interrupt_nodes list
    return (
        [task for task in tasks if ...]
        if any_updates_since_prev_interrupt
        else []
    )

これは should_interrupt:155-185 からの引用です。鍵は versions_seen[INTERRUPT] という特殊レコード——これはあるノードが読んだバージョン番号ではなく、「前回 interrupt が起きたときの全チャネルバージョン番号スナップショット」です。any_updates_since_prev_interrupt はこのスナップショットと現在の channel_versions を比較し、いずれかのチャネルバージョン番号が高ければ「新更新あり」とみなし、interrupt 判定に入ります。これにより resume 後に同じ名前のノードが tick で触発されても、すぐにまた interrupt しません——チャネルに本当に新書き込みがない限り。

apply_writes は収尾段階の中核で、まず versions_seen を更新し、次に読んだチャネルを消費し、最後に新値を書きます:

python

# update seen versions
for task in tasks:
    checkpoint["versions_seen"].setdefault(task.name, {}).update(
        {
            chan: checkpoint["channel_versions"][chan]
            for chan in task.triggers
            if chan in checkpoint["channel_versions"]
        }
    )

# Consume all channels that were read
for chan in {
    chan
    for task in tasks
    for chan in task.triggers
    if chan not in RESERVED and chan in channels
}:
    if channels[chan].consume() and next_version is not None:
        checkpoint["channel_versions"][chan] = next_version

# Apply writes to channels
updated_channels: set[str] = set()
for chan, vals in pending_writes_by_channel.items():
    if chan in channels:
        if channels[chan].update(vals) and next_version is not None:
            checkpoint["channel_versions"][chan] = next_version
            if channels[chan].is_available():
                updated_channels.add(chan)

これは apply_writes 核心:262-323 からの引用です。三段のセマンティクスは明快です:

versions_seen を更新——各 task の現在の trigger チャネルバージョン番号を「このノードはこの版のチャネルを見た」という証拠として記録。次回の _triggers 判定で未来のバージョン番号と比較し、更新があったときだけ再触発します。
読んだチャネルを consume()——Topic のような「一度読んだらクリア」チャネルは consume() で内部バッファを空にし、バージョン番号の進行だけを残します。これが Send fan-out のメッセージが消費された後に他ノードを再触発しない原理です。
update(vals) で新値を書く——本ステップの全 task の各チャネルへの writes を集めて一度にマージし、reducer / LastValue などのチャネルのマージロジックを触発します。is_available()(値があり finish されていない)チャネルだけが updated_channels に入ります——finish() 後のチャネルはバージョン番号が進みますが、どのノードも触発しません。

prepare_next_tasks は tick 冒頭で呼ばれ、本ステップで走らせる task を決めます:

python

# Consume pending tasks
tasks_channel = cast(Topic[Send] | None, channels.get(TASKS))
if tasks_channel and tasks_channel.is_available():
    for idx, _ in enumerate(tasks_channel.get()):
        if task := prepare_single_task(
            (PUSH, idx), None, ..., for_execution=for_execution, ...
        ):
            tasks.append(task)

# This section is an optimization that allows which nodes will be active
# during the next step.
if updated_channels and trigger_to_nodes:
    triggered_nodes: set[str] = set()
    for channel in updated_channels:
        if node_ids := trigger_to_nodes.get(channel):
            triggered_nodes.update(node_ids)
    candidate_nodes: Iterable[str] = sorted(triggered_nodes)
elif not checkpoint["channel_versions"]:
    candidate_nodes = ()
else:
    candidate_nodes = processes.keys()

for name in candidate_nodes:
    if task := prepare_single_task((PULL, name), None, ..., for_execution=for_execution, ...):
        tasks.append(task)
return {t.id: t for t in tasks}

これは prepare_next_tasks 核心:441-513 からの引用です。二種の task がここで合流します:

PUSH task:TASKS チャネルから Send オブジェクトを取る。これは前ステップのノードが return Send(...) で fan-out したもので、各 Send が一つの PUSH task に対応。
PULL task:updated_channels から trigger_to_nodes を逆引き——この mapping はコンパイル時に算出された「チャネル → それを購読するノードリスト」で、和集合を取れば本ステップが触発すべき候補ノードになります。sorted は task 順序を確定させるためで、実行結果には影響しないが checkpoint 再生に影響します。

各候補ノードはさらに _triggers で二次確認します(_triggers call:606-612)——「チャネルに新バージョンがあり、ノードがまだこの版を見ていない」場合だけ本当に task を生成します。このステップが PULL ノードの重複除去の鍵です: updated_channels にあっても、versions_seen[name] が既にこの版を記録していれば、task は再スケジュールされません。

algo 層が Pregel ループ内で占める位置:

境界と失敗

versions_seen[INTERRUPT] が更新されないと無限ループに:should_interrupt は any_updates_since_prev_interrupt が真のときだけ非空リストを返します。versions_seen[INTERRUPT] の更新は apply_writes 後の _put_checkpoint の暗黙ロジックで行われます。interrupt 後の resume で INTERRUPT チャネルの seen バージョンが正しく進まないと、同じ更新が繰り返し「新」と判定され、interrupt-resume-interrupt の無限ループになります(versions_seen INTERRUPT:163-168)。
bump_step は step 推進の唯一信号:any(t.triggers for t in tasks) が偽のときは bump しません——これは「null task はチャネルに書くだけで step を進めない」セマンティクスです(bump_step:259)。null task は tick 開始前に input writes を注入するためのもので、step カウンタを跳ばせるべきではありません。
finish() は「どのノードも触発されなかった」ときのみ呼ばれる:bump_step and updated_channels.isdisjoint(trigger_to_nodes)(finish condition:336)——updated_channels にいずれかのノードが購読するチャネルが残っていれば finish しません。つまり「最終ステップでグラフが死ぬとき」だけ finish 信号を送り、毎ステップでは送りません。
未知チャネルへの書き込みは warn だけでエラーにしない:pending_writes_by_channel で task が channels にないチャネル名に書いた場合は logger.warning だけ(unknown channel warn:310-313)。これは耐障害性のため——例えばノードが既に finish されたチャネルに書いても、グラフ全体を落とさないようにします。
local_read の fresh=True パスはチャネルをコピーする:fresh 読み出し時には各チャネルを channels[k].copy() し(fresh read copy:213-219)、副本に本 task の writes を適用します——これにより条件エッジ (conditional edge) ノードが自分が書いたばかりでまだ apply_writes されていないローカルビューを読め、他 task が見るグローバル状態に影響しません。
prepare_single_task の task id ハッシュアルゴリズム:checkpoint["v"] > 1 のとき _xxhash_str、それ以外は _uuid5_str(task_id_func:550)。これは checkpoint バージョン間の互換性です——旧 checkpoint が uuid5 で生成した task id を再生できなければなりません。

まとめ

_algo.py の三つの純関数 should_interrupt / apply_writes / prepare_next_tasks が Pregel のセマンティクス層をきれいに抽出します: いつ中断するか、書き戻しをどうマージするか、次ステップで誰を走らせるか。これらは状態も IO も持たず、PregelLoop.tick + after_tick に呼ばれて BSP ループ全体を駆動します。ループ駆動層は /pregel/loop、実行層は /pregel/runner、全体組立は /pregel/pregel、チャネルの update / consume / finish / is_available の各メソッドのセマンティクスは /channel/base-channel を参照してください。

公式資料:LangGraph ドキュメント · README

Pregel アルゴリズム層:中断・書き戻し・调度 ​

役割 ​

設計動機 ​

主要ファイル ​

データフロー ​

境界と失敗 ​

まとめ ​

Pregel アルゴリズム層:中断・書き戻し・调度

役割

設計動機

主要ファイル

データフロー

境界と失敗

まとめ