Stream Transformers: turning raw Pregel output into usable projections

源码版本1.2.9

Responsibilities

During each superstep, PregelLoop pushes (checkpoint_ns, mode, payload) triples into a SyncQueue via _emit (_emit:1380-1414) — these are raw "value", "update", "task", "checkpoint" events. That is the output shape of the v1 / v2 protocol: the user gets a stream of chunks. But in the v3 protocol, what users want is not chunks but "per-mode organized, single-consumer-iterable projections": run.values is a stream of complete state snapshots, run.messages is a ChatModelStream handle for LLM token streams, run.updates is a stream of node updates, and run.subgraphs is a handle for subgraph entry.

The logic that turns raw chunks into these projections lives in libs/langgraph/langgraph/stream/transformers.py (ValuesTransformer:28). Each mode has one transformer, all inheriting from the StreamTransformer abstract base class (StreamTransformer:44). They are registered by StreamMux, fed events by GraphRunStream, and each maintains its own internal StreamChannel as its projection. When you call graph.stream_events(version="v3"), what you actually get is a GraphRunStream (GraphRunStream:31); its .values / .messages / .updates / .subgraphs / .custom / .lifecycle attributes are the projection entry points of the native transformers (native attrs:78-79).

The whole transformer system is the v3 protocol's extension point — beyond the five built-ins, users can add their own transformer via Pregel.compile(stream_transformers=[...]) (stream_transformers param:783), passed to StreamMux as a factory alongside the built-ins inside _pregel_stream_v3 (factories list:3533-3544).

Design motivation

• Native vs extension projection dual track: A transformer with _native = True has its projection mounted as a direct run.<key> attribute (_native:50-52); non-native ones go into run.extensions and are only exposed via the dict. The five built-ins are all native; user-defined ones default to extensions, and must explicitly set _native = True for a direct attribute.
• required_stream_modes reverse-derives which modes the graph emits: A transformer declares which modes it needs (required_stream_modes:88-93), _collect_stream_modes takes the union (_collect_stream_modes:398-414), and when v3 calls the underlying stream() it passes that union as stream_mode (stream_mode=_collect_stream_modes:3549). The user does not specify stream_mode directly — instead "pulls on demand".
• Scope filtering avoids noise: ValuesTransformer.process checks params["namespace"] != self._scope_list (ValuesTransformer.process:70-82) and only accepts events from its own scope; subgraph events are left to the subgraph's own mini-mux, so root run.values is not polluted by nested graphs.
• Sync / async dual track with auto-detection: A transformer defaults to supports_sync = False; as long as it overrides aprocess / afinalize / afail, transformer_requires_async flags it as async-only (transformer_requires_async:308-330), and registering it on a sync mux raises immediately. To support both sync and async you must explicitly set supports_sync = True, e.g. SubgraphTransformer (supports_sync:689).
• SubgraphTransformer goes through SubgraphRunStream, not raw events: When a subgraph call is discovered, _on_started builds a SubgraphRunStream wrapping the mini-mux (SubgraphTransformer:670-705). What the user gets is not an event stream but a handle object, on which handle.values / handle.messages / handle.subgraphs can be recursed — isomorphic to the root run interface.

Key files

• StreamTransformer base class:44-115 — Abstract init / process / aprocess / finalize / fail + ClassVar config.
• init:128-138 — Returns the projection dict; keys go into extensions, and when _native=True also onto run.<key>.
• process:148-164 — Sync lane; defaults to raise NotImplementedError; subclasses must override.
• aprocess:166-184 — Async lane; defaults to delegating to process; override for async work.
• schedule:233-263 — An asyncio.Task bound to the mux lifecycle — gathered on aclose, cancelled on afail.
• transformer_requires_async:308-330 — Detects whether aprocess / afinalize / afail are overridden, deciding whether the transformer can run on a sync mux.
• ValuesTransformer:28-82 — The run.values projection; required_stream_modes=("values",).
• UpdatesTransformer:120-152 — The run.updates projection.
• MessagesTransformer:155-335 — The run.messages projection; packs LLM token streams into ChatModelStream handles and handles the v2 protocol's (payload, metadata) input.
• LifecycleTransformer:608-667 — The run.lifecycle projection; emits subgraph lifecycle events back into the main log as native protocol events visible to remote SDKs.
• SubgraphTransformer:670-705 — The run.subgraphs projection; builds SubgraphRunStream + mini-mux on subgraph discovery.
• GraphRunStream._pump_next:107-130 — The pump: pulls one chunk, runs convert_to_protocol_event, and feeds the mux.
• convert_to_protocol_event:10-32 — Maps v2 StreamPart fields to v3 ProtocolEvent.

Data flow

The entire v3 chain starts at PregelLoop._emit, passes through SyncQueue, GraphRunStream._pump_next, convert_to_protocol_event, and finally reaches the mux. convert_to_protocol_event directly converts v2's {type, ns, data, interrupts} into a v3 ProtocolEvent:

def convert_to_protocol_event(part: StreamPart) -> ProtocolEvent:
    part_dict = cast(dict[str, Any], part)
    params: _ProtocolEventParams = {
        "namespace": list(part_dict["ns"]),
        "timestamp": int(time.time() * 1000),
        "data": part_dict["data"],
    }
    if "interrupts" in part_dict:
        params["interrupts"] = part_dict["interrupts"]
    return {
        "type": "event",
        "method": part_dict["type"],
        "params": params,
    }

(convert body:20-32) Note that method is exactly v2's type — values / updates / messages / custom / tasks / checkpoints / debug. This is the dispatch key inside StreamTransformer.process; for instance, ValuesTransformer only acts on events with method == "values" (method check:70-71).

The v1 / v2 protocol takes a different path inside Pregel.stream: raw chunks come straight out of the SyncQueue and through the _output function (_output:4184-4243), which decides based on whether stream_mode is a string or a list whether to yield payload, (mode, payload), or (ns, mode, payload):

if version == "v2":
    ...
    yield {"type": mode, "ns": ns, "data": payload, "interrupts": ints}
elif stream_subgraphs and isinstance(stream_mode, list):
    yield (ns, mode, payload)
elif isinstance(stream_mode, list):
    yield (mode, payload)
elif stream_subgraphs:
    yield (ns, payload)
else:
    yield payload

(_output shape branches:4220-4243) This is why stream_mode=["updates", "values"] yields triples while a single string yields a single value — the same raw chunk gets shaped differently at _output. v3 does not go through _output; it goes through GraphRunStream._pump_next → mux.push → transformer.

Boundaries and failure modes

• process must be implemented; init must return a dict: init is an @abstractmethod by default; failing to implement it breaks startup (init abstract:127-138). process defaults to raise NotImplementedError (process default:162-164); not overriding it blows up on the first pushed event.
• Transformer order is sensitive: A transformer with before_builtins = True runs first (before_builtins:94-109), but the docstring warns that it sees tasks events before LifecycleTransformer / SubgraphTransformer consume them. Mutating event["params"]["namespace"] or the id / result / error / interrupts fields would corrupt the bookkeeping of those two — you may only observe and edit cold fields.
• MessagesTransformer is coupled to the v2 protocol: stream_events(version="v3") forces version="v2" and CONFIG_KEY_STREAM_MESSAGES_V2=True when calling the underlying stream() (_pregel_stream_v3 stream call:3545-3551), otherwise the messages projection does not get the right payload format.
• StreamChannel is single-consumer; iterating twice raises: run.values can only be consumed by one for loop (single-consumer note:38-40); to fan-out you must call projection.tee(n) (tee:245), at the cost of buffering.
• schedule(on_error="raise") turns the close path into fail: on_error="raise" makes a task exception push the mux into afail during aclose (on_error:254-257); the default "log" only logs without propagating — choosing wrong lets one transformer's failure take down the whole stream.
• SubgraphTransformer._should_track is strictly greater than scope: It only tracks subgraphs one level below the current scope (should_track:634-636); grandchild graphs are discovered recursively through handle.subgraphs. If the user skips handle.subgraphs and looks directly at root run.subgraphs, grandchild events will not appear there.

Summary

StreamTransformer is the extension point of the v3 streaming protocol: five built-in transformers turn raw (ns, mode, payload) into iterable typed projections, and users can add their own transformer via the stream_transformers parameter. For the overall dispatch mechanism see StreamMux; for how raw chunks are fed from PregelLoop into the SyncQueue see PregelLoop.

See official docs: LangGraph docs · README.