ストリーミング出力:astream と stream_mode

源码版本1.2.9

役割

Pregel.astream / Pregel.stream はコンパイル済みグラフが外部に晒す実行エントリです。入力を投げると PregelLoop を起動して何周も走らせ、その過程で各スーパーステップ (superstep) が生むデータを呼び出し側が指定した stream_mode で投影してストリームイベントにし、キューに詰め、呼び出し側がイテレータとして消費します。ainvoke / invoke は本質的に astream / stream を単一値に収束させたものです。Pregel.invoke の実装体は for chunk in self.stream(...) で最後の 1 片を取ります (Pregel.invoke body:3891)。

この層の責務は「実行」と「出力形状」を切り離すことです。実行は Pregel の BSP ループのままで、出力形状は stream_mode で決まります。同じグラフを 1 回走らせても、values(各ステップの完全状態)、updates(各ステップの差分)、messages(LLM トークンストリーム)、custom(ノード内のカスタム出力)、checkpoints(チェックポイントイベント)、tasks(タスクの開始終了)、debug(全部)など異なる投影を要求できます。さらに list を渡して複数モードを同時に取ることもできます。

設計動機

• 実行をブロックしない:出力はキューを通ります。runner.tick / atick はタスクを 1 つ完了するたびにイベントをキューに詰め、外側の for ループはイテレートしながら取り出します。グラフ全体を走ら終えてから一気に返すわけではありません。これは LLM トークンストリーム(messages)には必須で、ユーザーは生成しながら見る必要があります。
• 複数投影が同一の実行を再利用:runner.atick(...) の内部で多種のイベントが生じ、_output 関数(_output:4184)はフィルタ/ディスパッチだけを行い、グラフを再走させません。stream_mode=["values","updates"] を渡しても 2 回走りません。
• サブグラフ名前空間を統一:subgraphs=True を開くと、イベントは namespace プレフィックス((ns, mode, payload))を持ち、親グラフがサブグラフ内部のイベントを見えます。サブグラフ自身が別のストリームを開くのではなく、_output がデキュー時にタグを打ちます。
• v2 の型付け:新バージョンは version="v2" でイベントを {"type": mode, "ns": ..., "data": ..., "interrupts": ...} の dict にし、下流のプログラム的処理を容易にします。values モードはさらに __interrupt__ を interrupts フィールドとして別途抽出します。

主要ファイル

• Pregel class:450 — Pregel クラス定義。すべてのストリーミングエントリがここにあります。
• Pregel.stream:2655 — 同期ストリーミングエントリ。シグネチャに全 stream_mode 選択肢を列挙します。
• Pregel.astream:3063 — 非同期ストリーミングエントリ。真の非同期メインループはその実装体内に書かれます。
• Pregel.invoke:3836 — 同期収束エントリ。内部で for chunk in self.stream(...) し最後の values chunk を取ります。
• sync main loop:2964 — 同期メインループ:while loop.tick() → runner.tick → loop.after_tick()。
• async main loop:3437 — 非同期メインループ:3 段式は同じ。runner.atick が同期版を置き換えます。
• _output:4184 — デキューのフィルタ関数。stream_mode / print_mode / subgraphs に基づき各イベントをどう yield するか決めます。
• GraphRunStream:31 — 同期呼び出し側駆動のストリームラッパ。for ループがポンプで、バックグラウンドスレッドはいません。
• AsyncGraphRunStream:304 — 非同期対応物。複数投影(run.values / run.messages)が 1 つのポンプを共用します。
• stream_mode attr:709 — Pregel のデフォルト stream_mode="values"。stream(stream_mode=...) で上書きできます。

データフロー

非同期メインループの骨格は astream 実装体内にあります (async main loop:3437)。3 段式は Pregel エンジンのページで説明したとおりですが、各段の間に _output が挟まり、キューの中身を呼び出し側に吐き出します:

while loop.tick():
    for task in await loop.amatch_cached_writes():
        loop.output_writes(task.id, task.writes, cached=True)
    async for _ in runner.atick(
        [t for t in loop.tasks.values() if not t.writes],
        timeout=self.step_timeout,
        get_waiter=get_waiter,
        schedule_task=loop.aaccept_push,
    ):
        # emit output
        for o in _output(
            stream_mode,
            print_mode,
            subgraphs,
            stream.get_nowait,
            asyncio.QueueEmpty,
            version,
            _output_mapper,
            _state_mapper,
        ):
            yield o
    loop.after_tick()
    await aemit_graph_lifecycle_events(loop)
    # wait for checkpoint
    if durability_ == "sync":
        await cast(asyncio.Future, loop._put_checkpoint_fut)

runner.atick は内部で各タスクが生んだ writes を loop に書き戻すと同時に、(ns, mode, payload) の 3-tuple を stream(asyncio.Queue)に詰めます。外側の async for が実行権を得たとき、_output は stream.get_nowait() を呼んでキュー内の準備済みイベントをすべて引き抜き、stream_mode でフィルタしてからユーザーに yield します。_output の核心論理(_output:4184)はこうです:(ns, mode, payload) を取り、mode in print_mode なら先に print し、mode in stream_mode のときだけ外に yield します。つまり print_mode は「見るだけで送らない」デバッグスイッチで、実際に yield する内容には影響しません。

下の図は stream 呼び出しをエントリからイベントデキューまでなぞります:

境界と失敗

• デフォルト stream_mode:stream_mode=None のとき、サブグラフとして呼ばれている(CONFIG_KEY_TASK_ID が config にある)場合はデフォルト values、さもなくば self.stream_mode を使います (stream_mode default:2740)。サブグラフのデフォルト updates モードが親グラフの出力を汚さないようにします。
• ステップ数超過:メインループ退出後に loop.status を見て、out_of_steps なら GraphRecursionError を送出し、recursion_limit を上げるよう促します (out_of_steps:3002)。draining 状態は GraphDrained を送出し、制御を外部 RunControl に戻します。
• 複数モード同時出力:stream_mode に list を渡すと、各イベントは (mode, payload) の tuple になります。さらに subgraphs=True を重ねると (ns, mode, payload) になります (tuple mode:4240)。呼び出し側はタプル構造で受け取る必要があります。
• messages モードの継承処理:stream_mode="messages" かつ version="v1" のとき、上位層から継承した v2 messages handler を剥がします。v1 ストリームが content-block イベントプロトコルにルーティングされるのを防ぐためです (strip v2 handler:2773)。ただし subgraphs=True のとき内層の messages イベントを外層が観察できるよう、v1 handler は残します。
• durability と checkpoint の同期:durability="sync" のとき、メインループは各ステップ終了時に await loop._put_checkpoint_fut し、checkpoint がディスクに落ちてから次のステップに入ります。"async" はデフォルトで、永続化と次ステップを並行させます。"exit" は退出時だけディスクに落とします。
• v3 実験的 stream_events:stream_events(version="v3") は stream_mode と subgraphs パラメータを受け付けません。両者は mux が内部で接管し、明示的に渡すと _reject_v3_invariant_kwargs に拒否されます (_reject_v3_invariant_kwargs:387)。

まとめ

astream / stream は Pregel の BSP ループをイテレータインタフェースに包み、stream_mode の選択が出力投影を決め、_output がデキューのフィルタを行う唯一の咽喉です。実行の詳細は /pregel/pregel と /pregel/loop を、複数モードの mux は libs/langgraph/langgraph/stream/_mux.py を参照してください。公式資料:LangGraph 文档 · README。