Error Recovery Patterns with Pydantic AI: Complete Guide

Introduction to Error Recovery Patterns with Pydantic AI

Building production-grade applications with Large Language Models (LLMs) is rarely a "call once and done" affair. Networks fail, models hallucinate, rate limits kick in, and structured outputs sometimes come back malformed. Without a deliberate strategy for handling these failures, your AI-powered features will degrade silently or crash at the worst possible moment. Pydantic AI, a framework that brings type safety and structured validation to LLM workflows, offers several primitives that make robust error recovery not just possible, but ergonomic.

This guide walks through the most important error recovery patterns you can implement with Pydantic AI: from basic retries and validation fallbacks to multi-model fallbacks, tool-call recovery, and circuit breakers. By the end, you will have a toolkit of patterns you can drop directly into your own agents.

Why Error Recovery Matters

LLM applications face a broader and stranger set of failure modes than typical web services. A few of the most common include:

• Transient infrastructure errors — connection resets, DNS hiccups, 5xx responses from model providers.
• Rate limiting — 429 responses that require backoff and retry.
• Validation failures — the model returns JSON that does not match your expected schema, or returns text that cannot be parsed at all.
• Tool-call errors — the model invokes a tool with the wrong arguments, or the tool itself raises an exception.
• Context window overflow — prompts that exceed the model's token limit.
• Timeouts — long-running generations that never complete.

Each of these requires a different recovery strategy. Pydantic AI's design — built around typed agents, structured outputs, and a dependency injection system — gives you the hooks you need to implement all of them cleanly.

Prerequisites and Setup

Before diving into the patterns, make sure you have a working environment. Install Pydantic AI and an optional HTTP library for retries:

pip install pydantic-ai tenacity

You will also need at least one model provider API key. The examples below use OpenAI, but the patterns apply to any provider Pydantic AI supports.

export OPENAI_API_KEY="sk-..."

Pattern 1: Basic Retries with Exponential Backoff

The simplest recovery pattern is retrying transient failures. Pydantic AI agents will surface provider errors as exceptions, so you can wrap your agent calls in a retry loop. The tenacity library is a clean way to do this with exponential backoff and jitter.

from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from openai import APIConnectionError, RateLimitError, APITimeoutError

model = OpenAIModel("gpt-4o-mini")
agent = Agent(model, system_prompt="You are a helpful assistant.")

@retry(
    retry=retry_if_exception_type((APIConnectionError, RateLimitError, APITimeoutError)),
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=30),
    reraise=True,
)
async def run_with_retry(prompt: str) -> str:
    result = await agent.run(prompt)
    return result.output

# Usage
answer = await run_with_retry("Summarize the rules of chess in two sentences.")
print(answer)

Key points about this pattern:

• Only retry exceptions that are genuinely transient. Don't retry ValidationError or BadRequestError — those will fail the same way every time unless you change the input.
• Use jitter to avoid thundering-herd effects when many clients retry simultaneously.
• Cap both the number of attempts and the maximum wait time so a single call cannot hang forever.

Pattern 2: Recovering from Validation Failures

One of Pydantic AI's signature features is structured output: you declare a result type, and the agent ensures the model's response parses into that type. When the model produces output that does not validate, Pydantic AI raises a ValidationError. The recovery strategy here is different from a network retry — you want to give the model feedback about what went wrong and ask it to try again.

Pydantic AI supports this natively through its result_retries option. When set, the framework automatically feeds the validation error back to the model and asks it to correct its output.

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

class WeatherReport(BaseModel):
    location: str = Field(description="City and country")
    temperature_c: float = Field(description="Temperature in Celsius")
    conditions: str = Field(description="Brief weather description")
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence 0-1")

model = OpenAIModel("gpt-4o-mini")

# result_retries=3 means the agent will ask the model to fix validation
# errors up to 3 times before giving up.
agent = Agent(
    model,
    result_type=WeatherReport,
    result_retries=3,
    system_prompt="You produce structured weather reports. Always respond with valid JSON.",
)

async def get_report(city: str) -> WeatherReport:
    try:
        result = await agent.run(f"Give me a weather report for {city}.")
        return result.output
    except Exception as e:
        # After retries are exhausted, handle gracefully.
        print(f"Failed to produce a valid report: {e}")
        raise

report = await get_report("Tokyo, Japan")
print(report.model_dump_json(indent=2))

This pattern is powerful because the model sees the exact validation error message. It can learn, for example, that confidence must be between 0 and 1, and adjust its next attempt accordingly. For complex schemas, this self-correction loop is often more reliable than a single shot.

Pattern 3: Multi-Model Fallbacks

Sometimes a single model is not enough. A cheaper model may fail on hard inputs, or a specific provider may have an outage. A robust setup tries a primary model and falls back to one or more secondary models if the primary fails.

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.models.gemini import GeminiModel

class Summary(BaseModel):
    title: str
    body: str

primary_agent = Agent(
    OpenAIModel("gpt-4o"),
    result_type=Summary,
    result_retries=2,
    system_prompt="Summarize the input text concisely.",
)

fallback_agent_1 = Agent(
    AnthropicModel("claude-3-5-sonnet-latest"),
    result_type=Summary,
    result_retries=2,
    system_prompt="Summarize the input text concisely.",
)

fallback_agent_2 = Agent(
    GeminiModel("gemini-1.5-pro"),
    result_type=Summary,
    result_retries=2,
    system_prompt="Summarize the input text concisely.",
)

async def summarize_with_fallback(text: str) -> Summary:
    agents = [primary_agent, fallback_agent_1, fallback_agent_2]
    last_error = None
    for agent in agents:
        try:
            result = await agent.run(f"Summarize: {text}")
            return result.output
        except Exception as e:
            print(f"Agent {agent.model} failed: {e}")
            last_error = e
    raise RuntimeError(f"All models failed. Last error: {last_error}")

This pattern gives you resilience against provider outages and lets you trade cost for quality: try the cheap model first, fall back to a more capable one only when needed.

Pattern 4: Tool-Call Error Recovery

Agents that use tools face an additional failure mode: the model may call a tool with invalid arguments, or the tool itself may raise an exception. Pydantic AI lets you handle both gracefully by raising a ModelRetry exception from within a tool. When you do, Pydantic AI catches it and sends the error message back to the model so it can try again with corrected arguments.

from pydantic_ai import Agent, ModelRetry
from pydantic_ai.models.openai import OpenAIModel

agent = Agent(OpenAIModel("gpt-4o-mini"), system_prompt="You look up user accounts.")

@agent.tool
async def get_user(ctx, user_id: int) -> dict:
    """Look up a user by ID. IDs must be positive integers."""
    if user_id <= 0:
        # Tell the model what went wrong so it can correct itself.
        raise ModelRetry(f"user_id must be a positive integer, got {user_id}")
    # Simulate a database lookup
    if user_id == 42:
        return {"id": 42, "name": "Ada Lovelace", "email": "ada@example.com"}
    raise ModelRetry(f"No user found with id {user_id}. Try a different ID.")

result = await agent.run("Look up user 42 and tell me their email.")
print(result.output)

Using ModelRetry is preferable to letting raw exceptions bubble up. The model gets a human-readable explanation and can adjust its behavior — for example, by trying a different ID or asking the user for clarification. Combined with result_retries, this creates a tight feedback loop where the agent self-corrects both tool calls and final outputs.

Pattern 5: Circuit Breakers for Dependent Services

When your agent depends on external services — a vector database, a search API, a payment gateway — you do not want a single failing service to drag down your whole application. A circuit breaker wraps each dependency and stops calling it after a threshold of failures, returning a fallback or error immediately instead.

import time
from pydantic_ai import Agent, ModelRetry
from pydantic_ai.models.openai import OpenAIModel

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: float = 60.0):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure_time = 0.0
        self.state = "closed"  # closed, open, half-open

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"

    def allow_request(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open allows one trial request

search_breaker = CircuitBreaker(failure_threshold=3, reset_timeout=30.0)

agent = Agent(OpenAIModel("gpt-4o-mini"), system_prompt="You answer questions using web search.")

@agent.tool
async def web_search(ctx, query: str) -> str:
    """Search the web for information."""
    if not search_breaker.allow_request():
        raise ModelRetry("Web search is temporarily unavailable. Answer from your own knowledge if possible.")
    try:
        # Simulated search call
        result = await call_search_api(query)
        search_breaker.record_success()
        return result
    except Exception as e:
        search_breaker.record_failure()
        raise ModelRetry(f"Search failed: {e}. Try a simpler query or answer from general knowledge.")

async def call_search_api(query: str) -> str:
    # Replace with real implementation
    return f"Results for: {query}"

The circuit breaker here integrates cleanly with ModelRetry. When the breaker is open, the tool immediately signals the model to use a different strategy, rather than waiting for yet another timeout.

Pattern 6: Graceful Degradation with Cached or Default Responses

Not every failure needs to be retried. Sometimes the right move is to return a cached result, a default value, or a "best effort" response. This is especially true for user-facing features where a slightly stale or generic answer is better than an error screen.

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
import json
from pathlib import Path

class Recommendation(BaseModel):
    product: str
    reason: str

agent = Agent(
    OpenAIModel("gpt-4o-mini"),
    result_type=Recommendation,
    result_retries=2,
    system_prompt="You recommend products based on user preferences.",
)

cache_path = Path("last_recommendation.json")

async def recommend(user_input: str) -> Recommendation:
    try:
        result = await agent.run(user_input)
        recommendation = result.output
        # Cache successful results for fallback use
        cache_path.write_text(recommendation.model_dump_json())
        return recommendation
    except Exception as e:
        print(f"Generation failed: {e}")
        if cache_path.exists():
            print("Returning cached recommendation.")
            return Recommendation.model_validate_json(cache_path.read_text())
        # Final fallback: a generic recommendation
        return Recommendation(
            product="Generic Best Seller",
            reason="Our most popular item, recommended when personalization is unavailable.",
        )

This pattern layers recovery: first try the live model, then fall back to a cache, then fall back to a hardcoded default. Each layer increases in staleness but guarantees the user always gets a response.

Pattern 7: Combining Patterns with a Resilient Runner

In real applications, you will want to combine several of these patterns. Below is a more complete example that ties together retries, validation recovery, model fallback, and graceful degradation into a single resilient runner function.

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.models.anthropic import AnthropicModel
from openai import APIConnectionError, RateLimitError, APITimeoutError
from pydantic import BaseModel

class Answer(BaseModel):
    text: str
    sources: list[str] = []

def make_agent(model) -> Agent:
    return Agent(
        model,
        result_type=Answer,
        result_retries=3,
        system_prompt="You are a precise research assistant. Cite sources when possible.",
    )

agents = [
    make_agent(OpenAIModel("gpt-4o-mini")),
    make_agent(AnthropicModel("claude-3-5-sonnet-latest")),
]

@retry(
    retry=retry_if_exception_type((APIConnectionError, RateLimitError, APITimeoutError)),
    stop=stop_after_attempt(4),
    wait=wait_exponential_jitter(initial=1, max=20),
    reraise=True,
)
async def call_agent(agent: Agent, prompt: str) -> Answer:
    result = await agent.run(prompt)
    return result.output

async def resilient_answer(prompt: str) -> Answer:
    errors = []
    for agent in agents:
        try:
            return await call_agent(agent, prompt)
        except Exception as e:
            errors.append(f"{type(agent.model).__name__}: {e}")
            continue
    # All agents failed — degrade gracefully
    return Answer(
        text="I'm unable to generate an answer right now. Please try again shortly.",
        sources=[],
    )

# Run it
answer = await resilient_answer("What are the main causes of inflation?")
print(answer.model_dump_json(indent=2))

This runner is a template you can adapt. Swap in your own result types, add tools with ModelRetry, insert circuit breakers around external dependencies, and tune retry parameters to match your latency and cost budgets.

Best Practices

• Retry only idempotent, transient failures. Retrying a validation error without changing the input is wasteful. Use result_retries for validation issues and tenacity for network issues.
• Always set bounds. Cap retry counts, maximum wait times, and total timeouts. Unbounded retries can turn a small outage into a cascading failure.
• Log every recovery event. Retries and fallbacks are signals that something is wrong. Log them with enough context to diagnose the root cause later.
• Use ModelRetry for tool errors. It keeps the agent in control and lets the model adapt, rather than crashing the whole run.
• Design schemas to be forgiving. Use Optional, defaults, and Field constraints that match what models actually produce. Overly strict schemas cause unnecessary validation failures.
• Test failure paths explicitly. Inject faults — bad JSON, timeouts, tool exceptions — in your test suite to verify your recovery logic works.
• Monitor retry rates in production. A sudden spike in retries often precedes a larger incident. Alert on it.
• Prefer structured outputs. Pydantic AI's typed results make validation failures detectable and recoverable, which is far better than silently accepting malformed text.

Conclusion

Error recovery is not an afterthought in LLM applications — it is a core design concern. Pydantic AI gives you the building blocks to handle the full spectrum of failures: result_retries for validation self-correction, ModelRetry for tool-call recovery, typed agents for clean multi-model fallbacks, and standard Python exception handling for transient infrastructure issues. By combining these primitives thoughtfully — with bounded retries, circuit breakers for dependencies, and graceful degradation paths — you can build AI features that stay responsive and useful even when individual components fail. Start with the resilient runner template above, adapt it to your domain, and treat every recovery event as a signal to improve your prompts, schemas, and tool designs over time.