agent-builder-pydantic-ai（Agent 生成器 pydantic ai）

Pydantic AI Agent Builder

Purpose

Create production-ready AI agents with type safety, automatic validation, and minimal boilerplate using Pydantic AI framework.

When to Use

• Building FastAPI backend with AI capabilities
• Need strict type checking and validation
• Want auto-retry on malformed LLM responses
• Creating agents with custom tools

Architecture Pattern

Project Structure

backend/
├── agents/
│   ├── __init__.py
│   ├── base_agent.py          # Base agent class
│   └── [feature]_agent.py     # Feature-specific agents
├── tools/
│   ├── __init__.py
│   └── [tool_name].py         # Tool definitions
└── config/
    └── agent_config.py        # Agent configurations

Installation

pip install pydantic-ai httpx pydantic python-dotenv

Base Agent Pattern

from pydantic_ai import Agent
from pydantic import BaseModel
import os

class AgentResponse(BaseModel):
    result: str
    confidence: float

agent = Agent(
    model='openrouter:openai/gpt-4o',
    output_type=AgentResponse,
    tools=[tool1, tool2],
    system_prompt="You are a helpful AI assistant."
)

# Usage
result = await agent.run("user message")

Integration with OpenRouter

Setup

import os
from pydantic_ai.models import OpenRouterModel

model = OpenRouterModel(
    name='openai/gpt-4o',
    api_key=os.getenv('OPENROUTER_API_KEY'),
    http_referer=os.getenv('FRONTEND_URL')
)

Environment Variables

OPENROUTER_API_KEY=sk-or-v1-...
FRONTEND_URL=http://localhost:3000

Tool Definition Pattern

from pydantic import BaseModel, Field
from pydantic_ai import Agent, Tool

class GenerateImageArgs(BaseModel):
    prompt: str = Field(description="Image description")
    num_images: int = Field(ge=1, le=10, default=1)

async def generate_image_tool(args: GenerateImageArgs) -> dict:
    # Your implementation
    return {"images": [...]}

# Register tool
agent.add_tool(
    Tool(
        name="generate_image",
        description="Generate images using AI",
        parameters=GenerateImageArgs,
        execute=generate_image_tool
    )
)

Streaming Pattern

async def stream_response(agent, message):
    async for chunk in agent.stream(message):
        yield {
            "type": "text" if isinstance(chunk, str) else "tool_call",
            "content": chunk
        }

Error Handling & Retry

from pydantic_ai import Agent, RetryConfig

agent = Agent(
    model='openrouter:openai/gpt-4o',
    retry_config=RetryConfig(
        max_retries=3,
        retry_on=[ValidationError, TimeoutError]
    )
)

# Auto-retry on validation errors
try:
    result = await agent.run("user message")
except ValidationError as e:
    # Will retry automatically
    logger.error(f"Validation failed after retries: {e}")

FastAPI Integration

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    history: list = []

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    try:
        result = await agent.run(
            request.message,
            context={"history": request.history}
        )
        return {"response": result.result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Testing Pattern

import pytest
from pydantic_ai import Agent

@pytest.mark.asyncio
async def test_agent_response():
    agent = Agent(
        model='openrouter:openai/gpt-4o',
        system_prompt="You are a test assistant"
    )

    result = await agent.run("Say hello")
    assert "hello" in result.lower()

Best Practices

1. Type Safety: Always define Pydantic models for inputs/outputs
2. Dependency Injection: Use FastAPI-style DI for tools
3. Auto-Retry: Configure retry logic for robustness
4. Logging: Add structured logging for debugging
5. Testing: Write pytest tests for agent behaviors
6. Validation: Let Pydantic handle validation automatically
7. Context: Pass context dict for stateful conversations

Example: Complete Agent

from pydantic_ai import Agent, Tool
from pydantic import BaseModel, Field
import os

# Output type
class ChatResponse(BaseModel):
    message: str
    tool_used: str | None = None
    confidence: float = Field(ge=0, le=1)

# Tool definition
class WeatherArgs(BaseModel):
    city: str

async def get_weather(args: WeatherArgs) -> dict:
    # Your API call here
    return {"temp": 72, "condition": "sunny"}

# Create agent
agent = Agent(
    model='openrouter:openai/gpt-4o',
    output_type=ChatResponse,
    system_prompt="You are a helpful weather assistant."
)

# Register tool
agent.add_tool(
    Tool(
        name="get_weather",
        description="Get current weather for a city",
        parameters=WeatherArgs,
        execute=get_weather
    )
)

# Usage
if __name__ == "__main__":
    result = await agent.run("What's the weather in SF?")
    print(result.message)

Common Pitfalls

❌ Don't: Use any type ✅ Do: Define strict Pydantic models

❌ Don't: Handle retries manually ✅ Do: Configure RetryConfig

❌ Don't: Parse LLM output manually ✅ Do: Let Pydantic AI handle it

Resources

• Pydantic AI Docs
• OpenRouter Docs
• FastAPI Docs