StreamMux: Event-Verteiler für Multi-Modus-Streams

源码版本1.2.9

Verantwortung

Im v3-Streaming-Protokoll von LangGraph erhält der Nutzer bei graph.stream_events(version="v3") keine Rohfolge von (namespace, mode, payload)-Tupeln mehr, sondern ein GraphRunStream-Objekt mit Projektionen (projection) wie run.values / run.messages / run.updates / run.subgraphs / run.lifecycle (GraphRunStream:31-49). Diese Roh-Events nach Modus auf verschiedene Projektionen aufzuteilen und dem Nutzer über eine gewöhnliche for-Schleife zugänglich zu machen, ist Aufgabe des StreamMux – der Event-Verteilungszentrale. Der Code liegt in libs/langgraph/langgraph/stream/_mux.py (StreamMux:26).

StreamMux selbst implementiert keine modusspezifische Logik; es macht nur drei Dinge: Transformer registrieren, Events push-en und schließen. Beim Push wird für jeden registrierten StreamTransformer.process(event) in Registrierreihenfolge aufgerufen (push:269-296); der Transformer entscheidet, ob das Event in seinen eigenen projection-Channel aufgenommen wird. Danach hängt der Mux das Event an _events an – das Hauptlog ist eine «vollständige Audit-Kopie» des Roh-Streams. Die Projektionen, die ein Transformer öffnet, sind StreamChannel-Instanzen; benannte Channels werden vom Mux zusätzlich automatisch ins Hauptlog weitergeleitet, sodass die Seiteneffekte eines Transformers sowohl in seinem eigenen Channel als auch im Hauptlog sichtbar sind.

StreamMux ist außerdem die Grundlage für die Schachtelung von Untergraphen (subgraph). Sobald ein Subgraph entdeckt wird, ruft SubgraphTransformer mux._make_child(scope) auf (_make_child:193-225), um einen Mini-Mux für den Subgraphen zu erzeugen. Dieser Mini-Mux teilt sich denselben Satz an Factorys mit dem Eltern-Mux (jeder Transformer ist also pro Scope einmal vorhanden) und auch die Pump-Bindung des Eltern-Mux – sobald die Root-Pump sich bewegt, rücken alle Subgraph-Events um eine Position nach vorne.

Entwurfsmotivation

• Aufrufergetriebene Pump, kein Hintergrund-Thread: Der Iterator von GraphRunStream ist die Pump (_pump_next:107-130). Wenn der Nutzer for event in run.values ausführt, löst jede Iteration _pump_next aus und holt ein Roh-Event, das dem Mux zugeführt wird; es wird kein Hintergrund-Thread gestartet, und der Speicherverbrauch richtet sich nach der Konsumrate. bind_pump registriert diese Pump-Funktion im Mux (bind_pump:162-180) und wird über _make_child automatisch vererbt.
• Transformer in Registrierreihenfolge seriell: push nutzt for transformer in self._transformers: if not transformer.process(event): keep = False (push loop:288-292); die Seiteneffekte des vorigen Transformers finden zuerst statt, und der nächste sieht sie bereits; ein Transformer, der False zurückgibt, unterdrückt das Event nur im Hauptlog, die vorherigen Transformer haben es bereits konsumiert.
• native vs. extension Projektionen: Transformer mit _native = True werden vom Mux als direkte Attribute wie run.values an GraphRunStream angehängt (native attrs:78-79); nicht-native landen nur in run.extensions. Die Menge native_keys (native_keys:109) ist die Whitelist an «direkten Attributen», die das v3-Protokoll dem SDK freigibt.
• before_builtins macht Platz für inhaltsmodifizierende Transformer: Transformer für PII-Filterung oder Inhaltsmoderation müssen die rohen Textfelder vor den eingebauten Transformern wie MessagesTransformer sehen (before_builtins:94-109); der Mux teilt die Registrierung in zwei Lanes auf, wenn before_builtins = True (partition:131-148), aber die Reihenfolge innerhalb einer Lane bleibt erhalten.
• seq wird nur im Root-Mux vergeben: _assign_seq ist per Default True, der Child-Mux erhält über _make_child den Wert False (_assign_seq:215-220). So werden Subgraph-Events beim direkten Weiterleiten ins Root-Log nicht mit neuem Umschlag versehen, und seq bleibt monoton und entspricht der Schreibreihenfolge im Root.

Schlüsseldateien

• StreamMux Klasse:26-46 — Zentrale Scheduler-Attribute und Docstring.
• StreamMux.__init__:48-148 — Nimmt eine Transformer-Liste oder Factories, registriert nach before_builtins in zwei Lanes.
• _make_child:193-225 — Erzeugt beim Entdecken eines Subgraphen einen Mini-Mux und vererbt die Pump-Bindung.
• _register:227-267 — Ruft transformer.init() auf, um das projection-Dict zu holen, prüft Schlüsselkonflikte und bindet benannte StreamChannels.
• push:269-296 — Synchroner Pfad: ruft process in Reihenfolge auf und entscheidet dann über die Aufnahme ins Hauptlog.
• apush:351-378 — Asynchroner Pfad: aprocess wird seriell awaited, sodass nachfolgende Transformer die async-Ergebnisse der vorherigen sehen.
• close:298-324 — Synchrones Schließen; ruft finalize jedes Transformers auf; im Fehlerfall werden dennoch alle Channels geschlossen.
• aclose:380-404 — Asynchrones Schließen; zunächst gather aller geschedulten Tasks, dann afinalize.
• _collect_stream_modes:398-414 — Leitet aus den required_stream_modes der registrierten Transformer ab, welche Modi beim Graphen angefordert werden müssen.
• _pregel_stream_v3:3533-3558 — V3-Synchroner Einstieg; instanziiert StreamMux + GraphRunStream und reicht den vom Mux bestimmte stream_mode an das zugrunde liegende stream() weiter.

Datenfluss

Der Kern von StreamMux.push besteht nur aus wenigen Zeilen – die Transformer werden in Registrierreihenfolge durchlaufen, und wenn das Event überlebt, wird es ins Hauptlog eingefügt:

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) Es wirkt unscheinbar, aber in transformer.process kann ein Event in den eigenen StreamChannel gepusht werden, und benannte Channels werden vom Mux automatisch ins Hauptlog gewired (_register:227-267). So kann ein Roh-values-Event gleichzeitig von ValuesTransformer in run.values übernommen und von einem nutzerdefinierten Transformer (transformer) transformiert und zurück ins Hauptlog geschrieben werden. Events laufen über eine serielle Kette zwischen Transformern, die Reihenfolge bleibt strikt erhalten.

Grenzen und Fehler

• Async Transformer im sync-Modus wirft sofort: _register ruft transformer_requires_async auf (requires_async check:234-239); wenn ein Transformer aprocess überschreibt, aber nicht supports_sync=True setzt, schlägt die Registrierung am sync-Mux fehl und fordert astream() statt stream().
• projection-Schlüsselkonflikt wirft direkt ValueError: _register prüft set(projection) & set(self.extensions) (conflict check:246-256) und listet im Fehler die Konfliktschlüssel sowie die besitzenden Transformer auf; ein eigener Transformer darf keine Projection-Keys verwenden, die mit eingebauten kollidieren.
• _make_child lehnt vorgefertigte Transformer ab: Wird der Mux mit dem Parameter transformers= konstruiert, kann kein Child-Mux erzeugt werden (factory required:209-214), da vorgefertigte Instanzen sich in einem neuen Scope nicht neu initialisieren lassen. SubgraphTransformer läuft nur auf einem Mux, der mit factories= konstruiert wurde.
• Fehler in finalize blockiert nicht die übrige Aufräumung: close fängt die erste Ausnahme in first_error und schließt trotzdem alle weiteren Transformer und Channels, bevor die Ausnahme am Ende erneut geworfen wird (close error handling:312-324), damit keine Ressourcen lecken.
• bind_pump muss vor dem Lesen des Channels aufgerufen werden: GraphRunStream.__init__ mit wire_pump=True ruft per Default _wire_request_more auf (_wire_request_more:80-92); der Child-Mux übernimmt beim _make_child automatisch die Eltern-Bindung. Wer einen Mux selbst instanziiert und vergisst, die Pump zu binden, lässt StreamChannel.__iter__'s _request_more für immer None sein (_request_more:191-192), und die Iteration endet sofort.
• schedule(coro) funktioniert nur im async-Modus: StreamTransformer.schedule benötigt einen Event-Loop (schedule:233-263); der Mux sammelt in aclose alle geschedulten Tasks mit gather ein und ruft erst nach Abschluss afinalize auf; im sync-Modus schlägt schedule fehl.

Zusammenfassung

StreamMux ist der Knotenpunkt des v3-Streaming-Protokolls: Es implementiert selbst keine Modus-Logik, sondern ruft Transformer in Reihenfolge auf, baut pro Scope einen Child-Mux auf und nimmt sowohl Roh-Events als auch Transformer-Projektionen ins Hauptlog auf. Wie jeder Modus die Events in eine nutzerfreundliche Form wandelt, siehe stream/transformers; wie die Roh-Events aus dem PregelLoop eingespeist werden, siehe Pregel 引擎.

Siehe offizielle Dokumentation: LangGraph 文档 · README.