LangGraph Agent Orchestration

Build stateful, multi-step agents with 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

curl -fsSL --create-dirs -o ~/.claude/skills/langgraph-stateful-agents/SKILL.md https://claudeskil.com/skills/langgraph-stateful-agents/SKILL.md

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.