Streaming output: astream and stream_mode

源码版本1.2.9

Responsibilities

Pregel.astream / Pregel.stream are the execution entry points a compiled graph exposes. You feed in input, it starts a PregelLoop that runs round after round, and along the way it projects the data produced by each superstep into stream events according to the caller-specified stream_mode, pushes them into a queue, and lets the caller iterate and consume. ainvoke / invoke are essentially astream / stream collapsed to a single value — the body of Pregel.invoke is just for chunk in self.stream(...) taking the last chunk (Pregel.invoke body:3891).

This layer's responsibility is decoupling "execution" from "output shape": execution stays the Pregel BSP loop, while the output shape is decided by stream_mode. A single run of the same graph can ask for values (full state per step), updates (per-step deltas), messages (LLM token stream), custom (in-node custom output), checkpoints (checkpoint events), tasks (task start/stop), debug (everything), and other projections — you can even pass a list to pull multiple modes at once.

Design motivation

• Non-blocking execution: Output goes through a queue; runner.tick / atick pushes events in as each task finishes, and the outer for loop iterates and pulls rather than running the whole graph and returning at the end. This is a hard requirement for LLM token streams (messages) — users need to see tokens as they are generated.
• Multiple projections reuse one execution: A single runner.atick(...) produces multiple kinds of events internally. The _output function (_output:4184) only filters and dispatches; it does not re-run the graph. Passing stream_mode=["values","updates"] does not run the graph twice.
• Unified subgraph namespace: With subgraphs=True enabled, events carry a namespace prefix ((ns, mode, payload)); the parent graph sees subgraph-internal events, tagged by _output on dequeue rather than the subgraph opening its own stream.
• v2 typing: The new version turns events into {"type": mode, "ns": ..., "data": ..., "interrupts": ...} dicts via version="v2", for easier downstream programmatic handling; the values mode also extracts __interrupt__ into a separate interrupts field.

Key files

• Pregel class:450 — The Pregel class definition; all streaming entry points live here.
• Pregel.stream:2655 — Sync streaming entry; the signature lists every stream_mode option.
• Pregel.astream:3063 — Async streaming entry; the real async main loop is written inside its body.
• Pregel.invoke:3836 — Sync collapsing entry; internally for chunk in self.stream(...) taking the last values chunk.
• sync main loop:2964 — Sync main loop: while loop.tick() → runner.tick → loop.after_tick().
• async main loop:3437 — Async main loop: three phases as above, with runner.atick replacing the sync version.
• _output:4184 — Dequeue filter function; decides how each event is yielded based on stream_mode / print_mode / subgraphs.
• GraphRunStream:31 — Sync caller-driven stream wrapper; the for loop is the pump, no background thread.
• AsyncGraphRunStream:304 — Async counterpart; multiple projections (run.values / run.messages) share one pump.
• stream_mode attr:709 — Pregel defaults to stream_mode="values", overridable by stream(stream_mode=...).

Data flow

The async main loop skeleton lives in the astream body (async main loop:3437); the three phases are the same as on the Pregel engine page, the difference being _output is interleaved between phases to flush the queue to the caller:

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)

Inside runner.atick, writes produced by each task are written back to loop, and (ns, mode, payload) triples are pushed into stream (an asyncio.Queue). When the outer async for gets control, _output calls stream.get_nowait() to drain every ready event in the queue, filters by stream_mode, and then yields to the user. The core of _output (_output:4184): take the (ns, mode, payload), print it if mode in print_mode, and only yield outward if mode in stream_mode — so print_mode is a "look but don't emit" debug switch that does not affect what is actually yielded.

The diagram below walks a stream call from entry to event dequeue:

Boundaries and failure modes

• Default stream_mode: When stream_mode=None, if called as a subgraph (CONFIG_KEY_TASK_ID is in config), it defaults to values; otherwise it uses self.stream_mode (stream_mode default:2740), to avoid a subgraph's default updates mode polluting the parent's output.
• Step limit exceeded: After the main loop exits, inspect loop.status; if it is out_of_steps, raise GraphRecursionError telling the user to raise recursion_limit (out_of_steps:3002). A draining status raises GraphDrained, handing control back to the outer RunControl.
• Multiple modes output simultaneously: When stream_mode is a list, each event becomes a (mode, payload) tuple; layering subgraphs=True on top makes it (ns, mode, payload) (tuple mode:4240) — the caller must destructure accordingly.
• Inheritance handling for messages mode: With stream_mode="messages" and version="v1", the inherited v2 messages handler is stripped off to prevent v1 streams from being routed through the content-block event protocol (strip v2 handler:2773), but the v1 handler is kept so inner messages events can still be observed by outer layers when subgraphs=True.
• Durability and checkpoint sync: Under durability="sync", the main loop awaits loop._put_checkpoint_fut at the end of each step, ensuring the next step only starts after the checkpoint has landed. "async" is the default — persistence runs in parallel with the next step. "exit" only persists on exit.
• v3 experimental stream_events: stream_events(version="v3") does not accept stream_mode and subgraphs arguments — both are taken over by the mux internally; passing them explicitly is rejected by _reject_v3_invariant_kwargs (_reject_v3_invariant_kwargs:387).

Summary

astream / stream wrap Pregel's BSP loop in an iterable interface; the choice of stream_mode decides the output projection, and _output is the single throat for dequeue filtering. For execution details see /pregel/pregel and /pregel/loop; for multi-mode mux see libs/langgraph/langgraph/stream/_mux.py.

See official docs: LangGraph docs · README.