Stream Transformers: Roh-Ausgabe von Pregel in nutzbare Projektionen wandeln

源码版本1.2.9

Verantwortung

PregelLoop schiebt in jedem Superstep über _emit ein Tripel (checkpoint_ns, mode, payload) in eine SyncQueue (_emit:1380-1414); darin finden sich Roh-Events der Art «Werte», «Updates», «Tasks», «Checkpoints». Das ist die Ausgabeform der v1- / v2-Protokolle – der Nutzer erhält eine Folge von Chunks. Im v3-Protokoll möchte der Nutzer jedoch keine Chunks, sondern «nach Modus organisierte, einzeln konsumierbare Projektionen (projection)»: run.values ist ein Strom vollständiger Zustandssnapshots, run.messages ist ein Handle ChatModelStream auf einen LLM-Token-Strom, run.updates ist ein Strom von Knoten-Updates, run.subgraphs ist ein Handle für den Subgraph-Einstieg.

Die Logik, die Roh-Chunks in diese Projektionen überführt, liegt in libs/langgraph/langgraph/stream/transformers.py (ValuesTransformer:28); pro Modus existiert ein Transformer, alle gemeinsam erben von der abstrakten Basisklasse StreamTransformer (StreamTransformer:44). Sie werden von StreamMux registriert, von GraphRunStream mit Events gefüttert und verwalten jeweils einen internen StreamChannel als Projektion. Wenn der Nutzer graph.stream_events(version="v3") aufruft, erhält er faktisch ein GraphRunStream (GraphRunStream:31); dessen Attribute .values / .messages / .updates / .subgraphs / .custom / .lifecycle sind die Projektions-Eingänge der nativen Transformer (native attrs:78-79).

Das gesamte Transformer-System ist der Erweiterungspunkt des v3-Protokolls – neben den fünf eingebauten Transformern kann der Nutzer in Pregel.compile(stream_transformers=[...]) eigene Transformer hinzufügen (stream_transformers param:783); sie werden in _pregel_stream_v3 gemeinsam mit den eingebauten Transformern als Factory an StreamMux übergeben (factories list:3533-3544).

Entwurfsmotivation

Doppelte Projektions-Bahn native vs. extension: Projektionen von Transformern mit _native = True werden als direkte Attribute run.<key> angehängt (_native:50-52); nicht-native landen in run.extensions und sind nur über das Dict zugänglich. Die fünf eingebauten Transformer sind nativ; nutzerdefinierte landen per Default in extensions – für ein direktes Attribut muss explizit _native = True gesetzt werden.
required_stream_modes leitet ab, welche Modi der Graph senden muss: Jeder Transformer deklariert selbst, welche Modi er benötigt (required_stream_modes:88-93); _collect_stream_modes bildet die Vereinigung (_collect_stream_modes:398-414), und beim v3-Aufruf des zugrunde liegenden stream() wird diese Vereinigung als stream_mode übergeben (stream_mode=_collect_stream_modes:3549). Der Nutzer spezifiziert stream_mode nicht direkt, sondern «zieht bei Bedarf».
Scope-Filter vermeidet Lärm: ValuesTransformer.process prüft params["namespace"] != self._scope_list (ValuesTransformer.process:70-82) und akzeptiert nur Events aus dem eigenen Scope; die Events von Subgraphen überlässt er dem eigenen Mini-Mux des Subgraphen, sodass das Root-run.values nicht durch verschachtelte Graphen verschmutzt wird.
sync / async Doppel-Bahn mit automatischer Erkennung: Ein Transformer hat per Default supports_sync = False; sobald er aprocess / afinalize / afail überschreibt, stuft transformer_requires_async ihn als nur-async ein (transformer_requires_async:308-330), und die Registrierung an einem sync-Mux schlägt fehl. Wer sync und async unterstützen will, muss explizit supports_sync = True setzen, z. B. SubgraphTransformer (supports_sync:689).
SubgraphTransformer geht über SubgraphRunStream, nicht über nackte Events: Wird ein Subgraph-Aufruf entdeckt, baut _on_started ein SubgraphRunStream, das den Mini-Mux verpackt (SubgraphTransformer:670-705) – der Nutzer erhält keinen Event-Strom, sondern ein Handle-Objekt, auf dem rekursiv handle.values / handle.messages / handle.subgraphs verfügbar sind, eine isomorphe Schnittstelle zum Root-run.

Schlüsseldateien

StreamTransformer Basisklasse:44-115 — Abstrakte init / process / aprocess / finalize / fail + ClassVar-Konfiguration.
init:128-138 — Gibt das projection-Dict zurück; Keys landen in extensions, bei _native=True zusätzlich als run.<key>.
process:148-164 — sync-Lane, Default ist raise NotImplementedError; Subklassen müssen überschreiben.
aprocess:166-184 — async-Lane, Default delegiert an process; bei echter async-Arbeit überschreiben.
schedule:233-263 — Ein an die Mux-Lebensdauer gebundener asyncio.Task, der bei aclose mit gather eingesammelt und bei afail abgebrochen wird.
transformer_requires_async:308-330 — Erkennt, ob aprocess / afinalize / afail überschrieben wurden, und entscheidet, ob der Transformer am sync-Mux laufen darf.
ValuesTransformer:28-82 — run.values-Projektion, required_stream_modes=("values",).
UpdatesTransformer:120-152 — run.updates-Projektion.
MessagesTransformer:155-335 — run.messages-Projektion; bündelt LLM-Token-Streams in einem ChatModelStream-Handle und verarbeitet die (payload, metadata)-Eingabe des v2-Protokolls.
LifecycleTransformer:608-667 — run.lifecycle-Projektion; sendet Subgraph-Lebenszyklus-Events als native Protokoll-Events ans Hauptlog, sichtbar für Remote-SDKs.
SubgraphTransformer:670-705 — run.subgraphs-Projektion; baut beim Entdecken eines Subgraph-Aufrufs ein SubgraphRunStream + Mini-Mux auf.
GraphRunStream._pump_next:107-130 — Pump: holt einen Chunk, convert_to_protocol_event, übergibt an den Mux.
convert_to_protocol_event:10-32 — Feldabbildung v2 StreamPart → v3 ProtocolEvent.

Datenfluss

Die gesamte v3-Kette beginnt bei PregelLoop._emit und läuft über SyncQueue, GraphRunStream._pump_next, convert_to_protocol_event bis hin zum Mux. convert_to_protocol_event wandelt das v2-{type, ns, data, interrupts} direkt in ein v3-ProtocolEvent um:

python

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) Beachten Sie: method ist genau das type aus v2 – values / updates / messages / custom / tasks / checkpoints / debug – und dient in StreamTransformer.process als Grundlage des Dispatch, etwa wenn ValuesTransformer nur bei method == "values" aktiv wird (method check:70-71).

Die v1- / v2-Protokolle gehen in Pregel.stream einen anderen Weg: Die Roh-Chunks kommen direkt aus der SyncQueue und durchlaufen die Funktion _output (_output:4184-4243); je nachdem, ob stream_mode ein String oder eine Liste ist, wird entweder payload oder (mode, payload) oder (ns, mode, payload) ausgegeben:

python

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) Deshalb liefert stream_mode=["updates", "values"] Tripel, während ein einzelner String einen Einzelwert liefert – derselbe Roh-Chunk wird in _output in unterschiedliche Formen gebracht. v3 geht nicht über _output, sondern über GraphRunStream._pump_next → mux.push → Transformer.

Grenzen und Fehler

process muss implementiert sein, init muss ein Dict zurückgeben: init ist per @abstractmethod vorgeschrieben, und eine Subklasse ohne Implementierung schlägt beim Start fehl (init abstract:127-138); process hat den Default raise NotImplementedError (process default:162-164), und ohne Überschreibung explodiert der Aufruf beim ersten push.
Reihenfolge zwischen Transformern ist sensibel: Transformer mit before_builtins = True laufen zuerst (before_builtins:94-109), aber der Docstring warnt, dass sie tasks-Events sehen, bevor LifecycleTransformer / SubgraphTransformer diese konsumieren; wer event["params"]["namespace"] oder id / result / error / interrupts verändert, bringt die Buchführung der beiden anderen durcheinander – nur beobachten, «kalte» Felder ändern.
MessagesTransformer ist an das v2-Protokoll gekoppelt: stream_events(version="v3") übergibt beim Aufruf des zugrunde liegenden stream() zwingend version="v2" und CONFIG_KEY_STREAM_MESSAGES_V2=True (_pregel_stream_v3 stream call:3545-3551), sonst erhält die messages-Projektion nicht das richtige Payload-Format.
StreamChannel ist einzelverbraucher, zweimaliges Iterieren wirft: run.values darf nur von einer for-Schleife konsumiert werden (single-consumer note:38-40); für Fan-out muss projection.tee(n) verwendet werden (tee:245), was Puffer kostet.
schedule(on_error="raise") macht aus dem close-Pfad einen fail-Pfad: on_error="raise" bei schedule sorgt dafür, dass eine Task-Ausnahme in aclose den Mux in afail versetzt (on_error:254-257); der Default "log" protokolliert nur, ohne Weitergabe – eine falsche Wahl lässt einen einzelnen Transformer-Defekt den gesamten Stream mitreißen.
SubgraphTransformer._should_track ist strikt größer als der Scope: Es werden nur Subgraphen eine Ebene unter dem aktuellen Scope verfolgt (should_track:634-636); Enkel-Subgraphen werden rekursiv über handle.subgraphs des Child-Handles entdeckt; wer handle.subgraphs überspringt und direkt das Root-run.subgraphs liest, sieht dort keine Enkel-Events.

Zusammenfassung

StreamTransformer ist der Erweiterungspunkt des v3-Streaming-Protokolls: fünf eingebaute Transformer wandeln rohe (ns, mode, payload) in iterierbare typisierte Projektionen um, und der Nutzer kann eigene Transformer im Parameter stream_transformers ergänzen. Den gesamten Verteilungsmechanismus siehe StreamMux; wie die Roh-Chunks aus dem PregelLoop in die SyncQueue gelangen, siehe PregelLoop.

Siehe offizielle Dokumentation: LangGraph 文档 · README.

Stream Transformers: Roh-Ausgabe von Pregel in nutzbare Projektionen wandeln ​

Verantwortung ​

Entwurfsmotivation ​

Schlüsseldateien ​

Datenfluss ​

Grenzen und Fehler ​

Zusammenfassung ​