Created
December 1, 2025 01:05
-
-
Save dsfaccini/667e798f97436befdb2520daadc7f2cf to your computer and use it in GitHub Desktop.
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`.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """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()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment