Langgraph

LangGraph — stateful agent graphs

Overview

LangGraph (from the LangChain team, but usable without langchain) models an LLM workflow as an explicit state machine: a typed state schema, nodes (functions that return state updates), and edges (fixed or conditional routing). A built-in checkpointer persists state after every step, which is what enables thread-scoped memory, time travel, interrupts, and fault tolerance.

Reach for LangGraph when you need to see and control every transition — branching, retry loops, parallel fan-out, approval gates, multi-agent handoffs. For a simple "LLM + tools in a loop", a plain tool-calling loop or [[building-pydantic-ai-agents]] is less machinery; [[crewai]] fits role-based agent teams; [[dspy]] optimizes the prompts inside your nodes. Trace/eval graphs with [[langfuse]] and [[llm-observability-evals]].

uv add langgraph                 # verified against langgraph 1.2.11, Python >=3.10
uv add langchain                 # only for the prebuilt agent / model bindings

StateGraph fundamentals

from typing import Annotated, Literal
from typing_extensions import TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    text: str                              # no reducer: each update REPLACES
    steps: Annotated[list[str], add]       # reducer: updates are APPENDED

def classify(state: State):
    return {"steps": ["classify"]}         # nodes return partial updates, not full state

def route(state: State) -> Literal["short", "long"]:
    return "short" if len(state["text"]) < 10 else "long"

def short(state: State):
    return {"steps": ["short"], "text": state["text"].upper()}

def long(state: State):
    return {"steps": ["long"], "text": state["text"][:10]}

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("short", short)
builder.add_node("long", long)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route)   # router's return value = next node
builder.add_edge("short", END)
builder.add_edge("long", END)

graph = builder.compile()                  # MUST compile before use
graph.invoke({"text": "hi", "steps": []})
# {'text': 'HI', 'steps': ['classify', 'short']}

• State is a TypedDict (Pydantic models and dataclasses also work). A reducer (Annotated[type, fn]) defines how a node's update merges into the current value; without one the update overwrites. For chat agents use the message reducer: messages: Annotated[list, add_messages] (from langgraph.graph.message import add_messages).
• add_conditional_edges(source, router) routes to whatever node name the router returns; pass a dict as third arg to map return values to node names.
• A node can update state and route in one shot by returning a Command:

from langgraph.types import Command

def a(state: State) -> Command[Literal["b", "c"]]:      # type hint enables graph rendering
    return Command(update={"n": state["n"] + 1}, goto="b" if state["n"] < 5 else "c")

Persistence and thread memory

A checkpointer saves a snapshot after every super-step, keyed by thread_id. Same thread = continued conversation/state; new thread = clean slate.

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "user-42"}}

graph.invoke({"messages": [{"role": "user", "content": "hi, I'm Bob"}]}, config)
graph.invoke({"messages": [{"role": "user", "content": "my name?"}]}, config)  # remembers

snapshot = graph.get_state(config)          # inspect current state
history = list(graph.get_state_history(config))   # time travel

InMemorySaver dies with the process. For production install langgraph-checkpoint-sqlite (SqliteSaver.from_conn_string("state.db")) or langgraph-checkpoint-postgres (PostgresSaver; call .setup() once).

Human-in-the-loop interrupts

interrupt() pauses the graph mid-node, surfaces a payload, and waits indefinitely; resume later (any process, same thread) with Command(resume=...). Requires a checkpointer + thread_id.

from langgraph.types import interrupt, Command

def approval(state: State):
    ok = interrupt({"question": "approve?", "plan": state["plan"]})  # pauses here
    return {"approved": ok}

result = graph.invoke(inputs, config)
result["__interrupt__"]        # [Interrupt(value={'question': 'approve?', ...}, id=...)]

graph.invoke(Command(resume=True), config)   # True becomes interrupt()'s return value

On resume the node re-runs from its top (interrupt is not a coroutine yield) — keep side effects after the interrupt() call, or make code before it idempotent.

Streaming

for chunk in graph.stream(inputs, config, stream_mode="updates"):
    print(chunk)                     # {'node_name': {updated keys}} per step

Pass a list (stream_mode=["updates", "custom"]) to multiplex; chunks then arrive as (mode, data) tuples. astream is the async twin.

Prebuilt ReAct agent

langgraph.prebuilt.create_react_agent is deprecated in 1.x (removal planned for 2.0). The replacement is create_agent from the langchain package — same tools-in-a-loop graph, plus a middleware system:

from langchain.agents import create_agent

agent = create_agent(
    model="anthropic:claude-sonnet-4-5",     # provider string needs langchain-anthropic installed
    tools=[search, calculator],              # plain functions or @tool-decorated
    system_prompt="You are a terse research assistant.",   # was `prompt` in create_react_agent
    checkpointer=checkpointer,               # optional: thread memory for free
)
agent.invoke({"messages": [{"role": "user", "content": "..."}]}, config)

The result is a compiled LangGraph graph — stream, checkpointers, and interrupts all work on it. Drop to a hand-written StateGraph when the agent loop itself needs custom shape.

Gotchas

• Mutating state in a node does nothing — only the returned dict (merged via reducers) counts. Return {"key": new_value}, don't state["key"] = ....
• Missing reducer on list state silently overwrites accumulated history each step; parallel fan-out into the same key without a reducer raises InvalidUpdateError — annotate with add / add_messages.
• KeyError: '__interrupt__'-style confusion: interrupts need a checkpointer AND a thread_id; resuming must reuse the same thread_id.
• Runaway loops no longer fail fast: old LangGraph raised GraphRecursionError at 25 steps; current 1.2.x defaults to 10007 (LANGGRAPH_DEFAULT_RECURSION_LIMIT). Put a real stop condition in your router, or set a low safety rail: graph.invoke(inputs, config={"recursion_limit": 50}).
• Old tutorials say MemorySaver — it's InMemorySaver now (old name still aliased), and create_react_agent snippets should be ported to create_agent.
• Checkpoint growth is unbounded by default in Postgres/SQLite — prune old threads.

Resources

• Docs: https://docs.langchain.com/oss/python/langgraph/overview
• Source: https://github.com/langchain-ai/langgraph
• Reference: https://reference.langchain.com/python/