This is an example of how to implement automatic fallback for a Pydantic AI agent that repeatedly fails to produce valid output. Pydantic AI offers a `FallbackModel` that falls back on network errors, so this is an example of how to fall back on `ValidationErrors`.

pydantic-ai-fall-back-on-validation-errors.py

"""Example: Fallback on validation errors.

The FallbackModel only falls back on API errors (4xx/5xx), not validation errors.
This example shows how to fall back to another model when validation retries are exhausted.
"""
import asyncio
from collections.abc import Sequence

import logfire
from pydantic import BaseModel, Field, ValidationError

from pydantic_ai import Agent, Tool
from pydantic_ai.agent import AgentRunResult
from pydantic_ai.exceptions import UnexpectedModelBehavior
from pydantic_ai.messages import ModelMessage
from pydantic_ai.models import KnownModelName, Model
from pydantic_ai.output import NativeOutput

logfire.configure(send_to_logfire='if-token-present')
logfire.instrument_httpx(capture_all=True)


class StandupSummary(BaseModel):
    today_focus: str = Field(description='Main focus for today')
    yesterday_summary: str
    blockers: list[str]
    funniest_incident_title: str | None = None

async def get_recent_incidents(limit: int = 5) -> list[dict[str, str]]:
    """Return a few recent incidents from Logfire (title, service, severity)."""
    incidents = [
        {
            'title': 'Checkout 500s during peak traffic',
            'service': 'checkout-service',
            'severity': 'high',
        },
        {
            'title': 'Slow DB queries on user profile',
            'service': 'user-service',
            'severity': 'medium',
        },
        {
            'title': 'Background job retry storm',
            'service': 'worker-service',
            'severity': 'medium',
        },
        {
            'title': 'Suspicious spike in 4xx from API clients',
            'service': 'public-api',
            'severity': 'low',
        },
        {
            'title': 'Feature flag misconfiguration (again)',
            'service': 'feature-flags',
            'severity': 'low',
        },
    ]
    return incidents[:limit]

async def run_with_fallback[DepsT, OutputT](
    agent: Agent[DepsT, OutputT],
    prompt: str,
    models: Sequence[Model | KnownModelName | str],
    *,
    deps: DepsT = None,
    message_history: list[ModelMessage] | None = None,
) -> AgentRunResult[OutputT]:
    """Run an agent with fallback on validation errors (after retries are exhausted)."""
    errors: list[Exception] = []

    for model in models:
        try:
            return await agent.run(
                prompt,
                model=model,
                deps=deps,
                message_history=message_history,
            )
        except UnexpectedModelBehavior as e:
            if isinstance(e.__cause__, ValidationError):
                # Validation retries exhausted - try next model
                errors.append(e)
                continue
            raise  # Re-raise non-validation UMB errors

    raise ExceptionGroup('All models failed validation', errors)


agent: Agent[None, StandupSummary] = Agent(
    output_type=NativeOutput(StandupSummary),  # some models may not support both `NativeOutput` and `Tool.strict=True`
    tools=[Tool(get_recent_incidents, strict=True)],
    retries=3,
    instructions=(
        'You are an engineering manager who secretly prepares for standup by reading Logfire traces. '
        'Turn recent production activity into a concise standup summary for the team. '
        'Keep it honest but slightly roast-y, in a friendly way.'
    ),
)

MODELS = [
    'openrouter:openai/gpt-oss-120b:basent',  # 4fp quantized, not great structured output
    'openrouter:gpt-5-mini',  # reliable fallback
]

PROMPT = "Generate today's standup summary for the backend team."

async def main():
    with logfire.span('Standup Summary') as span:
        result = await run_with_fallback(agent, PROMPT, MODELS)
        span.set_attribute('result', result.output)
        print(f'Final Result: {result.output}')


if __name__ == '__main__':
    asyncio.run(main())