LangGraph: Stateful, Graph-Based Agent Orchestration

Plain agent loops (call → call → feed back) work until a process crashes and the conversation — and the money spent — vanishes. LangGraph gives you a different mental model: an explicit graph of nodes and edges, where state is a first-class citizen and is saved to a checkpointer after every super-step. That one change buys you durable execution, human-in-the-loop, and time travel for free.

Prerequisites

• Python 3.10+
• pip install langgraph langchain-openai (or any model provider)

State with reducers

LangGraph state is a TypedDict (or Pydantic model). Fields can carry reducers that say how new values combine with old ones — the key to conversational memory:

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]   # append, don't overwrite
    topic: str

add_messages appends new messages to the history instead of replacing them, so every turn the model sees the full conversation.

Building the graph

A graph is a set of nodes and edges describing how state flows:

from langgraph.graph import StateGraph, START, END

def analyze(state: State) -> dict:
    return {"messages": [{"role": "assistant", "content": f"Researching: {state['topic']}"}]}

def synthesize(state: State) -> dict:
    return {"messages": [{"role": "assistant", "content": "Final summary here."}]}

builder = StateGraph(State)
builder.add_node("analyze", analyze)
builder.add_node("synthesize", synthesize)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "synthesize")
builder.add_edge("synthesize", END)
graph = builder.compile()

The graph is deterministic where you want it (edges, hand-coded nodes) and LLM-driven where you need it (an agentic node choosing a tool). That mix of deterministic and model-driven steps in a single graph is LangGraph’s core strength.

Stateful execution with a checkpointer

The graph above forgets everything when invoke() returns. Attach a checkpointer to persist after every super-step (a single tick where all scheduled nodes run):

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()   # dev only — lost on restart
graph = builder.compile(checkpointer=checkpointer)

# thread_id is the persistence key
config = {"configurable": {"thread_id": "alice"}}
graph.invoke({"topic": "Transformers", "messages": []}, config)
graph.invoke({"topic": "Transformers", "messages": []}, config)
# second call resumes the SAME thread, appending via add_messages

Every super-step boundary now produces a checkpoint — a snapshot of the channel values plus which nodes were queued next. The thread_id selects which conversation history a run belongs to. Durable checkpoints (SQLite, Postgres, Redis instead of the in-memory save) mean a process crash is no longer data loss: a fresh process reattaches to the same thread_id and resumes from the last completed super-step.

Crash recovery is just an invoke

There is no special “resume” API. If execution crashed mid-graph, you call the graph again with the same thread_id. LangGraph loads the latest checkpoint and continues from the node that was next in the queue. The state they came to disk and back is indistinguishable from memory — durability and process-restart-resume are the same operation.

For reproduction after a wrong answer, re-run from an old checkpoint, or patch a historical checkpoint with graph.update_state() and continue down a corrected branch — git-branch-style for your agent runs.

Human-in-the-loop with interrupts

Checkpointing is what makes pausing for a human possible. An interrupt() saves a checkpoint and returns control; when the human’s input arrives, the graph resumes from that exact point. A wait can last minutes or days, well past the lifetime of any process, because the run’s state is durable on disk keyed by thread.

Production notes

• Pick the right checkpointer. MemorySaver loses everything on restart; SqliteSaver needs a persistent volume mount in containers (otherwise a restart means it’s effectively memory again); PostgresSaver for multi-process, concurrent threads.
• Keep checkpoints lean. If a single checkpoint exceeds ~50KB, move the payload to external storage (S3) and store only references in state; every field in state pays in serialization time and LLM context budget.
• One checkpointer for subgraphs. Only compile the parent graph with a checkpointer; subgraph checkpointers create duplicate namespaces and bloated state.

Putting It All Together

A stateful research agent that survives restarts and remembers across turns:

from langgraph.checkpoint.postgres import PostgresSaver

# durable, shareable across processes
checkpointer = PostgresSaver(conn)  # call checkpoint.setup() on first run

def research_agent(state): ...  # LLM tool-calling node
def review(state): ...          # deterministic gate

builder = StateGraph(State)
builder.add_node("research", research_agent)
builder.add_node("review", review)
# conditional edges route research -> review -> END
graph = builder.compile(checkpointer=checkpointer)

result = graph.invoke(
    {"messages": [], "topic": "Mixture of Experts"},
    {"configurable": {"thread_id": "alice"}},
)

Conclusion & Next Steps

LangGraph turns an agent into a graph whose state is crash-safe and inspectable. You get conversational memory through reducers, durability through checkpointers, and human-in-the-loop through interrupts — all from one checkpointer= argument. Next: pick a durable backend, keep your state schema lean, and wire LangSmith tracing so you can replay exactly what an agent did when it misbehaved.

References / Sources

• LangGraph overview. https://langchain-ai.github.io/langgraph/
• LangGraph GitHub. https://github.com/langchain-ai/langgraph
• Checkpointers. https://docs.langchain.com/oss/python/langgraph/checkpointers
• LangGraph state management & lasting production decisions. https://activewizards.com/blog/langgraph-state-management-checkpointing-recovery-and-the-persistence-layer-decision/