---

name: pydantic-ai description: "使用 PydanticAI 构建生产就绪的 AI 代理——类型安全的工具使用、结构化输出、依赖注入和多模型支持。" category: ai-agents risk: safe source: community date_added: "2026-03-18" author: suhaibjanjua tags: [pydantic-ai, ai-agents, llm, openai, anthropic, gemini, tool-use, structured-output, python] tools: [claude, cursor, gemini]

PydanticAI — Python 中的类型化 AI 代理

概述

PydanticAI 是 Pydantic 团队开发的 Python 代理框架，将 Pydantic 的类型安全和验证保证带入基于 LLM 的应用。它支持结构化输出（使用 Pydantic 模型验证）、依赖注入以实现可测试性、流式响应、多轮对话和工具使用——跨 OpenAI、Anthropic、Google Gemini、Groq、Mistral 和 Ollama。当构建正确性和可测试性至关重要的生产 AI 代理、聊天机器人或 LLM 管道时使用此技能。

适用场景

• 构建调用工具并返回结构化数据的 Python AI 代理时
• 需要经过验证的、类型化的 LLM 输出（而非原始字符串）时
• 想要编写不需要调用真实 LLM 的代理逻辑单元测试时
• 在不重写代理代码的情况下切换 LLM 提供商时
• 用户询问 Agent、@agent.tool、RunContext、ModelRetry 或 result_type 时

工作原理

步骤 1：安装

pip install pydantic-ai

# Install extras for specific providers
pip install 'pydantic-ai[openai]'       # OpenAI / Azure OpenAI
pip install 'pydantic-ai[anthropic]'    # Anthropic Claude
pip install 'pydantic-ai[gemini]'       # Google Gemini
pip install 'pydantic-ai[groq]'         # Groq
pip install 'pydantic-ai[vertexai]'     # Google Vertex AI

步骤 2：最小代理

from pydantic_ai import Agent

# Simple agent — returns a plain string
agent = Agent(
    'anthropic:claude-sonnet-4-6',
    system_prompt='You are a helpful assistant. Be concise.',
)

result = agent.run_sync('What is the capital of Japan?')
print(result.data)  # "Tokyo"
print(result.usage())  # Usage(requests=1, request_tokens=..., response_tokens=...)

步骤 3：使用 Pydantic 模型的结构化输出

from pydantic import BaseModel
from pydantic_ai import Agent

class MovieReview(BaseModel):
    title: str
    year: int
    rating: float  # 0.0 to 10.0
    summary: str
    recommended: bool

agent = Agent(
    'openai:gpt-4o',
    result_type=MovieReview,
    system_prompt='You are a film critic. Return structured reviews.',
)

result = agent.run_sync('Review Inception (2010)')
review = result.data  # Fully typed MovieReview instance
print(f"{review.title} ({review.year}): {review.rating}/10")
print(f"Recommended: {review.recommended}")

步骤 4：工具使用

使用 @agent.tool 注册工具——LLM 可以在运行期间调用它们：

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
import httpx

class WeatherReport(BaseModel):
    city: str
    temperature_c: float
    condition: str

weather_agent = Agent(
    'anthropic:claude-sonnet-4-6',
    result_type=WeatherReport,
    system_prompt='Get current weather for the requested city.',
)

@weather_agent.tool
async def get_temperature(ctx: RunContext, city: str) -> dict:
    """Fetch the current temperature for a city from the weather API."""
    async with httpx.AsyncClient() as client:
        r = await client.get(f'https://wttr.in/{city}?format=j1')
        data = r.json()
        return {
            'temp_c': float(data['current_condition'][0]['temp_C']),
            'description': data['current_condition'][0]['weatherDesc'][0]['value'],
        }

import asyncio
result = asyncio.run(weather_agent.run('What is the weather in Tokyo?'))
print(result.data)

步骤 5：依赖注入

将服务（数据库、HTTP 客户端、配置）注入代理以实现可测试性：

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel

@dataclass
class Deps:
    db: Database
    user_id: str

class SupportResponse(BaseModel):
    message: str
    escalate: bool

support_agent = Agent(
    'openai:gpt-4o-mini',
    deps_type=Deps,
    result_type=SupportResponse,
    system_prompt='You are a support agent. Use the tools to help customers.',
)

@support_agent.tool
async def get_order_history(ctx: RunContext[Deps]) -> list[dict]:
    """Fetch recent orders for the current user."""
    return await ctx.deps.db.get_orders(ctx.deps.user_id, limit=5)

@support_agent.tool
async def create_refund(ctx: RunContext[Deps], order_id: str, reason: str) -> dict:
    """Initiate a refund for a specific order."""
    return await ctx.deps.db.create_refund(order_id, reason, ctx.deps.user_id)

# Usage
async def handle_support(user_id: str, message: str):
    deps = Deps(db=get_db(), user_id=user_id)
    result = await support_agent.run(message, deps=deps)