Error Handling & Agent Reliability

Reliability Is Not Automatic

An agent that works correctly 80% of the time is dangerous in production. Unlike a web page that either loads or shows an error, an agent can fail in subtle ways: calling the wrong tool, producing a hallucinated response, taking an irreversible action, or getting stuck in an infinite loop.

Professional agent systems are built with reliability as a first-class concern, not an afterthought. This module covers the patterns that make agents behave consistently and fail gracefully.

Pattern 1: Retry with Exponential Backoff

The most common failure mode is transient: the API returned a 429 (rate limit) or 503 (temporary outage). These resolve themselves if you wait. The right response is to retry with increasing delays.

import time
import random
from functools import wraps

def with_retry(max_attempts: int = 4, base_delay: float = 1.0):
    """Decorator that retries a function with exponential backoff."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise   # out of retries: propagate the error

Now calculate the delay and decide whether to retry:

                    # Only retry on transient errors
                    error_str = str(e).lower()
                    is_transient = any(x in error_str for x in [
                        '429', '503', '500', 'rate limit', 'timeout',
                        'connection', 'temporarily unavailable'
                    ])
                    
                    if not is_transient:
                        raise   # permanent error: don't retry
                    
                    # Exponential backoff: 1s, 2s, 4s, 8s... with jitter
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
                    print(f"Attempt {attempt+1} failed ({e}). Retrying in {delay:.1f}s...")
                    time.sleep(delay)
        
        return wrapper
    return decorator

Use it as a decorator:

@with_retry(max_attempts=4, base_delay=1.0)
def call_llm(messages: list) -> str:
    response = client.chat.completions.create(
        model='gpt-5.6-sol',
        messages=messages,
    )
    return response.choices[0].message.content

@with_retry(max_attempts=3, base_delay=2.0)
def search_web(query: str) -> list:
    # ... your search implementation
    pass

Every tool call and every LLM call should have retry logic. Transient failures are normal in distributed systems.

Pattern 2: Fallback Chains

When your primary approach fails, the model is down, the tool is broken, the API key expired, a fallback chain automatically tries alternatives:

def call_llm_with_fallback(messages: list) -> str:
    """Try GPT-5.6 Sol first, fall back to cheaper tiers if unavailable."""
    
    models = ['gpt-5.6-sol', 'gpt-5.6-terra', 'gpt-5.6-luna']
    
    for model in models:
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                timeout=10,
            )
            if model != 'gpt-5.6-sol':
                print(f"⚠ Using fallback model: {model}")
            return response.choices[0].message.content
        
        except Exception as e:
            print(f"Model {model} failed: {e}")
            continue   # try next model
    
    raise RuntimeError("All fallback models exhausted")

The same pattern applies to search providers, email services, or any tool with an alternative:

def search_with_fallback(query: str) -> list:
    """Try Serper, fall back to DuckDuckGo."""
    
    try:
        return search_serper(query)
    except Exception:
        print("Serper failed, trying DuckDuckGo...")
        try:
            return search_duckduckgo(query)
        except Exception:
            print("Both search providers failed. Returning empty.")
            return []

Pattern 3: Output Validation

The agent produces an output: but is it correct? Does it follow the expected format? Is it safe to use downstream?

Schema validation catches structural errors:

from pydantic import BaseModel, validator

class AgentOutput(BaseModel):
    summary:     str
    action_items: list[str]
    confidence:  float   # 0.0 to 1.0
    
    @validator('confidence')
    def confidence_in_range(cls, v):
        if not 0.0 <= v <= 1.0:
            raise ValueError(f'confidence must be 0.0-1.0, got {v}')
        return v
    
    @validator('action_items')
    def has_action_items(cls, v):
        if not v:
            raise ValueError('action_items must not be empty')
        return v

def validated_llm_call(messages: list) -> AgentOutput | None:
    """Call LLM and validate the output."""
    raw = call_llm(messages)
    
    try:
        import json
        data = json.loads(raw)
        return AgentOutput(**data)
    except (json.JSONDecodeError, ValueError) as e:
        print(f"Output validation failed: {e}")
        print(f"Raw output: {raw[:200]}")
        return None

If validation fails, you can retry with a correction prompt:

def call_with_validation(messages: list, schema: type, max_retries: int = 2) -> object:
    for attempt in range(max_retries):
        raw = call_llm(messages)
        try:
            import json
            return schema(**json.loads(raw))
        except Exception as e:
            if attempt < max_retries - 1:
                messages.append({'role': 'assistant', 'content': raw})
                messages.append({'role': 'user', 'content': f'Your output had an error: {e}. Please fix it and respond with valid JSON matching the schema.'})
    raise ValueError(f"Failed to get valid output after {max_retries} attempts")

Pattern 4: Guardrails

Guardrails prevent agents from taking actions they shouldn’t. Think of them as a safety layer between the agent’s decision and actual execution:

DANGEROUS_PATTERNS = [
    'rm -rf', 'DROP TABLE', 'DELETE FROM', 'format c:',
    'sudo', '/etc/passwd', 'private key', 'api_key =',
]

def pre_action_guardrail(action_name: str, action_args: dict) -> tuple[bool, str]:
    """
    Check if an action should be allowed.
    Returns (allowed: bool, reason: str).
    """
    # Check for dangerous content in arguments
    args_str = str(action_args).lower()
    
    for pattern in DANGEROUS_PATTERNS:
        if pattern.lower() in args_str:
            return False, f"Blocked: action contains dangerous pattern '{pattern}'"
    
    # Limit file operations to specific directories
    if action_name in ('write_file', 'delete_file'):
        path = action_args.get('path', '')
        if not path.startswith('/tmp/') and not path.startswith('./output/'):
            return False, f"Blocked: file operations only allowed in /tmp/ and ./output/"
    
    # Rate limit expensive actions
    if action_name == 'send_email':
        # Check how many emails sent today (implementation omitted)
        emails_today = get_email_count_today()
        if emails_today >= 10:
            return False, f"Blocked: daily email limit reached ({emails_today}/10)"
    
    return True, "OK"

Integrate guardrails into the tool execution path:

def safe_execute_tool(name: str, args: dict) -> str:
    """Execute a tool only if it passes guardrail checks."""
    
    allowed, reason = pre_action_guardrail(name, args)
    if not allowed:
        return f"GUARDRAIL BLOCKED: {reason}"
    
    try:
        func = TOOL_REGISTRY[name]
        result = func(**args)
        return str(result)
    except Exception as e:
        return f"ERROR: {e}"

Guardrails should never be optional. Always route tool execution through the safety check.

Putting It Together: A Reliable Agent Template

def reliable_agent(goal: str) -> str:
    """A template for building reliable production agents."""
    messages = [
        {'role': 'system', 'content': AGENT_SYSTEM_PROMPT},
        {'role': 'user',   'content': goal},
    ]
    
    max_steps = 15
    for step in range(max_steps):
        
        # LLM call with retry
        response = call_llm_with_fallback(messages)
        
        # Check for tool calls
        if has_tool_calls(response):
            for tc in get_tool_calls(response):
                # Guardrail check
                allowed, reason = pre_action_guardrail(tc.name, tc.args)
                if not allowed:
                    messages.append({'role': 'system', 'content': f'Action blocked: {reason}'})
                    continue
                
                # Execute with retry
                try:
                    result = safe_execute_tool(tc.name, tc.args)
                except Exception as e:
                    result = f"Tool failed after retries: {e}"
                
                messages.append({'role': 'tool', 'content': result})
        else:
            # Final response: validate it
            return response
    
    return f"Reached max steps. Partial result available."

Exercise: Take an agent you built earlier and add all four reliability patterns: (1) wrap all LLM calls with @with_retry, (2) add a fallback model, (3) add output validation for a structured output, (4) add a simple guardrail that prevents the agent from calling more than 10 tool calls in one run. Test by simulating a rate limit error (temporarily pass a bad API key for the first call).