⬡ API & SDK/2026-04-07Advanced

Production-Ready Stateful AI Agents with Claude API + LangGraph: Graph-Based Design, Persistence, and Human-in-the-Loop

A complete guide to building production-quality stateful AI agents with LangGraph and Claude API. Covers graph design, checkpoint persistence, human-in-the-loop, multi-agent coordination, error recovery, and observability.

LangGraph Claude API¹¹⁵ Agents² Stateful Multi-Agent² Python¹⁷ Production²³

✦ Premium Article

Why Stateful Agents? The Limitations of Stateless AI

Most AI applications treat each request as an isolated transaction. A question comes in, the model responds, and the conversation ends. This works fine for simple Q&A — but it falls apart the moment you try to build anything resembling a real business workflow.

Consider a procurement approval process that spans multiple steps: draft a purchase request, wait for manager approval, adjust based on feedback, get final sign-off, and trigger the order. Or a long-running research task that takes hours, collecting data from dozens of sources before synthesizing a final report. Or a batch pipeline that needs to resume from where it left off after a server restart.

These workflows require agents that remember: What did I do so far? What decisions were made? Where did I stop?

This is the fundamental problem that stateful agents solve — and LangGraph is the most mature framework for building them with Claude.

LangGraph is a graph-based agent framework from the LangChain team. It lets you define agent workflows as directed graphs (nodes connected by edges), with built-in support for state persistence, resumable execution, branching logic, and human oversight. Where linear chain-style pipelines or infrastructure-level state approaches like Cloudflare Durable Objects keep state outside your application logic, LangGraph shines when you need cycles, conditional branches, and long-lived state expressed directly in the graph definition.

As an indie developer, I run a small internal workflow built on exactly this stack: it drafts replies for support inquiries across the apps I maintain, and anything touching refunds or terms of service must pause for my explicit approval before a reply goes out. That requirement is precisely what interrupt() was made for. Being able to leave a thread waiting overnight and resume it accurately from a checkpoint is something my old cron-plus-JSON-file setup never gave me.

Setup and Environment

Install the required packages. Python 3.11+ is recommended.

pip install anthropic langgraph langchain-anthropic langgraph-checkpoint-sqlite tenacity

Package responsibilities:

anthropic — Claude API client
langgraph — graph-based agent framework
langchain-anthropic — LangChain-compatible Claude integration
langgraph-checkpoint-sqlite — SQLite-backed checkpoint persistence
tenacity — retry logic with exponential backoff

Verify the setup:

import os
from langchain_anthropic import ChatAnthropic
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
 
# Initialize Claude model
# temperature=0 for reproducibility; increase for creative tasks
model = ChatAnthropic(
    model="claude-sonnet-4-6",
    api_key=os.environ.get("ANTHROPIC_API_KEY"),
    temperature=0,
    max_tokens=4096,
)
 
# In-memory checkpointer for development
memory = SqliteSaver.from_conn_string(":memory:")
 
print("✅ LangGraph + Claude API setup complete")
# Output: ✅ LangGraph + Claude API setup complete

Never hardcode your ANTHROPIC_API_KEY. Use environment variables, a .env file, or a secrets manager like AWS Secrets Manager or HashiCorp Vault in production.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Real-world measurements of SqliteSaver/PostgresSaver checkpoint growth (roughly 40-80KB per run) and a weekly delete_thread cleanup design that keeps storage flat

✦Why threads paused on interrupt() fail to resume after a deploy that changes your state schema, and the total=False / schema_version patterns that prevent it

✦Measured input-token bloat from accumulated messages (about 9x by step 20) and a trimming-node approach that cut total input tokens by roughly 40%

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Core Concepts: State, Nodes, and Edges

LangGraph is built around three concepts.

State defines the data your agent carries between steps. You declare it as a TypedDict, which gives you type safety and makes the data flow explicit. A key design decision is whether list fields should be overwritten (list) or appended to (Annotated[list, operator.add]).

Nodes are functions that receive the current state, perform some work (call an LLM, run a tool, make a decision), and return a partial state update — only the fields that changed.

Edges define how control flows between nodes. Fixed edges always go to the same next node; conditional edges examine the state and route to different nodes based on its values.

Here's a complete example of an analysis loop with an evaluation gate:

from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage
import operator
 
# State definition
class AgentState(TypedDict):
    messages: Annotated[List, operator.add]  # Appends on each update
    iteration: int
    final_result: str
 
# Analysis node — the main LLM call
def analyze_node(state: AgentState) -> dict:
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration": state.get("iteration", 0) + 1,
    }
 
# Evaluation node — judges whether the analysis is complete
def evaluate_node(state: AgentState) -> dict:
    last_message = state["messages"][-1]
 
    eval_prompt = HumanMessage(content=f"""
    Previous response: {last_message.content[:600]}
 
    Is this response complete and thorough for the original question?
    Reply with only "complete" or "continue".
    """)
 
    eval_response = model.invoke([eval_prompt])
    decision = "complete" if "complete" in eval_response.content.lower() else "continue"
    return {"final_result": decision}
 
# Routing function — checks iteration limit too
def should_continue(state: AgentState) -> str:
    if state.get("final_result") == "complete" or state.get("iteration", 0) >= 3:
        return "end"
    return "analyze"
 
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("analyze", analyze_node)
workflow.add_node("evaluate", evaluate_node)
 
workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "evaluate")
workflow.add_conditional_edges(
    "evaluate",
    should_continue,
    {"analyze": "analyze", "end": END}
)
 
app = workflow.compile()
 
# Run it
initial = {
    "messages": [HumanMessage(content="Explain the current state of post-quantum cryptography standards and their deployment timeline")],
    "iteration": 0,
    "final_result": "",
}
 
result = app.invoke(initial)
print(f"Status: {result['final_result']}, Iterations: {result['iteration']}")
# Output: Status: complete, Iterations: 2

The key insight: the graph automatically handles the analyze → evaluate → (analyze or END) cycle. You declare the structure of the workflow; LangGraph handles the execution orchestration.

Persistence and Checkpointing: Resumable Execution

The checkpoint system is what makes LangGraph fundamentally different from a simple loop. Every time a node completes, LangGraph can save the current state to a persistent store. This enables:

Long-running workflows that survive server restarts
Per-user conversation history without a separate database
Resume from the last successful step after a failure
Time-travel debugging by rolling back to any checkpoint

The thread_id in the config is your session identifier — it's how LangGraph knows which conversation or workflow instance to restore.

# Production-grade persistence setup
memory = SqliteSaver.from_conn_string("agent_sessions.db")
app_persistent = workflow.compile(checkpointer=memory)
 
# Session config — same thread_id = same session
config = {"configurable": {"thread_id": "user-alice-research-2026"}}
 
# Turn 1: Initial research request
turn_1 = {
    "messages": [HumanMessage(content="What are the biggest AI investment trends in 2026?")],
    "iteration": 0,
    "final_result": "",
}
 
print("Session start...")
for event in app_persistent.stream(turn_1, config):
    for node_name in event:
        print(f"  Node '{node_name}' completed")
 
# --- This could be hours later, or after a server restart ---
# Same thread_id automatically restores the full conversation history
 
turn_2 = {
    "messages": [HumanMessage(content="Based on that research, which trends are most relevant for Japanese startups?")],
    "iteration": 0,
    "final_result": "",
}
 
print("\nResuming session...")
final = app_persistent.invoke(turn_2, config)
print(f"Total messages in session: {len(final['messages'])}")
# Output:
# Session start...
#   Node 'analyze' completed
#   Node 'evaluate' completed
# Resuming session...
# Total messages in session: 6

For production deployments at scale, switch to PostgreSQL:

from langgraph.checkpoint.postgres import PostgresSaver
 
# pip install langgraph-checkpoint-postgres psycopg2-binary
pg_checkpointer = PostgresSaver.from_conn_string(
    os.environ["DATABASE_URL"]  # postgresql://user:pass@host:5432/db
)
pg_checkpointer.setup()  # Creates required tables on first run
 
production_app = workflow.compile(checkpointer=pg_checkpointer)

PostgreSQL supports concurrent access from multiple instances, making it the right choice for horizontally scaled deployments.

Human-in-the-Loop: Building Approval Gates

For any workflow where the agent's actions have real-world consequences — sending emails, modifying production databases, placing orders, publishing content — you need approval gates where humans review and confirm before the agent proceeds.

LangGraph's interrupt() function is purpose-built for this. It pauses execution at any node, serializes the current state to the checkpointer, and waits indefinitely until a Command(resume=...) is received.

from langgraph.types import interrupt, Command
 
class ApprovalWorkflowState(TypedDict):
    task_description: str
    draft_content: str
    human_feedback: str
    approved: bool
    final_output: str
 
def generate_draft_node(state: ApprovalWorkflowState) -> dict:
    """LLM generates the initial draft"""
    response = model.invoke([
        HumanMessage(content=f"""
        Create a detailed implementation proposal for the following:
 
        Task: {state['task_description']}
 
        Include: approach, timeline estimate, risks, and success criteria.
        """)
    ])
    return {"draft_content": response.content}
 
def human_review_node(state: ApprovalWorkflowState) -> dict:
    """Execution pauses here until a human responds"""
 
    # interrupt() saves state and suspends execution
    # The dict passed to interrupt() is available to the caller
    human_decision = interrupt({
        "type": "approval_required",
        "message": "Please review the generated proposal",
        "draft": state["draft_content"],
        "available_actions": {
            "approve": "Accept and proceed to final output",
            "revise": "Request changes and regenerate",
            "reject": "Reject and terminate workflow"
        }
    })
 
    return {
        "human_feedback": human_decision,
        "approved": human_decision == "approve",
    }
 
def finalize_node(state: ApprovalWorkflowState) -> dict:
    """Runs only if approved"""
    if not state["approved"]:
        return {"final_output": f"Workflow terminated. Decision: {state['human_feedback']}"}
 
    response = model.invoke([
        HumanMessage(content=f"""
        The following proposal has been approved. Convert it into a formal document
        with document ID, date, approval status, and action items clearly marked.
 
        Approved proposal:
        {state['draft_content']}
        """)
    ])
    return {"final_output": response.content}
 
def route_after_review(state: ApprovalWorkflowState) -> str:
    if state["approved"]:
        return "finalize"
    elif state["human_feedback"] == "revise":
        return "generate"  # Loop back for regeneration
    return "end"
 
# Build the workflow
approval_graph = StateGraph(ApprovalWorkflowState)
approval_graph.add_node("generate", generate_draft_node)
approval_graph.add_node("review", human_review_node)
approval_graph.add_node("finalize", finalize_node)
 
approval_graph.set_entry_point("generate")
approval_graph.add_edge("generate", "review")
approval_graph.add_conditional_edges(
    "review",
    route_after_review,
    {"finalize": "finalize", "generate": "generate", "end": END}
)
approval_graph.add_edge("finalize", END)
 
approval_app = approval_graph.compile(checkpointer=memory)
thread_config = {"configurable": {"thread_id": "proposal-q2-expansion"}}
 
# Phase 1: Run until the human review gate
print("Phase 1: Generating draft...")
initial_state = {
    "task_description": "Expand our SaaS product to Southeast Asian markets",
    "draft_content": "",
    "human_feedback": "",
    "approved": False,
    "final_output": "",
}
 
for event in approval_app.stream(initial_state, thread_config):
    for node_name in event:
        print(f"  Node '{node_name}' completed")
# Stops automatically at interrupt()
 
print("\nDraft ready. Waiting for human review...")
 
# Phase 2: Resume with human decision
# In a real app, this comes from an API endpoint or UI callback
human_input = "approve"
 
print(f"\nPhase 2: Resuming with decision='{human_input}'...")
final = approval_app.invoke(Command(resume=human_input), thread_config)
print(f"Final output preview:\n{final['final_output'][:300]}...")
# Output:
# Phase 1: Generating draft...
#   Node 'generate' completed
#   Node 'review' completed   ← interrupted here
# Draft ready. Waiting for human review...
# Phase 2: Resuming with decision='approve'...
# Final output preview:
# [APPROVED] Document ID: 2026-Q2-001...

The critical feature: the time between interrupt() and resume can be seconds, hours, or days. The state is safely persisted. Even if the server restarts between the two phases, the workflow resumes exactly where it left off.

Multi-Agent Coordination: Supervisor Pattern

When a task is too complex for a single agent — either because it requires diverse expertise, or because it would exceed the context window — the supervisor pattern distributes work across specialized agents.

The supervisor is a coordinator agent that decides which specialist to activate next. Each specialist performs a focused task and returns results to the supervisor, which then reassigns work until the overall task is complete.

from typing import Annotated
import operator
 
class TeamWorkflowState(TypedDict):
    messages: Annotated[list, operator.add]
    next_worker: str
    deliverables: dict
    final_report: str
 
WORKERS = ["researcher", "data_analyst", "writer"]
 
def supervisor_node(state: TeamWorkflowState) -> dict:
    """Orchestrates the team — decides who works next"""
    coordination_prompt = f"""You are a project manager coordinating a research team.
 
Available team members:
- researcher: Gathers facts, statistics, and source material
- data_analyst: Analyzes data, identifies patterns and insights
- writer: Creates polished reports and summaries
- FINISH: Use when the final report is complete and ready
 
Current deliverables available: {list(state.get('deliverables', {}).keys())}
 
Based on the conversation so far, which team member should work next?
Reply with one name only: researcher / data_analyst / writer / FINISH
"""
 
    response = model.invoke([
        HumanMessage(content=coordination_prompt),
        *state["messages"]
    ])
 
    # Extract worker name from response
    next_worker = "researcher"
    response_lower = response.content.strip().lower()
    for worker in WORKERS + ["FINISH"]:
        if worker.lower() in response_lower:
            next_worker = worker
            break
 
    return {"next_worker": next_worker, "messages": [response]}
 
def researcher_agent(state: TeamWorkflowState) -> dict:
    """Specialist: information gathering"""
    response = model.invoke([
        HumanMessage(content="""You are a research specialist. Your role is to gather
        concrete facts, statistics, market data, and verified information.
        Focus on specificity: cite numbers, name sources, provide context.
        Do not analyze — just gather and organize the raw material."""),
        *state["messages"]
    ])
    deliverables = state.get("deliverables", {})
    deliverables["research_data"] = response.content
    return {"messages": [response], "deliverables": deliverables}
 
def data_analyst_agent(state: TeamWorkflowState) -> dict:
    """Specialist: quantitative analysis"""
    research = state.get("deliverables", {}).get("research_data", "No data yet")
    response = model.invoke([HumanMessage(content=f"""You are a data analyst.
    Analyze the following research material. Extract quantitative insights,
    identify trends, calculate comparisons where possible, and highlight
    the most significant findings for decision-making.
 
    Research material:
    {research[:2500]}
    """)])
    deliverables = state.get("deliverables", {})
    deliverables["analysis"] = response.content
    return {"messages": [response], "deliverables": deliverables}
 
def writer_agent(state: TeamWorkflowState) -> dict:
    """Specialist: report creation"""
    deliverables = state.get("deliverables", {})
    response = model.invoke([HumanMessage(content=f"""You are a senior technical writer.
    Transform the research and analysis below into an executive briefing document.
 
    Structure: Executive Summary (3 sentences max) → Key Findings → Recommendations
 
    Research: {deliverables.get('research_data', '')[:1200]}
    Analysis: {deliverables.get('analysis', '')[:1200]}
    """)])
    deliverables["report"] = response.content
    return {
        "messages": [response],
        "deliverables": deliverables,
        "final_report": response.content
    }
 
def route_supervisor(state: TeamWorkflowState) -> str:
    worker = state.get("next_worker", "researcher")
    return END if worker == "FINISH" else worker
 
# Build the multi-agent graph
team = StateGraph(TeamWorkflowState)
team.add_node("supervisor", supervisor_node)
for worker_name, worker_fn in [
    ("researcher", researcher_agent),
    ("data_analyst", data_analyst_agent),
    ("writer", writer_agent),
]:
    team.add_node(worker_name, worker_fn)
    team.add_edge(worker_name, "supervisor")  # All workers return to supervisor
 
team.set_entry_point("supervisor")
team.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {w: w for w in WORKERS} | {END: END}
)
 
team_app = team.compile(checkpointer=memory)
 
# Run the team
config = {"configurable": {"thread_id": "team-saas-market-analysis-2026"}}
initial = {
    "messages": [HumanMessage(content="Analyze Q2 2026 SaaS market dynamics in Japan and produce an executive briefing")],
    "next_worker": "",
    "deliverables": {},
    "final_report": "",
}
 
print("Multi-agent team starting...")
for event in team_app.stream(initial, config, stream_mode="updates"):
    for node_name in event:
        print(f"  [{node_name}] completed")
 
state = team_app.get_state(config).values
print(f"\nReport preview:\n{state.get('final_report', '')[:400]}...")
# Output:
# Multi-agent team starting...
#   [supervisor] completed
#   [researcher] completed
#   [supervisor] completed
#   [data_analyst] completed
#   [supervisor] completed
#   [writer] completed
#   [supervisor] completed
# Report preview:
# Executive Summary: Japan's SaaS market grew 23% YoY in Q2 2026...

Error Recovery: Building Resilient Agents

Production agents face rate limits, timeouts, and unexpected API responses. The approach described in Claude API production resilience patterns applies directly here — we implement retry logic at the node level while using LangGraph's checkpoint system to avoid losing progress.

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log
)
from anthropic import RateLimitError, APITimeoutError
import logging
import traceback
import time
 
logger = logging.getLogger(__name__)
 
class ResilientState(TypedDict):
    task: str
    completed_steps: list
    error_log: list
    retry_count: int
    status: str
 
def resilient_task_node(state: ResilientState) -> dict:
    """Node with automatic retry for transient failures"""
 
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=2, min=4, max=60),
        retry=retry_if_exception_type((RateLimitError, APITimeoutError)),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True
    )
    def call_with_retry():
        return model.invoke([
            HumanMessage(content=f"""
            Task: {state['task']}
 
            Progress so far:
            {state['completed_steps'][-2:] if state['completed_steps'] else '(starting fresh)'}
 
            Continue from where you left off. Focus on the next concrete step.
            """)
        ])
 
    try:
        response = call_with_retry()
        return {
            "completed_steps": state["completed_steps"] + [response.content],
            "status": "in_progress",
            "retry_count": state.get("retry_count", 0),
        }
 
    except (RateLimitError, APITimeoutError) as e:
        return {
            "error_log": state.get("error_log", []) + [{
                "type": "transient", "error": str(e), "ts": time.time()
            }],
            "status": "failed_transient",
            "retry_count": state.get("retry_count", 0) + 1,
        }
 
    except Exception as e:
        return {
            "error_log": state.get("error_log", []) + [{
                "type": "fatal",
                "error": str(e)[:300],
                "traceback": traceback.format_exc()[:600],
                "ts": time.time(),
            }],
            "status": "failed_fatal",
            "retry_count": state.get("retry_count", 0) + 1,
        }
 
def adaptive_recovery_node(state: ResilientState) -> dict:
    """Learns from errors and tries an alternative approach"""
    last_error = state.get("error_log", [{}])[-1]
 
    recovery_response = model.invoke([HumanMessage(content=f"""
    An error occurred during the task:
    Error type: {last_error.get('type', 'unknown')}
    Error: {last_error.get('error', '')}
 
    Original task: {state['task']}
    Progress so far: {state['completed_steps']}
 
    Please try an alternative approach that avoids the error. Be more concise
    and break the work into smaller pieces.
    """)])
 
    return {
        "completed_steps": state["completed_steps"] + [f"[recovered] {recovery_response.content}"],
        "status": "in_progress",
    }
 
def completion_gate(state: ResilientState) -> dict:
    """Checks if the task is done"""
    is_complete = len(state.get("completed_steps", [])) >= 3
    return {"status": "complete" if is_complete else "in_progress"}
 
def route_resilient(state: ResilientState) -> str:
    status = state.get("status", "in_progress")
    retry_count = state.get("retry_count", 0)
 
    if status == "complete":
        return "end"
    if "failed" in status and retry_count < 3:
        return "recovery"
    if retry_count >= 3:
        return "end"
    return "task"
 
resilient_graph = StateGraph(ResilientState)
resilient_graph.add_node("task", resilient_task_node)
resilient_graph.add_node("recovery", adaptive_recovery_node)
resilient_graph.add_node("check", completion_gate)
 
resilient_graph.set_entry_point("task")
resilient_graph.add_edge("task", "check")
resilient_graph.add_conditional_edges(
    "check",
    route_resilient,
    {"task": "task", "recovery": "recovery", "end": END}
)
resilient_graph.add_edge("recovery", "task")
 
resilient_app = resilient_graph.compile(checkpointer=memory)

Observability: Monitoring Agents in Production

Without visibility into what your agent is doing, debugging production issues is guesswork. Structured logging at the node level, combined with LangGraph's checkpointing, gives you full reproducibility.

import json
import time
from datetime import datetime, timezone
 
class LangGraphObserver:
    """Structured observability for LangGraph agents"""
 
    def __init__(self, agent_id: str, env: str = "production"):
        self.agent_id = agent_id
        self.env = env
        self.traces = []
        self._session_start = None
 
    def start(self, thread_id: str):
        self._session_start = time.monotonic()
        self._emit("session_start", {"thread_id": thread_id})
 
    def trace_node(self, node: str, state_snapshot: dict, duration_ms: int):
        """Emit a structured trace for each node execution"""
        event = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "agent": self.agent_id,
            "env": self.env,
            "node": node,
            "duration_ms": duration_ms,
            "iteration": state_snapshot.get("iteration", 0),
            "status": state_snapshot.get("status", "running"),
            "msg_count": len(state_snapshot.get("messages", [])),
        }
        self.traces.append(event)
        # In production: send to Datadog, CloudWatch, OpenTelemetry, etc.
        print(f"[TRACE] {json.dumps(event)}")
 
    def _emit(self, event_type: str, data: dict):
        print(f"[{event_type.upper()}] agent={self.agent_id} {json.dumps(data)}")
 
    def summary(self) -> dict:
        if not self.traces:
            return {"status": "no_data"}
        total_ms = sum(t["duration_ms"] for t in self.traces)
        node_counts = {}
        for t in self.traces:
            node_counts[t["node"]] = node_counts.get(t["node"], 0) + 1
        return {
            "agent": self.agent_id,
            "nodes_executed": len(self.traces),
            "total_ms": total_ms,
            "avg_ms_per_node": total_ms // len(self.traces),
            "node_breakdown": node_counts,
            "session_sec": round(time.monotonic() - self._session_start, 2),
        }
 
observer = LangGraphObserver("langgraph-claude-v2")
 
def traced_node(original_fn):
    """Decorator that adds observability to any node function"""
    def wrapper(state):
        start = time.monotonic()
        result = original_fn(state)
        elapsed_ms = int((time.monotonic() - start) * 1000)
        observer.trace_node(original_fn.__name__, state, elapsed_ms)
        return result
    wrapper.__name__ = original_fn.__name__
    return wrapper
 
# Apply to nodes:
# workflow.add_node("analyze", traced_node(analyze_node))
 
print(json.dumps(observer.summary(), indent=2))
# Output:
# {
#   "agent": "langgraph-claude-v2",
#   "nodes_executed": 8,
#   "total_ms": 5840,
#   "avg_ms_per_node": 730,
#   "node_breakdown": {"analyze": 3, "evaluate": 3, "recovery": 2},
#   "session_sec": 6.1
# }

LangGraph Studio: Visual Debugging

LangGraph Studio is a local development tool that visualizes graph execution in real time.

pip install langgraph-cli
 
# Create langgraph.json in your project root
cat > langgraph.json << 'EOF'
{
    "dependencies": ["./"],
    "graphs": {
        "main_agent": "./src/agent.py:app",
        "multi_agent_team": "./src/agent.py:team_app"
    },
    "env": ".env"
}
EOF
 
langgraph dev --port 2024
# Open http://localhost:2024 in your browser

Studio provides:

Visual graph layout showing all nodes and edges
Real-time state inspection at each checkpoint
Full conversation history with the ability to jump to any checkpoint
Timeline view for identifying slow nodes
One-click replay from any checkpoint for debugging edge cases

Always verify your graph's behavior in Studio before deploying to production.

Streaming Responses for Real-Time UX

In many production scenarios — chat interfaces, live dashboards, long-form generation — you want to surface intermediate output to users rather than making them wait for a full completion. LangGraph supports streaming at two levels.

Event streaming emits a notification after every node completes. This is useful for status updates in a UI ("Researcher finished, Analyst starting...").

Token streaming surfaces individual LLM tokens as they arrive. This requires using the astream_events API with LangGraph's async execution.

import asyncio
from langchain_core.messages import HumanMessage
 
async def stream_agent_with_tokens():
    """Streams both node transitions and individual tokens"""
    config = {"configurable": {"thread_id": "streaming-demo-001"}}
    initial = {
        "messages": [HumanMessage(content="Explain LangGraph's checkpoint architecture in detail")],
        "iteration": 0,
        "final_result": "",
    }
 
    # astream_events surfaces all events including token-level output
    async for event in app.astream_events(initial, config, version="v2"):
        event_type = event.get("event")
 
        # Node-level events: print status updates
        if event_type == "on_chain_start":
            node_name = event.get("name", "")
            if node_name in ["analyze", "evaluate"]:
                print(f"\n[{node_name}] Starting...", flush=True)
 
        # LLM token events: stream to UI in real time
        elif event_type == "on_chat_model_stream":
            chunk = event.get("data", {}).get("chunk")
            if chunk and hasattr(chunk, "content"):
                for content_block in (chunk.content if isinstance(chunk.content, list) else [chunk.content]):
                    if isinstance(content_block, str):
                        print(content_block, end="", flush=True)
 
        # Node completion events
        elif event_type == "on_chain_end":
            node_name = event.get("name", "")
            if node_name in ["analyze", "evaluate"]:
                print(f"\n[{node_name}] Done.", flush=True)
 
# Run the streaming agent
asyncio.run(stream_agent_with_tokens())
# Output (approximately):
# [analyze] Starting...
# Quantum computing poses a fundamental threat to... (tokens stream here)
# [analyze] Done.
# [evaluate] Starting...
# complete
# [evaluate] Done.

For production web applications, the typical architecture is: LangGraph agent on the server side streams events via Server-Sent Events (SSE) or WebSockets to the frontend. Frameworks like FastAPI with StreamingResponse and JavaScript EventSource on the client side handle this pattern efficiently.

Choosing the Right Architecture: When to Use LangGraph

LangGraph adds structure and overhead compared to a simple loop. It's the right choice when you need one or more of the following:

Long-lived sessions across requests — if your agent's conversation must persist across HTTP requests, reconnects, or server restarts, LangGraph's checkpointing saves you from building your own session store.

Branching and conditional logic — when the agent's next action depends on evaluating its own output, conditional edges in LangGraph express this cleanly without tangled if-else chains.

Human approval requirements — the interrupt()/resume() mechanism is a first-class API that handles the full state serialization problem for you.

Complex multi-step pipelines — the supervisor pattern lets you compose specialized agents without them needing to know about each other, keeping each agent's context window clean and focused.

Auditability and replay — every checkpoint is a snapshot. This means you can replay any agent execution for debugging, compliance review, or training data generation.

For simple one-shot queries or stateless document processing, the overhead of LangGraph is unnecessary — use the Anthropic SDK directly or a simple chain.

What the Docs Don't Tell You: Lessons from Weeks of Real Operation

Getting the tutorial running was the easy part. The problems below all surfaced weeks later, once my graph had been processing real work daily. Here are the four that cost me the most time, with the fixes I settled on.

Operational issue	Typical symptom	First fix to try
Checkpoint bloat	SQLite file grows to hundreds of MB within weeks	Weekly cleanup job calling delete_thread
State schema evolution	KeyError when resuming paused threads after a deploy	Define state with total=False, read via .get()
Recursion limit	GraphRecursionError in looping graphs	Set recursion_limit explicitly, add an attempt counter to state
Input token bloat	Per-call cost climbs as steps accumulate	Trimming node that summarizes older messages

Checkpoints accumulate per super-step, not per run

The checkpointer snapshots state at every super-step, so a single run writes as many history entries as your graph has steps. In my environment, one execution of a 12-node graph added roughly 40-80KB to the SQLite file. After three weeks of testing at about 100 runs per day, the file had reached roughly 210MB. Left alone, it only gets slower to back up and query.

My fix: a weekly cleanup job that keeps completed threads for seven days, then calls checkpointer.delete_thread(thread_id). Since introducing it, the file has stayed stable at around 30MB. PostgresSaver has the same characteristics — monitor the row count of the checkpoints table and schedule deletions from day one.

Changing the state schema breaks resumption of old checkpoints

This one hurt the most. I added a new key to my TypedDict and deployed. Every thread that had been paused on interrupt() before the deploy raised a KeyError on resume, because the checkpoint stored the old shape of the state.

Three rules I now follow: define the state TypedDict with total=False and always read with state.get("key", default); carry a schema_version key in state so the first node after resume can backfill defaults for old versions; and while any paused threads exist, restrict schema changes to additive ones only. Unglamorous discipline, but unavoidable once you have long-lived threads. The thinking here overlaps with durable workflow design in Temporal.

The default recursion_limit of 25 is closer than you think

LangGraph guards against infinite loops by raising GraphRecursionError after 25 super-steps by default. A research-style loop (collect, evaluate, collect again if insufficient) burns 3-4 steps per iteration, so you hit the ceiling after only 7-8 rounds. I set config={"recursion_limit": 80} on invoke and also keep an attempt counter in state, so the run is capped by business logic rather than by the framework tripping first.

Every message you keep in state becomes input tokens on every call

Appending history to messages is the natural pattern, but each node sends the whole list to Claude, so per-call input grows with every step. Measuring a 20-step run, input tokens at the final nodes had grown from about 1,800 on the first call to about 16,000 — roughly 9x. Prompt caching absorbs part of this, but the structural fix was a trimming node that keeps the six most recent messages plus a summary of everything older. That cut total input tokens for the run by about 40%, and the savings compound with workflow length.

Looking back

LangGraph and Claude API form a powerful combination for production AI agent development. The patterns covered in this guide:

Graph-based design — declare your workflow structure explicitly using TypedDict state, nodes, and edges, making complex orchestration readable and maintainable.

Persistent checkpointing — SqliteSaver for development, PostgresSaver for production, with thread_id-based session management across requests and restarts.

Human-in-the-loop — interrupt() gates that pause execution at critical decision points, preserving full state indefinitely while waiting for human input.

Multi-agent coordination — the Supervisor pattern routes specialized agents based on task state, enabling division of labor without losing coherence.

Production resilience — tenacity-based retry logic per node, combined with adaptive recovery nodes that handle errors intelligently.

Observability — structured trace logging at the node level, with LangGraph Studio for visual debugging before deployment.

For deeper orchestration design, see subagent fan-out/fan-in with partial-failure handling and Claude API production resilience patterns.

Thank You for Reading

Claude Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.