Error Handling Patterns for AI Service Integrations
Most AI service errors fall into a small number of categories, and each category has the right handling strategy. Getting error handling right is the difference between a service that recovers gracefully and one that either retries indefinitely or surfaces cryptic stack traces to users.
The error taxonomy that actually matters
Provider API errors are not all the same. Treating them uniformly with a single retry policy is incorrect and expensive.
| Error type | HTTP status | Retry? | Strategy |
|---|---|---|---|
| Rate limit | 429 | Yes | Exponential backoff, respect Retry-After header |
| Service overload | 503 | Yes | Short backoff, 2-3 attempts |
| Server error | 500 | Sometimes | Retry once; if persistent, fallback or fail |
| Bad request | 400 | Never | Fix the request before retrying |
| Auth failure | 401 | Never | Alert operations, do not retry |
| Context overflow | 400 (specific) | Never | Trim the prompt, then retry |
| Timeout | — | Once | With a slightly longer timeout |
The critical distinction: do not retry errors caused by the request itself (400, 401, context overflow). Retrying them wastes quota and adds latency. Only retry errors caused by the provider's transient state (429, 503, intermittent 500).
Classifying errors in Python
from enum import Enum
class ErrorClass(Enum):
RETRYABLE = "retryable" # 429, 503, transient 500
FATAL = "fatal" # 400, 401, 403
TIMEOUT = "timeout" # asyncio.TimeoutError
CONTEXT_OVERFLOW = "overflow" # 400 with specific message
def classify_provider_error(exc: Exception) -> ErrorClass:
error_str = str(exc).lower()
if isinstance(exc, TimeoutError):
return ErrorClass.TIMEOUT
if hasattr(exc, "status_code"):
if exc.status_code == 429:
return ErrorClass.RETRYABLE
if exc.status_code in (500, 503):
return ErrorClass.RETRYABLE
if exc.status_code in (400, 401, 403):
if "context" in error_str or "token" in error_str or "length" in error_str:
return ErrorClass.CONTEXT_OVERFLOW
return ErrorClass.FATAL
return ErrorClass.RETRYABLE # conservative default: try once more
Retry with classification-aware policy
import asyncio
import random
import logging
logger = logging.getLogger(__name__)
async def call_with_policy(
client,
request: dict,
max_retries: int = 3,
base_delay: float = 1.0,
) -> dict:
last_exc = None
for attempt in range(max_retries + 1):
try:
return await asyncio.wait_for(client.generate(request), timeout=30.0)
except asyncio.TimeoutError as exc:
error_class = ErrorClass.TIMEOUT
last_exc = exc
if attempt == 0: # retry once with longer timeout
await asyncio.sleep(1.0)
continue
break # give up after second timeout
except Exception as exc:
error_class = classify_provider_error(exc)
last_exc = exc
if error_class == ErrorClass.FATAL:
logger.error("provider_fatal_error", extra={
"attempt": attempt, "error": str(exc), "class": error_class.value
})
raise # do not retry
if error_class == ErrorClass.CONTEXT_OVERFLOW:
raise # caller must trim the prompt
# RETRYABLE: exponential backoff with jitter
if attempt < max_retries:
delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
logger.warning("provider_retry", extra={
"attempt": attempt + 1,
"max_retries": max_retries,
"delay_s": round(delay, 2),
"error": str(exc),
})
await asyncio.sleep(delay)
raise last_exc
Fallback chains
When retries are exhausted, a fallback chain tries alternative options before returning a user-visible error:
async def generate_with_fallback(
request: dict,
primary: ProviderClient,
fallback: ProviderClient,
cache: TTLCache | None = None,
) -> dict:
# Check cache first
if cache:
cached = cache.get(request)
if cached is not None:
return {**cached, "source": "cache"}
# Try primary
try:
result = await call_with_policy(primary, request)
if cache:
cache.set(request, result)
return result
except Exception as primary_exc:
logger.warning("primary_provider_failed_trying_fallback", extra={"error": str(primary_exc)})
# Try fallback (no cache write — fallback results may be lower quality)
try:
return await call_with_policy(fallback, {**request, "model": "gpt-4o-mini"})
except Exception as fallback_exc:
logger.error("all_providers_failed", extra={"fallback_error": str(fallback_exc)})
raise RuntimeError("All providers unavailable") from fallback_exc
Surface meaningful errors without leaking internals
Users should see a useful message. Your logs should see the full context. Application code should see a typed exception.
class AIServiceError(Exception):
def __init__(self, user_message: str, internal_context: dict):
super().__init__(user_message)
self.user_message = user_message
self.internal_context = internal_context
def log(self, logger) -> None:
logger.error("ai_service_error", extra=self.internal_context)
# In the request handler:
try:
result = await generate_with_fallback(request, primary, fallback)
except RuntimeError as exc:
service_error = AIServiceError(
user_message="I'm having trouble connecting to the AI service. Please try again in a moment.",
internal_context={"request_id": request.get("id"), "error": str(exc)},
)
service_error.log(logger)
return {"error": service_error.user_message}
Common mistakes
One retry policy for all errors. Retrying a 400 Bad Request three times burns quota and adds 3-9 seconds of unnecessary latency. Classify first, then apply the right policy.
Swallowing exceptions without logging context. except Exception: return None hides failures. Always log the error class, request ID, attempt count, and provider name before returning a fallback.
Not respecting Retry-After headers. Many 429 responses include a Retry-After header with the seconds until the rate limit resets. Ignoring it and using a fixed backoff can still cause 429s if your backoff is shorter than the reset window.
Fallback to same model with same payload. A fallback that makes the exact same call to a different client often gets the same error. When falling back, also reduce the request (smaller model, fewer tokens, simpler prompt) to improve the fallback success rate.
The practical rule
Build your error handling with three answers ready before writing any code:
- What is retryable and what is fatal for this provider?
- What is the fallback when retries are exhausted?
- What does the user see, and what does the log record?