2. Error Handling for LLM APIs

Description

Before implementing retry or circuit breakers, you need to understand what can go wrong and why. LLM APIs have a different error taxonomy than traditional REST APIs: some are transient (retry resolves them), some are input errors (retry doesn't help — the same input will give the same error), and some are outages that require fallback. In this capsule you'll implement complete error handling for OpenAI: classify each error, decide whether it's worth retrying, and propagate the correct information to the rest of the system.

LLM API error taxonomy

TRANSIENT ERROR → Retry can help
├── Timeout (30s, 60s): the model took too long
│   Cause: very long prompt, busy model
│   Strategy: retry with short backoff (1-2s)
│
├── RateLimitError (429): too many requests
│   Cause: you exceeded the RPM or TPM limit
│   Strategy: retry with long backoff (30-60s) or Retry-After header
│
└── ServerError (500, 502, 503): error on OpenAI's side
    Cause: partial outage, faulty instance
    Strategy: retry with backoff, max 3-5 attempts

INPUT ERROR → No retry, change the input
├── TokenLimitExceeded: the input is too long
│   Cause: prompt + context > context window
│   Strategy: truncate the input, don't retry with the same input
│
├── BadRequest (400): invalid input
│   Cause: incorrect message format, invalid parameters
│   Strategy: fix the code, no retry
│
└── ContentPolicyViolation: content blocked by moderation
    Cause: the input or expected output violate the policies
    Strategy: apply pre-LLM guardrails (Module 4), no retry

PERMANENT ERROR → No retry, useless
├── AuthenticationError (401): invalid API key
│   Cause: expired key, wrong key
│   Strategy: configuration alert, no retry
│
└── PermissionError (403): no access to the resource/model
    Cause: the plan doesn't have access to the requested model
    Strategy: change the model in config, no retry

OUTPUT ERROR → Problem with the response, not the request
├── InvalidJSON: the LLM returned text instead of JSON
│   Cause: the model forgot the format, ambiguous prompt
│   Strategy: retry (max 1-2 attempts), fallback to default
│
└── TruncatedResponse: finish_reason="length"
    Cause: max_tokens insufficient for the full response
    Strategy: increase max_tokens, don't retry with the same config

OpenAI library errors

# The Python openai client exceptions

from openai import (
    OpenAI,
    # Exception hierarchy:
    APIError,              # Base for all API errors
    APIConnectionError,    # Couldn't connect (network, DNS)
    APITimeoutError,       # Request timeout
    RateLimitError,        # 429: rate limit exceeded
    AuthenticationError,   # 401: invalid API key
    PermissionDeniedError, # 403: no access
    NotFoundError,         # 404: resource not found
    UnprocessableEntityError,  # 422: unprocessable input
    InternalServerError,   # 5xx: OpenAI server error
    BadRequestError,       # 400: invalid request
)

# APIError has useful attributes:
# e.status_code: int (the HTTP status code)
# e.message: str (the error message)
# e.request_id: str (the request ID in OpenAI — useful for support)

Error classification function

# src/infrastructure/error_classifier.py
from openai import (
    APIError, APITimeoutError, RateLimitError, APIConnectionError,
    AuthenticationError, PermissionDeniedError, BadRequestError,
    InternalServerError, UnprocessableEntityError
)
from enum import Enum
from dataclasses import dataclass
from typing import Optional

class ErrorCategory(Enum):
    TRANSIENT = "transient"           # Retry can help
    INPUT_ERROR = "input_error"       # Change the input
    AUTH_ERROR = "auth_error"         # Config problem
    OUTAGE = "outage"                 # Service down
    OUTPUT_ERROR = "output_error"     # Problem with the response
    UNKNOWN = "unknown"               # Not classified

@dataclass
class ClassifiedError:
    category: ErrorCategory
    should_retry: bool
    retry_after_seconds: Optional[float]  # If the API indicates when to retry
    user_message: str                     # Message appropriate for the user
    log_level: str                        # "warning", "error", "critical"
    original_error: Exception

def classify_error(e: Exception) -> ClassifiedError:
    """
    Classify an LLM API error and determine the handling strategy.
    """
    
    if isinstance(e, APITimeoutError):
        return ClassifiedError(
            category=ErrorCategory.TRANSIENT,
            should_retry=True,
            retry_after_seconds=None,
            user_message="The service took too long. Retrying...",
            log_level="warning",
            original_error=e
        )
    
    if isinstance(e, RateLimitError):
        # Try to extract Retry-After from the header if available
        retry_after = _extract_retry_after(e)
        return ClassifiedError(
            category=ErrorCategory.TRANSIENT,
            should_retry=True,
            retry_after_seconds=retry_after or 30.0,
            user_message="Service temporarily saturated. Retrying...",
            log_level="warning",
            original_error=e
        )
    
    if isinstance(e, APIConnectionError):
        return ClassifiedError(
            category=ErrorCategory.OUTAGE,
            should_retry=True,
            retry_after_seconds=5.0,
            user_message="Couldn't connect to the service.",
            log_level="error",
            original_error=e
        )
    
    if isinstance(e, AuthenticationError):
        return ClassifiedError(
            category=ErrorCategory.AUTH_ERROR,
            should_retry=False,
            retry_after_seconds=None,
            user_message="Service configuration error.",
            log_level="critical",  # Immediate alert — broken config
            original_error=e
        )
    
    if isinstance(e, PermissionDeniedError):
        return ClassifiedError(
            category=ErrorCategory.AUTH_ERROR,
            should_retry=False,
            retry_after_seconds=None,
            user_message="No access to the requested model.",
            log_level="error",
            original_error=e
        )
    
    if isinstance(e, BadRequestError):
        # Distinguish between "prompt too long" and "malformed request"
        if e.status_code == 400 and "context_length" in str(e.message).lower():
            return ClassifiedError(
                category=ErrorCategory.INPUT_ERROR,
                should_retry=False,
                retry_after_seconds=None,
                user_message="The text is too long to process.",
                log_level="warning",
                original_error=e
            )
        return ClassifiedError(
            category=ErrorCategory.INPUT_ERROR,
            should_retry=False,
            retry_after_seconds=None,
            user_message="Invalid request.",
            log_level="error",
            original_error=e
        )
    
    if isinstance(e, InternalServerError):
        return ClassifiedError(
            category=ErrorCategory.TRANSIENT,
            should_retry=True,
            retry_after_seconds=10.0,
            user_message="The service is experiencing problems.",
            log_level="error",
            original_error=e
        )
    
    if isinstance(e, APIError):
        # Generic APIError — classify by status_code
        if e.status_code == 429:
            return ClassifiedError(
                category=ErrorCategory.TRANSIENT,
                should_retry=True,
                retry_after_seconds=30.0,
                user_message="Service temporarily saturated.",
                log_level="warning",
                original_error=e
            )
        if e.status_code and e.status_code >= 500:
            return ClassifiedError(
                category=ErrorCategory.TRANSIENT,
                should_retry=True,
                retry_after_seconds=10.0,
                user_message="Server error.",
                log_level="error",
                original_error=e
            )
    
    # Unknown error
    return ClassifiedError(
        category=ErrorCategory.UNKNOWN,
        should_retry=False,
        retry_after_seconds=None,
        user_message="Unexpected error.",
        log_level="error",
        original_error=e
    )

def _extract_retry_after(e: RateLimitError) -> Optional[float]:
    """Extract the Retry-After value if available."""
    try:
        # The header may be in the response
        if hasattr(e, "response") and e.response:
            retry_after = e.response.headers.get("Retry-After")
            if retry_after:
                return float(retry_after)
    except (AttributeError, ValueError):
        pass
    return None

OpenAIProvider with integrated classification

# src/infrastructure/openai_provider.py (updated)
import time
import structlog
from openai import APIError, APITimeoutError, RateLimitError
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
from src.infrastructure.error_classifier import classify_error, ErrorCategory
from src.logging_config import calculate_cost

log = structlog.get_logger()

class OpenAIProvider:
    def __init__(self, client, model: str, temperature: float,
                 max_tokens: int, seed: int = None):
        self._client = client
        self._model = model
        self._temperature = temperature
        self._max_tokens = max_tokens
        self._seed = seed
    
    def complete(self, messages: list[dict], **kwargs) -> str:
        """
        Make a call to the OpenAI API.
        
        Classifies errors before propagating them, converting
        OpenAI-specific exceptions into LLMProviderError
        with classification information.
        """
        start = time.time()
        
        try:
            response = self._client.chat.completions.create(
                model=self._model,
                messages=messages,
                temperature=self._temperature,
                max_tokens=self._max_tokens,
                seed=self._seed,
                **kwargs
            )
            
            duration_ms = (time.time() - start) * 1000
            cost_usd = calculate_cost(
                self._model,
                response.usage.prompt_tokens,
                response.usage.completion_tokens
            )
            
            log.info(
                "llm_call_completed",
                model=self._model,
                input_tokens=response.usage.prompt_tokens,
                output_tokens=response.usage.completion_tokens,
                cost_usd=cost_usd,
                duration_ms=round(duration_ms, 1),
                finish_reason=response.choices[0].finish_reason
            )
            
            # Warn if the response was truncated
            if response.choices[0].finish_reason == "length":
                log.warning(
                    "response_truncated_by_max_tokens",
                    max_tokens=self._max_tokens,
                    output_tokens=response.usage.completion_tokens
                )
            
            return response.choices[0].message.content
        
        except Exception as e:
            duration_ms = (time.time() - start) * 1000
            classified = classify_error(e)
            
            # Log with the appropriate level
            log_fn = getattr(log, classified.log_level)
            log_fn(
                "llm_call_failed",
                error_type=type(e).__name__,
                error_category=classified.category.value,
                should_retry=classified.should_retry,
                model=self._model,
                duration_ms=round(duration_ms, 1)
            )
            
            raise LLMProviderError(
                message=classified.user_message,
                original_error=e,
                category=classified.category,
                should_retry=classified.should_retry,
                retry_after=classified.retry_after_seconds
            )

Enriched LLMProviderError

# src/infrastructure/llm_provider.py (updated)
from src.infrastructure.error_classifier import ErrorCategory
from typing import Optional

class LLMProviderError(Exception):
    """
    Generic LLM provider error with classification information.
    
    Wraps provider-specific errors and adds:
    - category: error type for retry/fallback routing
    - should_retry: whether it's worth retrying
    - retry_after: how many seconds to wait before retrying
    """
    def __init__(
        self,
        message: str,
        original_error: Exception = None,
        category: ErrorCategory = ErrorCategory.UNKNOWN,
        should_retry: bool = False,
        retry_after: Optional[float] = None
    ):
        super().__init__(message)
        self.original_error = original_error
        self.category = category
        self.should_retry = should_retry
        self.retry_after = retry_after
    
    @property
    def is_transient(self) -> bool:
        return self.category == ErrorCategory.TRANSIENT
    
    @property
    def is_auth_error(self) -> bool:
        return self.category == ErrorCategory.AUTH_ERROR
    
    @property
    def is_input_error(self) -> bool:
        return self.category == ErrorCategory.INPUT_ERROR

Handling output errors: malformed responses

# src/processing/sentiment_parser.py (with robust error handling)
import json
import re
import structlog
from typing import Optional

log = structlog.get_logger()

class ParseResult:
    """Parsing result with quality information."""
    def __init__(self, data: dict, parse_strategy: str, warnings: list = None):
        self.data = data
        self.parse_strategy = parse_strategy  # "direct", "regex", "fallback"
        self.warnings = warnings or []
        self.is_fallback = parse_strategy == "fallback"

def parse_with_full_error_handling(raw: str, request_id: str = None) -> ParseResult:
    """
    Parse the LLM output with multiple strategies and full logging.
    """
    # Strategy 1: direct JSON
    stripped = raw.strip()
    if stripped.startswith("{"):
        try:
            data = json.loads(stripped)
            return ParseResult(data, "direct")
        except json.JSONDecodeError:
            pass
    
    # Strategy 2: JSON in a markdown code block
    match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
    if match:
        try:
            data = json.loads(match.group(1))
            log.warning("json_in_markdown", request_id=request_id)
            return ParseResult(data, "markdown_extraction",
                             warnings=["JSON extracted from markdown block"])
        except json.JSONDecodeError:
            pass
    
    # Strategy 3: Look for any JSON object
    match = re.search(r"\{[^{}]+\}", raw, re.DOTALL)
    if match:
        try:
            data = json.loads(match.group())
            log.warning("json_regex_extracted", request_id=request_id,
                       raw_preview=raw[:100])
            return ParseResult(data, "regex_extraction",
                             warnings=["JSON extracted with regex"])
        except json.JSONDecodeError:
            pass
    
    # Strategy 4: Total fallback
    log.error(
        "json_parse_completely_failed",
        request_id=request_id,
        raw_preview=raw[:200],
        raw_length=len(raw)
    )
    return ParseResult(
        data={"sentiment": "unknown", "score": 0.0, "confidence": 0.0},
        parse_strategy="fallback",
        warnings=["Couldn't extract valid JSON — using fallback"]
    )

Error classifier tests

# tests/unit/test_error_classifier.py
import pytest
from unittest.mock import MagicMock
from openai import APITimeoutError, RateLimitError, AuthenticationError
from src.infrastructure.error_classifier import classify_error, ErrorCategory

def test_timeout_is_transient_and_retryable():
    error = APITimeoutError.__new__(APITimeoutError)
    result = classify_error(error)
    assert result.category == ErrorCategory.TRANSIENT
    assert result.should_retry is True

def test_rate_limit_has_retry_delay():
    error = MagicMock(spec=RateLimitError)
    error.response = None
    result = classify_error(error)
    assert result.category == ErrorCategory.TRANSIENT
    assert result.should_retry is True
    assert result.retry_after_seconds is not None
    assert result.retry_after_seconds > 0

def test_auth_error_is_not_retryable():
    error = MagicMock(spec=AuthenticationError)
    result = classify_error(error)
    assert result.should_retry is False
    assert result.log_level == "critical"  # Immediate alert

def test_server_error_is_transient():
    from openai import InternalServerError
    error = MagicMock(spec=InternalServerError)
    result = classify_error(error)
    assert result.should_retry is True

Exercises

Exercise 1: Decision tree

For each error, draw the decision tree (retry/no retry, fallback/no fallback):

1. OpenAI returns 500 for the first time
2. OpenAI returns 401
3. The LLM returns an invalid JSON response for the third time in a row
4. The input has 150,000 tokens for a model with a 128K limit

Exercise 2: Complete the classify_error

The classify_error function doesn't handle UnprocessableEntityError (422). Write the missing block:

if isinstance(e, UnprocessableEntityError):
    return ClassifiedError(
        category=???,
        should_retry=???,
        retry_after_seconds=???,
        user_message=???,
        log_level=???,
        original_error=e
    )

if isinstance(e, UnprocessableEntityError):
    return ClassifiedError(
        category=ErrorCategory.INPUT_ERROR,
        should_retry=False,
        retry_after_seconds=None,
        user_message="Couldn't process the request.",
        log_level="warning",
        original_error=e
    )

Exercise 3: Improve the JSON parser

The parse_with_full_error_handling doesn't handle the case where the LLM responds with valid JSON but is missing required fields (for example, it has sentiment but not score). Write a post-parse validation function:

REQUIRED_FIELDS = {"sentiment", "score", "confidence"}

def validate_parsed_result(data: dict, request_id: str = None) -> ParseResult:
    """Validate that the parsed JSON has all the required fields."""
    missing = REQUIRED_FIELDS - set(data.keys())
    if not missing:
        return ParseResult(data, "validated")
    
    # Fill missing fields with defaults
    defaults = {"sentiment": "unknown", "score": 0.0, "confidence": 0.0}
    for field in missing:
        data[field] = defaults[field]
    
    log.warning("json_missing_fields", missing=list(missing), request_id=request_id)
    return ParseResult(data, "partial_with_defaults",
                      warnings=[f"Missing fields filled in: {missing}"])

Exercise 4: Test for unknown error

Write a test that verifies classify_error correctly handles a completely unknown exception (for example, a generic RuntimeError):

def test_unknown_error_is_not_retryable():
    """Unknown errors are not retried for safety."""
    error = RuntimeError("Something completely unexpected")
    result = classify_error(error)
    
    assert result.category == ErrorCategory.UNKNOWN
    assert result.should_retry is False
    assert result.log_level == "error"
    assert result.original_error is error

Troubleshooting

"My classify_error doesn't recognize OpenAI's exceptions"

Verify that you're importing from the correct module. The openai library in version 1.x changed the exception hierarchy:

# ❌ Incorrect (v0.x, deprecated)
from openai.error import RateLimitError

# ✅ Correct (v1.x+)
from openai import RateLimitError

Run pip show openai to check your version. You need >=1.0.

"The LLMProviderError doesn't have category — I get AttributeError"

Make sure you're creating the error with all the fields. If some old code does raise LLMProviderError("msg") without the classification arguments, category will be ErrorCategory.UNKNOWN by default, but retry_after will be None. Find all the raise LLMProviderError in your code and update them.

"The parser always falls to the fallback"

Check what the LLM is actually responding. Add a temporary log of the raw output:

log.debug("raw_llm_output", raw=raw[:500], length=len(raw))

Common causes: the model responds with an explanation before the JSON, uses single quotes instead of double quotes, or includes trailing commas that aren't valid JSON.

"How do I know if my error is from the openai library or the network?"

APIConnectionError is a network error (DNS, firewall, proxy). APITimeoutError is an HTTP request timeout. Both are transient, but APIConnectionError is more serious because it may indicate an infrastructure problem on your end (not OpenAI's). Check your connectivity before assuming it's OpenAI's fault.

Summary

• Classify before handling: not every error deserves a retry
• Transients: timeout, 429, 5xx → retry with backoff
• Input errors: token limit, 400, content policy → fix the input, no retry
• Auth errors: 401, 403 → configuration alert, no retry
• Output errors: invalid JSON → retry max 1 time, then fallback
• Enriched LLMProviderError: carries the category for retry/fallback routing

Additional resources

1. OpenAI Error Handling Guide — Official documentation
2. OpenAI API Reference — Errors — Complete list of errors
3. HTTP Status Codes — Reference
4. Retry-After Header — How to use it