LangGraph: Stateful Agentic Workflows

Problem Context

Most agent libraries assume your workflow is a tree of tool calls or a free-form conversation. Real production agents are neither — they are cyclic, stateful, conditionally branching, and they must survive process restarts. LangGraph models the agent as an explicit graph of nodes and edges over a shared, persisted state object, which gives you everything a workflow engine gives you (resumability, replay, observability) without leaving Python.

The Mental Model

LangGraph is built on three primitives:

• State — a typed dict (or Pydantic model) annotated with reducers (add_messages, operator.add).
• Nodes — pure functions state -> state_update (or async equivalents).
• Edges — static, conditional (function returns next node name), or fan-out/fan-in.

A StateGraph is compiled once and then executed against a checkpointer (SQLite, Postgres, Redis). Every node execution writes a checkpoint, which is what makes pause/resume and time-travel debugging possible.

flowchart TD
    S([START]) --> P[plan]
    P --> A{router}
    A -- search --> T[tool_call]
    A -- answer --> R[respond]
    T --> P
    R --> H{needs_human?}
    H -- yes --> HUM([interrupt → human])
    H -- no --> E([END])
    HUM --> R
      

Minimal Working Example

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 langchain_openai import ChatOpenAI

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

llm = ChatOpenAI(model="gpt-4o-mini")

def planner(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

def needs_tool(state: State) -> str:
    last = state["messages"][-1]
    return "tool" if last.tool_calls else "end"

def tool_node(state: State):
    # ... execute tool, append ToolMessage ...
    return {"messages": [...]}

g = StateGraph(State)
g.add_node("planner", planner)
g.add_node("tool", tool_node)
g.add_edge(START, "planner")
g.add_conditional_edges("planner", needs_tool, {"tool": "tool", "end": END})
g.add_edge("tool", "planner")

app = g.compile(checkpointer=SqliteSaver.from_conn_string("checkpoints.db"))
app.invoke({"messages": [("user", "What's the weather in Oslo?")]},
           config={"configurable": {"thread_id": "user-42"}})

What Makes LangGraph Different

• Durable checkpoints — every node-completion is a save point. Crash, restart, resume from the exact same state.
• Interrupts — interrupt_before=["tool"] pauses the graph for human approval and serialises state.
• Time travel — replay any thread from any checkpoint, change one input, watch a different branch unfold.
• Subgraphs — compose graphs as nodes; a multi-agent supervisor is a graph whose nodes are subgraphs.

The State Reducer Pattern

The single design choice that trips most teams up is the reducer. messages: listmeans "replace the whole list" on every update. messages: Annotated[list, add_messages]means "append, deduplicating by id". Get this wrong and you either overwrite history or duplicate it. The rule: any field two nodes write to needs an explicit reducer.

Failure Modes

• Implicit cycles — forget a termination edge and the graph loops until your recursion limit (default 25). Always wire an explicit END branch.
• Checkpointer pinned to SQLite in production. Use Postgres + the official langgraph-checkpoint-postgres with connection pooling.
• Huge state objects — checkpoints store the whole state every step. Keep large blobs in object storage and reference by URL.
• thread_id collisions — using user-id only means multiple sessions share state. Use {user_id}:{session_id}.

Production Checklist

• Postgres checkpointer + connection pool (PgBouncer in transaction mode).
• OpenTelemetry instrumentation via LangSmith or the OTLP exporter — every node becomes a span.
• Set recursion_limit conservatively (10–15) and surface the resulting error as a typed agent failure.
• Version your graph: ship graph_v3 alongside graph_v2 until all in-flight threads drain.

Key Takeaways

• LangGraph = workflow engine + LLM ergonomics. Treat it as Temporal-for-agents, not as LangChain v2.
• State reducers and explicit edges are the contract that makes the graph debuggable.
• Checkpoints unlock pause/resume, human-in-the-loop, and replay — features you cannot retrofit later.