LangGraph — Stateful Multi-Agent Workflows with Cycles, Branching, and Human-in-the-Loop

When a Chain Is Not Enough

A linear chain is a fine model for a single question and answer. The moment an agent has to loop — call a tool, read the result, decide whether to call another — a straight line stops describing the control flow. Real agents branch, retry, and sometimes wait for a human. That shape is a graph, not a pipeline.

LangGraph is an open-source library from the LangChain team for building exactly these stateful, cyclic workflows. You describe your agent as a graph of nodes and edges, LangGraph runs it as a state machine, and the hard parts — persistence, resumption, parallelism, pausing for approval — become configuration rather than plumbing you write by hand.

This guide walks through the core primitives: state and reducers, nodes and edges, conditional routing, durable execution with checkpointers, human-in-the-loop with interrupt(), the Send API for map-reduce, and the prebuilt agents that let you skip the boilerplate. It complements our broader piece on building AI agents that actually work.

State Is the Contract

Everything in a LangGraph workflow revolves around a shared state object. You declare its shape as a TypedDict, and every node receives the current state and returns a partial update. The framework merges each update back in — so a node only returns the keys it changed.

How updates merge is controlled by reducers. By default a returned key overwrites the old value. But for a running message history you want to append, not replace. LangGraph ships an add_messages reducer for exactly this, attached to a field with Annotated.

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class State(TypedDict):
    # add_messages appends new messages and merges by ID
    messages: Annotated[list, add_messages]
    # plain field: each write overwrites the previous value
    step_count: int

The add_messages reducer is smart about identity: it appends unseen messages but updates an existing one when the new message shares its ID. LangGraph also provides a ready-made MessagesState class with a single messages field already annotated this way, so most chat agents subclass it instead of redeclaring the field.

Nodes, Edges, and the Graph

A node is just a function that takes the state and returns a partial update. Edges connect nodes into a flow. You register nodes on a StateGraph builder, wire them with add_edge, mark the entry and exit with the sentinel START and END nodes, then call compile() to get an executable graph.

def call_model(state: State):
    # your LLM call goes here; return only the keys you change
    reply = {"role": "assistant", "content": "hello"}
    return {"messages": [reply], "step_count": state["step_count"] + 1}


builder = StateGraph(State)
builder.add_node("model", call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()

result = graph.invoke({"messages": [], "step_count": 0})

The builder is not the runtime. StateGraph is a construction API; only after compile() do you get a CompiledStateGraph that implements the standard Runnable interface — invoke, stream, ainvoke, and astream. That separation is what lets LangGraph validate the graph and attach persistence at compile time.

Conditional Edges: Where the Cycles Live

A fixed add_edge always routes A to B. To branch, you use add_conditional_edges: you give it a source node and a routing function that inspects the state and returns a key. A mapping translates that key into the next node — and because an edge can point back to an earlier node, this is how you build loops.

from langgraph.prebuilt import ToolNode

# ToolNode reads tool_calls from the latest message and runs them
builder.add_node("tools", ToolNode(tools))


def should_continue(state: State) -> str:
    last = state["messages"][-1]
    # if the model asked for a tool, run it; otherwise stop
    if getattr(last, "tool_calls", None):
        return "call_tools"
    return "done"


builder.add_conditional_edges(
    "model",
    should_continue,
    {"call_tools": "tools", "done": END},
)
# after tools run, loop back to the model to interpret the result
builder.add_edge("tools", "model")

That single conditional edge plus the return edge from tools to model is the entire ReAct loop: reason, act, observe, repeat until the model stops requesting tools. ToolNode is a prebuilt that reads the tool_calls on the most recent AI message, executes the matching functions, and wraps each result in a tool message the model can read on the next pass.

Durable Execution With Checkpointers

The feature that separates LangGraph from a hand-rolled loop is persistence. Attach a checkpointer at compile time and LangGraph snapshots the full state after every superstep. Each run is keyed by a thread_id, so you can pause, crash, or come back tomorrow and resume from the last checkpoint — the basis of both memory and fault tolerance.

from langgraph.checkpoint.memory import InMemorySaver

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

config = {"configurable": {"thread_id": "user-42"}}

# First turn — state is saved under thread "user-42"
graph.invoke({"messages": [{"role": "user", "content": "Hi, I'm Bob."}]}, config)

# Later turn, same thread — the graph remembers the conversation
graph.invoke({"messages": [{"role": "user", "content": "What's my name?"}]}, config)

InMemorySaver is perfect for tests and notebooks, but its state lives in RAM and is lost when the process restarts. For anything real you swap in a durable backend: a local SqliteSaver for single-node development, or PostgresSaver (from langgraph.checkpoint.postgres import PostgresSaver) for production, where many workers share one database.

from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://${PGUSER}:${PGPASSWORD}@localhost:5432/langgraph"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # creates the checkpoint tables on first run
    graph = builder.compile(checkpointer=checkpointer)
    graph.invoke(
        {"messages": [{"role": "user", "content": "resume where we left off"}]},
        {"configurable": {"thread_id": "user-42"}},
    )

The same durability discipline shows up across data infrastructure — checkpointing is to an agent what a write-ahead log is to a database. If you run these workflows on a scheduler, the persistence contract pairs naturally with a heavier orchestrator; see our guide to Apache Airflow in production for how to slot long-running graph runs into a broader DAG.

Human-in-the-Loop With interrupt()

Some actions should not happen without a human saying yes — sending an email, spending money, deleting records. LangGraph makes this a first-class primitive. Call interrupt() inside a node and the graph pauses, surfacing a value to the caller. A human reviews it, then you resume by passing a Command(resume=...).

from langgraph.types import interrupt, Command


def approval_node(state: State):
    # execution pauses here; the payload is shown to the reviewer
    decision = interrupt({"question": "Send this email?", "draft": state["draft"]})
    if decision == "approve":
        return {"status": "sent"}
    return {"status": "cancelled"}


# Requires a checkpointer — the pause is a persisted state snapshot
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "email-1"}}

graph.invoke({"draft": "Hi team..."}, config)      # runs, then pauses at interrupt
graph.invoke(Command(resume="approve"), config)    # human approves; node resumes

Two things are mandatory: a checkpointer (the pause is literally a saved snapshot) and the same thread_id for the pause and the resume. There is one important gotcha — when you resume, the node restarts from the top, re-running every line before the interrupt() call, which now returns the resume value instead of pausing.

Parallel Fan-Out With the Send API

Sometimes the number of subtasks is not known until runtime — a planner produces five subtopics on one run and twelve on the next. Hard-coded edges cannot adapt to that. The Send API lets a routing function dispatch a node many times in parallel, each with its own custom state. This is the map step of a map-reduce.

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send


class OverallState(TypedDict):
    subjects: list
    # operator.add lets parallel branches append without conflict
    results: Annotated[list, operator.add]


def fan_out(state: OverallState):
    # return a list of Send objects -> one parallel task per subject
    return [Send("do_work", {"subject": s}) for s in state["subjects"]]


builder = StateGraph(OverallState)
builder.add_node("plan", plan_node)
builder.add_node("do_work", do_work_node)
builder.add_node("reduce", reduce_node)

builder.add_edge(START, "plan")
builder.add_conditional_edges("plan", fan_out)   # map: dynamic fan-out
builder.add_edge("do_work", "reduce")            # fan-in barrier
builder.add_edge("reduce", END)

The routing function returns a list of Send objects rather than a string key; LangGraph runs them concurrently in the next superstep. The reduce step is a normal edge — LangGraph waits for every branch to finish before running it. The one requirement is a reducer (here operator.add) on the collecting field, so parallel writes concatenate instead of clobbering each other.

Skipping the Boilerplate: create_react_agent

The reason-act loop is so common that LangGraph ships it prebuilt. create_react_agent assembles the model node, the ToolNode, the conditional edge, and the loop for you, returning a compiled graph you can invoke immediately. Reach for it when your agent is a standard tool-caller; drop to a hand-built StateGraph when the control flow gets bespoke.

from langgraph.prebuilt import create_react_agent


def check_weather(city: str) -> str:
    """Return the weather for a city."""
    return f"It's sunny in {city}."


agent = create_react_agent(
    model="anthropic:claude-sonnet-4-5",
    tools=[check_weather],
    prompt="You are a concise assistant.",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in Warsaw?"}]}
)

The prebuilt accepts the same checkpointer argument as any compiled graph, so you get memory and human-in-the-loop for free. Under the hood it is the exact model/tools/conditional-edge structure from earlier — reading its source is one of the better ways to internalize the pattern before you write your own. Agents that call tools are also agents that need governance; pairing LangGraph tools with a Model Context Protocol server keeps tool definitions reusable across frameworks.

Streaming and Observability

A compiled graph exposes stream and astream so you can surface progress instead of blocking on a final answer. The stream_mode argument controls granularity: "values" emits the full state after each step, "updates" emits only what each node changed, and "messages" streams LLM tokens as they generate.

for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "Research and summarize X"}]},
    config,
    stream_mode="updates",
):
    # chunk is a dict keyed by the node that just ran
    print(chunk)

For deeper tracing across nodes, model calls, and tool executions, LangGraph integrates with LangChain's tracing tooling, and every step is already a discrete, inspectable state transition. Treating an agent run as a series of observable spans is the same discipline we apply to services in our piece on distributed tracing with Jaeger.

Composing Multiple Agents

A compiled graph is itself a Runnable, which means you can use one graph as a node inside another. That composability is how LangGraph models multi-agent systems: a supervisor graph routes to specialist subgraphs, each an agent with its own tools and prompt, and the supervisor decides who acts next based on the shared state.

• Supervisor. A central router node inspects the state and hands control to one of several worker agents, then loops back to decide the next hop.
• Swarm / hand-off. Agents transfer control directly to one another using Command(goto=...) to jump to a named node instead of routing through a hub.
• Subgraph as node. Wrap a fully compiled agent and add it with add_node; the parent state schema maps onto the child's, keeping each agent independently testable.

The key design decision is how much state agents share. A single shared message list is simple but leaks context; scoped subgraph state is cleaner but requires explicit mapping. Start with the supervisor pattern — it is the easiest to reason about — and only reach for direct hand-offs when the routing genuinely belongs to the workers.

Production Checklist

• Persist deliberately. Use PostgresSaver in production; reserve InMemorySaver for tests where losing state on restart is fine.
• Always set a recursion limit. Treat GraphRecursionError as a routing bug, not a reason to raise the ceiling blindly.
• Gate irreversible actions. Put an interrupt() before anything that spends money or sends messages, and keep pre-interrupt code side-effect-free.
• Use reducers for parallel writes. Any field written by fanned-out Send branches needs an append reducer like operator.add.
• Prefer the prebuilt first. Start with create_react_agent; graduate to a hand-built graph only when control flow demands it.
• Stream for UX and debugging. Ship stream_mode="updates" to the client so users see progress and you see which node ran.
• Keep one thread per conversation. A stable thread_id is what turns checkpoints into memory and makes resume-after-crash work.

LangGraph's bet is that agents are state machines, and that treating them as one — explicit state, reviewable transitions, durable checkpoints, and a place to pause for a human — is what turns a demo loop into something you can run in production. If you have been fighting a tangle of nested callbacks to make an agent loop and retry, modeling it as a graph is worth an afternoon on the prebuilt agent.