Reliable Agent Design

You've built agents. You've gotten them to loop, call tools, and produce answers. Now comes the hard part: making those agents reliable under real-world conditions — flaky APIs, long-running conversations, unexpected failures, and the need for human oversight.

This document codifies our team's standards for writing LangGraph code that runs in production. Every recommendation is opinionated, actionable, and grounded in the framework's capabilities. Treat this as a living standard — not a suggestion list.

---

What Makes an Agent "Reliable"

A production agent is not just one that gives correct answers on a happy path. It is a system that behaves predictably when things go wrong. Reliability across five dimensions:

Dimension	What It Means	What Happens Without It
Fault tolerance	Nodes fail without crashing the graph	Single tool timeout kills the entire conversation
State durability	Execution survives process crashes	Users lose context and must restart from scratch
Human oversight	Critical actions require approval	Agents take irreversible actions autonomously
Observability	You can see what the agent is doing	Debugging means adding print statements and redeploying
Testability	Behavior is verified, not hoped for	Regression bugs reach production undetected

These five dimensions map directly to LangGraph's core capabilities: error handling in nodes, checkpointers, interrupts, LangSmith tracing, and structured testing.

---

Error Handling Patterns

Agents call external APIs, query databases, and invoke LLMs — all of which fail. Your nodes must handle failure gracefully.

Try/Catch in Nodes

Every node is a function. Wrap the core logic in try/except so that a single failure does not terminate the entire graph:

from langchain.messages import ToolMessage


def execute_tools(state: AgentState) -> dict:
    """Execute tool calls with per-tool error handling."""
    results = []
    last_message = state["messages"][-1]

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_fn = tools_by_name.get(tool_name)

        if tool_fn is None:
            results.append(
                ToolMessage(
                    content=f"Unknown tool '{tool_name}'. Available tools: {list(tools_by_name.keys())}.",
                    tool_call_id=tool_call["id"],
                )
            )
            continue

        try:
            observation = tool_fn.invoke(tool_call["args"])
            results.append(
                ToolMessage(content=str(observation), tool_call_id=tool_call["id"])
            )
        except Exception as exc:
            results.append(
                ToolMessage(
                    content=(
                        f"Tool '{tool_name}' failed ({type(exc).__name__}): {exc}. "
                        "Try a different approach or rephrase your request."
                    ),
                    tool_call_id=tool_call["id"],
                )
            )

    return {"messages": results}

Never let exceptions bubble out of nodes

An unhandled exception in a node terminates graph execution. If you use a checkpointer, the checkpoint is still saved — but the agent stops responding. Always catch exceptions and return a state update, even if that update signals failure.

Retry Strategies

For transient failures (network timeouts, rate limits), retry with exponential backoff. LangGraph provides a @retry_decorator for this:

from langgraph.retry import retry_decorator
import httpx


@retry_decorator(
    max_retries=3,
    initial_backoff=1.0,
    backoff_factor=2.0,  # 1s, 2s, 4s
    jitter=True,
    retry_on=(httpx.TimeoutException, httpx.NetworkError),
)
def call_external_api(query: str) -> str:
    """Fetch data from an external API with automatic retry."""
    with httpx.Client(timeout=10.0) as client:
        response = client.get(f"https://api.example.com/search?q={query}")
        response.raise_for_status()
        return response.json()

When to use retries:

Failure Type	Retry?	Reason
Network timeout	Yes	Transient; server may recover
429 Rate limit	Yes (with backoff)	Provider will accept after delay
5xx server error	Yes	May be temporary
400 bad request	No	Retry won't fix invalid input
401/403 auth error	No	Credentials are wrong
Tool returned empty result	Maybe	Depends on whether re-invocation helps

Fallback Nodes

When a node fails repeatedly, route to an alternative path rather than retrying indefinitely. Use conditional edges to detect failure state:

from typing import Literal
from langgraph.graph import StateGraph, START, END


def route_after_search(
    state: AgentState,
) -> Literal["synthesize", "fallback_summary", END]:
    """Route based on search results quality."""
    search_results = state.get("search_results", [])

    if not search_results:
        # No results — try the fallback path
        return "fallback_summary"
    if len(search_results) < 2:
        # Weak results — still try synthesis but flag it
        return "synthesize"
    return "synthesize"


def fallback_summary_node(state: AgentState) -> dict:
    """Provide a best-effort answer when search fails."""
    return {
        "answer": (
            "I couldn't find specific information for your query. "
            "Here's what I know from general knowledge: ..."
        ),
        "confidence": "low",
    }


# Graph wiring
graph = StateGraph(AgentState)
graph.add_node("search", search_node)
graph.add_node("synthesize", synthesize_node)
graph.add_node("fallback_summary", fallback_summary_node)

graph.add_edge(START, "search")
graph.add_conditional_edges(
    "search",
    route_after_search,
    ["synthesize", "fallback_summary", END],
)
graph.add_edge("synthesize", END)
graph.add_edge("fallback_summary", END)

Graceful Degradation

Design agents that operate at reduced capability rather than failing completely. The key principle: a partial answer is better than no answer.

def research_node(state: AgentState) -> dict:
    """Research with graceful degradation across multiple sources."""
    findings = []

    # Primary source: web search
    try:
        web_results = web_search(state["query"])
        if web_results:
            findings.extend(web_results)
    except Exception:
        logger.warning("Web search unavailable, continuing with other sources")

    # Secondary source: internal knowledge base
    try:
        kb_results = knowledge_base_search(state["query"])
        if kb_results:
            findings.extend(kb_results)
    except Exception:
        logger.warning("Knowledge base unavailable")

    # Tertiary: LLM's own knowledge
    if not findings:
        return {
            "findings": [],
            "source": "llm_knowledge",
            "degraded": True,
        }

    return {
        "findings": findings,
        "source": "external",
        "degraded": False,
    }

---

Human-in-the-Loop

Production agents should not take irreversible actions without human approval. LangGraph provides interrupt() to pause execution at any point.

Interrupt Before Action

The interrupt() function halts graph execution and returns control to the caller. You can then inspect state, modify it, and resume:

from langgraph.types import interrupt


def send_email_node(state: AgentState) -> dict:
    """Pause before sending any email for human review."""
    draft = state["email_draft"]

    # Interrupt triggers a pause — execution stops here
    feedback = interrupt({
        "action": "send_email",
        "recipient": state["email_recipient"],
        "subject": state["email_subject"],
        "body_preview": draft[:200] + "...",
        "request": "Review and approve this email before sending.",
    })

    # After human approves (or modifies), execution resumes
    if feedback.get("approved"):
        sent_email = email_client.send(
            to=state["email_recipient"],
            subject=state["email_subject"],
            body=feedback.get("modified_body", draft),
        )
        return {"email_sent": True, "email_id": sent_email.id}

    return {"email_sent": False, "reason": "Rejected by reviewer"}

Approval Workflows

Use interrupts for any action that is costly, irreversible, or policy-sensitive:

Action	Interrupt?	Why
Sending external communications	Yes	Wrong messages damage reputation
Writing to production databases	Yes	Data corruption is hard to undo
Financial transactions	Yes	Obvious
Internal data analysis	No	Low risk, fast iteration
Drafting responses (not sending)	No	Human reviews the final output

Breaking Execution and Resuming

When an interrupt fires, the graph pauses. The caller receives the interrupt payload and can:

1. Approve — pass back a value that satisfies the interrupt and continues.
2. Modify — update state (e.g., edit the email body) before resuming.
3. Reject — signal that execution should not continue.

import uuid
from langgraph.checkpoint.memory import MemorySaver

# Compile with a checkpointer — required for interrupts
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Invoke with a thread ID
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

# This will pause at the interrupt() call
for event in app.stream({"messages": [HumanMessage(content="Send an email to...")]}, config):
    pass

# At this point, the graph is paused. Inspect state:
current_state = app.get_state(config)
print(f"Paused at node: {current_state.next}")
print(f"State: {current_state.values}")

# Resume with human feedback
for event in app.stream(
    None,  # No new input
    config,
    stream_mode="values",
):
    # The interrupt receives this as the return value
    pass

# Or modify state before resuming
app.update_state(config, {"email_draft": "Modified version of the email..."})

Interrupt Patterns in Practice

Simple Approval

from langgraph.types import interrupt

def approve_action(state: AgentState) -> dict:
    should_proceed = interrupt({
        "message": f"Agent wants to: {state['planned_action']}. Approve?",
        "default": False,
    })
    return {"approved": bool(should_proceed)}

Edit Before Proceed

from langgraph.types import interrupt

def review_draft(state: AgentState) -> dict:
    feedback = interrupt({
        "current_draft": state["draft"],
        "instructions": "Edit the draft if needed, then return the final version.",
    })
    return {"draft": feedback.get("final_draft", state["draft"])}

Multi-Step Review

from langgraph.types import interrupt

def two_phase_review(state: AgentState) -> dict:
    # Phase 1: Review the plan
    plan_approved = interrupt({
        "phase": "plan_review",
        "plan": state["execution_plan"],
    })

    # Phase 2: Review the result
    result_approved = interrupt({
        "phase": "result_review",
        "result": state["execution_result"],
    })

    return {
        "plan_approved": bool(plan_approved),
        "result_approved": bool(result_approved),
    }

---

Checkpointing and Persistence

Checkpointing is the mechanism by which LangGraph saves agent state between steps. It is the foundation for crash recovery, human-in-the-loop, time travel, and debugging.

Why Checkpointing Matters

Without checkpointing, every graph execution is ephemeral. A process crash, a network interruption, or a human review pause means losing all accumulated state. With checkpointing:

• Crash recovery — restart from the last saved step, not from the beginning.
• Human-in-the-loop — interrupts require a checkpointer to pause and resume.
• Time travel — inspect or revert to any previous state in the execution history.
• Debugging — replay execution from any checkpoint to reproduce issues.
• Conversation threading — maintain context across multiple user interactions.

In-Memory vs. Persistent Checkpointers

In-Memory (Development)

from langgraph.checkpoint.memory import MemorySaver

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

# State lives only in the Python process.
# Great for development and testing. Lost on restart.

SQLite (Small-Scale)

from langgraph.checkpoint.sqlite import SqliteSaver

with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    # State persists across process restarts.
    # Good for single-server deployments.

PostgreSQL (Production)

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg import Connection, Pipeline

conn = Connection.connect("postgresql://user:pass@localhost/langgraph")
with PostgresSaver(conn) as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
    # Production-grade persistence with concurrent access.

Redis (High Throughput)

from langgraph.checkpoint.redis import RedisSaver
from redis import Redis

redis = Redis.from_url("redis://localhost:6379")
checkpointer = RedisSaver(redis)
app = graph.compile(checkpointer=checkpointer)
# Low-latency checkpointing for high-throughput deployments.

Checkpointer Selection Guide

Scenario	Checkpointer
Local development, testing	MemorySaver
Single-server, low traffic	SqliteSaver
Multi-server, production	PostgresSaver
High-throughput, low latency	RedisSaver

Thread IDs and Conversation Threading

Every graph execution belongs to a thread, identified by a thread_id. The thread ID is how LangGraph associates checkpoints with a particular conversation or workflow:

import uuid

# Each conversation gets its own thread
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

# First turn
result = app.invoke(
    {"messages": [HumanMessage(content="Hello")]},
    config=config,
)

# Second turn — same thread, state is preserved
result = app.invoke(
    {"messages": [HumanMessage(content="Follow-up question")]},
    config=config,
)

# The checkpointer links both invocations to the same thread

Thread IDs are typically:

• User session IDs for chat applications.
• Task IDs for batch processing workflows.
• UUIDs generated at the start of each interaction.

Inspecting Checkpoint History

You can traverse the full execution history of any thread:

# Get the current state
current_state = app.get_state(config)
print(f"Current node: {current_state.next}")
print(f"State values: {current_state.values}")

# Get all checkpoints in the thread
for checkpoint in app.get_state_history(config):
    print(f"Checkpoint ID: {checkpoint.id}")
    print(f"  Next nodes: {checkpoint.next}")
    print(f"  Metadata: {checkpoint.metadata}")

# Get state at a specific checkpoint
specific_state = app.get_state(
    {"configurable": {"thread_id": thread_id, "checkpoint_id": "some-checkpoint-id"}}
)

Next: PostgreSQL Checkpointing

For detailed guidance on configuring PostgreSQL as your production checkpointer — including connection pooling, schema setup, and migration strategies — see the PostgreSQL Checkpointing & Memory Setup guide.

---

Debugging LangGraph Agents

When an agent behaves unexpectedly, you need tools to understand what happened. LangGraph integrates with LangSmith and provides built-in observability.

Using LangSmith for Debugging

Enable LangSmith tracing at the environment level. Every graph invocation, node execution, and tool call is automatically traced:

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="lsv2_..."
export LANGSMITH_PROJECT="my-agent-project"

# No additional code needed — LangSmith traces automatically
# when LANGSMITH_TRACING=true is set

# Optional: add metadata for filtering
config = {
    "configurable": {"thread_id": thread_id},
    "metadata": {
        "user_id": "user-123",
        "agent_version": "2.1.0",
        "environment": "production",
    },
}

In the LangSmith UI, you can:

• See the full execution trace with timing for each node.
• Inspect input and output state at every step.
• Filter traces by metadata (user, version, environment).
• Compare traces side-by-side to identify regressions.

Visualizing Graph Execution

LangGraph Studio provides a visual graph editor and execution debugger:

# Install LangGraph CLI
uv add langgraph-cli

# Start the studio server
langgraph dev

The studio shows:

• Graph structure — nodes, edges, and conditional routing.
• Execution path — which nodes ran and in what order.
• State at each step — inspect the full state after every node.
• Interactive replay — step through execution manually.

Inspecting State at Each Step

Use astream with stream_mode="values" to observe state changes in real time:

async for event in app.astream(
    {"messages": [HumanMessage(content="Research quantum computing")]},
    config=config,
    stream_mode="values",
):
    print(f"--- After step ---")
    print(f"Step count: {event.get('step_count', 'N/A')}")
    print(f"Messages so far: {len(event.get('messages', []))}")
    print(f"Search results: {len(event.get('search_results', []))}")

For more granular debugging, use stream_mode="updates" to see each node's contribution:

async for event in app.astream(
    {"messages": [HumanMessage(content="Research quantum computing")]},
    config=config,
    stream_mode="updates",
):
    # event is a dict: {"node_name": {"key": "value"}}
    for node_name, updates in event.items():
        print(f"[{node_name}] Updated: {list(updates.keys())}")

Logging Strategies Within Nodes

Add structured logging to nodes for production debugging:

import logging
import time

logger = logging.getLogger("agent")


def timed_node(node_func):
    """Decorator that logs execution time and status for any node."""
    def wrapper(state: AgentState) -> dict:
        start = time.perf_counter()
        node_name = node_func.__name__
        logger.info("Node %s starting", node_name)
        try:
            result = node_func(state)
            elapsed = time.perf_counter() - start
            logger.info("Node %s completed in %.2fs", node_name, elapsed)
            return result
        except Exception as exc:
            elapsed = time.perf_counter() - start
            logger.error(
                "Node %s failed after %.2fs: %s",
                node_name,
                elapsed,
                exc,
                exc_info=True,
            )
            raise
    return wrapper


@timed_node
def call_llm(state: AgentState) -> dict:
    """LLM call with automatic timing and error logging."""
    response = model_with_tools.invoke([system] + state["messages"])
    return {"messages": [response], "step_count": state.get("step_count", 0) + 1}

Getting Stack Traces from Failed Nodes

When a node fails, LangGraph does not swallow the exception. The full traceback is available:

try:
    result = app.invoke(
        {"messages": [HumanMessage(content="trigger error")]},
        config=config,
    )
except Exception as exc:
    # Full traceback is available
    import traceback
    traceback.print_exc()

    # If using LangSmith, the error is also captured in the trace
    # with full state context at the point of failure

For nodes that catch their own exceptions, log the traceback internally:

def resilient_node(state: AgentState) -> dict:
    try:
        return do_risky_thing(state)
    except Exception:
        logger.exception("Resilient node failed — details logged above")
        return {"error": "Node encountered an error", "recovered": True}

---

Graph Design Patterns

Different problems call for different graph architectures. Here are the patterns we use most.

Supervisor Pattern

A supervisor node orchestrates multiple specialist sub-agents. The supervisor decides which specialist to call, routes results, and determines when the task is complete.

from typing import Literal
from langgraph.graph import StateGraph, START, END


def supervisor_node(state: AgentState) -> dict:
    """Decide which specialist to call next."""
    last_response = state["messages"][-1]
    # The LLM acts as the supervisor, choosing the next specialist
    decision = supervisor_model.invoke([
        SystemMessage(content=supervisor_system_prompt),
        *state["messages"],
    ])
    return {
        "messages": [decision],
        "next_specialist": parse_specialist(decision),
    }


def route_to_specialist(state: AgentState) -> Literal["researcher", "coder", "writer", END]:
    """Route based on supervisor's decision."""
    specialist = state.get("next_specialist")
    if specialist == "done":
        return END
    return specialist or END


# Build the graph
graph = StateGraph(AgentState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("researcher", researcher_node)
graph.add_node("coder", coder_node)
graph.add_node("writer", writer_node)

graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
    "supervisor",
    route_to_specialist,
    ["researcher", "coder", "writer", END],
)
# Each specialist routes back to the supervisor
graph.add_edge("researcher", "supervisor")
graph.add_edge("coder", "supervisor")
graph.add_edge("writer", "supervisor")

When to use: Multi-step tasks requiring different expertise at different stages (research then code then write).

Map-Reduce Pattern

Process multiple inputs in parallel, then combine results. LangGraph's Send API enables fan-out execution:

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import Send


def map_node(state: ResearchState) -> dict:
    """Fan out: process each sub-query independently."""
    sub_queries = state["sub_queries"]
    # Send creates parallel execution branches
    return [
        Send("process_query", {"query": q, "query_index": i})
        for i, q in enumerate(sub_queries)
    ]


def process_query(state: QueryState) -> dict:
    """Process a single query (runs in parallel for each sub-query)."""
    results = search_tool.invoke(state["query"])
    return {
        "query_results": [
            {"query": state["query"], "results": results, "index": state["query_index"]}
        ],
    }


def reduce_node(state: ResearchState) -> dict:
    """Combine all parallel results into a unified answer."""
    all_results = state.get("query_results", [])
    combined = synthesize_model.invoke([
        SystemMessage(content="Synthesize these search results into a comprehensive answer."),
        HumanMessage(content="\n\n".join(str(r) for r in all_results)),
    ])
    return {"final_answer": combined.content}


graph = StateGraph(ResearchState)
graph.add_node("split_queries", split_queries_node)
graph.add_node("map", map_node)
graph.add_node("process_query", process_query)
graph.add_node("reduce", reduce_node)

graph.add_edge(START, "split_queries")
graph.add_edge("split_queries", "map")
graph.add_edge("reduce", END)

When to use: Processing multiple independent inputs (documents, queries, data sources) and combining results.

Pipeline Pattern

Sequential processing with early exit. Each stage validates its output before passing to the next:

from typing import Literal
from langgraph.graph import StateGraph, START, END


def validate_input(state: PipelineState) -> Literal["extract", "error"]:
    """Gate: reject bad input early."""
    if not state.get("input_text") or len(state["input_text"]) < 10:
        return "error"
    return "extract"


def validate_extraction(
    state: PipelineState,
) -> Literal["analyze", "error"]:
    """Gate: check extraction quality before expensive analysis."""
    entities = state.get("entities", [])
    if not entities:
        return "error"
    return "analyze"


graph = StateGraph(PipelineState)
graph.add_node("extract", extract_entities_node)
graph.add_node("analyze", analyze_sentiment_node)
graph.add_node("summarize", summarize_node)
graph.add_node("error", error_handler_node)

graph.add_edge(START, "extract")
graph.add_conditional_edges("extract", validate_extraction, ["analyze", "error"])
graph.add_edge("analyze", "summarize")
graph.add_edge("summarize", END)
graph.add_edge("error", END)

When to use: Multi-stage processing where early validation saves cost (extract → validate → analyze → summarize).

Reflection Pattern

The agent generates output, critiques it, and iteratively improves:

from typing import Literal
from langgraph.graph import StateGraph, START, END


def generate(state: ReflectionState) -> dict:
    """Generate or regenerate the answer."""
    prompt = state["prompt"]
    if state.get("critique"):
        prompt += f"\n\nPrevious critique: {state['critique']}\nImprove your answer:"
    response = model.invoke([SystemMessage(content=system), HumanMessage(content=prompt)])
    return {"answer": response.content, "iteration": state.get("iteration", 0) + 1}


def critique(state: ReflectionState) -> dict:
    """Evaluate the generated answer."""
    critique_response = critic_model.invoke([
        SystemMessage(content=critic_system_prompt),
        HumanMessage(content=f"Answer to critique:\n{state['answer']}"),
    ])
    return {"critique": critique_response.content}


def should_improve(state: ReflectionState) -> Literal["generate", END]:
    """Stop if the critique is satisfied or max iterations reached."""
    if state.get("iteration", 0) >= MAX_ITERATIONS:
        return END
    if "approve" in state.get("critique", "").lower():
        return END
    return "generate"


graph = StateGraph(ReflectionState)
graph.add_node("generate", generate)
graph.add_node("critique", critique)

graph.add_edge(START, "generate")
graph.add_edge("generate", "critique")
graph.add_conditional_edges("critique", should_improve, ["generate", END])

When to use: Tasks where quality matters more than speed (writing, code generation, analysis).

Pattern Selection Guide

Pattern	Best For	Complexity	LLM Calls
Supervisor	Multi-expert orchestration	Medium	High (supervisor + specialists)
Map-Reduce	Parallel processing	Medium	High (many parallel + one reduce)
Pipeline	Sequential stages with validation	Low	Moderate
Reflection	Quality-critical generation	Low-Medium	High (iterate until satisfied)

---

Performance Optimization

Agents are expensive. Every LLM call costs money and adds latency. Optimize aggressively.

Parallel Node Execution

LangGraph executes nodes in parallel when they are independent. Use the Send API to fan out work:

from langgraph.graph.message import Send


def parallel_research(state: AgentState) -> dict:
    """Fan out multiple research queries in parallel."""
    queries = [
        "LangGraph architecture overview",
        "LangGraph checkpointing mechanisms",
        "LangGraph streaming patterns",
    ]
    # Each Send creates an independent execution branch
    return [
        Send("research_node", {"query": q})
        for q in queries
    ]

Streaming Responses

Stream output to reduce perceived latency. LangGraph supports multiple streaming modes:

Stream State Values

# Stream updated state after each step
for event in app.stream(input_data, config=config, stream_mode="values"):
    last_message = event.get("messages", [])[-1]
    if hasattr(last_message, "content"):
        print(last_message.content, end="\n\n")

Stream Messages

# Stream individual messages as they are generated
for event in app.stream(input_data, config=config, stream_mode="messages"):
    message, metadata = event
    if message.type == "ai":
        print(f"[AI] {message.content}")
    elif message.type == "tool":
        print(f"[Tool] {message.content[:100]}...")

Stream Events

# Granular event stream — see every internal event
for event in app.stream(input_data, config=config, stream_mode="events"):
    event_type = event.get("event")
    if event_type == "on_chat_model_stream":
        # Token-level streaming from the LLM
        token = event["data"]["chunk"].content
        print(token, end="", flush=True)
    elif event_type == "on_chain_start":
        print(f"\n[{event['name']}] starting...")
    elif event_type == "on_chain_end":
        print(f" [{event['name']}] done")

Caching LLM Calls

Cache repeated LLM calls to eliminate redundant API charges:

from langchain.globals import set_llm_cache
from langchain_community.cache import SQLiteCache

# Set up persistent cache
set_llm_cache(SQLiteCache(database_path=".llm_cache.db"))

# First call: hits the API
result1 = model.invoke(messages)

# Second identical call: returns cached result instantly
result2 = model.invoke(messages)  # No API call

Caching Strategy

Cache aggressively for deterministic prompts (system prompts, fixed instructions). Avoid caching for prompts that include dynamic user input — the cache key includes the full prompt, so user-specific prompts rarely hit.

Reducing Unnecessary State Updates

State is checkpointed at every step. Minimize what you store:

# Bad: storing large intermediate data in state
def bad_node(state: AgentState) -> dict:
    raw_data = fetch_huge_dataset()  # 10MB of data
    processed = extract_insights(raw_data)
    return {
        "raw_data": raw_data,       # Don't checkpoint this
        "insights": processed,
    }

# Good: only store what you need
def good_node(state: AgentState) -> dict:
    raw_data = fetch_huge_dataset()
    processed = extract_insights(raw_data)
    return {
        "insights": processed,
        "data_summary": f"Processed {len(raw_data)} records",  # Lightweight
    }

State optimization checklist:

• Remove large data from state after processing.
• Store references (IDs, paths) instead of full objects.
• Use state fields only for data shared across multiple nodes.
• Keep per-node intermediate data local to the node function.

---

Testing LangGraph Agents

Untested agents are production liabilities. Test at three levels: nodes in isolation, graph execution with mocks, and full integration with fake models.

Unit Testing Individual Nodes

Test each node function independently. Nodes are pure functions that take state and return updates:

import pytest
from langchain.messages import HumanMessage, AIMessage, ToolMessage


def test_execute_tools_handles_unknown_tool():
    """Tool node returns error message for unknown tool names."""
    state = {
        "messages": [
            AIMessage(
                content="",
                tool_calls=[{"name": "nonexistent_tool", "args": {}, "id": "call_1"}],
            )
        ]
    }
    result = execute_tools(state)
    assert len(result["messages"]) == 1
    assert "Unknown tool" in result["messages"][0].content


def test_execute_tools_handles_tool_error():
    """Tool node catches exceptions and returns error ToolMessage."""
    state = {
        "messages": [
            AIMessage(
                content="",
                tool_calls=[{"name": "failing_tool", "args": {}, "id": "call_2"}],
            )
        ]
    }
    result = execute_tools(state)
    assert len(result["messages"]) == 1
    assert "failed" in result["messages"][0].content.lower()


def test_route_after_llm_routes_to_tools():
    """Routing function sends to tool node when LLM requests tools."""
    state = {
        "messages": [
            AIMessage(content="", tool_calls=[{"name": "search", "args": {}}]),
        ]
    }
    assert route_after_llm(state) == "execute_tools"


def test_route_after_llm_routes_to_end():
    """Routing function ends when LLM has no tool calls."""
    state = {
        "messages": [AIMessage(content="The answer is 42.")]
    }
    assert route_after_llm(state) == END

Integration Testing Full Graph Execution

Test the complete graph with mocked LLM calls:

from unittest.mock import MagicMock, patch
from langchain.messages import AIMessage, HumanMessage
from langchain_core.language_models import BaseChatModel


@pytest.fixture
def mock_llm():
    """Mock LLM that returns predetermined responses."""
    mock = MagicMock(spec=BaseChatModel)
    mock.invoke.side_effect = [
        # First call: request a tool
        AIMessage(
            content="",
            tool_calls=[{"name": "web_search", "args": {"query": "test"}, "id": "call_1"}],
        ),
        # Second call: final answer
        AIMessage(content="Based on the search results, here is the answer."),
    ]
    return mock


def test_agent_full_execution(mock_llm):
    """Agent completes a full research cycle."""
    with patch("my_agent.model_with_tools", mock_llm):
        app = build_agent_graph()
        result = app.invoke({
            "messages": [HumanMessage(content="What is LangGraph?")],
            "step_count": 0,
        })

        assert result["step_count"] == 2
        assert len(result["messages"]) >= 3  # Human + AI + Tool + AI

Mocking LLM and Tool Calls

Never call real models in tests. Mock at the right level:

from langchain_community.chat_models import FakeListChatModel
from langchain.messages import AIMessage


def test_agent_with_fake_model():
    """Test agent behavior using FakeListChatModel."""
    fake_model = FakeListChatModel(responses=[
        AIMessage(content="", tool_calls=[
            {"name": "calculator", "args": {"expression": "2+2"}, "id": "call_1"}
        ]),
        AIMessage(content="The answer is 4."),
    ])

    app = build_agent_graph(model=fake_model)
    result = app.invoke({
        "messages": [HumanMessage(content="What is 2+2?")],
    })

    assert "4" in result["messages"][-1].content

Testing Checklist

• Every node function has a unit test for the happy path.
• Every node function has a test for the error path.
• Every conditional edge function has tests for each branch.
• The full graph has an integration test with mocked LLM.
• Interrupt behavior is tested (pause and resume).
• Checkpointing is tested (state survives across invocations).
• No test calls a real LLM API.

---

Summary

Area	Standard
Error Handling	Every node catches exceptions. Transient failures retry with backoff. Unrecoverable failures route to fallback nodes.
Human-in-the-Loop	Irreversible actions use interrupt(). Approval workflows for emails, writes, and financial operations.
Checkpointing	Always compile with a checkpointer in production. MemorySaver for dev, PostgresSaver for production.
Debugging	LangSmith tracing enabled in all environments. Structured logging in nodes. stream_mode for real-time observation.
Graph Design	Choose pattern by problem: Supervisor for orchestration, Map-Reduce for parallelism, Pipeline for sequential validation, Reflection for quality.
Performance	Stream to users. Cache deterministic calls. Parallelize independent work. Minimize state size.
Testing	Unit test nodes. Integration test graphs. Mock LLMs always. Test error paths and interrupt behavior.

Follow these patterns and your LangGraph agents will be reliable, observable, and maintainable. Break them only with a documented reason.

---

Further Reading

• LangGraph Getting Started — core concepts and your first agent.
• PostgreSQL Checkpointing & Memory Setup — production-grade state persistence.
• LangGraph Official Docs — comprehensive API reference.
• LangSmith Observability — tracing, debugging, and evaluation.
• LangChain Best Practices — patterns for the broader LangChain ecosystem.