Back to Details

agent-builder-pydantic-ai

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Pydantic AI Agent Builder

Pydantic AI Agent 构建器

Purpose

用途

Create production-ready AI agents with type safety, automatic validation, and minimal boilerplate using Pydantic AI framework.
使用Pydantic AI框架创建具备类型安全、自动验证和极少样板代码的生产级AI Agent。

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
  • 构建具备AI能力的FastAPI后端
  • 需要严格的类型检查与验证
  • 希望对格式错误的LLM响应自动重试
  • 创建带有自定义工具的Agent

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
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

安装

bash
pip install pydantic-ai httpx pydantic python-dotenv
bash
pip install pydantic-ai httpx pydantic python-dotenv

Base Agent Pattern

基础Agent模式

python
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."
)
python
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

Usage

result = await agent.run("user message")
undefined
result = await agent.run("user message")
undefined

Integration with OpenRouter

与OpenRouter集成

Setup

设置

python
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')
)
python
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

环境变量

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

Tool Definition Pattern

工具定义模式

python
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": [...]}
python
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

Register tool

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

Streaming Pattern

流处理模式

python
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
        }
python
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

错误处理与重试

python
from pydantic_ai import Agent, RetryConfig

agent = Agent(
    model='openrouter:openai/gpt-4o',
    retry_config=RetryConfig(
        max_retries=3,
        retry_on=[ValidationError, TimeoutError]
    )
)
python
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

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}")
undefined
try: result = await agent.run("user message") except ValidationError as e: # Will retry automatically logger.error(f"Validation failed after retries: {e}")
undefined

FastAPI Integration

FastAPI集成

python
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))
python
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

测试模式

python
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()
python
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
  1. 类型安全：始终为输入/输出定义Pydantic模型
  2. 依赖注入：对工具使用FastAPI风格的依赖注入
  3. 自动重试：配置重试逻辑以提升鲁棒性
  4. 日志记录：添加结构化日志用于调试
  5. 测试：为Agent行为编写pytest测试
  6. 验证：让Pydantic自动处理验证工作
  7. 上下文：传递上下文字典以支持有状态对话

Example: Complete Agent

示例：完整的Agent

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

Output type

Output type

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

Tool definition

Tool definition

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

Create agent

Create agent

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

Register tool

Register tool

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

Usage

Usage

if name == "main": result = await agent.run("What's the weather in SF?") print(result.message)
undefined
if name == "main": result = await agent.run("What's the weather in SF?") print(result.message)
undefined

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
❌ 不要：使用
any
类型 ✅ 要：定义严格的Pydantic模型
❌ 不要：手动处理重试逻辑 ✅ 要：配置RetryConfig
❌ 不要：手动解析LLM输出 ✅ 要：让Pydantic AI自动处理

Resources

相关资源