LangGraph in Python: Stateful LLM Agents with Checkpointing and Human-in-the-Loop (2026)

Updated: August 12, 2026

LangGraph is a Python library from LangChain Inc. that models an LLM agent as a stateful directed graph, where nodes are functions that read and write typed state, edges control transitions, and every step is checkpointed to a pluggable backend (memory, SQLite, or Postgres). That checkpoint layer is what turns a fragile agent loop into a durable one: you can pause on a human approval, resume after a crash, replay from any prior state, and run the same graph across many workers. Version 1.2.0 shipped on May 11, 2026 and hardened the persistence and streaming APIs that most production teams now rely on.

• LangGraph 1.2.0 (May 2026) is the durable execution engine underneath LangChain's create_agent; use it directly when you need branching, loops, or interrupts.
• Every node transition is written to a checkpointer (MemorySaver, SqliteSaver, PostgresSaver) keyed by thread_id, giving you free pause/resume and time-travel debugging.
• Human-in-the-loop is a first-class primitive via interrupt() and Command(resume=...). No custom queueing layer needed.
• Stateful agents typically cost 1.3–2x the tokens of a linear chain because tool loops re-consume history; caching, message trimming, and structured outputs pay for themselves fast.
• LangGraph is not a replacement for LangChain; it composes with it. Use LangChain integrations (chat models, retrievers, tool wrappers) inside LangGraph nodes.
• Deploy the OSS library on any Python runtime (FastAPI, Ray, Modal). LangGraph Platform is optional and mostly buys you managed persistence and Studio-based debugging.

What is LangGraph used for?

LangGraph is used to build LLM agents that need to loop, branch, wait for human input, or survive a process restart mid-run. If your workflow is a straight line (retrieve context, call the model once, return the answer), you don't need it. But the moment you add a tool-calling loop with a stopping condition, an approval gate before a destructive action, or a long-running research task that has to survive a redeploy, you are re-inventing what LangGraph already ships. I've watched teams write three custom versions of this durable-state layer before switching; the fourth time it stuck.

The framework crossed roughly 33,100 monthly Google searches in early 2026 and has become the execution substrate at Klarna, LinkedIn, Uber, and Replit for agent workflows. That popularity matters practically: the checkpointer schema is stable, the Python and JS runtimes track each other, and the LangChain team merged the runtime under LangChain 1.0 on October 22, 2025, so create_agent in LangChain already runs on LangGraph under the hood. You are not betting on a side project.

Concrete jobs LangGraph does well:

• Tool-calling agents with retries, guardrails, and per-tool timeouts.
• Multi-agent supervisor patterns, meaning a router graph that delegates to specialist subgraphs.
• Human approval workflows for anything that spends money, sends email, or writes to prod.
• Long-running research or ETL agents where a single run may take minutes and must resume after a Kubernetes eviction.
• Deterministic replay for debugging a bad answer three days later without re-paying for the LLM calls.

LangGraph vs LangChain: what's actually different?

LangChain is a toolkit (chat model wrappers, retrievers, output parsers, tool decorators, LCEL for simple pipelines). LangGraph is an execution engine: a stateful directed graph with a checkpointer and an interrupt mechanism. They are not competitors; since the joint 1.0 release the recommended path is both: LangChain gives you the integrations, LangGraph runs the agent loop. The confusion in older blog posts is a hangover from when LangChain shipped its own AgentExecutor, which is now soft-deprecated in favor of create_agent (a thin wrapper on LangGraph).

The mental model I use: LangChain is what to call, LangGraph is how the calling proceeds over time. State in a LangChain agent loop is implicit, hidden inside AgentExecutor's scratchpad and lost when the process dies. State in LangGraph is a typed TypedDict or Pydantic model that you own, and it is written to a checkpointer on every node transition. That single change unlocks everything else: resume, time travel, human-in-the-loop, multi-worker scaling. If you have ever debugged a stuck agent by adding print statements between LangChain calls, you already know why explicit state matters.

If you're just starting a chatbot with retrieval, LangChain's create_agent is fine; you get the LangGraph runtime for free without touching graph APIs. Reach for LangGraph directly when you need conditional edges based on state, subgraphs, or fine-grained control over what gets checkpointed. For LLM-side concerns like cost tracking across providers, our guide on the LiteLLM Python gateway pairs well with either approach.

Building a StateGraph: nodes, edges, and typed state

A LangGraph program has three parts: a state schema, a set of node functions, and edges that wire them together. Here is a minimal research agent that searches, summarizes, and decides whether to search again. That's the shape 80% of production agents end up in.

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

class ResearchState(TypedDict):
    question: str
    findings: Annotated[list[str], add]  # reducer: append across nodes
    iterations: int
    answer: str

llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)

def search_node(state: ResearchState) -> dict:
    # Replace with a real search tool (Tavily, Exa, Brave)
    query = state["question"]
    result = f"Simulated finding for: {query} (iteration {state['iterations']})"
    return {"findings": [result], "iterations": state["iterations"] + 1}

def summarize_node(state: ResearchState) -> dict:
    prompt = [
        SystemMessage("Summarize the findings into a single answer. If insufficient, say NEED_MORE."),
        HumanMessage(f"Question: {state['question']}\nFindings:\n" + "\n".join(state["findings"])),
    ]
    response = llm.invoke(prompt)
    return {"answer": response.content}

def route(state: ResearchState) -> str:
    if "NEED_MORE" in state["answer"] and state["iterations"] < 3:
        return "search"
    return END

graph = StateGraph(ResearchState)
graph.add_node("search", search_node)
graph.add_node("summarize", summarize_node)
graph.add_edge(START, "search")
graph.add_edge("search", "summarize")
graph.add_conditional_edges("summarize", route, {"search": "search", END: END})

checkpointer = SqliteSaver.from_conn_string("checkpoints.sqlite")
app = graph.compile(checkpointer=checkpointer)

A few things to notice. The Annotated[list[str], add] declaration tells LangGraph how to reduce concurrent writes to the same key (in this case, append). Without a reducer, two nodes writing to findings would overwrite each other. Reducers are how you get safe parallel fan-out; they are also how the checkpointer knows how to replay history without corrupting state.

Conditional edges are the second thing production people underestimate. The route function inspects state and returns the next node name (or END). It's plain Python (no DSL, no YAML), so you can unit-test it without a network call. Every LangGraph agent I've shipped has ended up with a router node whose test suite has more branches than the rest of the app combined; that's a good sign, not a bad one.

How does LangGraph checkpointing work?

Every time a node returns, LangGraph writes a snapshot of the full state to the configured checkpointer, keyed by thread_id. Invoking the compiled graph with the same thread_id resumes from the last snapshot; invoking with a new thread_id starts fresh. That's the whole model. The three built-in savers are MemorySaver (dev only, evaporates on process exit), SqliteSaver (fine for single-node deployments up to a few QPS), and PostgresSaver (the production default).

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

DB_URI = "postgresql://user:[email protected]:5432/agents"
pool = ConnectionPool(DB_URI, max_size=20, kwargs={"autocommit": True})

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # creates tables on first run; idempotent

app = graph.compile(checkpointer=checkpointer)

# First call, starts a new thread
config = {"configurable": {"thread_id": "user-42-session-1"}}
result = app.invoke({"question": "What is LangGraph 1.2?", "findings": [], "iterations": 0, "answer": ""}, config)

# Later, from a different process, resumes automatically
resumed = app.invoke(None, config)  # None means "continue from checkpoint"

# Inspect the whole history
for snap in app.get_state_history(config):
    print(snap.metadata["step"], list(snap.values.keys()))

Sizing note from a real deployment: a single agent turn with a ~2 KB message list and 4 tool results writes about 8–12 KB per checkpoint. At 100 concurrent agents each doing 6 node transitions per turn, that's ~7 MB of Postgres writes per turn, which is trivial for a small RDS instance but worth a TTL cleanup job if your threads are long-lived. Ship a nightly DELETE FROM checkpoints WHERE created_at < NOW() - INTERVAL '30 days' or your storage bill will surprise you six weeks in.

Time travel is the underrated feature. app.get_state_history(config) returns every prior snapshot; passing a checkpoint id back into invoke lets you rerun from that point with modified state. When a customer reports "the agent gave me a weird answer at 3pm," you don't guess, you replay from the checkpoint immediately preceding the bad node and step through with LangSmith. Pair this with the patterns in our LLM observability comparison and you have real production debuggability.

Human-in-the-loop: interrupt and Command(resume)

Human-in-the-loop in LangGraph is a two-line change. Inside any node, call interrupt(value), and the graph pauses, persists state, and returns control. Your API layer serves the interrupt value to a human (approve/reject, edit a draft, pick from options). When the human replies, you resume with app.invoke(Command(resume=reply), config). No custom queue, no polling, no state machine you have to write yourself.

from langgraph.types import interrupt, Command

def approve_purchase(state: dict) -> dict:
    decision = interrupt({
        "type": "approval",
        "action": "purchase",
        "amount": state["amount"],
        "vendor": state["vendor"],
    })
    if decision["approved"]:
        return {"status": "approved", "reviewer": decision["reviewer"]}
    return {"status": "rejected", "reason": decision.get("reason", "")}

graph.add_node("approve_purchase", approve_purchase)
# ... edges ...
app = graph.compile(checkpointer=checkpointer, interrupt_before=[])

# First call, hits the interrupt and pauses
config = {"configurable": {"thread_id": "purchase-8821"}}
state = app.invoke({"amount": 4200, "vendor": "AcmeCloud"}, config)
snapshot = app.get_state(config)
if snapshot.tasks and snapshot.tasks[0].interrupts:
    pending = snapshot.tasks[0].interrupts[0].value
    # Serve `pending` to a reviewer, wait for their decision...

# Later, when the reviewer approves
final = app.invoke(
    Command(resume={"approved": True, "reviewer": "[email protected]"}),
    config,
)

Two production notes. First, always set a maximum age on interrupts. A checkpoint that never resumes is a memory leak in your ops rotation. I run a cron that fails any interrupt older than 48 hours by resuming with a synthetic "expired" reply. Second, treat the resume payload as untrusted input; validate it against a Pydantic model before merging into state. It's coming from a web form, not from the LLM. For structuring those payloads consistently, our post on structured LLM outputs covers the schema patterns.

Streaming tokens and intermediate state

LangGraph supports four stream modes: values (full state after each node), updates (just the diff per node), messages (raw LLM tokens), and debug (everything, verbose). In a chat UI you almost always want messages for token-by-token rendering plus updates so you can display "searching…", "summarizing…" status chips as nodes run.

async for chunk in app.astream(
    {"question": "How do checkpointers scale?", "findings": [], "iterations": 0, "answer": ""},
    config={"configurable": {"thread_id": "chat-1"}},
    stream_mode=["updates", "messages"],
):
    mode, payload = chunk
    if mode == "updates":
        for node, delta in payload.items():
            print(f"[node:{node}] wrote keys: {list(delta.keys())}")
    elif mode == "messages":
        token, meta = payload
        print(token.content, end="", flush=True)

Wiring the stream through an SSE endpoint is straightforward. The async iterator yields per chunk, so you push each into an EventSourceResponse. For the FastAPI patterns (heartbeats, Nginx buffering, cancellation on disconnect) I lean on our guide to FastAPI streaming for LLM APIs; the same disconnection handling applies verbatim.

Production checklist: latency, cost, and on-call

The demos look magical. The bills and the pager tell a different story. Here's the checklist I run through before any LangGraph agent goes to production.

Latency budget

Each node transition is one network round trip to the checkpointer plus (usually) one LLM call. A 6-node agent turn against Postgres + GPT-4.1 is realistically 3.5–8 seconds. If you promised sub-second responses, either reduce nodes (collapse deterministic steps into one), use a faster model for router nodes, or move the checkpointer to a local SQLite replica and async-replicate to Postgres. Don't pretend you can hide it behind a spinner forever.

Cost per prediction

Tool-calling loops re-send the growing message history on every iteration. A 5-iteration ReAct loop with 2 KB of tool output per step can easily 4x the token cost of a linear pipeline. Mitigations that actually work: trim the message history to the last N turns before each LLM call, use structured outputs to keep tool results compact, and cache tool responses by input hash (LangGraph won't do this for you, so a functools.lru_cache on the tool function is often enough).

Concurrency and connection pools

PostgresSaver holds a connection per invocation. Under 100 concurrent agents you need at least a 100-connection pool or you will deadlock waiting for a checkpoint write. Use psycopg_pool.AsyncConnectionPool, set max_size generously, and monitor pool_stats(). This is the single most common production incident I've seen with LangGraph. The framework itself is fine, the pgbouncer sizing is not.

Observability

Turn on LangSmith or Langfuse from day one. Every node transition, every tool call, every LLM invocation should have a trace. When an agent goes wrong the checkpoint tells you what state it was in; the trace tells you why. You want both.

Kill switch

Every graph needs a global step limit. Use graph.compile(checkpointer=cp, ...) then app.invoke(input, config={"recursion_limit": 25, ...}). Without it, a broken router will loop until you notice on the invoice. 25 is my default; some pipelines need more, but if yours needs 100 you probably have a router bug.

Deploying LangGraph: FastAPI, Docker, and Platform

You have three roughly-viable deployment shapes in 2026. Which one you pick depends on how much of your infra team's time you want back.

Option 1: FastAPI + Postgres. The default. Wrap app.astream() in an SSE endpoint, put it behind Uvicorn workers, terminate TLS at a load balancer. You own the Postgres. You get to reuse your existing auth, logging, and secrets. This is what I ship on almost every project.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
        await checkpointer.setup()
        app.state.agent = graph.compile(checkpointer=checkpointer)
        yield

app = FastAPI(lifespan=lifespan)

@app.post("/agent/{thread_id}/invoke")
async def invoke(thread_id: str, body: dict):
    config = {"configurable": {"thread_id": thread_id}, "recursion_limit": 25}
    return await app.state.agent.ainvoke(body, config)

Option 2: LangGraph Platform (managed). LangChain Inc.'s hosted product. You push a langgraph.json manifest, they run the workers, checkpointer, and provide LangGraph Studio for debugging. Priced per node execution as of mid-2026. Reasonable if your team has no Postgres ops capacity; less compelling if you already run Django on RDS.

Option 3: Self-hosted LangGraph Server. Same Docker image LangGraph Platform runs, deployed to your Kubernetes cluster. You get Studio and the platform APIs, but you own the DB and observability. Requires a LangSmith license for Studio. Good middle ground for regulated workloads.

LangGraph vs CrewAI vs AutoGen vs plain LangChain

The multi-agent framework landscape is crowded and half of it is marketing. Here's how I actually compare them when advising teams. The dimensions matter more than the row labels; your workload will punish the wrong choice on persistence or debuggability long before ease of demo saves you.

Dimension	LangGraph 1.2	LangChain create_agent	CrewAI 0.100+	AutoGen 0.4+
Execution model	Explicit state graph	Wrapper on LangGraph	Role-based crew abstraction	Event-driven actor model
Persistence	Built-in (SQLite, Postgres)	Inherited from LangGraph	External storage only	Runtime state store (v0.4+)
Human-in-the-loop	First-class (interrupt)	Via LangGraph primitives	Task-level callbacks	UserProxyAgent pattern
Time travel / replay	Yes, via checkpoint history	Yes (same runtime)	No	Partial via event log
Streaming granularity	Tokens + state deltas	Tokens + state deltas	Task-level events	Message-level events
Observability	LangSmith, Langfuse, OTel	Same	CrewAI Cloud, OTel	OTel, Autogen Studio
Learning curve	Medium (graph mental model)	Low	Low (role metaphor)	Medium-high (async actors)
Best fit	Durable production agents	Quick tool-calling agents	Persona-driven task crews	Async multi-agent research

Practical read: if you need durable execution and interrupts, LangGraph wins on capability today. CrewAI is genuinely nicer for "give me a marketing crew that writes then edits then posts" style prototypes but you'll end up bolting on your own state store once real customers show up. AutoGen 0.4's actor model is elegant and I'd choose it for research settings with many concurrent agents. I wouldn't choose it for a customer-facing product where every misfire is a support ticket. For programmatic prompt optimization on top of any of these, our writeup on DSPy 3 for programming LLMs is a useful adjacent read.

A few closing opinions from on-call

The biggest LangGraph mistake I see is over-graphing. Not every workflow needs a graph. If your agent is "call an LLM, if it returned a tool call run the tool, loop until done", that's literally what create_agent gives you in three lines. Reach for a hand-rolled StateGraph when you have real branching, real interrupts, or subgraphs. Otherwise you're paying the graph tax (schemas, reducers, edges) for no benefit.

The second biggest is treating the checkpointer as free. It is not. Every node transition is a database write. If you have a 20-node pipeline running at 50 QPS, you're doing 1,000 writes per second, so plan the Postgres accordingly, or split hot state into a Redis cache and only checkpoint the durable bits. The official LangGraph documentation and the LangGraph GitHub repository are both actively maintained and worth bookmarking for the checkpointer schema and migration notes. For the framework's positioning around the 1.0 release, LangChain's 1.0 announcement post is the canonical source.

The third is skipping the recursion limit. I've watched a $6,000 GPT-4 bill accumulate over a weekend because a router node returned itself under a specific tool-error condition. Set the limit. Alarm on it. Sleep better.

Frequently Asked Questions

Is LangGraph production-ready in 2026?

Yes. LangGraph 1.0 shipped alongside LangChain 1.0 on October 22, 2025, and 1.2.0 (May 2026) hardened the persistence and streaming APIs. It is running in production at Klarna, LinkedIn, Uber, and Replit. The library itself is stable; the operational maturity you need to add is standard Postgres tuning and observability.

Do I need LangChain to use LangGraph?

No. LangGraph is a standalone execution engine. You can call any LLM SDK (OpenAI, Anthropic, LiteLLM, raw HTTP) from inside a node. Most teams do use LangChain's chat model wrappers and tool decorators because they normalize the interfaces, but there is no hard dependency at the graph level.

How does LangGraph handle failures and retries?

Node exceptions bubble up and the graph halts, but the state is already checkpointed at the previous successful node, so you can re-invoke with the same thread_id and resume from there. For automatic retries within a node, wrap the risky call in tenacity or use LangGraph's RetryPolicy when adding the node: graph.add_node("call_api", fn, retry=RetryPolicy(max_attempts=3)).

Can LangGraph agents run in parallel?

Yes, both across threads and within a single graph. Different thread_id values are fully independent, so scale them horizontally behind a load balancer. Within one graph, adding multiple outgoing edges from a node fans out execution in parallel; reducers on your state keys tell LangGraph how to merge the concurrent writes safely.

What's the difference between LangGraph OSS and LangGraph Platform?

LangGraph OSS is the MIT-licensed Python (and JS) library, free forever, self-host anywhere. LangGraph Platform is LangChain Inc.'s hosted product that runs the same runtime plus a managed checkpointer, task queue, and the Studio debugger. You pay for the ops convenience; the graphs you write are identical.

How do I test a LangGraph agent without hitting real LLMs?

Two layers. Unit-test node functions directly by calling them with a fixture state dict; they're just Python functions. For integration tests, use langchain_core.language_models.fake.FakeListLLM to script model responses, and use an in-memory MemorySaver checkpointer so tests are fast and hermetic.