Streaming-Ausgabe: astream und stream_mode

源码版本1.2.9

Verantwortung

Pregel.astream / Pregel.stream sind die vom kompilierten Graphen nach außen gelegten Einstiegspunkte für die Ausführung. Man wirft eine Eingabe hinein, die PregelLoop startet und läuft Runde für Runde; dabei werden die in jedem Superstep (superstep) erzeugten Daten gemäß dem vom Aufrufer angegebenen stream_mode zu Stream-Events projiziert, in eine Queue gesteckt und vom Aufrufer iterativ konsumiert. ainvoke / invoke sind im Wesentlichen astream / stream, die zu einem Einzelwert konvergieren – der Rumpf von Pregel.invoke ist for chunk in self.stream(...) und nimmt das letzte Stück (Pregel.invoke body:3891).

Die Verantwortung dieser Schicht ist es, «Ausführung» und «Ausgabeform» zu entkoppeln: Die Ausführung folgt weiterhin der BSP-Schleife von Pregel, die Ausgabeform wird durch stream_mode bestimmt – derselbe Graph, einmal ausgeführt, kann verschiedene Projektionen liefern: values (voller Zustand pro Schritt), updates (inkrementell pro Schritt), messages (LLM-Token-Strom), custom (selbstdefinierte Ausgaben innerhalb eines Knotens), checkpoints (Checkpoint-Events), tasks (Task-Beginn/-Ende), debug (alles); es kann sogar eine Liste übergeben werden, um mehrere Modi gleichzeitig zu beziehen.

Entwurfsmotivation

• Ausführung nicht blockieren: Die Ausgabe läuft über eine Queue; runner.tick / atick schreibt ein Event, sobald ein Task fertig ist, und die äußere for-Schleife nimmt es bei der Iteration ab, statt erst den gesamten Graphen fertig zu rechnen und dann alles auf einmal zurückzugeben. Für LLM-Token-Streams (messages) ist das zwingend – der Nutzer möchte Token für Token sehen, während sie generiert werden.
• Mehrere Projektionen aus derselben Ausführung: Innerhalb eines runner.atick(...) entstehen mehrere Events; die Funktion _output (_output:4184) führt nur Filterung/Verteilung durch und rechnet den Graphen nicht erneut. stream_mode=["values","updates"] führt nicht zu zwei Durchläufen.
• Einheitlicher Subgraph-Namensraum: Mit subgraphs=True erhalten Events einen namespace-Präfix ((ns, mode, payload)); der Eltern-Graph sieht die Events aus dem Inneren des Subgraphen. Das Tagging passiert beim Verlassen der Queue durch _output, der Subgraph öffnet keinen eigenen Stream.
• v2 typisiert: In der neuen Version formt version="v2" Events zu einem Dict {"type": mode, "ns": ..., "data": ..., "interrupts": ...} um, was die programmgesteuerte Verarbeitung erleichtert. Im values-Modus wird __interrupt__ zusätzlich in ein separates interrupts-Feld ausgekoppelt.

Schlüsseldateien

• Pregel class:450 — Die Pregel-Klassendefinition; alle Streaming-Einstiegspunkte liegen hier.
• Pregel.stream:2655 — Synchroner Streaming-Einstieg; die Signatur listet alle stream_mode-Optionen auf.
• Pregel.astream:3063 — Asynchroner Streaming-Einstieg; die eigentliche asynchrone Hauptschleife steht in seinem Rumpf.
• Pregel.invoke:3836 — Synchrone Konvergenz; intern for chunk in self.stream(...), nimmt den letzten values-Chunk.
• sync main loop:2964 — Synchrone Hauptschleife: while loop.tick() → runner.tick → loop.after_tick().
• async main loop:3437 — Asynchrone Hauptschleife: dreiteilig wie oben, runner.atick ersetzt die synchrone Variante.
• _output:4184 — Dequeue-Filter-Funktion; entscheidet anhand von stream_mode / print_mode / subgraphs, wie jedes Event yielded wird.
• GraphRunStream:31 — Synchrone, aufrufergetriebene Stream-Verpackung; die for-Schleife ist die Pump, ohne Hintergrund-Thread.
• AsyncGraphRunStream:304 — Asynchrones Gegenstück; mehrere Projektionen (run.values / run.messages) teilen sich eine Pump.
• stream_mode attr:709 — Pregel hat den Default stream_mode="values", kann durch stream(stream_mode=...) überschrieben werden.

Datenfluss

Das Skelett der asynchronen Hauptschleife steht im Rumpf von astream (async main loop:3437); der dreiteilige Aufbau entspricht dem auf der Pregel-Engine-Seite Beschriebenen, mit dem Unterschied, dass zwischen den Teilen _output die Inhalte der Queue an den Aufrufer weitergibt:

while loop.tick():
    for task in await loop.amatch_cached_writes():
        loop.output_writes(task.id, task.writes, cached=True)
    async for _ in runner.atick(
        [t for t in loop.tasks.values() if not t.writes],
        timeout=self.step_timeout,
        get_waiter=get_waiter,
        schedule_task=loop.aaccept_push,
    ):
        # emit output
        for o in _output(
            stream_mode,
            print_mode,
            subgraphs,
            stream.get_nowait,
            asyncio.QueueEmpty,
            version,
            _output_mapper,
            _state_mapper,
        ):
            yield o
    loop.after_tick()
    await aemit_graph_lifecycle_events(loop)
    # wait for checkpoint
    if durability_ == "sync":
        await cast(asyncio.Future, loop._put_checkpoint_fut)

runner.atick schreibt intern die von jedem Task erzeugten Writes zurück ins loop und legt gleichzeitig die Tripel (ns, mode, payload) in den stream (eine asyncio.Queue); wenn die äußere async for-Schleife die Kontrolle erhält, ruft _output stream.get_nowait() auf, um alle bereiten Events aus der Queue zu entleeren, filtert sie nach stream_mode und yieldet sie an den Nutzer. Die Kernlogik von _output (_output:4184): Holt (ns, mode, payload), gibt bei mode in print_mode zunächst eine print-Kopie aus und yieldet nach außen nur, wenn mode in stream_mode – print_mode ist also ein Debug-Schalter «nur anzeigen, nicht senden» und beeinflusst den tatsächlich geyieldeten Inhalt nicht.

Das folgende Diagramm durchläuft den stream-Aufruf vom Einstieg bis zum Verlassen der Queue:

Grenzen und Fehler

• Default stream_mode: Bei stream_mode=None ist der Default values, wenn der Aufruf als Subgraph erfolgt (CONFIG_KEY_TASK_ID im Config), andernfalls gilt self.stream_mode (stream_mode default:2740), damit der Subgraph die Eltern-Ausgabe nicht mit seinem Default updates verschmutzt.
• Schrittzahl überschritten: Nach dem Verlassen der Hauptschleife wird loop.status geprüft; ist es out_of_steps, wirft der Code GraphRecursionError und empfiehlt, recursion_limit zu erhöhen (out_of_steps:3002); der Status draining wirft GraphDrained und gibt die Kontrolle an das externe RunControl zurück.
• Mehrere Modi gleichzeitig: Wenn stream_mode eine Liste ist, wird jedes Event zum Tupel (mode, payload); zusammen mit subgraphs=True ergibt sich (ns, mode, payload) (tuple mode:4240), und der Aufrufer muss das Tupel entsprechend entpacken.
• Vererbungsbehandlung im messages-Modus: Bei stream_mode="messages" und version="v1" wird der vom übergeordneten Layer vererbte v2-messages-Handler entfernt, damit der v1-Stream nicht in das Content-Block-Event-Protokoll geroutet wird (strip v2 handler:2773); der v1-Handler bleibt aber erhalten, damit bei subgraphs=True innere messages-Events nach außen sichtbar sind.
• durability und Checkpoint-Synchronisation: Bei durability="sync" wartet die Hauptschleife nach jedem Schritt auf await loop._put_checkpoint_fut, also darauf, dass der Checkpoint persistiert ist, bevor der nächste Schritt beginnt; "async" ist der Default – Persistenz und nächster Schritt laufen parallel; "exit" persistiert nur beim Beenden.
• Experimentelles v3 stream_events: stream_events(version="v3") akzeptiert keine Parameter stream_mode und subgraphs, diese werden intern vom Mux übernommen; explizites Übergeben wird von _reject_v3_invariant_kwargs abgelehnt (_reject_v3_invariant_kwargs:387).

Zusammenfassung

astream / stream verpacken die BSP-Schleife von Pregel als iterierbare Schnittstelle; stream_mode bestimmt die Ausgabeprojektion, und _output ist der einzige Engpass für die Dequeue-Filterung. Für Ausführungsdetails siehe /pregel/pregel und /pregel/loop; für Multi-Modus-Mux siehe libs/langgraph/langgraph/stream/_mux.py. Siehe offizielle Dokumentation: LangGraph 文档 · README.