AI Agent Design Patterns for Production Reliability

You have built an AI agent that works perfectly on your laptop. Every tool call succeeds. The routing logic behaves exactly as intended. You deploy to production, and within an hour, the agent enters an infinite loop, your on-call alerts fire, and you spend the next two days staring at logs trying to understand what changed.

Nothing changed. The production environment just revealed what was always broken.

After running agents in production across several systems, I have found that failures almost always fall into one of three categories: unhandled tool call errors, runaway state growth, and over-reliance on nondeterministic LLM output. These are not obscure edge cases. They are structural problems that emerge predictably once real traffic hits your agent.

This article walks through each failure pattern, explains why it happens, and shows the specific design changes that prevent it.

Pattern 1: Tool Errors That Propagate and Kill the Agent

In local development, your external API calls almost always succeed. The network is stable, you are nowhere near rate limits, and the test payloads are clean. That environment hides a class of failures that only surfaces at scale.

In production, APIs time out. Third-party services return 503 for a few seconds during a deploy. Search endpoints return empty result sets for unusual queries. When any of these happen and your tool function does not handle the exception, one of two things occurs: the exception propagates up through the agent loop and the whole process stops, or you catch the exception somewhere upstream and pass None or an empty string into the next LLM call, which the model interprets as a valid result and continues processing toward a nonsensical conclusion.

The second case is the dangerous one because nothing in your logs looks like a failure.

The fragile implementation

import requests
import json

def fetch_product_data(product_id: str) -> str:
    # No timeout. No error handling. Will block indefinitely on slow responses.
    response = requests.get(f"https://api.internal.example.com/products/{product_id}")
    return response.json()["data"]["description"]  # IndexError or KeyError kills the agent

def dispatch_tool(tool_call) -> dict:
    args = json.loads(tool_call.function.arguments)
    # Any exception here propagates to the agent loop
    result = fetch_product_data(**args)
    return {
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result,
    }

Three specific problems here. First, requests.get has no timeout, so a slow upstream service will block the agent indefinitely. Second, the key access chain on the response will raise KeyError or IndexError on any unexpected response shape. Third, dispatch_tool does nothing to contain these exceptions, so the agent loop crashes.

The resilient implementation

The design principle is simple: tool functions do not raise exceptions. They return strings. Errors become data that the LLM can reason about.

import requests
import json
from typing import Any

TOOL_ERROR_PREFIX = "TOOL_ERROR"

def fetch_product_data(product_id: str) -> str:
    """
    Always returns a string. Errors are encoded as TOOL_ERROR: prefixed messages.
    The caller never needs to handle exceptions from this function.
    """
    try:
        response = requests.get(
            f"https://api.internal.example.com/products/{product_id}",
            timeout=10,
        )
        response.raise_for_status()

        payload = response.json()
        description = payload.get("data", {}).get("description")
        if description is None:
            return f"{TOOL_ERROR_PREFIX}: Product {product_id!r} not found or missing description field."

        return description

    except requests.Timeout:
        return f"{TOOL_ERROR_PREFIX}: Request timed out after 10 seconds for product {product_id!r}."
    except requests.HTTPError as exc:
        return f"{TOOL_ERROR_PREFIX}: HTTP {exc.response.status_code} from product API."
    except requests.ConnectionError:
        return f"{TOOL_ERROR_PREFIX}: Network unreachable when fetching product {product_id!r}."
    except (KeyError, ValueError, TypeError) as exc:
        return f"{TOOL_ERROR_PREFIX}: Unexpected response shape — {type(exc).__name__}: {exc}"
    except Exception as exc:
        return f"{TOOL_ERROR_PREFIX}: Unhandled error — {type(exc).__name__}: {exc}"

def dispatch_tool(tool_call) -> dict:
    args = json.loads(tool_call.function.arguments)
    result = fetch_product_data(**args)  # Never raises
    return {
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result,  # LLM sees TOOL_ERROR: prefix and can decide what to do next
    }

The TOOL_ERROR: prefix is load-bearing. You need the LLM to be able to identify error responses and act on them. Include guidance in your system prompt:

When a tool returns a message beginning with TOOL_ERROR:, do not treat that as a successful result.
Attempt an alternative approach or different parameters. After three failed attempts on the same
subtask, report the failure clearly to the user and stop.

For transient failures like timeouts and rate limit responses, add a retry wrapper with exponential backoff:

import time
from functools import wraps

def retry_on_tool_error(max_attempts: int = 3, base_delay: float = 1.0):
    """Decorator that retries a tool function if it returns a TOOL_ERROR string."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_result = ""
            for attempt in range(max_attempts):
                last_result = func(*args, **kwargs)
                if not last_result.startswith(TOOL_ERROR_PREFIX):
                    return last_result
                if attempt < max_attempts - 1:
                    wait = base_delay * (2 ** attempt)
                    time.sleep(wait)
            return last_result  # Return the final error string after exhausting retries
        return wrapper
    return decorator

@retry_on_tool_error(max_attempts=3, base_delay=1.0)
def fetch_product_data(product_id: str) -> str:
    # Same implementation as above
    ...

Takumi's Take: The error-as-string pattern feels unnatural the first time you see it, and that discomfort is valid. You are deliberately choosing to encode failure information in band rather than using Python's exception system. The reason is that LLM tool-calling interfaces treat tool responses as messages, not return values. An exception escaping a tool function does not reach the LLM at all — it crashes the Python process or, worse, gets swallowed silently by a framework. The string approach keeps the LLM in the reasoning loop, which is exactly where you want it when recoverable errors happen. That said, do not retry 400 Bad Request responses from the tool API. Those are permanent failures caused by malformed arguments, and retrying them wastes tokens and time. Distinguish retryable errors (5xx, timeout, connection) from permanent ones (4xx client errors) in your retry logic.

Pattern 2: State Growth That Exceeds the Context Window

With LangGraph and similar frameworks, agent state tends to start small and accumulate weight over time. The initial TypedDict is clean. Then someone adds raw search results to help with debugging. Then intermediate steps for traceability. Then document contents because they were needed in a downstream node. Within a few iterations, the state object serializes to tens of thousands of tokens before it even reaches the LLM.

When you hit the context window limit, you get one of two outcomes: a clean API error that at least tells you what happened, or silent truncation where the model receives a partial view of the conversation history and produces output that seems coherent but is missing context from several turns ago. The second outcome is the one that generates user complaints two weeks after deployment.

The overloaded state definition

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class BloatedAgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # Unbounded accumulation
    all_search_results: list[str]        # Appended every step
    intermediate_steps: list[dict]       # Grows without limit
    retrieved_document_chunks: list[str] # Full document text, kept indefinitely
    debug_trace: Annotated[list[str], lambda a, b: a + b]  # Debug logs in state

def search_node(state: BloatedAgentState) -> dict:
    docs = vector_store.similarity_search(state["messages"][-1].content, k=10)
    return {
        # Appends to existing results instead of replacing them
        "all_search_results": state.get("all_search_results", []) + [d.page_content for d in docs],
        "retrieved_document_chunks": state.get("retrieved_document_chunks", []) + [d.page_content for d in docs],
    }

On step five of an agent loop with ten retrieved documents per step, retrieved_document_chunks alone contains fifty full document texts. Add the conversation history and you are well past the context limit for most models.

Minimal state with explicit trimming

The discipline required here is treating each field in the state as a deliberate choice. Ask: does this field need to be visible across nodes, or does only one node need it? Is the latest value sufficient, or do I genuinely need history?

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, SystemMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # Trimmed before LLM calls
    current_search_results: list[str]  # Overwritten each search step, not accumulated
    task: str                          # Set once at initialization
    step_count: int
    is_complete: bool

def trim_messages_to_budget(
    messages: list[BaseMessage],
    token_budget: int = 6000,
) -> list[BaseMessage]:
    """
    Removes oldest non-system messages until the estimated token count
    fits within the budget. Always preserves SystemMessage instances.
    Uses character count / 4 as a token approximation.
    In production, replace with tiktoken for accurate counts.
    """
    system_msgs = [m for m in messages if isinstance(m, SystemMessage)]
    other_msgs = [m for m in messages if not isinstance(m, SystemMessage)]

    def estimate(msgs: list[BaseMessage]) -> int:
        return sum(len(m.content) // 4 for m in msgs)

    system_budget_used = estimate(system_msgs)
    remaining = token_budget - system_budget_used

    trimmed = list(other_msgs)
    while trimmed and estimate(trimmed) > remaining:
        trimmed.pop(0)

    return system_msgs + trimmed

def agent_node(state: AgentState) -> dict:
    trimmed = trim_messages_to_budget(state["messages"], token_budget=6000)

    system_content = (
        f"You are an agent completing the following task: {state['task']}\n"
        f"Current step: {state['step_count']}\n"
        "When the task is finished, state that explicitly."
    )

    messages_for_llm = [SystemMessage(content=system_content)] + trimmed
    response = llm.invoke(messages_for_llm)

    return {
        "messages": [response],
        "step_count": state["step_count"] + 1,
    }

def search_node(state: AgentState) -> dict:
    query = state["messages"][-1].content
    docs = vector_store.similarity_search(query, k=5)  # k=5, not k=10
    return {
        # Overwrite, do not append
        "current_search_results": [doc.page_content for doc in docs],
    }

Periodic summarization for long-running agents

For agents that run more than ten steps, message trimming alone may not be enough. The tail of the conversation history still grows, and you lose context from earlier steps when you trim. A summarization node solves this by compressing old history into a single context message.

from langchain_core.messages import HumanMessage

def summarize_history_node(state: AgentState) -> dict:
    """
    Compresses all but the four most recent messages into a summary.
    Call this on a conditional edge every N steps.
    """
    messages = state["messages"]
    if len(messages) < 8:
        return {}  # Nothing to compress yet

    to_summarize = messages[:-4]
    recent = messages[-4:]

    history_text = "\n".join(
        f"{m.__class__.__name__}: {m.content[:300]}" for m in to_summarize
    )
    prompt = (
        "Summarize the following conversation history in three to five sentences. "
        "Include key decisions made, search results used, and current progress.\n\n"
        + history_text
    )

    summary_response = llm.invoke([HumanMessage(content=prompt)])
    summary_message = SystemMessage(
        content=f"[Summary of prior conversation]\n{summary_response.content}"
    )

    return {"messages": [summary_message] + recent}

def should_summarize(state: AgentState) -> str:
    if state["step_count"] > 0 and state["step_count"] % 8 == 0:
        return "summarize"
    return "agent"

The graph structure connects these nodes so that every eighth step routes through summarize_history_node before returning to agent_node. The key is that summarization runs before the next LLM call, not after, so the model always receives a compressed history rather than a truncated one.

One practical note on the token estimation function above: dividing character count by four is a rough heuristic. For English text it is usually close enough. For content with many numbers, URLs, or code snippets, accuracy degrades. In any production system, use tiktoken with the correct encoding for your model. The cost of running token estimation is negligible compared to the cost of hitting a context limit mid-conversation.

Pattern 3: Nondeterministic Output Controlling Deterministic Flow

LLM outputs are probabilistic. The same prompt run twice will produce different outputs. For free-form generation tasks, this is exactly what you want. For routing decisions inside an agent graph, it is a reliability hazard.

Consider an agent that asks the model to decide what to do next and expects a response like search, generate, or complete. In testing, the model reliably returns one of those three strings. In production, with slightly different input context, it returns Search with a capital letter, or search for more information, or 検索する if the task description was in Japanese. Your conditional edge matches only lowercase exact strings. None of these match. The default branch fires, and the agent loops.

This is the source of failures that appear intermittent and cannot be reproduced locally. The model's output distribution shifts slightly with different inputs, and your routing logic was never robust to that shift.

The brittle routing approach

def routing_node(state: AgentState) -> dict:
    prompt = (
        f"Task: {state['task']}\n"
        f"Search results available: {len(state.get('current_search_results', []))}\n\n"
        "Reply with one word only: search / generate / complete"
    )
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"next_action": response.content}

def route_to_next_node(state: AgentState) -> str:
    action = state.get("next_action", "")
    if action == "search":
        return "search"
    elif action == "generate":
        return "generate"
    elif action == "complete":
        return "complete"
    else:
        return "search"  # Default that causes loops when model output drifts

Enforcing output structure with Pydantic

The fix is to remove free-form string output from any decision that affects control flow. Use with_structured_output to force the model to return a validated object.

from pydantic import BaseModel, Field
from typing import Literal
from langchain_openai import ChatOpenAI

base_llm = ChatOpenAI(model="gpt-4o", temperature=0)

class AgentRoutingDecision(BaseModel):
    """Structured output for agent routing decisions."""
    action: Literal["search", "generate", "complete"] = Field(
        description=(
            "Next action. 'search': need more information. "
            "'generate': ready to produce final output. "
            "'complete': task is fully done."
        )
    )
    reasoning: str = Field(
        description="One or two sentences explaining this choice."
    )
    confidence: float = Field(
        ge=0.0,
        le=1.0,
        description="Confidence in this decision from 0.0 to 1.0.",
    )

routing_llm = base_llm.with_structured_output(AgentRoutingDecision)

def routing_node(state: AgentState) -> dict:
    prompt = (
        f"Task: {state['task']}\n"
        f"Steps completed: {state['step_count']}\n"
        f"Search results available: {len(state.get('current_search_results', []))}\n\n"
        "Decide the next action."
    )

    decision: AgentRoutingDecision = routing_llm.invoke(
        [HumanMessage(content=prompt)]
    )

    return {
        "next_action": decision.action,
        "routing_reasoning": decision.reasoning,
        "routing_confidence": decision.confidence,
    }

def route_to_next_node(state: AgentState) -> str:
    # Literal type guarantees this is one of three valid values
    return state.get("next_action", "search")

With with_structured_output, the action field is validated against Literal["search", "generate", "complete"] before the function returns. If the model produces something that cannot be coerced into one of those values, Pydantic raises a validation error. You can handle that exception explicitly in the routing node, rather than having an invalid string silently propagate through your graph.

The reasoning and confidence fields are not just nice-to-have. When you are debugging a production trace in LangSmith or reviewing logs, knowing why the model chose search over generate at step seven saves substantial time. The confidence score also lets you add a guard: if confidence drops below 0.4 on a routing decision, escalate to a human or take a safer default action.

Apply the same principle to tool argument generation. When an LLM generates arguments for a tool call, constrain the output schema:

class DocumentSearchArgs(BaseModel):
    query: str = Field(
        min_length=1,
        max_length=200,
        description="Search query string.",
    )
    result_limit: int = Field(
        default=5,
        ge=1,
        le=20,
        description="Maximum number of results to return.",
    )
    search_mode: Literal["semantic", "keyword", "hybrid"] = Field(
        default="semantic",
        description="Retrieval strategy to use.",
    )

Defining input schemas for tools accomplishes two things. The LLM receives precise instructions about what arguments are valid. And if the model generates an out-of-range value like result_limit=100, Pydantic catches it at the boundary before the tool function even runs.

Enforcing a Step Limit

All three patterns contribute to infinite loops, but the most direct protection is a hard step ceiling. LangGraph exposes this through recursion_limit:

from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)
# ... node and edge definitions ...

app = graph.compile()

result = app.invoke(
    {
        "task": "Research and summarize recent developments in X.",
        "messages": [],
        "step_count": 0,
        "is_complete": False,
        "current_search_results": [],
    },
    config={"recursion_limit": 25},
)

Pass recursion_limit in the config argument to invoke or stream, not as a parameter to compile. The limit applies per-run, which means you can vary it based on task complexity if needed.

Decide before deployment what happens when the limit is reached. Three options: raise an exception and let the caller handle it, return whatever partial state the agent has accumulated, or implement Human-in-the-Loop by checkpointing state and waiting for external input. Leaving this undefined means production surprises.

Pre-Deployment Checklist

Before shipping an agent to production, verify these items for each of the three patterns.

Area	Check
Tool errors	Every tool function catches all exceptions and returns strings
Tool errors	Explicit timeouts on all external HTTP calls
Tool errors	TOOL_ERROR: prefix on error strings, referenced in system prompt
Tool errors	Retry logic uses exponential backoff, not fixed delay
State growth	State fields audited: accumulating only what is necessary
State growth	messages trimmed before every LLM invocation
State growth	Token budget set to 70-80% of model's context window
State growth	Summarization node added for flows exceeding ten steps
Nondeterminism	All routing and completion decisions use with_structured_output
Nondeterminism	Routing actions constrained with Literal types
Nondeterminism	temperature=0 used for deterministic decisions
Loop control	recursion_limit configured per invocation
Loop control	Behavior at limit defined: error, partial result, or escalation
Production env	Error traces exported to LangSmith or equivalent observability tool

What to Fix First

If you are looking at an existing agent and deciding where to start, tool error handling gives the highest return on investment. Unhandled exceptions in tool functions are the single most common cause of complete agent failure in production. Add try/except to every tool function today, encode errors as TOOL_ERROR: strings, and update your system prompt with error handling instructions. That change alone eliminates the most visible class of production failures.

Message trimming is the second priority because context window violations tend to appear after several days of production traffic, when the agent has accumulated enough state to hit the limit. A simple token budget check before each LLM call prevents the problem entirely.

Structured output for routing decisions is the third fix because the symptoms are subtle. The agent does not crash. It takes unexpected paths, produces surprising outputs, and the failures are difficult to reproduce. Once you have Pydantic schemas on all routing decisions, this category of failure disappears from your traces.

The combination of these three changes does not make agents perfectly reliable. LLMs still hallucinate, external APIs still have outages, and context understanding still degrades on very long tasks. But these three design choices close the gap between "works in demo" and "runs stably under real load" for the vast majority of production agents.