Implementation:Langchain ai Langgraph Command Resume

Property	Value
API	`Command(*, graph=None, update=None, resume=None, goto=())`
Type	API Doc
Workflow	Human_in_the_Loop_Agent
Pipeline Stage	Resumption
Repository	Langchain_ai_Langgraph
Source	`libs/langgraph/langgraph/types.py:L367-418`

Overview

Command is a frozen dataclass that encodes instructions to update graph state, resume from an interrupt, and navigate to specific nodes. It is the primary mechanism for providing human input to a paused graph. When passed as the input to stream() or invoke(), it tells the Pregel execution loop how to resume from the last checkpoint.

Command is generic over a type variable N (bound to Hashable) representing node names, and also extends ToolOutputMixin for compatibility with LangChain tool output protocols.

Description

The Command dataclass has four fields:

graph: Specifies which graph to target. None targets the current graph; Command.PARENT (the string "__parent__") targets the closest parent graph. This is used in subgraph scenarios where a child graph needs to send a command to its parent.
update: A state update to apply before resuming. This is equivalent to calling update_state() before the resume. Can be a dict, a list of key-value tuples, or a dataclass/Pydantic model.
resume: The value to provide to the pending interrupt() call. Can be a single value (resolves the next interrupt) or a mapping of interrupt IDs to values (resolves specific interrupts).
goto: Navigation instructions. Can be a node name, a sequence of node names, a Send object, or a sequence of Send objects.

The Command also provides a class variable PARENT set to the string literal "__parent__", used as a constant for targeting parent graphs.

Internally, Command provides _update_as_tuples() to normalize the update field into a sequence of (key, value) tuples for the state update machinery.

Usage

from langgraph.types import Command

# Simple resume with a value
graph.stream(Command(resume="approved"), config)

# Resume with a state update
graph.stream(Command(update={"status": "reviewed"}, resume="yes"), config)

# Resume and navigate to a specific node
graph.stream(Command(resume="done", goto="summary"), config)

# Resume with interrupt ID mapping
graph.stream(Command(resume={"abc123": "value1", "def456": "value2"}), config)

Code Reference

Source Location

Item	Path
`Command` dataclass	`libs/langgraph/langgraph/types.py`, Lines 367-418
`Command.PARENT`	`libs/langgraph/langgraph/types.py`, Line 417
`_update_as_tuples()`	`libs/langgraph/langgraph/types.py`, Lines 402-415

Signature

@dataclass(**{"kw_only": True, "slots": True, "frozen": True})
class Command(Generic[N], ToolOutputMixin):
    graph: str | None = None
    update: Any | None = None
    resume: dict[str, Any] | Any | None = None
    goto: Send | Sequence[Send | N] | N = ()

    PARENT: ClassVar[Literal["__parent__"]] = "__parent__"

Import

from langgraph.types import Command

I/O Contract

Field	Type	Default	Description
`graph`	None	`None`	Target graph. `None` = current graph. `Command.PARENT` (`"__parent__"`) = closest parent graph.
`update`	None	`None`	State update to apply. Supports dict, list of tuples, dataclass, or Pydantic model.
`resume`	Any \| None	`None`	Resume value for pending `interrupt()`. A single value resolves the next interrupt. A dict maps interrupt IDs to values.
`goto`	Sequence[Send \| N] \| N	`()`	Navigation target. Node name(s) or `Send` object(s) to route to next.

Returns: N/A (dataclass, used as input to stream()/invoke())

Properties:

Frozen (immutable after construction)
Slots-based for memory efficiency
Keyword-only construction

Usage Examples

Basic interrupt-resume cycle

import uuid
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.constants import START
from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command

class State(TypedDict):
    foo: str
    human_value: str | None

def node(state: State):
    answer = interrupt("what is your age?")
    return {"human_value": answer}

builder = StateGraph(State)
builder.add_node("node", node)
builder.add_edge(START, "node")

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": uuid.uuid4()}}

# Phase 1: Run until interrupt
for chunk in graph.stream({"foo": "abc"}, config):
    print(chunk)
# {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06'),)}

# Phase 2: Resume with human input
for chunk in graph.stream(Command(resume="some input from a human!!!"), config):
    print(chunk)
# > Received an input from the interrupt: some input from a human!!!
# {'node': {'human_value': 'some input from a human!!!'}}

Resume with state update and navigation

# Modify state and resume, then navigate to a specific node
command = Command(
    update={"status": "human_approved"},
    resume="approved",
    goto="execute_action",
)
for chunk in graph.stream(command, config):
    print(chunk)

Resume with interrupt ID targeting

# When multiple interrupts are pending, target each by ID
state = graph.get_state(config)
interrupt_ids = {intr.id: intr.value for intr in state.interrupts}
# Provide specific resume values for each interrupt
resume_map = {id_: f"response to {val}" for id_, val in interrupt_ids.items()}
for chunk in graph.stream(Command(resume=resume_map), config):
    print(chunk)

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment