Command: the resume / goto / update three-in-one primitive

源码版本1.2.9

Responsibilities

Command is LangGraph's "cross-graph control primitive", defined in libs/langgraph/langgraph/types.py (Command class:759). A single Command instance can carry three kinds of semantics at once: resume (resume execution from an interrupt), goto (specify which node to jump to next), and update (write a value into the graph's state channel). A node function returns Command(...), or the caller passes it as input to graph.invoke(Command(...)). Pregel unpacks it into pending writes and feeds them into channels.

It sits between the node and the Pregel main loop: the node produces a Command, and Pregel uses map_command (map_command:56) to break its fields into (task_id, channel, value) triples written into pending writes. The next prepare_next_tasks naturally sees those writes and schedules accordingly: goto becomes a Send written into the TASKS channel, update writes directly to the corresponding channel, and resume goes through the RESUME channel to be consumed by interrupt().

Design motivation

• Unified expression of side effects: besides writing to state, a node may want to "jump to a non-default next node", "pass a message to the parent graph", or "resume an interrupt". These used to be scattered APIs; Command packs them into a single return type. The node signature -> Command[Literal["node_a", "node_b"]] also lets the type checker validate legal jump targets at compile time.
• Command.PARENT crosses graph layers: a node inside a subgraph can return Command(graph=Command.PARENT, update=...) to write updates into the parent graph's state. map_command sees graph == "__parent__" (PARENT guard:58) and raise InvalidUpdateError("There is no parent graph"); the actual handoff to the parent happens in the parent's _first when it processes writes reported by the subgraph.
• goto accepts Send: Command(goto=Send("node", arg)) is equivalent to fanning out a PUSH task; goto="node" is equivalent to goto=Send("node", START) (triggering that node's PULL). So Command embeds the capability of conditional edges into the node return value, with no need for a separate add_conditional_edges.
• resume as the single entry point: Command(resume=...) is the only resume path for interrupt(). It supports both single values (Command(resume="answer"), auto-routed to the unique pending interrupt) and dicts (Command(resume={interrupt_id: value, ...})) for resuming multiple interrupts concurrently.
• update honors reducers: Command(update={"messages": [msg]}) still flows through the channel's reducer. The messages channel is configured with operator.add, so this update appends rather than overwrites. The Overwrite wrapper bypasses the reducer (Overwrite:938) and takes the direct-write path on a BinaryOperatorAggregate channel.

Key files

• Command class:759 — @dataclass class Command(graph, update, resume, goto), generic N denotes the type of the goto node name.
• Command fields:780 — definitions and docstrings for graph / update / resume / goto.
• _update_as_tuples:791 — splits update into [(channel, value), ...], supporting dict, list[tuple], and annotated dataclass forms.
• map_command:56 — core function that breaks a Command into a sequence of pending writes.
• PARENT guard:58 — raises "no parent graph" when cmd.graph == Command.PARENT.
• goto as sends:62 — cmd.goto is decomposed into a list of Sends; Send is written into the TASKS channel, str is written into branch:to:<node> to trigger START.
• resume write:72 — cmd.resume is written into the RESUME channel, to be read by interrupt() on re-entry.
• update writes:74 — cmd.update is split into (channel, value) pairs written into the corresponding channels, going through the reducer.
• Command resume mapping:904 — entry point where the parent graph, upon receiving Command(resume=...), routes resume to the corresponding pending interrupt.
• Overwrite:938 — Overwrite(value=...) wrapper, bypasses the reducer when used with a BinaryOperatorAggregate channel.

Data flow

map_command is the core of Command parsing (map_command:56):

def map_command(cmd: Command) -> Iterator[tuple[str, str, Any]]:
    """Map input chunk to a sequence of pending writes in the form (channel, value)."""
    if cmd.graph == Command.PARENT:
        raise InvalidUpdateError("There is no parent graph")
    if cmd.goto:
        if isinstance(cmd.goto, (tuple, list)):
            sends = cmd.goto
        else:
            sends = [cmd.goto]
        for send in sends:
            if isinstance(send, Send):
                yield (NULL_TASK_ID, TASKS, send)
            elif isinstance(send, str):
                yield (NULL_TASK_ID, f"branch:to:{send}", START)
            else:
                raise TypeError(
                    f"In Command.goto, expected Send/str, got {type(send).__name__}"
                )
    if cmd.resume is not None:
        yield (NULL_TASK_ID, RESUME, cmd.resume)
    if cmd.update:
        for k, v in cmd._update_as_tuples():
            yield (NULL_TASK_ID, k, v)

task_id is always NULL_TASK_ID — these writes do not belong to any specific task, they are graph-level "outbound" writes: the Send from goto goes into the TASKS channel (the next prepare_next_tasks will schedule it as a PUSH task), the str form of goto goes into the branch:to:<node> channel (a channel reserved for conditional edges; writing START means "trigger that node's entry"); resume goes into the RESUME channel, consumed by interrupt() when the node re-enters; each (k, v) from update goes into the corresponding state channel and is merged by the reducer.

When Command is the input to graph.stream(...), PregelLoop._first (Command resume mapping:904) first converts it into pending writes, then groups them by task_id into checkpoint_pending_writes, and prepare_next_tasks handles them naturally. When Command is a node return value, attach_node attaches a _get_updates mapper to the node's writers (attach_node:1431) which extracts updates from Command._update_as_tuples() and writes them into state channels; goto and resume are routed to the branch:to: channel and RESUME via the _control_branch mapper.

Boundaries and failure modes

• Command.PARENT raises in the root graph: map_command directly raise InvalidUpdateError("There is no parent graph") (PARENT guard:58) — Command.PARENT is only meaningful inside a subgraph node; calling it from the root graph raises immediately.
• goto type restrictions: goto accepts only Send / str or a sequence of them; anything else raise TypeError (goto type:69) — passing an integer or dict directly is not allowed.
• Single pending interrupt enforced when resume has no id: Command(resume="single") with multiple pending interrupts raise RuntimeError (multi interrupt error:917); you must switch to Command(resume={id: value, ...}).
• update goes through the reducer and cannot write directly: a BinaryOperatorAggregate channel (e.g. Annotated[list, operator.add]) applies the reducer by default and merges multiple updates; to bypass the reducer use Overwrite(value=...) (Overwrite:938), but multiple Overwrites to the same channel within the same superstep raise InvalidUpdateError.
• Command and entrypoint.final are mutually exclusive: an @entrypoint function returns entrypoint.final(value=..., save=...) which internally uses the END / PREVIOUS channels; returning Command uses the control channels — the two cannot be combined. An entrypoint function should return either Command or final, not both.
• Empty Command raises: when all three fields of Command() are None / (), map_command yields nothing and _first detects the empty writes and raise EmptyInputError("Received empty Command input") (empty command:928).

Summary

Command is the three-in-one primitive for resume / goto / update. Through map_command it is decomposed into channel writes, and the next prepare_next_tasks handles them naturally. It unifies conditional edges, interrupt resume, and state updates in a single return value. See /interrupt/interrupt for the companion interrupt(), and /subgraph/send-command for Send fan-out semantics. See official docs: LangGraph docs · README.