From Storyboard to Runtime

StateGraph solves a concrete problem: coordinating many functions that evolve a shared state over time. Instead of hard-coding call chains, you draw a storyboard where each node is a function, edges define what can run next, and the script is a shared state object that every node can read and partially update.

CompiledStateGraph then turns that storyboard into a running production using a Pregel-style engine: nodes wake up when their input channels change, emit updates, and control where execution flows next. The entire system behaves like a conversation where nodes talk only through a constrained, typed medium: the state channels.

langgraph/
  graph/
    state.py        <- StateGraph & CompiledStateGraph (this file)
    _node.py        <- StateNodeSpec definitions
    _branch.py      <- BranchSpec for conditional edges
  channels/
    base.py         <- BaseChannel abstraction
    last_value.py   <- LastValue, LastValueAfterFinish
    ephemeral_value.py
    named_barrier_value.py
  pregel/
    __init__.py     <- Pregel runtime
    _read.py        <- ChannelRead
    _write.py       <- ChannelWrite, ChannelWriteEntry
  managed/
    base.py         <- ManagedValueSpec
  checkpoint/
    base.py         <- Checkpoint interface

User code
  -> builds StateGraph(StateSchema, ContextSchema)
  -> adds nodes/edges/branches
  -> calls .compile() -> CompiledStateGraph (Pregel-based)
  -> invokes graph via Runnable interface

The core abstraction is simple: every node is a function that takes the current state (and optional context) and returns a partial update to that state. Internally, this becomes a message-passing system of channels and triggers. The interesting design work in this file is how it hides that machinery while keeping strong guarantees about types, routing, and backward compatibility.

Channels: The Conveyor Belts of State

Once we think of nodes as scenes, the next question is how they talk. In this design, the answer is channels. A channel is like a conveyor belt in a factory: each belt carries values for one state key between machines (nodes), and the belt type determines how values are buffered or reduced.

Instead of asking you to wire those belts manually, StateGraph infers them from your schemas. You define your state as a TypedDict- or Pydantic-like model, and the graph turns each annotated field into a specific channel type.

def _get_channels(
    schema: type[dict],
) -> tuple[dict[str, BaseChannel], dict[str, ManagedValueSpec], dict[str, Any]]:
    if not hasattr(schema, "__annotations__"):
        return (
            {"__root__": _get_channel("__root__", schema, allow_managed=False)},
            {},
            {},
        )

    type_hints = get_type_hints(schema, include_extras=True)
    all_keys = {
        name: _get_channel(name, typ)
        for name, typ in type_hints.items()
        if name != "__slots__"
    }
    return (
        {k: v for k, v in all_keys.items() if isinstance(v, BaseChannel)},
        {k: v for k, v in all_keys.items() if is_managed_value(v)},
        type_hints,
    )

The helper _get_channel decides what kind of belt each field gets. If you use Annotated metadata to tag a field with a channel type or a reducer, that metadata is interpreted here. Otherwise, you get a default LastValue channel that simply holds the latest value.

The function returns three things:

• A mapping from state keys to BaseChannel implementations.
• A mapping from keys to ManagedValueSpec for values that are stored externally.
• The resolved type hints for later validation and JSON-schema generation.

The effect is that you describe your state once, using types, and the system builds a consistent, type-aware transport layer around it. State schemas become the source of truth for both data shape and wiring.

The Builder–Runtime Split

With channels in place, the file leans on a strict separation between declaring the graph and running it. This builder–runtime split is one of the strongest architectural choices here.

The StateGraph class is a pure builder. It tracks:

• The node specs and their names (self.nodes).
• The edges and conditional branches between nodes.
• The schemas for state, input, output, and context.
• The inferred channels and managed values for each schema.

None of that builder code executes the workflow. Execution lives in CompiledStateGraph, which subclasses a Pregel runtime. The bridge between the two worlds is compile(), which freezes the declarative structure into an efficient, reusable runtime.

def compile(
    self,
    checkpointer: Checkpointer = None,
    *,
    cache: BaseCache | None = None,
    store: BaseStore | None = None,
    interrupt_before: All | list[str] | None = None,
    interrupt_after: All | list[str] | None = None,
    debug: bool = False,
    name: str | None = None,
) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
    checkpointer = ensure_valid_checkpointer(checkpointer)
    interrupt_before = interrupt_before or []
    interrupt_after = interrupt_after or []

    self.validate(
        interrupt=(
            (interrupt_before if interrupt_before != "*" else []) + interrupt_after
            if interrupt_after != "*"
            else []
        )
    )

    output_channels = (
        "__root__"
        if len(self.schemas[self.output_schema]) == 1
        and "__root__" in self.schemas[self.output_schema]
        else [
            key
            for key, val in self.schemas[self.output_schema].items()
            if not is_managed_value(val)
        ]
    )

    compiled = CompiledStateGraph(
        builder=self,
        schema_to_mapper={},
        context_schema=self.context_schema,
        nodes={},
        channels={
            **self.channels,
            **self.managed,
            START: EphemeralValue(self.input_schema),
        },
        input_channels=START,
        stream_mode="updates",
        output_channels=output_channels,
        stream_channels=...,  # simplified here
        checkpointer=checkpointer,
        interrupt_before_nodes=interrupt_before,
        interrupt_after_nodes=interrupt_after,
        auto_validate=False,
        debug=debug,
        store=store,
        cache=cache,
        name=name or "LangGraph",
    )

compile() validates the graph, derives which channels represent the output, and then instantiates a CompiledStateGraph with:

• All data channels and managed values.
• An ephemeral input channel (START).
• Configured interruption points, checkpointer, cache, and store.

From that point on, callers interact with the compiled graph through a Runnable-style interface. Build-time is where types, schemas, and topology are resolved once; runtime is where message passing and node execution happen repeatedly.

Commands, Branches, and Joins

Real workflows do more than run straight lines. They branch, loop, and often need to wait for multiple paths to complete before moving on. This file encodes all of that control flow as data on channels, rather than ad-hoc conditionals buried inside node bodies.

Normalizing node outputs

Nodes in user code can return many shapes: plain dicts of updates, Command objects, lists combining both, or objects with Annotated metadata. Internally, the runtime needs a single, strict representation: a sequence of (key, value) updates targeting known channels.

def attach_node(self, key: str, node: StateNodeSpec[Any, ContextT] | None) -> None:
    if key == START:
        output_keys = [
            k
            for k, v in self.builder.schemas[self.builder.input_schema].items()
            if not is_managed_value(v)
        ]
    else:
        output_keys = list(self.builder.channels) + [
            k for k, v in self.builder.managed.items()
        ]

    def _get_updates(
        input: None | dict | Any,
    ) -> Sequence[tuple[str, Any]] | None:
        if input is None:
            return None
        elif isinstance(input, dict):
            return [(k, v) for k, v in input.items() if k in output_keys]
        elif isinstance(input, Command):
            if input.graph == Command.PARENT:
                return None
            return [
                (k, v) for k, v in input._update_as_tuples() if k in output_keys
            ]
        elif (
            isinstance(input, (list, tuple))
            and input
            and any(isinstance(i, Command) for i in input)
        ):
            updates: list[tuple[str, Any]] = []
            for i in input:
                if isinstance(i, Command):
                    if i.graph == Command.PARENT:
                        continue
                    updates.extend(
                        (k, v) for k, v in i._update_as_tuples() if k in output_keys
                    )
                else:
                    updates.extend(_get_updates(i) or ())
            return updates
        elif (t := type(input)) and get_cached_annotated_keys(t):
            return get_update_as_tuples(input, output_keys)
        else:
            msg = create_error_message(
                message=f"Expected dict, got {input}",
                error_code=ErrorCode.INVALID_GRAPH_NODE_RETURN_VALUE,
            )
            raise InvalidUpdateError(msg)

_get_updates sits on the hot path: every node return flows through it. It filters out unknown keys, ignores commands targeting parent graphs, and raises a dedicated InvalidUpdateError when a node produces an unexpected shape.

Without this central funnel, loosely-typed workflows quickly become fragile. A single misbehaving node could corrupt shared state in subtle ways. Here, one function enforces output invariants and concentrates error handling.

Commands and branch channels

Control flow itself is also data. LangGraph’s Command and Send objects let nodes say “go here next” or “enqueue this extra task.” This file translates those objects into writes on special control channels.

def _control_branch(value: Any) -> Sequence[tuple[str, Any]]:
    if isinstance(value, Send):
        return ((TASKS, value),)
    commands: list[Command] = []
    if isinstance(value, Command):
        commands.append(value)
    elif isinstance(value, (list, tuple)):
        for cmd in value:
            if isinstance(cmd, Command):
                commands.append(cmd)
    rtn: list[tuple[str, Any]] = []
    for command in commands:
        if command.graph == Command.PARENT:
            raise ParentCommand(command)

        goto_targets = (
            [command.goto] if isinstance(command.goto, (Send, str)) else command.goto
        )

        for go in goto_targets:
            if isinstance(go, Send):
                rtn.append((TASKS, go))
            elif isinstance(go, str) and go != END:
                rtn.append((_CHANNEL_BRANCH_TO.format(go), None))
    return rtn

The constant _CHANNEL_BRANCH_TO = "branch:to:{}" defines a naming convention: every node has a corresponding branch:to: channel that means “please run this node now.” Edges and commands ultimately become writes to these channels, and each node listens to its own branch channel as a trigger.

Joins as barrier channels

Joins—“run C only after A and B finish”—are implemented as named barriers. When you add a multi-start edge like add_edge(["A", "B"], "C"), the compiled graph inserts an intermediate channel that waits for all predecessors.

def attach_edge(self, starts: str | Sequence[str], end: str) -> None:
    if isinstance(starts, str):
        if end != END:
            self.nodes[starts].writers.append(
                ChannelWrite(
                    (ChannelWriteEntry(_CHANNEL_BRANCH_TO.format(end), None),)
                )
            )
    elif end != END:
        channel_name = f"join:{'+'.join(starts)}:{end}"
        if self.builder.nodes[end].defer:
            self.channels[channel_name] = NamedBarrierValueAfterFinish(
                str, set(starts)
            )
        else:
            self.channels[channel_name] = NamedBarrierValue(str, set(starts))
        self.nodes[end].triggers.append(channel_name)
        for start in starts:
            self.nodes[start].writers.append(
                ChannelWrite((ChannelWriteEntry(channel_name, start),))
            )

Each predecessor writes its own name into the join channel. The barrier channel knows the full set of required predecessors ({"A", "B"} in this example) and only emits when it has seen all of them. At that point, the downstream node’s trigger fires and C can run.

Staying Correct Over Time

A graph runtime like this lives a long time in production. That introduces two hard requirements: you must be able to evolve internal representations without breaking existing workflows, and you must be able to operate and debug complex graphs safely.

Migrations and long-lived checkpoints

LangGraph persists checkpoints that record per-channel values and versions. Earlier versions of the system used different channel naming schemes (e.g., start:, branch:source:cond:node, or just node). CompiledStateGraph carries migration logic that upgrades these to the current conventions.

def _migrate_checkpoint(self, checkpoint: Checkpoint) -> None:
    super()._migrate_checkpoint(checkpoint)

    values = checkpoint["channel_values"]
    versions = checkpoint["channel_versions"]
    seen = checkpoint["versions_seen"]

    if not versions:
        return

    if checkpoint["v"] >= 3:
        return

    # Migrate from start:node to branch:to:node
    for k in list(versions):
        if k.startswith("start:"):
            node = k.split(":")[1]
            if node not in self.nodes:
                continue
            new_k = f"branch:to:{node}"
            new_v = (
                max(versions[new_k], versions.pop(k))
                if new_k in versions
                else versions.pop(k)
            )
            for ss in (seen.get(node, {}), seen.get(INTERRUPT, {})):
                if k in ss:
                    s = ss.pop(k)
                    if new_k in ss:
                        ss[new_k] = max(s, ss[new_k])
                    else:
                        ss[new_k] = s
            if new_k not in values and k in values:
                values[new_k] = values.pop(k)
            versions[new_k] = new_v

    # (similar loop for branch:source:cond:node -> branch:to:node)
    # ...

For each renamed channel, the code updates:

• channel_versions, preserving the highest version number.
• versions_seen for both node execution and interrupts.
• channel_values, moving stored data to the new key when needed.

It also short-circuits when the checkpoint’s version is already new enough and skips channels that refer to nodes that no longer exist. This is the level of care you need if your workflows effectively become part of user-visible conversation history.

Schemas and guardrails for operability

On the operability side, this file exposes the graph’s contract via JSON Schema and enforces a set of invariants that make production failures easier to reason about.

Because StateGraph tracks typed schemas, the compiled graph can generate JSON Schema for its inputs and outputs:

def get_input_jsonschema(self, config: RunnableConfig | None = None) -> dict[str, Any]:
    return _get_json_schema(
        typ=self.builder.input_schema,
        schemas=self.builder.schemas,
        channels=self.builder.channels,
        name=self.get_name("Input"),
    )

Internally, _get_json_schema handles three cases: direct Pydantic models, TypedDict-style structures, and “other” types where it synthesizes a Pydantic model from channel update types. That keeps the external contract aligned with the internal wiring.

The file also chooses to fail fast in several places:

• validate() rejects edges that reference unknown nodes or missing entry points.
• _add_schema() stops “managed” values from entering input/output schemas, which would blur the line between internal and external state.
• _get_updates() raises InvalidUpdateError with a structured ErrorCode when a node returns an invalid shape.

These guardrails make graphs safer to operate. Combined with metrics such as graph_invalid_update_errors_total and graph_checkpoint_size_bytes, they give you a clear signal when changes in graph design or node behavior start to stress the system.

Design Lessons You Can Steal

Stepping back, this file is a compact demonstration of how to turn a set of functions into a distributed conversation over shared state without losing control. Everything revolves around one principle: treat the workflow as a typed graph of nodes talking through explicit channels, not as a tangle of ad-hoc calls.

If you’re designing your own workflow engine, orchestration layer, or stateful AI runtime, a few concrete steps emerge from this design:

1. Let types drive wiring. Use TypedDict, Pydantic, or similar schemas not just as documentation, but as the source of truth for channels, reducers, and managed values.
2. Separate declaration from execution. Keep a clean builder API and compile into a runtime that can optimize, checkpoint, and schedule independently of user code.
3. Normalize outputs in one place. Design a single funnel (like _get_updates) that every node output passes through. Enforce invariants there and emit structured errors.
4. Encode control flow as data. Use explicit channels and command objects for branches, joins, and background tasks instead of burying that logic inside node bodies.
5. Be deliberate about naming and versions. Choose clear channel naming conventions, centralize them, and add explicit migration logic when you evolve them.

As your systems grow from a handful of functions into rich, stateful conversations, treating them as graphs of nodes talking through well-defined, typed channels—exactly what StateGraph and CompiledStateGraph do here—can be the difference between an orchestration layer that scales gracefully and one that collapses under its own complexity.