Graph API overview - Docs by LangChain

The Graph API defines agent workflows as a graph of state, nodes, and edges. Send and Command control how the graph runs work: Send dynamically invokes nodes for fan-out, and Command updates state while routing to the next node (including resume after an interrupt). Use the Graph API when you want explicit shared state and a visual control flow. For step-by-step decisions while building, see the Build checklist for the Graph API. For a guided walkthrough, see Thinking in the Graph API. To choose between this and the Functional API, see Choosing between Graph and Functional APIs.

Graphs

LangGraph models agent workflows as graphs built from three parts:

State: A shared snapshot of the application. Usually a typed schema.
Nodes: Functions that read state, do work, and return updates.
Edges: Fixed or conditional transitions that determine the next node.

Compose nodes and edges to build sequences, branches, and loops. Nodes and edges are ordinary functions. They can call an LLM or run other code. In short: nodes do the work; edges decide what runs next.

Super-steps

LangGraph runs graphs with message passing, in the style of Google’s Pregel system. Execution advances in discrete super-steps:

Nodes that run in parallel share one super-step.
Nodes that run in sequence use separate super-steps.
A node becomes active when it receives input on an incoming edge.
After a super-step, nodes with no pending messages become inactive.
The run ends when every node is inactive and no messages remain in transit.

StateGraph

StateGraph is the main graph builder. Parameterize it with your state schema.

Compiling your graph

Build a graph by defining state, adding nodes and edges, then compiling. Compilation checks the graph structure (for example for orphaned nodes). It is also where you attach runtime options such as checkpointers and breakpoints. You must compile the graph before you invoke or stream it. This is the smallest end-to-end example:

from typing_extensions import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    text: str


def node_a(state: State):
    return {"text": state["text"] + "a"}


graph = (
    StateGraph(State)
    .add_node("node_a", node_a)
    .add_edge(START, "node_a")
    .add_edge("node_a", END)
    .compile()
)

graph.invoke({"text": ""})
# {'text': 'a'}

State

State is the shared data that nodes read and update. LangGraph stores that data in channels: named slots in the state schema (the keys of your state object). Each channel holds a value and defines how updates merge into it. Most channels use last-write-wins. Channels with a reducer combine values instead, for example by appending to a list. State has two parts:

A schema that declares the channels in the graph
Optional reducers that control how updates merge into each channel

Nodes return partial updates. LangGraph applies those updates through each channel’s reducer. Define state with a TypedDict, a dataclass (for defaults), or a Pydantic BaseModel when you need validation. Prefer TypedDict or dataclass for performance.

create_agent state schemas must be TypedDict. Pydantic models are not supported.

Schema

By default, the graph input and output schemas match the full state. You can narrow them when some keys are internal only. See Define input and output schemas.

Multiple schemas

Most graphs share one state schema. Use multiple schemas when you need either of the following:

Private channels that internal nodes share, but that are not part of graph input or output
Explicit input and output schemas that are subsets of the full internal state

A node can write to any channel in the graph state, including channels outside its declared input schema. Nodes can also introduce private channels when those schemas are defined. Input and output schemas constrain what invoke accepts and returns. They do not hide channels from streaming by default. See Private channels and streaming. For examples, see Define input and output schemas and Pass private state between nodes.

Reducers

Reducers control how node updates merge into state. Each key has its own reducer. With no reducer, updates overwrite the previous value.

Reducer arguments

A reducer is a binary function:

Left: The current value in state for that key
Right: The update returned by a node for that key

new_value = reducer(left=current_state[key], right=node_update[key])

Define a reducer and attach it to a state key with Annotated:

from typing import Annotated

from typing_extensions import TypedDict

def append_strings(left: list[str], right: list[str]) -> list[str]:
    """Combine the existing state value (left) with a node update (right)."""
    return left + right

class State(TypedDict):
    tags: Annotated[list[str], append_strings]

Calling the reducer with left and right shows how values combine:

append_strings(left=["draft"], right=["review"])  # returns ["draft", "review"]

Default reducer

The default reducer replaces the left value with the right. Nodes return partial updates, not the full state. For example, start with {"foo": 1, "bar": ["hi"]}. An update of {"foo": 2} yields {"foo": 2, "bar": ["hi"]}. A later {"bar": ["bye"]} yields {"foo": 2, "bar": ["bye"]}.

Custom reducers

Use a custom reducer to combine values, such as appending to a list.

import operator
from typing import Annotated

from typing_extensions import TypedDict


class State(TypedDict):
    foo: int
    bar: Annotated[list[str], operator.add]

Here foo overwrites. bar appends with operator.add. From {"foo": 1, "bar": ["hi"]}, an update of {"bar": ["bye"]} becomes {"foo": 1, "bar": ["hi", "bye"]}.

Overwrite

To bypass a merging reducer and replace a value (including clearing it), use Overwrite. See Bypass reducers with Overwrite.

Resetting a reducer field

With a merging reducer, returning an empty value keeps prior data. That matters for error buffers or retry counters that must clear between attempts. See Resetting a reducer field and Bypass reducers with Overwrite.

Working with messages in graph state

Why use messages?

Chat models take a list of messages as input. LangChain’s chat model interface accepts message objects such as HumanMessage and AIMessage. See the Messages conceptual guide.

Using messages in your graph

Store conversation history as a list of messages in state, with a reducer that appends new messages and updates existing ones by ID. Use add_messages on a messages key. Plain operator.add only appends, so it cannot update an existing message in place during human-in-the-loop edits.

from typing import Annotated

from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict


class GraphState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

Serialization

Message reducers also deserialize common wire formats into LangChain message objects.

# Supported
{"messages": [HumanMessage(content="message")]}

# Also supported
{"messages": [{"type": "human", "content": "message"}]}

Access attributes with dot notation, for example state["messages"][-1].content. See LangChain serialization.

MessagesState

MessagesState is a prebuilt state with a single messages key that uses add_messages. Subclass it to add fields:

from langgraph.graph import MessagesState


class State(MessagesState):
    documents: list[str]

Nodes

Nodes are sync or async functions that implement a step of the graph. A node can accept:

state: The graph state
config: A RunnableConfig with values such as thread_id and tags
runtime: A Runtime object with runtime context, store, stream_writer, execution_info, and related helpers

Add nodes with add_node:

from dataclasses import dataclass

from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
from typing_extensions import TypedDict

class State(TypedDict):
    input: str
    results: str

@dataclass
class Context:
    user_id: str

builder = StateGraph(State)

def plain_node(state: State):
    return state

def node_with_runtime(state: State, runtime: Runtime[Context]):
    return {"results": f"Hello, {state['input']}!"}

builder.add_node("plain_node", plain_node)
builder.add_node("node_with_runtime", node_with_runtime)

Behind the scenes, LangGraph wraps functions as RunnableLambda so they gain batching, async support, and tracing. If you omit a node name, LangGraph uses the function name.

builder.add_node(my_node)
# Reference the node as "my_node" in edges

Re-execution and idempotency

With a checkpointer, LangGraph saves checkpoints at super-step boundaries, not mid-function inside a node. After an interrupt or retry, the affected node runs again from the start of its function. Side effects before the pause run again. Design node logic so re-execution is safe. Prefer idempotency keys, upserts, or read-before-write checks. For effects around interrupt(), see Side effects called before interrupt must be idempotent. You can add or remove nodes and edges without breaking resume for existing threads. Resumed runs use saved state and the graph you compile now. If a node calls tasks or interrupt, stricter determinism rules apply on resume. Completed task results are restored from the checkpointer. Changing task or interrupt order before the resume point can mismatch cached values. See Idempotency.

Using tasks in nodes

When one node runs several operations that should checkpoint independently, wrap each operation in a task instead of splitting the work across nodes. With a checkpointer, completed task results are saved, so a resumed thread can skip finished work inside the node. For patterns and examples, see Tasks and Common pitfalls.

`START` node

START is the virtual entry that delivers user input into the graph. Use it to mark which nodes run first.

from langgraph.graph import START

graph.add_edge(START, "node_a")

`END` node

END marks a terminal edge. Use it when a path should stop.

from langgraph.graph import END

graph.add_edge("node_a", END)

Node caching

LangGraph can cache node (and task) results by input. Pass a cache when you compile the graph, and attach a CachePolicy on the node. For options, end-to-end examples, and graph-wide defaults, see Add node caching.

Edges

Edges decide where control goes next:

Normal edges: Always go from one node to the next
Conditional edges: Call a function to determine the next node(s)
Entry point: First node after user input (START → node)
Conditional entry point: Route from START with a function

A node can have multiple outgoing edges. Destinations on those edges run in parallel in the next super-step.

For each node, use either static edges or dynamic routing (conditional edges / Command), not both. Mixing them can run both paths.

Normal edges

If you always want to go from node A to node B, use add_edge:

graph.add_edge("node_a", "node_b")

Conditional edges

If you want to optionally route to one or more nodes (or terminate), use add_conditional_edges. Pass the source node and a routing function to call after that node runs:

graph.add_conditional_edges("node_a", routing_function)
# Or with an explicit map:
graph.add_conditional_edges(
    "node_a",
    routing_function,
    {"path_a": "node_b", "path_b": "node_c"},
)

The routing function receives the current graph state. By default, its return value is the name of the next node (or a list of names). Those nodes run in parallel in the next super-step. You can also pass a dictionary that maps the function’s output to node names. The routing function can return a node name, END, a list of names (fan-out), or Send objects for map-reduce. See Build sequences, branches, and loops.

Use Command instead of conditional edges when you want to combine state updates and routing in a single function.

Entry point

The entry point is the first node (or nodes) that run when the graph starts. Add an edge from the virtual START node to the first node to execute:

graph.add_edge(START, "node_a")

Conditional entry point

A conditional entry point lets you start at different nodes depending on custom logic. Use add_conditional_edges from START:

graph.add_conditional_edges(START, routing_function)

You can optionally provide a dictionary that maps the routing function’s output to the next node name.

`Send`

Use Send when you need to run the same node once per item, each with its own input, and the number of items is only known at runtime. That shape is often called map-reduce or orchestrator-worker: one node produces a list, workers process each item in parallel, then a later node reads the combined results. Normal edges cannot do this. They choose which nodes run next, but every chosen node still receives the shared graph state, and the topology is fixed when you build the graph. Return one or more Send objects from a conditional edge. Each Send names a target node and passes a custom argument as that invocation’s input. LangGraph runs those invocations in parallel. When workers write to shared state keys, reducers merge their updates, for example by concatenating lists. The first argument to Send is the target node name. The second is the state (or partial state) for that invocation.

from langgraph.types import Send

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

graph.add_conditional_edges("node_a", continue_to_jokes)

For a full map-reduce example, see Use Send for map-reduce.

`Command`

Use Command when a single step must both update state and choose what runs next. Normally those jobs are separate: a node returns state updates, and edges (or conditional edges) decide the next node. Return Command from a node when the next hop depends on work you just did in that same function. Command also covers two other cases that plain returns and edges do not: resume a run after an interrupt, and update or route from inside a tool. For routing alone, prefer conditional edges. Reach for Command when you need the update and the route together (or one of the interrupt / tool cases above). Command accepts:

update: State updates (like a normal node return)
goto: Next node(s) (like a conditional edge)
graph: Parent graph target when navigating from a subgraph
resume: Value used to resume after an interrupt

Use Command in three places:

Return from nodes: update, goto, and graph
Input to invoke or stream: resume after an interrupt
Return from tools: Updates and routing from inside a tool

Return from nodes

`update` and `goto`

Return Command to update state and route in one step:

def my_node(state: CommandState) -> Command[Literal["my_other_node"]]:
    return Command(
        update={"foo": "bar"},
        goto="my_other_node",
    )

Annotate the return type with the allowed destinations, for example Command[Literal["my_other_node"]]. That tells LangGraph which edges to render. Use Command when you need both an update and a route. For routing alone, prefer conditional edges.

Command adds dynamic edges. Static edges from the same node still run. Do not mix Command routing with add_edge from one node.

See Combine control flow and state updates with Command.

`graph`

From a subgraph node, navigate to a node in the parent graph with Command.PARENT:

def my_node(state: CommandState) -> Command[Literal["other_subgraph"]]:
    return Command(
        update={"foo": "bar"},
        goto="other_subgraph",  # where `other_subgraph` is a node in the parent graph
        graph=Command.PARENT,
    )

When you update a parent key that the subgraph also uses, define a reducer for that key in the parent. See Navigate to a node in a parent graph.

Input to `invoke` or `stream`

Command(resume=...) is the only Command pattern meant as input to invoke() / stream() (optionally with update=... while resuming). Do not pass Command(update=...) alone to continue a conversation. Any Command as input resumes from the latest checkpoint, not from START, so a finished thread can look stuck. Pass a plain dict to start a new turn:

# Wrong: resumes from the latest checkpoint
graph.invoke(Command(update={
    "messages": [{"role": "user", "content": "follow up"}]
}), config)

# Correct: plain dict restarts from START
graph.invoke({
    "messages": [{"role": "user", "content": "follow up"}]
}, config)

`resume`

Use Command(resume=...) after an interrupt. The value becomes the return value of the paused interrupt() call:

graph.invoke(Command(resume="yes"), config)

See Interrupts for full patterns.

Return from tools

Return Command from tools to update state and route after the tool finishes. goto from a tool adds a dynamic edge. Static edges on the calling node still run, so do not mix both. See Use inside tools.

Graph migrations

You can change a graph’s nodes, edges, and state schema even when a checkpointer already stores thread state. What is safe depends on whether the thread has finished or is paused. Topology (nodes and edges)

Finished threads (not interrupted): you can add, remove, or rename nodes and edges freely.
Interrupted threads: you can change edges and add nodes. Do not rename or remove a node the thread is about to enter. On resume, LangGraph looks up that node by name, and a missing name fails the run.

State schema

Adding or removing keys is backwards and forwards compatible.
Renaming a key drops the saved value for that key in existing threads.
Incompatible type changes on an existing key (for example, narrowing or requiring a field that old checkpoints omit) can break older thread state.

For logic changes that stay schema-compatible (for example, a new tool set), see Business compatibility.

Runtime context

Runtime context passes dependencies into nodes without putting them in graph state (for example, model name or a database handle).

from dataclasses import dataclass

from langgraph.graph import END, START, StateGraph
from langgraph.runtime import Runtime
from typing_extensions import TypedDict


class State(TypedDict):
    input: str
    output: str


@dataclass
class ContextSchema:
    llm_provider: str = "openai"


def get_llm(provider: str) -> str:
    return provider


def node_a(state: State, runtime: Runtime[ContextSchema]):
    llm = get_llm(runtime.context.llm_provider)
    return {"output": llm}


graph = (
    StateGraph(State, context_schema=ContextSchema)
    .add_node("node_a", node_a)
    .add_edge(START, "node_a")
    .add_edge("node_a", END)
    .compile()
)
graph.invoke({"input": "hi"}, context={"llm_provider": "anthropic"})

See Add runtime configuration.

Recursion limit

The recursion limit caps how many super-steps one run may execute. When the limit is exceeded, LangGraph raises GraphRecursionError. Starting in version 1.1.4, the default is 10,007 super-steps. Between 1.0.6 and 1.1.3, the default is 10,000 steps. Override it with the LANGGRAPH_DEFAULT_RECURSION_LIMIT environment variable, or pass recursion_limit as a top-level config key (not under configurable). For recipes (catching the error, reading langgraph_step, and Python RemainingSteps), see Impose a recursion limit and Access and handle the recursion counter.

Learn more

Connect these docs to your agent of choice via MCP for real-time answers.

Edit this page on GitHub or file an issue.

​Graphs

​Super-steps

​StateGraph

​Compiling your graph

​State

​Schema

​Multiple schemas

​Reducers

​Reducer arguments

​Default reducer

​Custom reducers

​Overwrite

​Resetting a reducer field

​Working with messages in graph state

​Why use messages?

​Using messages in your graph

​Serialization

​MessagesState

​Nodes

​Re-execution and idempotency

​Using tasks in nodes

​START node

​END node

​Node caching

​Edges

​Normal edges

​Conditional edges

​Entry point

​Conditional entry point

​Send

​Command

​Return from nodes

​update and goto

​graph

​Input to invoke or stream

​resume

​Return from tools

​Graph migrations

​Runtime context

​Recursion limit

​Learn more

Graphs

Super-steps

StateGraph

Compiling your graph

State

Schema

Multiple schemas

Reducers

Reducer arguments

Default reducer

Custom reducers

Overwrite

Resetting a reducer field

Working with messages in graph state

Why use messages?

Using messages in your graph

Serialization

MessagesState

Nodes

Re-execution and idempotency

Using tasks in nodes

`START` node

`END` node

Node caching

Edges

Normal edges

Conditional edges

Entry point

Conditional entry point

`Send`

`Command`

Return from nodes

`update` and `goto`

`graph`

Input to `invoke` or `stream`

`resume`

Return from tools

Graph migrations

Runtime context

Recursion limit

Learn more