---
name: langgraph-stateful-agents
description: "Build stateful multi-step agents with LangGraph — StateGraph, checkpointing, streaming, and human-in-the-loop interrupts. Production LangGraph: StateGraph with typed state and reducers (add_messages), conditional edges and Command routing, persistence via checkpointers (MemorySaver, SqliteSaver, Postgres) keyed by thread_id, interrupt() and Command(resume=...) for human-in-the-loop, time-travel with get_state_history and state forking, token/node/values streaming modes, ToolNode and create_react_agent prebuilts, and subgraphs for multi-agent supervisor patterns."
license: MIT
---

# LangGraph Agent Orchestration

Build stateful, multi-step agents with [LangGraph](https://langchain-ai.github.io/langgraph/) instead of hand-rolling control flow around raw LLM calls. This skill gives you working patterns for `StateGraph`, durable checkpointing, token/step streaming, and human-in-the-loop interrupts — the pieces you need when an agent has to pause, resume, branch, or be audited. It's aimed at engineers moving a prototype past a single `while` loop into something that survives restarts and reviews.

## When to use it

- A workflow has real branching: route to a tool, a retriever, or a clarifying question based on model output.
- You need conversations to persist across requests (a chat backend, a long-running research agent) keyed by user/session.
- A step requires human approval — refunds, code execution, sending email — before the graph continues.
- You want to inspect or rewind agent runs to debug why it took a path ("time travel").
- You're building RAG or tool-using agents where retries and partial state matter.

## How it works

You declare state as a `TypedDict` (or Pydantic model) and annotate fields with **reducers** — `Annotated[list, add_messages]` appends rather than overwrites. Nodes are plain functions `(state) -> dict` that return partial updates. Wire them with `graph.add_node`, `graph.add_edge`, and `graph.add_conditional_edges`, or return a `Command(goto=..., update=...)` to route and update in one step.

Compile with a **checkpointer** to get persistence: `MemorySaver` for tests, `SqliteSaver` / `PostgresSaver` (and async `AsyncPostgresSaver`) for production. Every invocation passes a `config={"configurable": {"thread_id": ...}}`; state is snapshotted per super-step. Call `interrupt(payload)` inside a node to pause and surface data to a human, then resume with `Command(resume=value)`. Use `graph.stream(..., stream_mode="updates"|"messages"|"values")` for incremental output, and `graph.get_state` / `graph.get_state_history(config)` to inspect or fork from any past checkpoint.

## Example

```python
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.types import interrupt, Command

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

def approve(state: State):
    ok = interrupt({"action": "send_email", "to": state["messages"][-1]})
    return {"messages": [f"sent={ok}"]}

g = StateGraph(State)
g.add_node("approve", approve)
g.add_edge(START, "approve"); g.add_edge("approve", END)

app = g.compile(checkpointer=SqliteSaver.from_conn_string("ckpt.db"))
cfg = {"configurable": {"thread_id": "user-42"}}
app.invoke({"messages": ["draft"]}, cfg)        # pauses at interrupt
app.invoke(Command(resume=True), cfg)           # resumes same thread
```

## Tips & gotchas

- A node without a reducer **overwrites** that key — forgetting `add_messages` silently drops history.
- `interrupt()` re-runs the whole node from the top on resume; keep side effects after the interrupt, not before.
- Every persisted run needs a `thread_id`; reusing one continues that conversation, a new one starts fresh.
- `SqliteSaver.from_conn_string` is a context manager — hold it open for the app's lifetime, don't recreate per request.

## FAQ

**LangGraph vs. plain LangChain agents?** LangChain's `AgentExecutor` hides the loop; LangGraph makes the graph explicit, so you get checkpointing, interrupts, and per-step streaming you can't easily bolt on afterward.

**How do I edit state during an interrupt?** Pass `update=` to `Command` (or call `graph.update_state(config, values)`) before resuming to override what the human reviewed.

**Does it need a server?** No — it's a library. LangGraph Platform/Server is optional for hosting; the graph runs anywhere Python does.