Structured LLM Errors

Handle LLM failures with provider-aware error classification and structured recovery routing

LLM failures use provider-aware error classification with structured recovery routing, enabling intelligent retry policies, context compression, credential rotation, and model fallback strategies.

from praisonaiagents import Agent

agent = Agent(
    name="Assistant",
    instructions="Answer briefly.",
    on_error=lambda e: print(e.context.get("user_message", e.message)),
)

agent.start("Hello")

The user sends a prompt; structured errors route retries, compression, or model fallback.

Quick Start

1. Simple Agent with Structured Errors

   from praisonaiagents import Agent
   
   agent = Agent(
       name="Error Handler",
       instructions="Process user requests with automatic error recovery",
       on_error=lambda error: print(f"Error: {error.context['user_message']}")
   )
   
   result = agent.start("Hello world")

2. Advanced Error Classification

   from praisonaiagents import Agent
   from praisonaiagents.llm.error_classifier import classify_llm_error
   
   def handle_structured_error(error):
       """Handle errors with structured classification"""
       category = error.context.get("error_category", "unknown")
       user_msg = error.context.get("user_message", "Unknown error")
   
       print(f"Category: {category}")
       print(f"Message: {user_msg}")
   
       if category == "rate_limit":
           print("Rate limit hit - will retry with backoff")
       elif category == "context_overflow":
           print("Context too large - will compress and retry")
       elif category == "auth":
           print("Authentication failed - check API keys")
   
   agent = Agent(
       name="Advanced Handler",
       instructions="Process requests with detailed error handling",
       on_error=handle_structured_error
   )

Error Categories

The new classifier recognizes seven distinct error categories with specific recovery actions:

How Errors Are Classified

The classifier picks a category using three signals, in order of precedence — stable signals first, message text last.

1. HTTP status code — read from error.status_code, error.http_status, error.status, an integer error.code, or error.response.status_code.
2. Exception class name — matched case-insensitively across the class hierarchy (MRO), so provider SDK subclasses are caught without importing those SDKs.
3. Message regex — used only when steps 1 and 2 don’t match.

This makes classification reliable across providers whose error messages don’t include the expected keyword — a 429 rate limit whose message never says “rate” is still classified as rate_limit.

Status code mapping

Exception class mapping

Provider SDK exception classes are matched by class name across the MRO, so subclasses in provider SDKs are handled without importing those SDKs.

Ambiguous status codes (HTTP 400)

Some providers overload HTTP 400 — OpenAI returns context_length_exceeded as a 400, and some auth failures also come back as 400. The classifier refines a broad 400 using message signals before falling back to invalid_request.

LLMErrorClassification Structure

The new structured classification provides detailed recovery routing information:

from praisonaiagents.llm.error_classifier import classify_llm_error

classification = classify_llm_error(
    exc,                  # The exception
    provider="openai",    # "openai" | "anthropic" | "azure"
    model="gpt-4",
    prompt_tokens=0,      # optional
    context_length=0,     # optional
    retry_depth=0,        # optional
)

print(f"Category: {classification.error_category}")
print(f"Should compress: {classification.should_compress_context}")
print(f"Backoff time: {classification.backoff_seconds}")

Provider-Aware Backoff

Different providers have different rate limiting patterns, so the classifier uses provider-specific base delays:

Backoff times include ±50% jitter to prevent thundering herd problems when multiple agents hit rate limits simultaneously.

Retry-After header

The classifier reads a provider’s Retry-After signal before falling back to message parsing, in this order:

1. error.response.headers["retry-after"] (or Retry-After — case-insensitive)
2. error.headers["retry-after"]
3. error.retry_after attribute
4. Regex over the message text

A numeric value is used only when it is finite and non-negative (zero and fractional values are kept). nan, +inf, -inf and any negative number are ignored and extraction falls through to the next signal — so an invalid high-priority header does not suppress a valid lower-priority hint, and a malformed header never produces a NaN backoff or parks a scheduler job on a past/NaN instant. Finite numeric values are capped at 300 s for in-tick backoff. RFC 7231 HTTP-date Retry-After values are still parsed and their delay-from-now (if positive) is used.

Negative numbers echoed into the error message (e.g. "retry after -3600 seconds", "-90 second cooldown") are also ignored, and a later valid hint in the same message is picked up correctly. Phrasings with words between the number and "second" (e.g. "retry in 30 more seconds") are intentionally not honoured so a status code can’t be misread as a delay.

Recovery Actions

Jittered Backoff

The retry system uses exponential backoff with ±50% jitter to avoid thundering herd problems:

graph TB
    subgraph "Backoff Calculation"
        Attempt[Attempt Number] --> Exponential[base × 2^(attempt-1)]
        Exponential --> Cap[Cap at maximum]
        Cap --> Jitter[Add ±50% jitter]
        Jitter --> Result[Final delay]
    end

    subgraph "Example Delays"
        A1[Attempt 1<br/>~2.5-7.5s] 
        A2[Attempt 2<br/>~5.0-15.0s]
        A3[Attempt 3<br/>~10.0-30.0s]
    end

    classDef formula fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef example fill:#F59E0B,stroke:#7C90A0,color:#fff

    class Attempt,Exponential,Cap,Jitter,Result formula
    class A1,A2,A3 example

from praisonaiagents.llm.retry_utils import jittered_backoff

# Calculate delay with jitter
delay = jittered_backoff(attempt=1, base=5.0, cap=120.0)  # ~2.5-7.5 seconds
delay = jittered_backoff(attempt=2, base=5.0, cap=120.0)  # ~5.0-15.0 seconds
delay = jittered_backoff(attempt=3, base=5.0, cap=120.0)  # ~10.0-30.0 seconds

Advanced Classification Usage

For users who want direct access to the classifier:

from praisonaiagents.llm.error_classifier import classify_llm_error
from praisonaiagents.llm.retry_utils import calculate_backoff_with_retry_after

def custom_retry_loop(exc, provider="openai", model="gpt-4"):
    """Custom retry logic using structured classification"""
    classification = classify_llm_error(
        exc,
        provider=provider,
        model=model,
        retry_depth=0
    )

    print(f"Error category: {classification.error_category}")
    print(f"User message: {classification.user_message}")

    if not classification.is_retryable:
        print("Error is not retryable")
        return False

    if classification.should_compress_context:
        print("Context compression needed")
        # Implement context compression logic

    if classification.should_fallback_model:
        print("Model fallback suggested")
        # Implement model fallback logic

    if classification.backoff_seconds > 0:
        print(f"Waiting {classification.backoff_seconds:.1f} seconds...")
        import time
        time.sleep(classification.backoff_seconds)

    return True  # Proceed with retry

How It Works

Async Support

Structured error classification works with both sync and async agents:

import asyncio
from praisonaiagents import Agent

async def async_error_example():
    agent = Agent(
        name="Async Agent",
        instructions="Process requests asynchronously",
        on_error=lambda error: print(f"Async error: {error.context['error_category']}")
    )

    result = await agent.astart("Hello async world")
    return result

# The same structured classification applies to async flows
# Rate limiting, context compression, and other recovery actions
# are handled automatically with asyncio.sleep for delays

Typed Error Classification (AgentErrorKind)

LLM errors are automatically classified into typed AgentErrorKind categories for precise handling. For the complete system, see LLM Error Classification.

Quick Reference

• Retryable: rate_limit, overloaded, idle_timeout, auth
• Non-retryable: auth_permanent, model_not_found, format_error, context_overflow, billing
• Limited retry: unknown, empty_response

Legacy Support

The simple two-bucket classification (retryable/non-retryable) remains available for backward compatibility, but typed categories provide much more control.

LLMError Structure

The LLMError class provides structured error information:

Error Context

The on_error handler receives enhanced context information:

def enhanced_error_handler(error):
    """Access structured error information"""
    context = error.context

    # New structured fields
    category = context.get("error_category", "unknown")
    user_message = context.get("user_message", "")

    # Original fields still available
    model = error.model_name
    agent_id = error.agent_id
    retryable = error.is_retryable

    print(f"Category: {category}")
    print(f"Model: {model}")
    print(f"User-friendly message: {user_message}")
    print(f"Can retry: {retryable}")

agent = Agent(
    name="Enhanced Error Agent",
    instructions="Process with enhanced error context",
    on_error=enhanced_error_handler
)

Retry Depth Limits

The recursive recovery loop (after context compression and after transient-error backoff) is bounded by the agent’s configured retry policy:

• Default / retry= set: the loop honours RetryBackoffConfig.max_retries (e.g. default and retry=True → 3, max_retries=0 → recursion disabled).
• With retry=False: the loop falls back to 2, preserving prior behaviour.
• Context compression: triggered on the context_overflow category.

Unimplemented Recovery Actions

Some recovery actions are planned but not yet implemented in the core system:

Migration from Binary Classification

If you were previously checking error.is_retryable only, you can now access richer classification:

Before (binary classification):

def simple_handler(error):
    if error.is_retryable:
        print("Will retry")
    else:
        print("Won't retry")

After (structured classification):

def structured_handler(error):
    category = error.context.get("error_category", "unknown")
    user_msg = error.context.get("user_message", "")

    # Still works
    if error.is_retryable:
        print(f"Will retry ({category}): {user_msg}")
    else:
        print(f"Won't retry ({category}): {user_msg}")

Typed Error Categories (New):

from praisonaiagents import LLMError

try:
    response = agent._chat_completion(messages)
except LLMError as e:
    # Use typed categories instead of string parsing
    if e.error_category == "billing":
        handle_quota_exceeded()
    elif e.error_category == "auth_permanent":
        handle_invalid_api_key()
    elif e.error_category == "rate_limit":
        # Auto-retry handles this
        raise

Best Practices

def smart_error_handler(error):
    """Handle errors using structured categories"""
    category = error.context.get("error_category", "unknown")

    # Better: Use structured category
    if category == "rate_limit":
        print("Rate limit detected - will retry with provider-specific backoff")
    elif category == "context_overflow":
        print("Context too large - will compress and retry")
    elif category == "auth":
        print("Authentication failed - check credentials")

    # Avoid: Pattern matching on error message
    # if "rate limit" in error.message.lower():
    #     ...

error_counts = {
    "rate_limit": 0,
    "context_overflow": 0, 
    "auth": 0,
    "overloaded": 0,
    "model_error": 0,
    "permanent": 0,
    "unknown": 0
}

def track_structured_errors(error):
    category = error.context.get("error_category", "unknown")
    error_counts[category] += 1

    # Send structured metrics to monitoring
    send_metric(f"llm.error.{category}", 1, {
        "provider": error.context.get("provider", "unknown"),
        "model": error.model_name
    })

def advanced_recovery_handler(error):
    category = error.context.get("error_category", "unknown")
    user_msg = error.context.get("user_message", "")

    if category == "auth":
        # Custom credential rotation
        print("Attempting credential rotation...")
        rotate_api_keys()

    elif category == "overloaded":
        # Custom model fallback
        print("Primary model overloaded, switching to backup...")
        switch_to_backup_model()

    elif category == "context_overflow":
        # Log compression metrics
        log_compression_event(error.context.get("prompt_tokens", 0))

    print(f"User message: {user_msg}")

def provider_aware_handler(error):
    category = error.context.get("error_category", "unknown")
    provider = error.context.get("provider", "unknown")

    if category == "rate_limit":
        if provider == "openai":
            print("OpenAI rate limit - 60s base delay with jitter")
        elif provider == "anthropic":
            print("Anthropic rate limit - 20s base delay with jitter")
        elif provider == "azure":
            print("Azure rate limit - 45s base delay with jitter")

    elif category == "overloaded" and provider == "anthropic":
        print("Anthropic service overloaded - consider Claude alternatives")

Related