title: "AI Agent Error Handling: Best Practices for Resilient Systems" slug: "ai-agent-error-handling-best-practices" description: "Production-grade error handling for AI agents — retry strategies, circuit breakers, graceful degradation, fallback chains, and the patterns that keep agents running when everything breaks." category: "Engineering" publishedAt: "2026-07-22"

AI Agent Error Handling: Best Practices for Resilient Systems

AI agents in production face a gauntlet of failure modes: model timeouts, tool crashes, rate limits, hallucinations, network partitions, and unexpected input formats. Without robust error handling, a single failure can cascade into a user-facing disaster. This guide covers every pattern you need to build AI agents that fail gracefully, recover automatically, and maintain service even when dependencies are down.

Understanding AI Agent Failure Modes

Before writing error handling code, you need to understand what can go wrong:

Model-Level Failures

Tool-Level Failures

System-Level Failures

• Network partitions between agent and model
• Database connection failures
• Memory exhaustion from large contexts
• Concurrent request limits exceeded
• Deployments that break running sessions

The Error Handling Architecture

Production-grade error handling has four layers:

Layer 1: Input Validation

Catch errors before they reach the model:

from pydantic import BaseModel, validator

class AgentInput(BaseModel):
    query: str
    context: dict = {}
    session_id: str
    
    @validator('query')
    def validate_query(cls, v):
        if not v or not v.strip():
            raise ValueError("Query cannot be empty")
        if len(v) > 10000:
            raise ValueError("Query exceeds maximum length")
        # Check for potential injection
        suspicious_patterns = ['ignore previous', 'system:', '<script']
        lowered = v.lower()
        for pattern in suspicious_patterns:
            if pattern in lowered:
                logger.warning(f"Potential injection attempt: {pattern}")
        return v
    
    @validator('context')
    def validate_context_size(cls, v):
        total_size = len(json.dumps(v))
        if total_size > 50000:
            raise ValueError("Context exceeds 50KB limit")
        return v

Layer 2: Execution Protection

Wrap every agent execution in comprehensive error handling:

from functools import wraps
import asyncio
from typing import TypeVar, Callable, Optional

T = TypeVar('T')

class AgentErrorHandler:
    def __init__(self, config: dict):
        self.max_retries = config.get("max_retries", 3)
        self.retry_delay = config.get("retry_delay", 1.0)
        self.timeout = config.get("timeout", 30)
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60
        )
    
    async def execute_with_protection(
        self,
        operation: Callable,
        *args,
        fallback: Optional[Callable] = None,
        **kwargs
    ) -> T:
        # Check circuit breaker
        if not self.circuit_breaker.can_execute():
            if fallback:
                return await fallback(*args, **kwargs)
            raise ServiceUnavailableError(
                "Circuit breaker is open. Service temporarily unavailable."
            )
        
        last_error = None
        for attempt in range(self.max_retries):
            try:
                # Execute with timeout
                result = await asyncio.wait_for(
                    operation(*args, **kwargs),
                    timeout=self.timeout
                )
                
                # Success — reset circuit breaker
                self.circuit_breaker.record_success()
                return result
                
            except asyncio.TimeoutError:
                last_error = TimeoutError(
                    f"Operation timed out after {self.timeout}s (attempt {attempt + 1})"
                )
                logger.warning(f"Timeout on attempt {attempt + 1}/{self.max_retries}")
                
            except RateLimitError as e:
                wait_time = e.retry_after or (self.retry_delay * (2 ** attempt))
                logger.warning(f"Rate limited. Waiting {wait_time}s before retry.")
                await asyncio.sleep(wait_time)
                last_error = e
                
            except ModelUnavailableError as e:
                logger.error(f"Model unavailable: {e}")
                if fallback:
                    return await fallback(*args, **kwargs)
                last_error = e
                
            except ValidationError as e:
                # Don't retry validation errors — they'll fail every time
                logger.error(f"Validation error (not retrying): {e}")
                raise
                
            except Exception as e:
                last_error = e
                logger.error(f"Unexpected error on attempt {attempt + 1}: {e}")
            
            # Exponential backoff with jitter
            if attempt < self.max_retries - 1:
                delay = self.retry_delay * (2 ** attempt) + random.uniform(0, 0.5)
                await asyncio.sleep(delay)
        
        # All retries exhausted
        self.circuit_breaker.record_failure()
        
        if fallback:
            logger.info("All retries exhausted. Using fallback.")
            return await fallback(*args, **kwargs)
        
        raise AgentExecutionError(
            f"Operation failed after {self.max_retries} attempts"
        ) from last_error

Layer 3: Output Validation

Validate model outputs before returning them:

class OutputValidator:
    def __init__(self, schema: dict, safety_checker=None):
        self.schema = schema
        self.safety_checker = safety_checker
    
    async def validate(self, output: str) -> ValidationResult:
        errors = []
        
        # Schema validation
        try:
            parsed = json.loads(output)
            for field, rules in self.schema.items():
                if field not in parsed:
                    errors.append(f"Missing required field: {field}")
                elif not self._validate_field(parsed[field], rules):
                    errors.append(f"Invalid value for field: {field}")
        except json.JSONDecodeError:
            if self.schema:
                errors.append("Output is not valid JSON")
        
        # Safety check
        if self.safety_checker:
            safety_result = await self.safety_checker.check(output)
            if not safety_result.safe:
                errors.append(f"Safety check failed: {safety_result.reason}")
        
        # Length check
        if len(output) > 10000:
            errors.append("Output exceeds maximum length")
        
        return ValidationResult(
            valid=len(errors) == 0,
            errors=errors,
            sanitized_output=self._sanitize(output) if not errors else output
        )
    
    def _sanitize(self, output: str) -> str:
        # Remove potential sensitive data
        sanitized = re.sub(r'\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b', '[EMAIL]', output, flags=re.IGNORECASE)
        sanitized = re.sub(r'\b\d{16,19}\b', '[CARD_NUMBER]', sanitized)
        return sanitized

Layer 4: Fallback Chains

When the primary approach fails, have a plan B, C, and D:

class FallbackChain:
    def __init__(self):
        self.strategies = []
    
    def add(self, name: str, handler: Callable, condition: Callable = None):
        self.strategies.append({
            "name": name,
            "handler": handler,
            "condition": condition or (lambda e: True)
        })
        return self
    
    async def execute(self, input_data):
        errors = []
        
        for strategy in self.strategies:
            try:
                result = await strategy["handler"](input_data)
                if result and result.is_valid():
                    if errors:
                        logger.info(f"Recovered after fallback: {strategy['name']}")
                    return result
            except Exception as e:
                errors.append((strategy["name"], str(e)))
                logger.warning(f"Fallback '{strategy['name']}' failed: {e}")
                continue
        
        # All strategies failed
        raise AllStrategiesExhaustedError(
            f"All {len(self.strategies)} strategies failed: {errors}"
        )


# Usage
fallback = FallbackChain() \
    .add("primary_model", call_gpt4) \
    .add("secondary_model", call_glm5) \
    .add("cheap_model", call_glm_flash) \
    .add("cached_response", get_cached) \
    .add("static_fallback", lambda _: StaticResponse("I'm having trouble right now. Please try again."))

Tool-Specific Error Handling

Handling Tool Timeouts

class ToolExecutor:
    async def call_with_timeout(self, tool, params, timeout=10):
        try:
            result = await asyncio.wait_for(
                tool.execute(params),
                timeout=timeout
            )
            return ToolResult(success=True, data=result)
            
        except asyncio.TimeoutError:
            logger.warning(f"Tool '{tool.name}' timed out after {timeout}s")
            
            # Try a simpler version of the tool if available
            if hasattr(tool, 'fast_mode'):
                logger.info(f"Trying fast mode for '{tool.name}'")
                result = await asyncio.wait_for(
                    tool.fast_mode(params),
                    timeout=timeout / 2
                )
                return ToolResult(success=True, data=result, degraded=True)
            
            return ToolResult(
                success=False,
                error="The request took too long. I'll try a different approach."
            )

Handling Rate Limits

class RateLimitHandler:
    def __init__(self):
        self.limits = {}  # {provider: {remaining, reset_time}}
    
    async def execute(self, provider: str, operation: Callable):
        # Check if we're rate limited
        if provider in self.limits:
            limit = self.limits[provider]
            if limit["remaining"] <= 0:
                wait_time = limit["reset_time"] - time.time()
                if wait_time > 0:
                    logger.info(f"Rate limited on {provider}. Waiting {wait_time:.0f}s")
                    await asyncio.sleep(wait_time)
        
        try:
            result = await operation()
            return result
        except RateLimitError as e:
            # Update rate limit info
            self.limits[provider] = {
                "remaining": e.remaining,
                "reset_time": time.time() + (e.retry_after or 60)
            }
            
            # Try alternative provider
            alternative = self.get_alternative_provider(provider)
            if alternative:
                logger.info(f"Switching from {provider} to {alternative}")
                return await self.execute(alternative, operation)
            
            raise

Circuit Breaker Pattern

Circuit breakers prevent cascading failures by stopping requests to failing services:

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing — reject all requests
    HALF_OPEN = "half_open"  # Testing if service recovered

class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
    
    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        elif self.state == CircuitState.OPEN:
            # Check if recovery time has passed
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
                logger.info("Circuit breaker entering half-open state")
                return True
            return False
        elif self.state == CircuitState.HALF_OPEN:
            return True
    
    def record_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                logger.info("Circuit breaker closed — service recovered")
        elif self.state == CircuitState.CLOSED:
            self.failure_count = 0
    
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
            logger.warning("Circuit breaker reopened — service still failing")
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            logger.error(
                f"Circuit breaker opened after {self.failure_count} failures"
            )

Graceful Degradation

When full capability isn't available, provide reduced service instead of no service:

class GracefulDegradation:
    LEVELS = {
        3: "full",         # All features available
        2: "reduced",      # Some features unavailable, core works
        1: "minimal",      # Only basic functionality
        0: "unavailable"   # Service down
    }
    
    def __init__(self):
        self.service_level = 3
    
    async def assess_and_adjust(self):
        """Check system health and adjust service level"""
        health = await self.check_health()
        
        if health.model_available and health.tools_available:
            self.service_level = 3
        elif health.model_available:
            self.service_level = 2  # Model works but some tools down
        elif health.fallback_model_available:
            self.service_level = 1  # Use simpler model
        else:
            self.service_level = 0  # Everything down
    
    async def handle_request(self, request):
        await self.assess_and_adjust()
        
        if self.service_level == 0:
            return Response(
                text="I'm experiencing technical difficulties. Please try again in a few minutes.",
                degraded=True
            )
        
        if self.service_level == 1:
            # Use fallback model with simpler prompts
            return await self.process_with_fallback_model(request)
        
        if self.service_level == 2:
            # Skip unavailable tools
            available_tools = [t for t in request.tools if t.available]
            return await self.process_with_tools(request, available_tools)
        
        # Full capability
        return await self.process_full(request)

User-Facing Error Messages

Technical errors should never reach users. Transform them into helpful messages:

class UserFriendlyErrors:
    MESSAGES = {
        "timeout": "I'm taking longer than expected. Let me try a simpler approach.",
        "rate_limit": "I'm handling a lot of requests right now. Please try in a moment.",
        "tool_unavailable": "I can't access that tool right now, but I can still help with other questions.",
        "model_error": "I had trouble processing that. Could you rephrase?",
        "validation_error": "I didn't quite understand that. Could you provide more detail?",
        "context_limit": "This conversation is getting long. Let me start fresh to help you better.",
    }
    
    def transform(self, error: Exception) -> str:
        if isinstance(error, TimeoutError):
            return self.MESSAGES["timeout"]
        elif isinstance(error, RateLimitError):
            return self.MESSAGES["rate_limit"]
        elif isinstance(error, ToolUnavailableError):
            return self.MESSAGES["tool_unavailable"]
        elif isinstance(error, ModelError):
            return self.MESSAGES["model_error"]
        elif isinstance(error, ValidationError):
            return self.MESSAGES["validation_error"]
        else:
            return "Something went wrong. I'll try again."
    
    def with_recovery_action(self, error: Exception) -> dict:
        message = self.transform(error)
        return {
            "message": message,
            "recovery_action": self.get_recovery_action(error),
            "can_retry": True
        }

Monitoring Error Patterns

Track errors to identify systemic issues:

class ErrorMonitor:
    def __init__(self):
        self.error_counts = {}  # {error_type: count}
        self.error_rates = {}   # {endpoint: {timestamp, rate}}
    
    async def record_error(self, error: Exception, context: dict):
        error_type = type(error).__name__
        
        self.error_counts[error_type] = self.error_counts.get(error_type, 0) + 1
        
        # Alert on critical patterns
        if self.error_counts.get(error_type, 0) > 20:
            await self.send_alert(
                f"High error count: {error_type} ({self.error_counts[error_type]} occurrences)"
            )
        
        # Detect error spikes
        current_rate = await self.calculate_error_rate(window_minutes=5)
        if current_rate > 0.10:  # >10% error rate
            await self.send_alert(
                f"Error rate spike: {current_rate:.1%} in last 5 minutes"
            )
        
        # Store for analysis
        await self.store_error({
            "type": error_type,
            "message": str(error),
            "context": context,
            "timestamp": datetime.utcnow().isoformat(),
            "stack_trace": traceback.format_exc()
        })

Conclusion

Error handling is what separates toy demos from production systems. The patterns in this guide — input validation, execution protection, circuit breakers, fallback chains, graceful degradation, and user-friendly error messages — are not optional. Every production AI agent needs all of them.

The investment in robust error handling pays back immediately: fewer support tickets, higher user satisfaction, lower operational cost, and the ability to sleep through the night without being paged. Build resilient agents from day one, and you'll never have to explain why your AI went rogue at 3 AM.