When Graph Runtimes Stay Sane

Setting the stage: actors, channels, and steps

The file we’re dissecting is LangGraph’s Pregel runtime: a graph execution engine where nodes are actors and edges are channels, all driven in discrete steps. LangGraph itself is a framework for building LLM applications as stateful graphs — tools, models, and services wired together with clear data flow.

langgraph/
  pregel/
    main.py        # Pregel runtime (this file)
    _loop.py       # SyncPregelLoop, AsyncPregelLoop
    _algo.py       # prepare_next_tasks, apply_writes, local_read
    _checkpoint.py # checkpoint creation, migration helpers
    _messages.py   # StreamMessagesHandler, v2
    _tools.py      # StreamToolCallHandler
    _runner.py     # PregelRunner (task execution)

User code
  |
  v
StateGraph / entrypoint APIs
  |
  v
Pregel(nodes, channels, ...)
  |        ^
  |        | get_state, bulk_update_state, stream_events
  v        |
SyncPregelLoop / AsyncPregelLoop  <---- BaseCheckpointSaver / BaseCache / BaseStore
  |
  v
Nodes (PregelNode) + Channels (BaseChannel)
  |
  v
LLMs / Tools / External services

Think of this file as the control tower for your LLM app: it doesn’t do the flying, but it decides which plane (node) takes off when, with which messages (channel writes), and how everything is logged (checkpoints and streams).

The runtime exposes two main entrypoints:

• Pregel: the engine that runs a graph, handles checkpoints, retries, and streaming.
• NodeBuilder: a small DSL to declare what each node listens to, does, and writes.

A node is declared structurally — what it subscribes to and what it writes — and the runtime owns the when and how of execution:

node1 = (
    NodeBuilder().subscribe_only("a")
    .do(lambda x: x + x)
    .write_to("b")
)

This says: when channel a changes, run this function, then write its result to channel b. The runtime decides when to run it, how its writes become visible to other nodes, how they’re persisted, and how they’re streamed out to callers.

The rest of the design boils down to three rules:

1. Time advances in discrete steps.
2. All graph state lives in checkpoints.
3. Streams are read‑only views on that state and its events.

Those rules are what keep the runtime sane as it grows: they make concurrency predictable, persistence centralized, and streaming separable from execution.

Rule #1: time moves in steps

Once we know what nodes do, the critical question becomes when they see each other’s outputs. This runtime commits to a strong answer: time advances in discrete steps, and writes from step N are only visible at step N+1. That’s the Bulk Synchronous Parallel (Pregel) model, enforced as an invariant:

• “Channel updates from step N are not visible to tasks in the same step; they become visible only at step N+1.”

You can see this in the core sync loop:

while loop.tick():
    for task in loop.match_cached_writes():
        loop.output_writes(task.id, task.writes, cached=True)
    for _ in runner.tick(
        [t for t in loop.tasks.values() if not t.writes],
        timeout=self.step_timeout,
        get_waiter=get_waiter,
        schedule_task=loop.accept_push,
    ):
        # emit output
        yield from _output(
            stream_mode,
            print_mode,
            subgraphs,
            stream.get,
            queue.Empty,
            version,
            _output_mapper,
            _state_mapper,
        )
    loop.after_tick()
    emit_graph_lifecycle_events(loop)
    if durability_ == "sync":
        loop._put_checkpoint_fut.result()

Each loop iteration is one step:

1. Plan: loop.tick() decides which tasks run this step.
2. Execute: runner.tick(...) runs them and accumulates writes.
3. Update: loop.after_tick() applies those writes for the next step.

Within a step, every node sees a stable view of the world. No node can observe half‑applied updates from its neighbors; you always trade a bit of latency for determinism.

Rule #2: state lives in checkpoints

Steps give us time, but we still need somewhere to store what happened and reconstruct it later. In this runtime, state lives in checkpoints, not in long-lived objects.

A checkpoint is a structured snapshot: channel values and versions, pending writes, tasks, metadata such as the step number, and timestamps. All state APIs — get_state, get_state_history, bulk_update_state — are thin views over this structure.

Reconstructing state from a checkpoint

When callers inspect state, the runtime pulls a CheckpointTuple from the configured saver and turns it into a StateSnapshot via _prepare_state_snapshot:

def _prepare_state_snapshot(
    self,
    config: RunnableConfig,
    saved: CheckpointTuple | None,
    recurse: BaseCheckpointSaver | None = None,
    apply_pending_writes: bool = False,
) -> StateSnapshot:
    if not saved:
        return StateSnapshot(
            values={},
            next=(),
            config=config,
            metadata=None,
            created_at=None,
            parent_config=None,
            tasks=(),
            interrupts=(),
        )

    self._migrate_checkpoint(saved.checkpoint)

    step = saved.metadata.get("step", -1) + 1
    channels, managed = channels_from_checkpoint(...)
    next_tasks = prepare_next_tasks(...)
    ...
    tasks_with_writes = tasks_w_writes(...)
    return StateSnapshot(
        read_channels(channels, self.stream_channels_asis),
        tuple(t.name for t in next_tasks.values() if not t.writes),
        patch_checkpoint_map(saved.config, saved.metadata),
        saved.metadata,
        saved.checkpoint["ts"],
        patch_checkpoint_map(saved.parent_config, saved.metadata),
        tasks_with_writes,
        tuple([i for task in tasks_with_writes for i in task.interrupts]),
    )

A few design choices stand out:

• Migrations are localized in _migrate_checkpoint, so schema changes don’t leak into callers.
• Subgraphs use namespaces (checkpoint_ns) and can be delegated to nested Pregel instances.
• Tasks and interrupts are derived from checkpoints and pending writes, not global mutable state.

Editing history safely with bulk updates

On top of this checkpoint model, the runtime exposes bulk_update_state / abulk_update_state. These APIs let you “edit” a graph’s state as if certain nodes had produced specific writes, still grounded in checkpoints.

That unlocks concrete workflows:

• Apply corrective updates to a running conversation or workflow.
• Simulate inputs (as_node == INPUT) without replaying the whole graph.
• Clear or fork state using special markers like END and "__copy__".

But the implementation never steps outside the core model: it starts from a checkpoint, uses the same helpers (apply_writes, prepare_next_tasks), and persists a new checkpoint at the end.

There’s also a small but important affordance around as_node. When you omit it, the runtime tries to infer a node from:

• Whether there’s only one node in the graph.
• Whether any node has ever updated the state (versions_seen).
• Which node most recently updated the state.

If it can’t pick a unique node, it raises InvalidUpdateError("Ambiguous update, specify as_node"). The API is convenient when the intent is obvious, and explicit when it isn’t.

Rule #3: streams are views, not side effects

The runtime also needs to expose what’s happening in real time: values changing, messages being produced, lifecycle events, interrupts. It does this with streaming — but without letting streaming own any business logic. Streams are derived views over internal events, not sources of truth.

This file carries three generations of streaming:

• v1: legacy, more ad‑hoc event structures.
• v2: typed StreamPart dicts with cleaner shapes and explicit interrupts.
• v3: an experimental mux-based protocol that builds multiple projections on top of v2.

The streaming choke point: _output

Both sync and async streaming funnel through a single helper, _output. This function is the last place an event passes through before it leaves the runtime:

def _output(
    stream_mode: StreamMode | Sequence[StreamMode],
    print_mode: StreamMode | Sequence[StreamMode],
    stream_subgraphs: bool,
    getter: Callable[[], tuple[tuple[str, ...], str, Any]],
    empty_exc: type[Exception],
    version: Literal["v1", "v2"] = "v1",
    output_mapper: Callable[[Any], Any] | None = None,
    state_mapper: Callable[[Any], Any] | None = None,
) -> Iterator:
    while True:
        try:
            ns, mode, payload = getter()
        except empty_exc:
            break
        if mode in print_mode:
            ...  # debug printing
        if mode in stream_mode:
            if version == "v2":
                if mode == "values":
                    ints: tuple[Interrupt, ...] = ()
                    if isinstance(payload, dict):
                        ints = payload.pop(INTERRUPT, ())
                        if output_mapper:
                            payload = output_mapper(payload)
                    yield {"type": mode, "ns": ns, "data": payload, "interrupts": ints}
                elif mode in ("checkpoints", "debug"):
                    if state_mapper:
                        _coerce_checkpoint_values(payload, state_mapper)
                    yield {"type": mode, "ns": ns, "data": payload}
                else:
                    yield {"type": mode, "ns": ns, "data": payload}
            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

This adapter decides:

1. The public shape (plain payloads vs typed dicts with type / ns / data).
2. How interrupts are surfaced (separate interrupts field in v2).
3. How subgraphs are represented (namespaces included or not).

Equally important is what it does not do: no scheduling, no checkpoint changes, no routing of tasks. It’s a pure projection over an internal event queue.

v1 vs v2: evolving formats safely

The public invoke / ainvoke helpers show how the runtime evolves formats without rewriting the engine.

For v2, invoke simply consumes v2 stream(...) events and aggregates value and interrupts into a GraphOutput:

if version == "v2":
    for chunk in self.stream(..., version=version):
        if stream_mode == "values":
            latest = chunk["data"]
            if chunk_ints := chunk.get("interrupts", ()):  # explicit field
                interrupts.extend(chunk_ints)
        else:
            chunks.append(chunk)
    return GraphOutput(value=latest, interrupts=tuple(interrupts))

For v1, it reads the same internal events but extracts interrupts from "updates" payloads and merges them back under the legacy INTERRUPT key. New code uses structured types; old code keeps working on top of the same stream.

v3: streaming as a multiplexed bus

The most advanced layer is v3 streaming, built around a StreamMux and transformers. A transformer subscribes to certain event modes and emits a structured view: “values only”, “messages with tokens”, “lifecycle events”, or any custom projection.

parent_ns = _resolve_parent_ns(self.config, config)
mux = StreamMux(
    factories=[
        ValuesTransformer,
        MessagesTransformer,
        LifecycleTransformer,
        SubgraphTransformer,
        *compiled_factories,
        *extra_factories,
    ],
    scope=parent_ns,
    is_async=False,
)

graph_iter = iter(
    self.stream(
        input,
        patch_configurable(config, {CONFIG_KEY_STREAM_MESSAGES_V2: True}),
        stream_mode=_collect_stream_modes(mux),
        subgraphs=True,
        version="v2",
        ...,
    )
)
return GraphRunStream(graph_iter, mux)

To keep v3 predictable, _reject_v3_invariant_kwargs blocks callers from overriding internal streaming invariants like stream_mode or subgraphs. If you opt into v3, the runtime owns how streams are wired; you only choose which projections you care about.

Operating this engine at scale

The three rules — stepped time, checkpointed state, and projection-only streams — also make operations measurable. The file is explicit about hot paths, and they line up cleanly with the design:

• stream / astream: main execution loops, cost ≈ steps × tasks per step.
• bulk_update_state / abulk_update_state: hot under migrations or batched corrections.
• _prepare_state_snapshot / _aprepare_state_snapshot: hit on every state or history read.

These translate almost directly into metrics worth tracking:

• pregel_steps_per_run to detect graphs edging towards recursion limits or infinite loops.
• pregel_tasks_per_step to spot sudden fan‑out that will burn CPU.
• checkpoint_write_latency_ms to understand the cost of durability, especially with durability="sync".

Each while loop.tick() is a step, each runner.tick(...) processes tasks within that step, and checkpoint writes are driven by the configured BaseCheckpointSaver and durability mode. Because execution, persistence, and streaming are cleanly separated, you can tune and instrument each dimension independently.

Practical takeaways

Underneath all the details, this file is about one core lesson: you can keep a complex graph runtime sane by enforcing simple, global rules for time, state, and streams. Everything else is an application of that idea.

1. Make time discrete when coordinating many workers.

   Use step-based semantics so each worker sees a stable view of the world during a step. This makes reasoning about concurrency tractable, especially when orchestrating async tools, LLM calls, or background jobs.

2. Treat checkpoints as your only source of truth.

   Centralize persistent state in a single schema, and route all mutation and inspection through it. That’s what enables safe migrations, history introspection, and features like bulk_update_state without hidden mutable objects.

3. Separate execution from streaming.

   Implement execution loops and persistence without caring about external formats, then build streaming as a projection layer on top. A tiny adapter like _output should be the only place where you commit to shapes and versions.

LangGraph’s Pregel implementation shows these rules applied consistently across a large codebase: steps govern visibility, checkpoints anchor state, and streams are strictly views. That’s what keeps the engine understandable as it gains features like bulk updates, subgraphs, and multiple streaming versions.

If you’re building serious LLM applications or any graph‑shaped system, internalizing these patterns is the difference between a clever demo and an engine you can run in production for years.