StreamMux: the event dispatcher for multi-mode streaming

源码版本1.2.9

Responsibilities

In the LangGraph v3 streaming protocol, calling graph.stream_events(version="v3") no longer returns a string of raw (namespace, mode, payload) tuples. Instead you get a GraphRunStream object exposing projections like run.values / run.messages / run.updates / run.subgraphs / run.lifecycle (GraphRunStream:31-49). Splitting those raw events by mode into different projections, and letting the user pull from them with an ordinary for loop, is the job of StreamMux — the event dispatch center. The code lives in libs/langgraph/langgraph/stream/_mux.py (StreamMux:26).

StreamMux itself implements no mode-specific logic. It does three things: register transformers, push events, and close. On push, it calls each StreamTransformer.process(event) in registration order (push:269-296); each transformer decides whether to drop the event into its own projection channel. The mux then appends the event to _events, the main log — a complete audit copy of the raw stream. The projections owned by transformers are StreamChannel instances; named channels are also auto-forwarded by the mux into the main log, so a transformer's side effects show up both in its own channel and in the main log.

StreamMux is also the foundation for nested subgraphs. When a subgraph is discovered, SubgraphTransformer calls mux._make_child(scope) (_make_child:193-225) to build a mini-mux for the subgraph. That mini-mux shares the same set of factories as the parent (so transformer instances are per-scope), and it inherits the parent mux's pump binding — when the root pump moves, every subgraph's events advance one step.

Design motivation

• Caller-driven pump, no background thread: The GraphRunStream iterator IS the pump (_pump_next:107-130). When the user does for event in run.values, each iteration triggers _pump_next which pulls one raw event and feeds it to the mux; no background thread is spawned, and memory usage is bounded by the consumption rate. bind_pump registers this pump function with the mux (bind_pump:162-180) and is automatically inherited by mini-mux children via _make_child.
• Transformers run in registration order, serially: push does for transformer in self._transformers: if not transformer.process(event): keep = False (push loop:288-292). Earlier transformers' side effects happen first and are visible to later ones; a transformer returning False only suppresses the event from the main log — earlier transformers still consumed it.
• Native vs extension projection split: A transformer with _native = True has its projection mounted as a direct attribute like run.values on GraphRunStream (native attrs:78-79); non-native ones only go into run.extensions. The native_keys set (native_keys:109) is the "direct attribute" whitelist the v3 protocol exposes to SDKs.
• before_builtins gives content-rewriting transformers right of way: Transformers doing PII filtering or content moderation must see the raw text fields before built-in transformers like MessagesTransformer (before_builtins:94-109). At registration the mux re-partitions into two lanes by before_builtins = True (partition:131-148); order within each lane is preserved.
• seq is only assigned at the root mux: _assign_seq defaults to True, and child muxes pass False via _make_child (_assign_seq:215-220). This way subgraph events forwarded directly to the root log do not have their envelope rewritten, keeping seq monotonic and aligned to the root's write order.

Key files

• StreamMux class:26-46 — Central scheduler attributes and docstring.
• StreamMux.__init__:48-148 — Accepts a transformers list or factories, and registers them in two lanes split by before_builtins.
• _make_child:193-225 — Builds a mini-mux when a subgraph is discovered, inheriting the pump binding.
• _register:227-267 — Calls transformer.init() to obtain the projection dict, checks for key conflicts, and wires named StreamChannels.
• push:269-296 — Sync path: calls process in order, then decides whether to enter the main log.
• apush:351-378 — Async path: serially awaits aprocess so later transformers see earlier async results.
• close:298-324 — Sync close; calls every transformer's finalize, and guarantees all channels close even on error.
• aclose:380-404 — Async close; first gathers all scheduled tasks, then afinalize.
• _collect_stream_modes:398-414 — Reverse-derives which modes to request from the graph based on the required_stream_modes of registered transformers.
• _pregel_stream_v3:3533-3558 — Sync v3 entry; instantiates StreamMux + GraphRunStream and feeds the mux's stream_mode back into the underlying stream().

Data flow

The core of StreamMux.push is just a few lines — walk the transformers in registration order, and if the event survives, append it to the main log:

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) It looks plain, but transformer.process may push into its own StreamChannel, and named channels are auto-wired into the main log by the mux (_register:227-267). So a single raw values event may simultaneously land in run.values via ValuesTransformer and be transformed by a user-defined transformer and pushed back into the main log. Events walk a serial chain across transformers, with order strictly preserved.

Boundaries and failure modes

• Async transformers raise immediately under sync mode: _register calls transformer_requires_async to check (requires_async check:234-239). If a transformer overrides aprocess but does not set supports_sync=True, registering it on a sync mux raises an error advising astream() instead of stream().
• Projection key conflicts raise ValueError directly: _register checks set(projection) & set(self.extensions) (conflict check:246-256); the error lists the conflicting keys and which transformer owns them. Custom transformers cannot reuse built-in projection key names.
• _make_child rejects pre-built transformers: A mux constructed with the transformers= argument cannot build child muxes (factory required:209-214), because pre-built instances cannot be re-initialized in a fresh scope. SubgraphTransformer only runs on a mux built with factories=.
• Finalize errors do not block other cleanup: close collects the first exception via first_error and keeps closing the remaining transformers and channels, then re-raises (close error handling:312-324), ensuring no resource leak.
• bind_pump must be called before channels are read: GraphRunStream.__init__'s wire_pump=True default runs _wire_request_more (_wire_request_more:80-92), and a child mux inherits the parent binding through _make_child. But if you construct a mux by hand and forget bind_pump, StreamChannel.__iter__'s _request_more will forever be None (_request_more:191-192), and iteration ends immediately.
• schedule(coro) only works in async mode: StreamTransformer.schedule requires an event loop (schedule:233-263). The mux gathers all scheduled tasks during aclose and only afinalizes once they finish; calling schedule in sync mode fails.

Summary

StreamMux is the hub of the v3 streaming protocol: it implements no mode logic itself, only dispatches transformers in order, builds child muxes per scope, and collects both raw events and transformer projections into the main log. For details on how each mode turns events into user-friendly shapes, see stream/transformers; for how raw events are fed in from PregelLoop, see the Pregel engine.

See official docs: LangGraph docs · README.