StreamMux:複数モードストリームのイベントディスパッチャ

源码版本1.2.9

役割

LangGraph v3 のストリーミングプロトコルでは、ユーザーが graph.stream_events(version="v3") を呼んで取得するのは生の (namespace, mode, payload) タプルの羅列ではなく、GraphRunStream オブジェクトです。その上に run.values / run.messages / run.updates / run.subgraphs / run.lifecycle といった「投影 (projection)」がぶら下がります (GraphRunStream:31-49)。これらの生イベントをモードごとに異なる投影に振り分け、ユーザーが普通の for ループで引っ張れるようにするのが StreamMux、イベントディスパッチセンターです。コードは libs/langgraph/langgraph/stream/_mux.py にあります (StreamMux:26)。

StreamMux 自身はどのモードの具体的論理も実装しません。3 つのことだけを行います:transformer の登録、イベントの push、クローズ。push 時は登録順に各 StreamTransformer.process(event) を呼び (push:269-296)、transformer がイベントを自身の projection channel に詰めるかどうかを決めます。その後 mux 自身がイベントを _events という主ログに追加します。主ログは生ストリームの「完全な監査副本」です。transformer が開いた projection はすべて StreamChannel インスタンスで、名前付き channel は mux によって自動的に主ログに転送されます。したがって transformer の副作用は自身の channel にも現れ、主ログにも残ります。

StreamMux はサブグラフのネストの基盤でもあります。subgraph が発見されるたび、SubgraphTransformer は mux._make_child(scope) (_make_child:193-225) を呼んでサブグラフ用に mini-mux を建てます。この mini-mux は親 mux と同じ factory 群を共有し(transformer インスタンスは scope ごとに 1 つ)、親 mux のポンプバインディングも共有します。ルートポンプが動くと、すべてのサブグラフのイベントが 1 格子前に進みます。

設計動機

• caller-driven ポンプ、バックグラウンドスレッドなし:GraphRunStream のイテレータがポンプです (_pump_next:107-130)。ユーザーが for event in run.values すると、各イテレートが _pump_next を発火し、生イベントを 1 つ引いて mux に与えます。バックグラウンドスレッドを開かず、メモリは消費速率で決まります。bind_pump はこのポンプ関数を mux に登(bind_pump:162-180)し、mini-mux のサブグラフは _make_child で自動的に継承します。
• transformer は登録順に直列:push は for transformer in self._transformers: if not transformer.process(event): keep = False (push loop:288-292)。前の transformer の副作用が先に起き、後の transformer がそれを見えます。ある transformer が False を返しても主ログへの追記を抑制するだけで、前の transformer はすでにイベントを受け取っています。
• native vs extension 投影の区別:_native = True の transformer 投影は mux によって run.values のような直接属性として GraphRunStream にぶら下げられます (native attrs:78-79)。非 native は run.extensions にしか入りません。native_keys 集合 (native_keys:109) は v3 プロトコルが SDK に晒す「直接属性」ホワイトリストです。
• before_builtins がコンテンツ書き換え transformer に道を譲る:PII フィルタやコンテンツ審査のような transformer は MessagesTransformer などの組込み transformer より前に生の text フィールドを見る必要があります (before_builtins:94-109)。mux は登録時に before_builtins = True で 2 レーンに再構成し (partition:131-148)、レーン内の順序は保ったままです。
• seq はルート mux だけが割り当て:_assign_seq はデフォルト True で、child mux は _make_child で False を渡されます (_assign_seq:215-220)。これによりサブグラフイベントがルートログに直接転送されるとき envelope が改変されず、seq がルートの書き込み順に単調に対応します。

主要ファイル

• StreamMux 类:26-46 — 中心スケジューラのプロパティと docstring。
• StreamMux.__init__:48-148 — transformers list または factories を受け取り、before_builtins でレーン分けして登録します。
• _make_child:193-225 — サブグラフ発見時に mini-mux を建て、ポンプバインディングを継承します。
• _register:227-267 — transformer.init() を呼んで projection dict を取り、key 衝突を検査し、名前付き StreamChannel をバインドします。
• push:269-296 — 同期パス:順序よく process を呼び、主ログに入れるか決めます。
• apush:351-378 — 非同期パス:aprocess を直列に await し、後続 transformer が前の async 結果を見えるようにします。
• close:298-324 — 同期クローズ。各 transformer の finalize を呼び、エラーでも全 channel を閉じることを保証します。
• aclose:380-404 — 非同期クローズ。schedule された task をすべて gather してから afinalize します。
• _collect_stream_modes:398-414 — mux に登録された transformer の required_stream_modes から、グラフに要求するモードを逆推します。
• _pregel_stream_v3:3533-3558 — v3 同期エントリ。StreamMux + GraphRunStream をインスタンス化し、mux の stream_mode を下層 stream() に逆流します。

データフロー

StreamMux.push の核心はこれだけです。登録順に transformer を走らせ、全部通ってまだ生きていれば主ログに詰めます:

def push(self, event: ProtocolEvent) -> None:
    keep = True
    for transformer in self._transformers:
        if not transformer.process(event):
            keep = False
    if keep:
        if self._assign_seq:
            self._seq += 1
            event["seq"] = self._seq
        self._events.push(event)

(push body:288-296) 質素に見えますが、transformer.process の中で自身の StreamChannel に push することがあり、名前付き channel は mux によって自動的に主ログに wire されます (_register:227-267)。したがって 1 つの生 values イベントが ValuesTransformer で run.values に詰められると同時に、ユーザー定義 transformer で変換されて主ログに戻される、ということが起きます。イベントは transformer 間を直列リンクで流れ、順序は厳密に保たれます。

境界と失敗

• async transformer は sync モードで即座に raise:_register は transformer_requires_async 検査を呼び (requires_async check:234-239)、transformer が aprocess をオーバーライドしているのに supports_sync=True を設定していない場合、sync mux への登録はエラーになります。stream() ではなく astream() を使うよう促します。
• projection key 衝突は即 ValueError:_register は set(projection) & set(self.extensions) を検査し (conflict check:246-256)、衝突する key と帰属する transformer 名を列挙します。カスタム transformer の projection key は組込みと同名にできません。
• _make_child は pre-built transformers を拒絶:mux が transformers= パラメータで構築されたときは子 mux を建てられません (factory required:209-214)。pre-built インスタンスは新しい scope 下で再初期化できないためです。SubgraphTransformer は factories= で構築された mux 上でしか走れません。
• finalize のエラーは他のクリーンアップをブロックしない:close は first_error で最初の例外を取り、残りの transformer と channel を閉じ続け、最後に raise し (close error handling:312-324)、リソース漏れを防ぎます。
• bind_pump は channel 読み取り前に呼ぶ必要あり:GraphRunStream.__init__ の wire_pump=True はデフォルトで _wire_request_more を走らせ (_wire_request_more:80-92)、子 mux は _make_child で親のバインディングが自動的に伝わります。しかしユーザーが自分で mux インスタンスを組み立てるとき、pump の bind 忘れは StreamChannel.__iter__ の _request_more が永遠に None になり (_request_more:191-192)、イテレートが即座に終了します。
• schedule(coro) は async モードでしか使えない:StreamTransformer.schedule はイベントループを必要とし (schedule:233-263)、mux は aclose 時に schedule された task をすべて gather し、async が走り終わってから afinalize します。sync モードで schedule を呼ぶと失敗します。

まとめ

StreamMux は v3 ストリーミングプロトコルの中枢です。モード論理は一切実装せず、順序よく transformer をスケジュールし、scope ごとに子 mux を建て、生イベントと transformer 投影を主ログに収録するだけです。各モードがどうイベントをユーザーフレンドリーな形状に変えるかは stream/transformers を、生イベントが PregelLoop からどう流れ込むかは Pregel エンジン を参照してください。

公式資料:LangGraph 文档 · README