Introduction

Building AI agents that maintain state across sessions, recover from failures, and execute intricate multi-step reasoning is crucial for real-world applications. Claude, with its superior reasoning and long-context capabilities, paired with LangGraph's graph-based orchestration, delivers exactly that: persistent, resilient agents.

In this guide, we'll walk through creating such agents step-by-step. Expect practical Python code, Claude-specific optimizations, and examples like a research workflow that checkpoints progress and retries on errors.

Why Claude + LangGraph?

LangGraph, from the LangChain ecosystem, models agent logic as graphs with nodes (actions/tools) and edges (transitions). It shines for stateful, cyclical workflows.

Claude excels here due to:

• Superior reasoning: Handles complex chains better than shorter-context models.
• Tool use: Native support via Messages API for structured outputs.
• Long context: Up to 200K tokens in Opus, ideal for memory-heavy agents.

Benefits include:

• Persistence: Checkpoints save state to resume later.
• Error recovery: Built-in retries and human-in-loop.
• Scalability: Deploy as APIs for production workflows.

Prerequisites

• Python 3.10+
• Anthropic API key (from console.anthropic.com)
• Familiarity with async Python and LangChain basics

Step 1: Install Dependencies

pip install langgraph langchain-anthropic python-dotenv

Create a .env file:

ANTHROPIC_API_KEY=your_key_here

Step 2: Set Up Claude Client

import os
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic

load_dotenv()
claude = ChatAnthropic(
    model="claude-3-5-sonnet-20240620",
    api_key=os.getenv("ANTHROPIC_API_KEY"),
    temperature=0.1
)

Sonnet balances speed and reasoning; swap to Opus for deeper tasks.

Step 3: Define Agent State

State tracks messages, memory, and custom fields:

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

class AgentState(TypedDict):
    messages: Annotated[List[dict], add_messages]
    research_summary: str
    next_step: str

This enables persistent memory across runs.

Step 4: Build Core Nodes

Nodes are Claude-powered functions. Example: researcher node.

async def researcher(state: AgentState) -> AgentState:
    msg = state["messages"][-1]["content"]
    prompt = f"""
    Research '{msg}'. Provide a summary and next action.
    Output JSON: {{"summary": "...", "next_step": "search|analyze|finish"}}
    """
    response = await claude.ainvoke(prompt)
    return {
        "messages": [{"role": "assistant", "content": response.content}],
        "research_summary": response.content,  # Parse JSON in prod
        "next_step": "search"  # Simplified
    }

Add tools node:

def tools_node(state: AgentState):
    # Simulate tool calls (e.g., web search)
    return {"messages": [{"role": "tool", "content": "Search results..."}]}

Step 5: Construct the Graph

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

workflow = StateGraph(AgentState)

workflow.add_node("researcher", researcher)
workflow.add_node("tools", tools_node)

workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "tools")
workflow.add_conditional_edges(
    "tools",
    lambda s: s["next_step"],
    {"search": "researcher", "finish": END}
)
workflow.add_edge("researcher", END)  # Simplified

# Compile without persistence yet
graph = workflow.compile()

Step 6: Add Persistent Memory with Checkpointers

Persistence via MemorySaver (in-memory) or Postgres for prod.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = workflow.compile(checkpointer=checkpointer)

# Run with thread_id for session persistence
config = {"configurable": {"thread_id": "agent_1"}}

input_message = {"messages": [{"role": "user", "content": "Research quantum computing"}]}

for chunk in graph.stream(input_message, config, stream_mode="values"):
    print(chunk)

Resume later:

# Same config resumes from checkpoint
resumed_input = {"messages": [{"role": "user", "content": "Continue research"}]}
graph.stream(resumed_input, config)

Claude's context window preserves full history effortlessly.

Step 7: Implement Error Recovery

Wrap nodes in retries:

import asyncio

async def robust_researcher(state: AgentState) -> AgentState:
    for attempt in range(3):
        try:
            return await researcher(state)
        except Exception as e:
            if attempt == 2:
                raise
            await asyncio.sleep(2 ** attempt)
    return state

# Or use LangGraph's built-in retry_policy
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["researcher"],
    retry_policy={"researcher": {"max_attempts": 3}}
)

Human-in-loop: Interrupt on error, inspect graph.get_state(config), then resume.

Real-World Example: Multi-Step Research Agent

Full workflow for topic research: search → summarize → analyze → report.

class ResearchState(TypedDict):
    messages: Annotated[List[dict], add_messages]
    summary: str
    analysis: str
    status: str  # 'searching|summarizing|analyzing|done'

# Nodes
async def search_node(state):
    # Fake Tavily search or real integration
    return {"summary": "Quantum bits enable superposition...", "status": "summarizing"}

async def summarize_node(state):
    prompt = f"Summarize: {state['summary']}"
    resp = await claude.ainvoke(prompt)
    return {"messages": [resp], "status": "analyzing"}

async def analyze_node(state):
    prompt = f"Analyze implications: {state['summary']}"
    resp = await claude.ainvoke(prompt)
    return {"analysis": resp.content, "status": "done"}

# Graph
research_graph = StateGraph(ResearchState)
research_graph.add_node("search", search_node)
research_graph.add_node("summarize", summarize_node)
research_graph.add_node("analyze", analyze_node)

research_graph.set_entry_point("search")
research_graph.add_edge("search", "summarize")
research_graph.add_edge("summarize", "analyze")
research_graph.add_edge("analyze", END)

checkpointer = MemorySaver()
research_agent = research_graph.compile(checkpointer=checkpointer)

# Usage
config = {"configurable": {"thread_id": "research_1"}}
result = await research_agent.ainvoke(
    {"messages": [{"role": "user", "content": "Start quantum research"}]},
    config
)
print(result["analysis"])

This checkpoints after each step—pause mid-research, resume anytime.

Step 8: Deploy for Production Workflows

• API Server: Use FastAPI + LangGraph Server.

from langgraph.deploy.fastapi import create_app
app = create_app(graph)

• Integrations: Hook to n8n/Zapier via webhooks.
• Scaling: PostgresSaver for SQLite/Postgres checkpointers.

Best Practices for Claude + LangGraph

1. Prompt Engineering: Use XML tags for Claude: <thinking>Reason step-by-step</thinking>.
2. State Pruning: Compress old messages with Claude summarization node.
3. Model Selection: Haiku for fast tools, Sonnet/Opus for reasoning.
4. Monitoring: Log checkpoints, track token usage.
5. Security: Validate tool inputs, rate-limit API calls.
6. Testing: Unit test nodes, simulate failures.
7. Cost Optimization: Cache common subgraphs.

Common Pitfalls and Fixes

• State Bloat: Implement a "reflect" node to summarize history.
• Infinite Loops: Add max iterations in edges.
• Tool Errors: Claude's structured outputs prevent parsing fails.

Conclusion

Claude + LangGraph empowers agents that think like teams: persistent, recoverable, and smart. Start with the research example, adapt to HR onboarding, sales pipelines, or code reviews.

Experiment in Colab, deploy to prod. Share your graphs on Claude Directory forums!

(Word count: ~1450)