The Reality of Production Agent Reliability
A typical production agent makes 10-50 LLM API calls and 5-20 tool calls per task. At each of these calls:
- The API might return a 429 (rate limited)
- The network might timeout
- The API might return a 500 (server error)
- The model might return an unparseable response
- A tool might encounter an unexpected error
Without deliberate reliability engineering, these transient failures cascade into complete task failures. With it, most of them are invisible to the user.
Retry Logic with Tenacity
# src/reliability/retry.py
from tenacity import (
retry, stop_after_attempt, wait_exponential, wait_random_exponential,
retry_if_exception_type, before_sleep_log, after_log
)
from openai import RateLimitError, APITimeoutError, APIConnectionError
from anthropic import APIStatusError
import logging
logger = logging.getLogger(__name__)
# Standard retry policy for LLM API calls: applied as a decorator via @llm_retry
llm_retry = retry(
stop=stop_after_attempt(5), # give up after 5 total attempts (1 original + 4 retries)
# wait_random_exponential adds jitter: prevents all retrying clients hitting the API at the same moment
wait=wait_random_exponential(multiplier=1, min=1, max=60),
# only retry on transient errors: don't retry on 400 Bad Request (that's a bug, not a blip)
retry=retry_if_exception_type((
RateLimitError, # 429, API is throttling us; backoff and retry
APITimeoutError, # request timed out, network issue, safe to retry
APIConnectionError, # couldn't reach the API, safe to retry
)),
before_sleep=before_sleep_log(logger, logging.WARNING), # log each retry attempt with wait time
reraise=True, # after max attempts, re-raise the original exception (don't swallow it)
)
@llm_retry
async def call_openai_with_retry(messages: list, **kwargs) -> str:
response = await openai_client.chat.completions.create(
messages=messages,
**kwargs,
)
return response.choices[0].message.content
# Separate, more aggressive policy for tool calls: cheaper to retry, shorter max wait
tool_retry = retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=0.5, min=0.5, max=5), # max 5s wait vs 60s for LLM
retry=retry_if_exception_type((TimeoutError, ConnectionError)),
)
@tool_retry
async def resilient_web_search(query: str) -> list[dict]:
return await search_web(query)Circuit Breaker Implementation
# src/reliability/circuit_breaker.py
from enum import Enum
import time
from collections import deque
from dataclasses import dataclass, field
class CircuitState(Enum):
CLOSED = "closed" # normal operation, all calls pass through
OPEN = "open" # failing fast, dependency is down, skip all calls immediately
HALF_OPEN = "half_open" # HALF_OPEN state: let one request through to probe if the service recovered before reopening
@dataclass
class CircuitBreaker:
name: str
failure_threshold: int = 5 # open the circuit after this many failures
recovery_timeout: float = 60.0 # seconds to wait in OPEN before trying HALF_OPEN
half_open_max_calls: int = 1 # how many probe calls allowed in HALF_OPEN before closing/reopening
_state: CircuitState = field(default=CircuitState.CLOSED, init=False)
# deque(maxlen=10) keeps only the last 10 failure timestamps: auto-evicts older ones
_failures: deque = field(default_factory=lambda: deque(maxlen=10), init=False)
_last_failure_time: float = field(default=0.0, init=False)
_half_open_calls: int = field(default=0, init=False)
@property
def state(self) -> CircuitState:
# Auto-transition from OPEN → HALF_OPEN when recovery_timeout has elapsed
if self._state == CircuitState.OPEN:
if time.time() - self._last_failure_time > self.recovery_timeout:
self._state = CircuitState.HALF_OPEN
self._half_open_calls = 0
return self._state
def call_succeeded(self):
# Success in any state resets the circuit: we're healthy again
self._failures.clear()
self._state = CircuitState.CLOSED
def call_failed(self):
self._failures.append(time.time())
self._last_failure_time = time.time()
# Once failure_threshold is reached, open the circuit to stop wasting time on a dead dependency
if len(self._failures) >= self.failure_threshold:
self._state = CircuitState.OPEN
logger.warning(f"Circuit breaker '{self.name}' OPENED after {self.failure_threshold} failures")
async def call(self, func, *args, **kwargs):
state = self.state
# OPEN: fail immediately without calling the dependency: saves latency and load
if state == CircuitState.OPEN:
raise Exception(f"Circuit '{self.name}' is OPEN, dependency is unavailable")
if state == CircuitState.HALF_OPEN:
# Allow only half_open_max_calls probes through; reject the rest while probing
if self._half_open_calls >= self.half_open_max_calls:
raise Exception(f"Circuit '{self.name}' is HALF_OPEN, testing recovery")
self._half_open_calls += 1
try:
result = await func(*args, **kwargs)
self.call_succeeded() # probe succeeded, close the circuit
return result
except Exception as e:
self.call_failed() # probe failed, go back to OPEN
raise
# Usage
search_circuit = CircuitBreaker(name="tavily_search", failure_threshold=5, recovery_timeout=60)
async def protected_search(query: str) -> list[dict]:
# All calls go through the circuit breaker: it decides whether to forward or fail fast
return await search_circuit.call(search_tavily, query)Provider Fallback Strategy
Never depend on a single LLM provider. Implement automatic fallback:
# src/reliability/llm_fallback.py
from dataclasses import dataclass
from typing import Callable, Any
@dataclass
class LLMProvider:
name: str
call_func: Callable
circuit_breaker: CircuitBreaker # each provider gets its own circuit breaker
is_primary: bool = True
class FallbackLLMClient:
"""LLM client with automatic provider failover."""
def __init__(self, providers: list[LLMProvider]):
# providers list is ordered: primary first, then fallbacks in preference order
self.providers = providers
async def invoke(self, messages: list, **kwargs) -> str:
last_error = None
for provider in self.providers:
try:
# circuit_breaker.call() will raise immediately if that provider's circuit is OPEN
result = await provider.circuit_breaker.call(
provider.call_func,
messages,
**kwargs,
)
if not provider.is_primary:
# Log when falling back so you can track primary provider health in dashboards
logger.info(f"Used fallback provider: {provider.name}")
return result
except Exception as e:
last_error = e
logger.warning(f"Provider {provider.name} failed: {e}")
continue # try the next provider in the list
# All providers failed: surface the last error for diagnostics
raise Exception(f"All LLM providers failed. Last error: {last_error}")
# Configure with primary + fallbacks: ordered by preference
llm_client = FallbackLLMClient(providers=[
LLMProvider(
name="openai-gpt4o",
call_func=call_openai,
circuit_breaker=CircuitBreaker("openai", failure_threshold=3),
is_primary=True,
),
LLMProvider(
name="anthropic-claude",
call_func=call_anthropic,
# failure_threshold=3 means 3 failures → open circuit → skip to next fallback
circuit_breaker=CircuitBreaker("anthropic", failure_threshold=3),
is_primary=False,
),
LLMProvider(
name="google-gemini",
call_func=call_gemini,
circuit_breaker=CircuitBreaker("gemini", failure_threshold=3),
is_primary=False,
),
])Iteration Limits and Cost Guards
# src/reliability/guards.py
from dataclasses import dataclass, field
@dataclass
class AgentBudget:
max_iterations: int = 20 # hard cap on LLM calls per run
max_cost_usd: float = 1.00 # dollar ceiling before aborting
max_wall_seconds: float = 300.0 # 5 minutes, prevents hanging agents blocking resources
# These counters are mutable state tracked at runtime: init=False means not set in __init__
iterations: int = field(default=0, init=False)
cost_usd: float = field(default=0.0, init=False)
start_time: float = field(default_factory=time.time, init=False)
def check(self) -> None:
# Call this at the top of every LLM node: raises BudgetExceeded before the API call
if self.iterations >= self.max_iterations:
raise BudgetExceeded(f"Iteration limit ({self.max_iterations}) reached")
if self.cost_usd >= self.max_cost_usd:
raise BudgetExceeded(f"Cost limit (${self.max_cost_usd:.2f}) reached, spent ${self.cost_usd:.4f}")
elapsed = time.time() - self.start_time
if elapsed >= self.max_wall_seconds:
raise BudgetExceeded(f"Time limit ({self.max_wall_seconds}s) reached after {elapsed:.0f}s")
def record_llm_call(self, input_tokens: int, output_tokens: int, model: str = "gpt-5.6-sol"):
self.iterations += 1
# Prices are per-token (not per-million): divide by 1e6 to convert
pricing = {
"gpt-5.6-sol": (5.0/1e6, 15.0/1e6), # $5/$15 per M tokens
"gpt-5.6-luna": (0.15/1e6, 0.6/1e6),
"claude-sonnet-5": (3.0/1e6, 15.0/1e6),
}
input_price, output_price = pricing.get(model, (5.0/1e6, 15.0/1e6))
self.cost_usd += input_tokens * input_price + output_tokens * output_price
class BudgetExceeded(Exception):
pass
# Integrate into LangGraph node
def guarded_llm_call(state: AgentState, budget: AgentBudget) -> dict:
budget.check() # raises BudgetExceeded before the API call if any limit is already hit
response = llm.invoke(state["messages"])
# response_metadata contains token counts from the provider: use them for accurate cost tracking
if hasattr(response, 'response_metadata'):
usage = response.response_metadata.get('token_usage', {})
budget.record_llm_call(
input_tokens=usage.get('prompt_tokens', 0),
output_tokens=usage.get('completion_tokens', 0),
)
return {"messages": [response]}Graceful Degradation
When things go wrong, degrade gracefully rather than failing completely:
async def agent_with_degradation(query: str) -> dict:
"""Agent that degrades gracefully when components fail."""
# Tier 1: full agentic response with tools and multi-step reasoning
try:
result = await full_agent.ainvoke({"messages": [HumanMessage(content=query)]})
return {"response": result["messages"][-1].content, "mode": "full"}
except BudgetExceeded as e:
logger.warning(f"Budget exceeded: {e}")
# Tier 2: single LLM call without tools: still useful, just no external data
response = llm.invoke([HumanMessage(content=query)])
return {"response": response.content, "mode": "no_tools", "warning": str(e)}
except Exception as e:
logger.error(f"Agent failed: {e}")
# Tier 3: last resort: always return something rather than a 500 to the user
return {
"response": "I encountered an issue processing your request. Please try again or contact support.",
"mode": "error",
"error": str(e), # include error in response dict for logging/alerting upstream
}Reliability engineering isn’t glamorous, but it’s what separates demo agents from production systems. These patterns, retry, circuit breaker, fallback, budget guards, are the difference between an agent that works and one that works reliably.