⬡ API & SDK/2026-04-07Advanced

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

Implementation patterns for running stateful LangGraph + Claude API agents in production: why from_conn_string fails at compile(), how the recursion_limit default changed, and the schema-evolution trap that breaks interrupt() resumption.

LangGraphClaude API¹²⁸AgentsStatefulMulti-AgentPython²² 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
import sqlite3
 
# 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,
)
 
# Checkpointer for development
#
# ⚠️ SqliteSaver.from_conn_string() is a *context manager*.
#    Writing `memory = SqliteSaver.from_conn_string(":memory:")` hands you a
#    _GeneratorContextManager, not a SqliteSaver, and the later call to
#    compile(checkpointer=memory) fails with a TypeError. Details below.
#
# Because the same checkpointer is reused throughout this article, we own the
# connection explicitly and pass it to the constructor.
conn = sqlite3.connect(":memory:", check_same_thread=False)
memory = SqliteSaver(conn)
 
print("✅ LangGraph + Claude API setup complete")
print(type(memory).__name__)
# Output:
# ✅ LangGraph + Claude API setup complete
# SqliteSaver

check_same_thread=False is not optional. LangGraph may execute nodes on worker threads, and a default SQLite connection raises ProgrammingError the moment it crosses one.

For a short-lived script, with SqliteSaver.from_conn_string(":memory:") as memory: is the safest form. The connection closes when the block exits, though, so a long-running web process is easier to reason about with the constructor.

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

✦Why SqliteSaver/PostgresSaver from_conn_string makes compile() raise TypeError, and the two correct shapes to use depending on process lifetime

✦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

✦How the recursion_limit default moving from 25 to 10007 inverts the runaway-loop risk, and a two-cap design that stops runs on business logic

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
conn = sqlite3.connect("agent_sessions.db", check_same_thread=False)
memory = SqliteSaver(conn)
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
 
# ⚠️ The dependency is psycopg 3, not psycopg2. The module opens with
#    `from psycopg import Capabilities, Connection, ...`, so installing
#    psycopg2-binary leaves you with an ImportError.
# pip install "langgraph-checkpoint-postgres" "psycopg[binary]"
 
# from_conn_string is a context manager here too. Assigning the return value
# and calling .setup() on it raises AttributeError.
with PostgresSaver.from_conn_string(
    os.environ["DATABASE_URL"]  # postgresql://user:pass@host:5432/db
) as pg_checkpointer:
    pg_checkpointer.setup()  # Creates required tables; must be called explicitly
    production_app = workflow.compile(checkpointer=pg_checkpointer)
    # Run the graph inside the block — the connection closes on exit

PostgreSQL supports concurrent access from multiple instances, making it the right choice for horizontally scaled deployments. In a long-lived web process, manage a connection pool yourself and construct PostgresSaver(pool) directly rather than wrapping everything in a with. Because setup() runs migrations, treat it as a deploy step you execute once, not something your app calls on every boot.

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 default changed	Runaway loops keep spending tokens instead of stopping	Set recursion_limit to a business-driven value, 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.

Assigning from_conn_string to a variable fails at compile()

This was the first wall I hit. I copied a tutorial fragment, wrote memory = SqliteSaver.from_conn_string("agent.db"), passed it to workflow.compile(checkpointer=memory), and got:

TypeError: Invalid checkpointer provided. Expected an instance of
`BaseCheckpointSaver`, `True`, `False`, or `None`.
Received _GeneratorContextManager.
Pass a proper saver (e.g., InMemorySaver, AsyncPostgresSaver).

from_conn_string is decorated with @contextmanager and typed to return Iterator[SqliteSaver]. Without a with statement you hold a generator wrapper, not a saver. PostgresSaver is implemented the same way, which is why pg_checkpointer.setup() raises AttributeError there.

What makes it slippery is that the assignment itself never complains. The failure surfaces dozens of lines later at compile(), and the message suggests InMemorySaver — which sends you hunting through imports and dependencies instead of looking at the line that actually caused it. I lost half a day here.

Two shapes, one rule: with for short-lived scripts, own the connection and use the constructor for anything long-running.

The default recursion_limit is no longer 25

For a long time the standard explanation was that LangGraph raises GraphRecursionError after 25 super-steps. This article said the same thing. Checking against LangGraph 1.2.11 on my machine, the default is now 10007 (DEFAULT_RECURSION_LIMIT in langgraph/_internal/_config.py, overridable via the LANGGRAPH_DEFAULT_RECURSION_LIMIT environment variable). A loop running 60 iterations under default settings passes straight through.

	Long-standing assumption (0.x era)	Observed behavior (1.2.11)
Default recursion_limit	25	10007
What a research loop does	Aborts early, after 7-8 rounds	Keeps going
What that costs you	Work stops halfway	API spend climbs quietly

The direction of the risk has flipped, and that matters. The old worry was a ceiling low enough to cut you off mid-task. The new one is a ceiling so distant that a bug in your routing condition circles indefinitely without anyone noticing. GraphRecursionError was, in effect, doing safety work for you.

So I stopped relying on the default. I set config={"recursion_limit": 40} — a number chosen from the business logic — on every invoke, and keep an attempt counter in state as a second, independent cap.

class ResearchState(TypedDict, total=False):
    messages: Annotated[list, operator.add]
    attempts: int          # business-level cap, separate from graph recursion
    schema_version: int
 
MAX_ATTEMPTS = 8
 
def should_continue(state: ResearchState) -> str:
    # Cap #1: business logic
    if state.get("attempts", 0) >= MAX_ATTEMPTS:
        return END
    if state.get("final_result"):
        return END
    return "collect"
 
# Cap #2: always state the graph-level limit explicitly
result = app.invoke(
    initial_state,
    config={
        "configurable": {"thread_id": "research-001"},
        "recursion_limit": 40,   # don't inherit the 10007 default
    },
)

Since the default moves between versions, writing it down also documents your intent. In practice, knowing which cap stopped a run matters more in operations than the value of either one.

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

What prompted this rewrite was discovering that the checkpointer code published here didn't run at all. Assigning the return value of from_conn_string to a variable is exactly what you get when you stitch together official snippets, and the failure only shows up much later at compile(). Not a gap in the documentation so much as a gap between its pieces.

If you check one thing in your own code, make it this:

# One line tells you whether your checkpointer is real
from langgraph.checkpoint.base import BaseCheckpointSaver
assert isinstance(memory, BaseCheckpointSaver), type(memory).__name__

If that passes, persistence and interrupt()-based resumption behave as designed. If it doesn't, pick one of the two shapes — wrap it in with, or hand the connection to the constructor.

Then look at recursion_limit. With the default now at 10007, shipping a looping graph on default settings is close to running with the safety catch removed. One line on invoke is all it takes.

The longer an agent runs, the more these unglamorous checks matter compared to clever architecture. I'm still finding my footing with plenty of this, but if it saves even one person the half day I lost, that's worth writing down.

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.

✦Copy-paste ready implementation code
✦New advanced guides published daily
✦$5/mo or $10 for lifetime access

View Membership →