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

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

# 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

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

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...

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...

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)

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
# }

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

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.

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

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
    },
)

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