PydanticAI 深度拆解：当 Python 决定「干掉 LangChain」——一个 26K Star 的 AI Agent 框架如何用类型安全重新定义智能体开发的工程哲学

引言：AI Agent 开发的「解析地狱」

2026 年，AI Agent 从概念验证全面走向生产落地。但一个残酷的现实是：大多数 AI 应用不是死在模型能力上，而是死在「结果能不能用」上。

你有没有经历过这样的场景：

import openai
import json
import re

response = openai.ChatCompletion.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "分析这份销售数据"}]
)

# 解析地狱开始
try:
    data = json.loads(response.choices[0].message.content)
    revenue = float(data["total_revenue"])  # KeyError: 'total_revenue'
except (json.JSONDecodeError, KeyError, ValueError):
    # 正则兜底...
    match = re.search(r'"total_revenue":\s*([\d.]+)', response.choices[0].message.content)
    if match:
        revenue = float(match.group(1))
    else:
        revenue = 0.0  # 鬼知道对不对

这段代码不是段子，这是无数生产系统的真实写照。LLM 返回的 JSON 字段名可能从 conversion_rate 变成 conv_rate_pct，数值可能多一个百分号，结构可能嵌套错层。你在用正则表达式和 try/except 给 AI 的自由发挥擦屁股。

更糟糕的是，这种「解析地狱」在生产环境中会引发连锁故障：

1. 数据管道断裂：下游系统期望 total_revenue 是 float，收到 string 后直接崩溃
2. 静默错误：正则兜底提取到错误数值，报表数据全部失真
3. 调试困难：错误发生在 LLM 输出和业务逻辑之间，很难定位是模型问题还是解析问题
4. 测试覆盖盲区：无法对「LLM 输出解析」写有效的单元测试

PydanticAI 的出现，正是为了终结这种「解析地狱」。它不是又一个 LLM 封装库，而是一个类型驱动（Schema-first） 的 AI Agent 框架——用 Python 类型系统从根源上约束 LLM 的输出，让 AI 的返回结果像 int 一样可靠。

一、PydanticAI 是什么？——从 Pydantic 到 Agent 的自然进化

1.1 起源：Pydantic 生态的 AI 延伸

Pydantic 是 Python 生态中最主流的数据验证库，基于类型提示实现声明式数据校验。几乎所有现代 Python Web 框架（FastAPI、Litestar、Strawberry）都以 Pydantic 作为数据层核心。Pydantic v2 用 Rust 重写了核心校验引擎，性能提升了 5-50 倍，成为 Python 数据验证的事实标准。

PydanticAI 是 Pydantic 团队的官方 AI Agent 框架，核心理念一脉相承：用类型系统约束不确定的输入输出。在 LLM 场景下，这意味着用 Pydantic Model 定义你期望的输出结构，框架自动将 LLM 的自由文本输出校验、转换为强类型数据。

项目由 Samuel Colvin（Pydantic 创始人）主导开发，2024年底首次发布，2025年迅速获得社区关注，2026年已迭代至 v2.12.0，成为 Python AI Agent 生态中最受关注的框架之一。

1.2 核心定位：类型驱动 vs Prompt 驱动

PydanticAI 与传统 LLM 框架的根本区别在于设计哲学：

1.3 GitHub 数据与社区

• Star 数：26,000+
• 提交数：2,600+
• 贡献者：200+
• 最新版本：v2.12.0（2026年7月）
• 许可证：MIT
• 核心依赖：pydantic >= 2.0, httpx
• PyPI 月下载：500,000+

二、核心架构：四层抽象的设计哲学

PydanticAI 的内部架构分为四层，每一层都有明确的职责边界：

┌─────────────────────────────────────────┐
│         用户层（Agent + Pydantic Model）   │
├─────────────────────────────────────────┤
│         运行层（RunContext + 依赖注入）     │
├─────────────────────────────────────────┤
│         模型层（Model 抽象 + 多供应商）     │
├─────────────────────────────────────────┤
│         工具层（Tool + Toolset + MCP）     │
└─────────────────────────────────────────┘

2.1 Agent 核心对象

Agent 是 PydanticAI 的一等公民。每个 Agent 实例绑定一个系统提示、一个输出类型、和一组工具：

from pydantic_ai import Agent
from pydantic import BaseModel, Field

class SalesAnalysis(BaseModel):
    """销售数据分析结果"""
    total_revenue: float = Field(ge=0, description="总收入，单位：万元")
    growth_rate: float = Field(ge=-1, le=10, description="同比增长率，如 0.15 表示 15%")
    top_products: list[str] = Field(min_length=1, max_length=5, description="TOP5产品")
    risk_factors: list[str] = Field(max_length=3, description="风险因素")
    confidence: float = Field(ge=0, le=1, description="分析置信度")

sales_agent = Agent(
    'openai:gpt-4o',
    system_prompt='你是一个专业的销售数据分析师，输出结构化的分析报告。',
    result_type=SalesAnalysis,  # 强类型输出
    retries=3,  # 校验失败自动重试
)

关键设计点：

1. result_type 参数：直接传入 Pydantic Model 类，框架自动生成 JSON Schema 并注入到 LLM 的 system prompt 中，约束输出格式
2. retries 参数：当 LLM 输出不符合 Schema 时，框架自动将校验错误反馈给 LLM 并重试，而不是直接抛异常
3. 类型提示自动推导：调用 sales_agent.run() 的返回值自动带有正确的类型标注，IDE 可以直接推断 result.data.total_revenue 是 float

2.2 RunContext：依赖注入的 Agent 版

这是 PydanticAI 最独特的设计之一。RunContext 允许你在 Agent 运行时注入依赖，类似 FastAPI 的 Depends 但专为 Agent 场景优化：

from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class DatabaseDeps:
    """数据库依赖"""
    db_url: str
    api_key: str

agent = Agent(
    'openai:gpt-4o',
    result_type=SalesAnalysis,
    deps_type=DatabaseDeps,  # 声明依赖类型
)

@agent.tool
def query_sales_data(ctx: RunContext[DatabaseDeps], product: str) -> str:
    """查询指定产品的销售数据"""
    # ctx.deps 自动获得 DatabaseDeps 类型，IDE 自动补全
    conn = connect(ctx.deps.db_url)
    result = conn.execute(f"SELECT * FROM sales WHERE product = '{product}'")
    return format_sales(result)

# 运行时注入依赖
result = agent.run_sync(
    "分析所有产品的销售趋势",
    deps=DatabaseDeps(db_url="postgres://...", api_key="sk-xxx")
)

为什么 RunContext 这么重要？

• 测试友好：Mock 依赖即可测试 Agent 逻辑，不需要真正调用 LLM
• 环境隔离：开发/测试/生产环境通过依赖注入切换，Agent 代码不变
• 类型安全：ctx.deps 有完整的类型提示，IDE 自动补全
• 生命周期管理：依赖可以有初始化和清理逻辑

2.3 模型层：多供应商统一抽象

PydanticAI 不绑定任何特定 LLM 供应商。通过 Model 抽象层，同一套 Agent 代码可以在不同模型间无缝切换：

from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.models.anthropic import AnthropicModel
from pydantic_ai.models.google import GoogleModel
from pydantic_ai.models.mistral import MistralModel

# 同一个 Agent，切换模型只需改一行
model = OpenAIModel('gpt-4o')
# model = AnthropicModel('claude-sonnet-4-20250514')
# model = GoogleModel('gemini-2.5-pro')
# model = MistralModel('mistral-large-latest')

agent = Agent(model, result_type=SalesAnalysis)

支持的模型供应商（截至 v2.12.0）：

2.4 工具层：声明式工具注册

PydanticAI 的工具注册采用声明式风格，工具的参数类型直接从 Python 函数签名推导：

from pydantic_ai import Agent, RunContext

agent = Agent('openai:gpt-4o', result_type=str)

@agent.tool
def get_weather(ctx: RunContext[None], city: str, unit: str = "celsius") -> str:
    """获取指定城市的天气信息
    
    Args:
        city: 城市名称，如"北京"、"上海"
        unit: 温度单位，celsius 或 fahrenheit
    """
    return f"{city}的天气：25°{unit[0].upper()}"

@agent.tool(retries=2)
def search_products(ctx: RunContext[None], query: str, max_results: int = 5) -> list[dict]:
    """搜索商品"""
    return [{"name": f"商品{i}", "price": 99.9} for i in range(max_results)]

关键特性：

1. 零样板代码：不需要手动写 JSON Schema，框架从类型提示自动生成
2. 工具级重试：每个工具可以独立设置重试策略
3. 依赖透传：通过 RunContext 访问 Agent 级别的依赖
4. 异步支持：工具可以是 async 函数，框架自动处理并发

三、结构化输出：PydanticAI 的杀手锏

3.1 为什么 Structured Output 不够？

OpenAI 的 Structured Output 和 Anthropic 的 Tool Use 都支持约束 JSON 输出，但它们有明显局限：

# OpenAI Structured Output 的问题
response = openai.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[...],
    response_format=SalesAnalysis,  # 只约束顶层结构
)

# 问题1：嵌套对象的字段校验不严格
# 问题2：校验失败直接报错，没有重试机制
# 问题3：无法在校验失败时给 LLM 反馈具体错误
# 问题4：不支持自定义校验逻辑
# 问题5：不支持 Union 类型、Optional 字段的复杂校验

PydanticAI 在此基础上做了关键增强：

from pydantic import BaseModel, Field, field_validator, model_validator

class SalesAnalysis(BaseModel):
    """销售数据分析结果"""
    total_revenue: float = Field(ge=0, description="总收入，单位：万元")
    growth_rate: float = Field(ge=-1, le=10, description="同比增长率，如 0.15 表示 15%")
    top_products: list[str] = Field(min_length=1, max_length=5)
    risk_factors: list[str] = Field(max_length=3)
    confidence: float = Field(ge=0, le=1)
    
    @field_validator('growth_rate')
    @classmethod
    def validate_growth_rate(cls, v: float) -> float:
        if abs(v) > 5:
            raise ValueError(f'增长率异常：{v}，请检查数据源')
        return v
    
    @model_validator(mode='after')
    def validate_consistency(self):
        """交叉校验：高置信度不应有高风险"""
        if self.confidence > 0.8 and self.risk_level == "high":
            raise ValueError('高置信度分析不应标记为高风险，请重新评估')
        return self

agent = Agent(
    'openai:gpt-4o',
    result_type=SalesAnalysis,
    retries=3,  # 校验失败自动重试，最多3次
)

关键增强：

1. Field 约束：ge、le、min_length、max_length、pattern 等约束直接生效
2. 自定义校验器：@field_validator 和 @model_validator 可以写任意校验逻辑
3. 校验失败重试：框架将 Pydantic 校验错误（包括自定义校验器的错误消息）反馈给 LLM，让它修正输出
4. 置信度追踪：每次重试都会记录，可以监控 LLM 输出质量
5. 错误消息传递：校验错误的具体原因会作为反馈传给 LLM，而不是模糊的"输出格式错误"

3.2 校验失败重试的内部机制

当 LLM 输出不符合 Pydantic Schema 时，PydanticAI 的重试流程如下：

LLM 输出 → JSON 解析 → Pydantic 校验 → 
  ├─ 通过 → 返回结果
  └─ 失败 → 构造错误反馈 → 重新调用 LLM → ...

错误反馈的格式经过精心设计：

# 假设 LLM 返回了这样的 JSON：
{
    "total_revenue": "一百万",  # 应该是 float
    "growth_rate": 0.15,
    "top_products": ["A", "B", "C", "D", "E", "F"],  # 超过 max_length=5
    "risk_factors": [],
    "confidence": 1.5  # 超过 le=1
}

# PydanticAI 会构造这样的反馈：
"""
The previous output was invalid. Please fix the following issues:

1. total_revenue: Input should be a valid number (expected float, got string "一百万")
2. top_products: List should have at most 5 items (got 6)
3. confidence: Input should be less than or equal to 1 (got 1.5)

Please output the correct JSON matching the schema.
"""

3.3 多步输出（Multi-step Output）

对于需要多步推理的复杂任务，PydanticAI 支持定义输出依赖链：

from pydantic import BaseModel
from pydantic_ai import Agent

class ResearchPlan(BaseModel):
    """研究计划"""
    topic: str
    key_questions: list[str]
    data_sources: list[str]
    estimated_time_hours: float

class ResearchResult(BaseModel):
    """研究结果"""
    summary: str
    findings: list[dict]
    confidence: float
    follow_up_questions: list[str]

# 第一步：生成研究计划
plan_agent = Agent('openai:gpt-4o', result_type=ResearchPlan)

# 第二步：根据计划执行研究
result_agent = Agent('openai:gpt-4o', result_type=ResearchResult)

async def research_pipeline(topic: str):
    # Step 1: 生成计划
    plan = await plan_agent.run(f"为以下主题制定研究计划：{topic}")
    
    # Step 2: 根据计划执行研究（计划的结构化输出作为下一步输入）
    result = await result_agent.run(
        f"根据以下研究计划执行研究：\n"
        f"主题：{plan.data.topic}\n"
        f"关键问题：{', '.join(plan.data.key_questions)}\n"
        f"数据来源：{', '.join(plan.data.data_sources)}\n"
        f"预计时间：{plan.data.estimated_time_hours}小时"
    )
    
    return result.data  # 自动类型：ResearchResult

3.4 流式输出与结构化

PydanticAI 支持流式输出，同时保持类型安全：

from pydantic_ai import Agent
from pydantic import BaseModel

class StreamingAnalysis(BaseModel):
    """流式分析结果"""
    step: str
    content: str
    progress: float  # 0-1

agent = Agent('openai:gpt-4o', result_type=StreamingAnalysis)

async def stream_analysis(topic: str):
    async with agent.run_stream(topic) as result:
        async for chunk in result.stream():
            # chunk 是 StreamingAnalysis 类型，自动校验
            print(f"[{chunk.progress:.0%}] {chunk.step}: {chunk.content}")
        
        # 最终结果
        final = await result.get_data()
        return final

3.5 Union 类型与多态输出

PydanticAI 支持复杂的 Union 类型，让同一个 Agent 可以返回不同结构的数据：

from pydantic import BaseModel
from typing import Union

class SuccessResult(BaseModel):
    """成功结果"""
    status: str = "success"
    data: dict
    confidence: float

class ErrorResult(BaseModel):
    """错误结果"""
    status: str = "error"
    error_code: str
    error_message: str
    suggested_fix: str

class AnalysisOutput(BaseModel):
    """分析输出（可能是成功或失败）"""
    result: Union[SuccessResult, ErrorResult]

agent = Agent('openai:gpt-4o', result_type=AnalysisOutput)

# Agent 可以根据情况返回 SuccessResult 或 ErrorResult
# PydanticAI 会自动校验 Union 类型的正确性

四、MCP 集成：Agent 的工具生态

4.1 什么是 MCP？

MCP（Model Context Protocol）是 Anthropic 推出的开放协议，标准化了 AI Agent 与外部工具的交互方式。它定义了工具发现、参数传递、结果返回的标准格式，让不同的 MCP 服务器可以被任何 MCP 客户端使用。

PydanticAI 是最早原生支持 MCP 的 Agent 框架之一，提供了两种 MCP 传输方式：

1. stdio 传输：通过标准输入/输出与 MCP 服务器通信（本地进程）
2. SSE 传输：通过 HTTP Server-Sent Events 与 MCP 服务器通信（远程服务）

4.2 PydanticAI 的 MCP 集成

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio, MCPServerSSE

# 方式1：stdio 传输（本地进程）
mcp_server_stdio = MCPServerStdio(
    'npx', ['-y', '@modelcontextprotocol/server-filesystem', '/tmp']
)

# 方式2：SSE 传输（远程服务）
mcp_server_sse = MCPServerSSE(
    'http://localhost:8080/sse',
    headers={'Authorization': 'Bearer sk-xxx'}
)

agent = Agent(
    'openai:gpt-4o',
    result_type=str,
    mcp_servers=[mcp_server_stdio, mcp_server_sse],  # 可以同时连接多个
)

# Agent 自动发现 MCP 服务器提供的所有工具
result = agent.run_sync("列出 /tmp 目录下的所有文件，并查询数据库中的用户信息")

4.3 MCP + 本地工具混合

PydanticAI 允许同时使用本地工具和 MCP 工具，框架自动合并工具列表：

from pydantic_ai import Agent, RunContext
from pydantic_ai.mcp import MCPServerStdio

mcp_server = MCPServerStdio('uvx', ['mcp-server-sqlite', '--db-path', '/tmp/data.db'])

agent = Agent(
    'openai:gpt-4o',
    result_type=str,
    mcp_servers=[mcp_server],
)

@agent.tool
def calculate_metrics(ctx: RunContext[None], data: list[float]) -> dict:
    """计算统计指标：均值、最大值、最小值、标准差"""
    import statistics
    return {
        "mean": statistics.mean(data),
        "max": max(data),
        "min": min(data),
        "stdev": statistics.stdev(data) if len(data) > 1 else 0,
    }

@agent.tool
def export_to_csv(ctx: RunContext[None], data: list[dict], filename: str) -> str:
    """将数据导出为 CSV 文件"""
    import csv
    with open(filename, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=data[0].keys())
        writer.writeheader()
        writer.writerows(data)
    return f"已导出到 {filename}"

# Agent 既有 MCP 提供的数据库工具，也有本地的计算和导出工具
result = agent.run_sync(
    "查询数据库中的销售数据，计算统计指标，然后导出为 CSV"
)

4.4 MCP 工具的类型安全

MCP 工具的参数也会经过 Pydantic 校验：

# 假设 MCP 服务器提供了这样一个工具：
# {
#   "name": "query_database",
#   "description": "执行 SQL 查询",
#   "inputSchema": {
#     "type": "object",
#     "properties": {
#       "sql": {"type": "string"},
#       "params": {"type": "array"}
#     },
#     "required": ["sql"]
#   }
# }

# PydanticAI 会自动将 MCP 工具的 JSON Schema 转换为 Pydantic Model
# LLM 调用 MCP 工具时，参数会经过 Pydantic 校验
# 如果参数不符合 Schema，会自动重试

五、生产级实战：从零构建销售分析 Agent

5.1 项目结构

sales-analyzer/
├── main.py              # Agent 入口
├── models.py            # Pydantic 数据模型
├── tools.py             # 工具函数
├── deps.py              # 依赖注入
├── config.py            # 配置管理
└── tests/
    ├── test_agent.py    # Agent 单元测试
    ├── test_tools.py    # 工具单元测试
    └── conftest.py      # 测试 fixtures

5.2 完整代码实现

config.py — 配置管理：

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """应用配置"""
    openai_api_key: str
    database_url: str
    max_retries: int = 3
    default_model: str = "openai:gpt-4o"
    
    class Config:
        env_file = ".env"
        env_prefix = "SALES_"

settings = Settings()

models.py — 定义数据模型：

from pydantic import BaseModel, Field, field_validator, model_validator
from enum import Enum
from datetime import date

class RiskLevel(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class SalesRecord(BaseModel):
    """单条销售记录"""
    product: str = Field(description="产品名称")
    revenue: float = Field(ge=0, description="收入（万元）")
    quantity: int = Field(ge=0, description="销售数量")
    sale_date: date = Field(description="销售日期")
    region: str = Field(description="销售区域")
    channel: str = Field(default="online", description="销售渠道")

class RegionalBreakdown(BaseModel):
    """区域销售明细"""
    region: str
    revenue: float = Field(ge=0)
    growth_rate: float
    market_share: float = Field(ge=0, le=1)

class AnalysisResult(BaseModel):
    """分析结果"""
    total_revenue: float = Field(ge=0, description="总收入（万元）")
    total_quantity: int = Field(ge=0, description="总销售数量")
    avg_order_value: float = Field(ge=0, description="平均客单价")
    growth_rate: float = Field(ge=-1, le=10, description="同比增长率")
    top_products: list[str] = Field(min_length=1, max_length=5, description="TOP5产品")
    regional_breakdown: list[RegionalBreakdown] = Field(min_length=1, description="区域明细")
    risk_level: RiskLevel = Field(description="风险等级")
    risk_factors: list[str] = Field(max_length=5, description="风险因素")
    recommendations: list[str] = Field(min_length=1, max_length=5, description="建议")
    confidence: float = Field(ge=0, le=1, description="分析置信度")
    analysis_period: str = Field(description="分析时间段")

    @field_validator('avg_order_value')
    @classmethod
    def validate_avg_order(cls, v: float) -> float:
        if v > 100000:
            raise ValueError(f'平均客单价异常：{v}，请检查数据')
        return v

    @model_validator(mode='after')
    def validate_consistency(self):
        """交叉校验"""
        if self.confidence > 0.9 and self.risk_level == RiskLevel.HIGH:
            raise ValueError('高置信度分析不应标记为高风险，请重新评估')
        if self.total_quantity > 0 and abs(self.avg_order_value - self.total_revenue / self.total_quantity) > 0.01:
            raise ValueError('平均客单价与总收入/总数量不一致')
        return self

deps.py — 定义依赖：

from dataclasses import dataclass, field
from typing import Optional
import asyncpg
from config import settings

@dataclass
class DatabaseDeps:
    """数据库依赖"""
    pool: asyncpg.Pool
    api_key: str = field(default_factory=lambda: settings.openai_api_key)
    region: str = "all"

    @classmethod
    async def create(cls, db_url: Optional[str] = None) -> "DatabaseDeps":
        url = db_url or settings.database_url
        pool = await asyncpg.create_pool(url, min_size=2, max_size=10)
        return cls(pool=pool)

    async def close(self):
        await self.pool.close()

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

tools.py — 定义工具：

from pydantic_ai import RunContext
from deps import DatabaseDeps
from models import SalesRecord
from datetime import date

async def query_sales(
    ctx: RunContext[DatabaseDeps],
    start_date: str,
    end_date: str,
    product: str | None = None,
    region: str | None = None,
) -> list[SalesRecord]:
    """查询指定时间范围的销售数据
    
    Args:
        start_date: 开始日期，格式 YYYY-MM-DD
        end_date: 结束日期，格式 YYYY-MM-DD
        product: 可选，筛选特定产品
        region: 可选，筛选特定区域
    """
    query = """
        SELECT product, revenue, quantity, date, region, channel
        FROM sales 
        WHERE date BETWEEN $1 AND $2
    """
    params: list = [start_date, end_date]
    param_idx = 3
    
    if product:
        query += f" AND product = ${param_idx}"
        params.append(product)
        param_idx += 1
    
    if region:
        query += f" AND region = ${param_idx}"
        params.append(region)
        param_idx += 1
    
    query += " ORDER BY date DESC"
    
    async with ctx.deps.pool.acquire() as conn:
        rows = await conn.fetch(query, *params)
    
    return [
        SalesRecord(
            product=row['product'],
            revenue=float(row['revenue']),
            quantity=row['quantity'],
            sale_date=row['date'],
            region=row['region'],
            channel=row.get('channel', 'online'),
        )
        for row in rows
    ]

async def get_product_ranking(
    ctx: RunContext[DatabaseDeps],
    top_n: int = 10,
    days: int = 30,
) -> list[dict]:
    """获取产品销售排名
    
    Args:
        top_n: 返回前N名，默认10
        days: 统计最近N天，默认30
    """
    query = """
        SELECT product, 
               SUM(revenue) as total_revenue, 
               SUM(quantity) as total_quantity,
               COUNT(*) as order_count
        FROM sales
        WHERE date >= CURRENT_DATE - INTERVAL '%d days'
        GROUP BY product
        ORDER BY total_revenue DESC
        LIMIT $1
    """ % days
    
    async with ctx.deps.pool.acquire() as conn:
        rows = await conn.fetch(query, top_n)
    
    total = sum(float(r['total_revenue']) for r in rows) if rows else 1
    
    return [
        {
            "rank": i+1, 
            "product": r['product'], 
            "revenue": float(r['total_revenue']),
            "quantity": r['total_quantity'],
            "order_count": r['order_count'],
            "market_share": float(r['total_revenue']) / total,
        }
        for i, r in enumerate(rows)
    ]

async def get_regional_stats(
    ctx: RunContext[DatabaseDeps],
    start_date: str,
    end_date: str,
) -> list[dict]:
    """获取区域销售统计"""
    query = """
        SELECT region,
               SUM(revenue) as total_revenue,
               SUM(quantity) as total_quantity
        FROM sales
        WHERE date BETWEEN $1 AND $2
        GROUP BY region
        ORDER BY total_revenue DESC
    """
    
    async with ctx.deps.pool.acquire() as conn:
        rows = await conn.fetch(query, start_date, end_date)
    
    total = sum(float(r['total_revenue']) for r in rows) if rows else 1
    
    return [
        {
            "region": r['region'],
            "revenue": float(r['total_revenue']),
            "quantity": r['total_quantity'],
            "market_share": float(r['total_revenue']) / total,
        }
        for r in rows
    ]

main.py — Agent 入口：

from pydantic_ai import Agent, RunContext
from models import AnalysisResult
from deps import DatabaseDeps
from tools import query_sales, get_product_ranking, get_regional_stats
from config import settings

# 创建 Agent
analyzer = Agent(
    settings.default_model,
    system_prompt="""你是一个资深销售数据分析师，擅长从数据中发现趋势和风险。

分析规则：
1. 总收入 = 所有记录的 revenue 之和
2. 平均客单价 = 总收入 / 总销售数量
3. 风险等级判定：
   - HIGH: 有任何产品收入同比下降超过30%
   - MEDIUM: 有产品收入同比下降10-30%
   - LOW: 所有产品收入稳定或增长
4. 建议必须具体可执行，不能是空话（如"加强营销"）
5. 置信度根据数据完整性和一致性评估
6. 区域分析要关注市场份额变化
7. 如果数据不足，明确说明置信度较低的原因""",
    result_type=AnalysisResult,
    deps_type=DatabaseDeps,
    retries=3,
)

# 注册工具
analyzer.tool(query_sales)
analyzer.tool(get_product_ranking)
analyzer.tool(get_regional_stats)

async def analyze_sales(start_date: str, end_date: str) -> AnalysisResult:
    """分析指定时间范围的销售数据"""
    async with await DatabaseDeps.create() as deps:
        result = await analyzer.run(
            f"分析 {start_date} 到 {end_date} 的销售数据。\n"
            f"请重点关注：\n"
            f"1. 整体收入趋势和增长情况\n"
            f"2. TOP产品的表现和变化\n"
            f"3. 各区域的销售分布\n"
            f"4. 潜在风险和改进建议",
            deps=deps,
        )
        return result.data

# 使用
import asyncio
result = asyncio.run(analyze_sales("2026-01-01", "2026-06-30"))
print(f"总收入：{result.total_revenue}万元")
print(f"增长趋势：{result.growth_rate:.1%}")
print(f"风险等级：{result.risk_level.value}")
print(f"TOP产品：{', '.join(result.top_products)}")
print(f"建议：{result.recommendations}")

5.3 单元测试

PydanticAI 的一大优势是不调用真实 LLM 就能测试 Agent 逻辑：

import pytest
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from models import AnalysisResult, RiskLevel

def test_analysis_result_structure():
    """验证输出结构完整性"""
    test_model = TestModel()
    
    agent = Agent(
        test_model,
        result_type=AnalysisResult,
    )
    
    @agent.tool
    def mock_query(ctx, start_date, end_date):
        return [
            {"product": "A", "revenue": 100.0, "quantity": 10, 
             "sale_date": "2026-01-15", "region": "华东"},
            {"product": "B", "revenue": 80.0, "quantity": 8,
             "sale_date": "2026-01-16", "region": "华南"},
        ]
    
    @agent.tool
    def mock_ranking(ctx, top_n=5, days=30):
        return [
            {"rank": 1, "product": "A", "revenue": 100.0, 
             "quantity": 10, "order_count": 5, "market_share": 0.55},
        ]
    
    @agent.tool
    def mock_regional(ctx, start_date, end_date):
        return [
            {"region": "华东", "revenue": 100.0, "quantity": 10, "market_share": 0.55},
        ]
    
    result = agent.run_sync("分析销售数据", deps=None)
    
    # 验证结构化输出
    assert isinstance(result.data, AnalysisResult)
    assert result.data.total_revenue >= 0
    assert 0 <= result.data.confidence <= 1
    assert result.data.risk_level in [RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.HIGH]
    assert len(result.data.top_products) >= 1
    assert len(result.data.recommendations) >= 1

def test_field_validation():
    """验证字段校验"""
    from pydantic import ValidationError
    from models import AnalysisResult
    
    # 测试无效数据 - 负数收入
    with pytest.raises(ValidationError):
        AnalysisResult(
            total_revenue=-1,
            total_quantity=100,
            avg_order_value=50.0,
            growth_rate=0.15,
            top_products=["A"],
            regional_breakdown=[{"region": "华东", "revenue": 100, "growth_rate": 0.1, "market_share": 0.5}],
            risk_level="low",
            risk_factors=[],
            recommendations=["继续"],
            confidence=0.8,
            analysis_period="2026-Q1",
        )

def test_model_validator():
    """验证交叉校验"""
    from pydantic import ValidationError
    from models import AnalysisResult
    
    # 高置信度 + 高风险应该报错
    with pytest.raises(ValidationError):
        AnalysisResult(
            total_revenue=100,
            total_quantity=10,
            avg_order_value=10.0,
            growth_rate=0.15,
            top_products=["A"],
            regional_breakdown=[{"region": "华东", "revenue": 100, "growth_rate": 0.1, "market_share": 0.5}],
            risk_level="high",
            risk_factors=["下降"],
            recommendations=["改进"],
            confidence=0.95,  # 高置信度
            analysis_period="2026-Q1",
        )

六、与主流框架的性能基准对比

6.1 延迟对比

测试环境：Python 3.12, Apple M3 Max, 32GB RAM, OpenAI API

6.2 内存占用

6.3 Token 效率

PydanticAI 的系统提示更精简，因为它利用 JSON Schema 约束而非冗长的 prompt 指令：

6.4 校验成功率

在1000次随机 LLM 输出测试中：

七、PydanticAI 的局限与选型建议

7.1 局限性

1. Python 绑定：目前只支持 Python，不像 LangChain 有 JS/TS 版本
2. 生态相对年轻：第三方集成和插件数量不如 LangChain 丰富
3. 复杂编排场景：对于需要复杂有向图编排（如 LangGraph 的 StateGraph）的场景，PydanticAI 的多 Agent 编排能力还在发展中
4. 可视化调试：没有 LangSmith 那样的可视化追踪平台（虽然集成了 OpenTelemetry）
5. 学习曲线：对于不熟悉 Pydantic 的开发者，需要额外学习成本

7.2 选型决策树

需要 AI Agent 框架？
├─ 只用 OpenAI 模型？
│  └─ 是 → OpenAI Agents SDK（极简）
├─ 需要多供应商支持？
│  ├─ 重视类型安全和测试？ → PydanticAI
│  └─ 需要丰富生态和编排？ → LangChain/LangGraph
├─ 需要复杂多步编排？
│  └─ LangGraph（StateGraph）
├─ 需要快速原型？
│  └─ OpenAI Agents SDK 或 PydanticAI
└─ 生产级部署？
   └─ PydanticAI（最佳工程实践）

7.3 各框架适用场景

选 PydanticAI 当：

• 你需要生产级的结构化输出
• 你的团队熟悉 Python 类型系统
• 你重视测试覆盖率和类型安全
• 你不想被框架锁定（PydanticAI 很薄，替换成本低）
• 你需要 MCP 工具集成
• 你需要依赖注入模式

选 LangChain 当：

• 你需要丰富的第三方集成生态（200+ 集成）
• 你需要复杂的多步编排（LangGraph）
• 你需要成熟的可观测性工具（LangSmith）
• 你的团队已经深度使用 LangChain 生态
• 你需要 JS/TS 支持

选 OpenAI Agents SDK 当：

• 你的场景只用 OpenAI 模型
• 你追求极简 API，不需要复杂功能
• 你是快速原型验证阶段
• 你需要 Handoff（Agent 间交接）能力

八、总结与展望

PydanticAI 的出现，代表了 AI Agent 开发从「Prompt Engineering」走向「Software Engineering」的趋势。它的核心贡献不是发明了新的 AI 技能，而是把软件工程的最佳实践（类型安全、依赖注入、单元测试、错误处理）引入了 AI Agent 开发。

关键洞察

1. Schema-first > Prompt-first：用类型系统约束输出，比用 prompt 约束输出更可靠、更可测试
2. 校验驱动重试：让 LLM 从校验错误中学习，而不是盲目重试
3. 依赖注入的 Agent 版：让 Agent 逻辑与基础设施解耦
4. 薄框架哲学：PydanticAI 不试图做所有事，而是做好一件事——让 LLM 输出像 int 一样可靠

未来方向

• 多模态结构化输出：图片、音频的结构化描述
• Agent-to-Agent 通信：标准化的 Agent 间协作协议
• 边缘部署优化：更小的运行时体积，支持 WASM 部署
• 可视化调试平台：基于 OpenTelemetry 的 Agent 执行追踪
• Python 3.13+ 优化：利用 free-threaded 模式提升并发性能

最后的思考

PydanticAI 告诉我们：AI Agent 的未来不在于模型有多聪明，而在于工程有多扎实。当你的 Agent 能像调用 int() 一样可靠地获取结构化数据时，真正有价值的 AI 应用才刚刚开始。

在这个 AI Agent 爆发的时代，选择一个正确的框架，不是选择一个最花哨的工具，而是选择一个最能帮你把 AI 能力转化为可靠产品的基础设施。PydanticAI，正是这样的基础设施。

九、内部架构深度剖析：PydanticAI 是怎么工作的？

9.1 Agent 运行时的完整生命周期

当你调用 agent.run() 时，PydanticAI 内部经历了以下步骤：

1. 构造系统提示
   └─ 将 Pydantic Model 的 JSON Schema 注入到 system prompt
   └─ 添加工具定义（从函数签名自动生成）
   └─ 添加用户消息

2. 调用 LLM
   └─ 通过 Model 抽象层发送请求
   └─ 收到 LLM 响应

3. 解析输出
   └─ 提取 LLM 返回的 JSON
   └─ 用 Pydantic Model 校验

4. 校验结果
   ├─ 通过 → 返回 AgentRunResult
   └─ 失败 → 构造错误反馈 → 回到步骤2（重试）

5. 返回结果
   └─ AgentRunResult 包含 data（结构化输出）、usage（token 用量）等

9.2 JSON Schema 生成机制

PydanticAI 利用 Pydantic v2 的 JSON Schema 生成能力，自动将 Python 类型转换为 LLM 可理解的约束：

from pydantic import BaseModel, Field
import json

class SalesAnalysis(BaseModel):
    total_revenue: float = Field(ge=0, description="总收入")
    growth_rate: float = Field(ge=-1, le=10, description="增长率")
    top_products: list[str] = Field(min_length=1, max_length=5)

# Pydantic 自动生成的 JSON Schema：
schema = SalesAnalysis.model_json_schema()
print(json.dumps(schema, indent=2, ensure_ascii=False))
"""
{
  "properties": {
    "total_revenue": {
      "description": "总收入",
      "exclusiveMinimum": 0,
      "title": "Total Revenue",
      "type": "number"
    },
    "growth_rate": {
      "description": "增长率",
      "maximum": 10,
      "minimum": -1,
      "title": "Growth Rate",
      "type": "number"
    },
    "top_products": {
      "description": "TOP5产品",
      "items": {"type": "string"},
      "maxItems": 5,
      "minItems": 1,
      "title": "Top Products",
      "type": "array"
    }
  },
  "required": ["total_revenue", "growth_rate", "top_products"],
  "title": "SalesAnalysis",
  "type": "object"
}
"""

这个 Schema 会被注入到 LLM 的 system prompt 中，告诉模型：

1. 输出必须是 JSON 格式
2. 必须包含 total_revenue、growth_rate、top_products 三个字段
3. total_revenue 必须是大于 0 的数字
4. growth_rate 必须在 -1 到 10 之间
5. top_products 必须是 1-5 个字符串的数组

9.3 错误反馈的构造

当 Pydantic 校验失败时，PydanticAI 会构造一个精确的错误反馈：

# 假设 LLM 返回了：
{
    "total_revenue": "一百万",
    "growth_rate": 0.15,
    "top_products": ["A", "B", "C", "D", "E", "F"]
}

# PydanticAI 构造的错误反馈：
"""
The previous output was invalid. Please fix the following issues:

1. total_revenue: Input should be a valid number (expected float, got string "一百万")
   - Hint: Use numeric values like 100.0, not Chinese text

2. top_products: List should have at most 5 items (got 6)
   - Hint: Limit to the top 5 products

Please output the correct JSON matching the schema:
{
  "total_revenue": number (required, > 0),
  "growth_rate": number (required, -1 <= x <= 10),
  "top_products": string[] (required, 1-5 items)
}
"""

这种精确的错误反馈让 LLM 能够理解问题所在并修正输出，而不是盲目重试。

十、高级模式：多 Agent 协作与编排

10.1 Agent 链（Agent Chain）

PydanticAI 支持将多个 Agent 串联起来，形成处理流水线：

from pydantic_ai import Agent
from pydantic import BaseModel

class InputData(BaseModel):
    raw_text: str

class ExtractedInfo(BaseModel):
    entities: list[str]
    relationships: list[dict]
    summary: str

class AnalysisReport(BaseModel):
    key_findings: list[str]
    risk_assessment: str
    recommendations: list[str]

# Agent 1: 信息提取
extractor = Agent('openai:gpt-4o', result_type=ExtractedInfo)

# Agent 2: 分析报告
analyst = Agent('openai:gpt-4o', result_type=AnalysisReport)

async def process_document(text: str) -> AnalysisReport:
    # Step 1: 提取信息
    extraction = await extractor.run(f"从以下文本中提取实体和关系：\n{text}")
    
    # Step 2: 生成分析报告（使用 Step 1 的结构化输出）
    report = await analyst.run(
        f"基于以下提取的信息生成分析报告：\n"
        f"实体：{extraction.data.entities}\n"
        f"关系：{extraction.data.relationships}\n"
        f"摘要：{extraction.data.summary}"
    )
    
    return report.data

10.2 并行 Agent 执行

对于独立的任务，可以并行执行多个 Agent：

import asyncio
from pydantic_ai import Agent
from pydantic import BaseModel

class MarketAnalysis(BaseModel):
    market_size: float
    growth_trend: str
    competitors: list[str]

class TechAnalysis(BaseModel):
    tech_stack: list[str]
    architecture_score: float  # 0-10
    scalability_assessment: str

class FinancialAnalysis(BaseModel):
    revenue_projection: float
    cost_breakdown: dict
    roi_estimate: float

market_agent = Agent('openai:gpt-4o', result_type=MarketAnalysis)
tech_agent = Agent('openai:gpt-4o', result_type=TechAnalysis)
financial_agent = Agent('openai:gpt-4o', result_type=FinancialAnalysis)

async def comprehensive_analysis(company_info: str) -> dict:
    # 并行执行三个分析
    market_task = market_agent.run(f"分析市场：{company_info}")
    tech_task = tech_agent.run(f"分析技术：{company_info}")
    financial_task = financial_agent.run(f"分析财务：{company_info}")
    
    # 等待所有任务完成
    market, tech, financial = await asyncio.gather(
        market_task, tech_task, financial_task
    )
    
    return {
        "market": market.data,
        "tech": tech.data,
        "financial": financial.data,
    }

10.3 条件路由

根据输入动态选择不同的 Agent：

from pydantic_ai import Agent
from pydantic import BaseModel

class ClassificationResult(BaseModel):
    category: str  # "technical", "business", "legal"
    confidence: float

class TechnicalReport(BaseModel):
    architecture: str
    performance_metrics: dict
    optimization_suggestions: list[str]

class BusinessReport(BaseModel):
    market_opportunity: str
    revenue_potential: float
    go_to_market_strategy: str

class LegalReport(BaseModel):
    compliance_status: str
    risk_areas: list[str]
    required_actions: list[str]

# 分类 Agent
classifier = Agent('openai:gpt-4o', result_type=ClassificationResult)

# 专业 Agent
technical_agent = Agent('openai:gpt-4o', result_type=TechnicalReport)
business_agent = Agent('openai:gpt-4o', result_type=BusinessReport)
legal_agent = Agent('openai:gpt-4o', result_type=LegalReport)

async def smart_analysis(document: str) -> dict:
    # Step 1: 分类
    classification = await classifier.run(f"分析文档类别：{document}")
    
    # Step 2: 根据类别选择专业 Agent
    category = classification.data.category
    
    if category == "technical":
        result = await technical_agent.run(f"技术分析：{document}")
    elif category == "business":
        result = await business_agent.run(f"商业分析：{document}")
    elif category == "legal":
        result = await legal_agent.run(f"法律分析：{document}")
    else:
        raise ValueError(f"未知类别：{category}")
    
    return {
        "category": category,
        "confidence": classification.data.confidence,
        "report": result.data,
    }

十一、真实案例：从 LangChain 迁移到 PydanticAI

11.1 迁移动机

某电商平台的订单分析系统原来使用 LangChain，面临以下问题：

1. 输出不稳定：LLM 返回的 JSON 格式不一致，下游系统频繁报错
2. 测试困难：无法对 Agent 逻辑写有效的单元测试
3. 维护成本高：prompt 工程需要反复调试，每次修改都可能破坏现有功能
4. 性能问题：LangChain 的抽象层引入了不必要的开销

11.2 迁移前的 LangChain 代码

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import JsonOutputParser
from pydantic import BaseModel

class OrderAnalysis(BaseModel):
    total_orders: int
    avg_order_value: float
    top_products: list[str]
    risk_level: str

# LangChain 方式
llm = ChatOpenAI(model="gpt-4o")
parser = JsonOutputParser(pydantic_object=OrderAnalysis)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个订单分析师。\n{format_instructions}"),
    ("human", "分析以下订单数据：{data}")
])

chain = prompt | llm | parser

# 问题：parser 只做基础 JSON 解析，不做 Pydantic 校验
# 如果 LLM 返回的字段名不对，不会报错，只是静默失败
result = chain.invoke({
    "data": order_data,
    "format_instructions": parser.get_format_instructions()
})
# result 可能是 dict，不是 OrderAnalysis 类型
# 需要手动校验

11.3 迁移后的 PydanticAI 代码

from pydantic_ai import Agent
from pydantic import BaseModel, Field

class OrderAnalysis(BaseModel):
    total_orders: int = Field(ge=0, description="总订单数")
    avg_order_value: float = Field(ge=0, description="平均客单价")
    top_products: list[str] = Field(min_length=1, max_length=5, description="TOP产品")
    risk_level: str = Field(description="风险等级：low/medium/high")

agent = Agent(
    'openai:gpt-4o',
    system_prompt='你是一个订单分析师，输出结构化的分析报告。',
    result_type=OrderAnalysis,
    retries=3,
)

@agent.tool
def query_orders(ctx, start_date: str, end_date: str) -> str:
    """查询订单数据"""
    # 实际的数据库查询逻辑
    return "订单数据..."

# 使用
result = agent.run_sync("分析Q2订单数据")
# result.data 是 OrderAnalysis 类型，自动校验通过
print(result.data.total_orders)  # int 类型
print(result.data.avg_order_value)  # float 类型

11.4 迁移效果

十二、性能优化实战

12.1 减少 Token 消耗

# 不好的做法：冗长的系统提示
bad_prompt = """
你是一个数据分析专家。你的任务是分析销售数据并生成报告。
请确保你的报告包含以下字段：
1. total_revenue (number): 总收入，必须大于等于0
2. growth_rate (number): 增长率，在-1到10之间
3. top_products (array of strings): TOP产品，1到5个
4. risk_factors (array of strings): 风险因素，最多3个
5. confidence (number): 置信度，在0到1之间

请严格按照以上格式输出 JSON。
"""

# 好的做法：让 PydanticAI 自动处理
good_agent = Agent(
    'openai:gpt-4o',
    system_prompt='你是一个数据分析专家。',  # 简洁的提示
    result_type=SalesAnalysis,  # Schema 自动注入
)

12.2 并发控制

import asyncio
from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', result_type=str)

async def process_batch(items: list[str], max_concurrent: int = 5):
    """批量处理，控制并发数"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_one(item: str):
        async with semaphore:
            result = await agent.run(f"处理：{item}")
            return result.data
    
    # 并发执行，但最多 max_concurrent 个
    tasks = [process_one(item) for item in items]
    return await asyncio.gather(*tasks)

12.3 缓存策略

from functools import lru_cache
from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', result_type=str)

# 对相同输入缓存结果
@lru_cache(maxsize=1000)
def cached_analysis(query: str) -> str:
    result = agent.run_sync(query)
    return result.data

# 使用
result1 = cached_analysis("分析Q1数据")  # 调用 LLM
result2 = cached_analysis("分析Q1数据")  # 直接返回缓存

十三、最佳实践总结

13.1 模型设计原则

1. 字段命名清晰：使用 total_revenue 而不是 tr
2. 添加描述：每个字段都加 description，帮助 LLM 理解
3. 设置合理约束：ge=0、max_length=5 等约束既帮助 LLM 又帮助校验
4. 使用 Enum：对于有限选项，用 Enum 而不是 str
5. 交叉校验：用 @model_validator 校验字段间的一致性

13.2 工具设计原则

1. 函数签名即文档：参数类型和默认值就是最好的文档
2. 添加 docstring：LLM 会读取工具的描述来决定何时使用
3. 异常处理：工具内部处理异常，返回友好的错误消息
4. 幂等性：工具应该是幂等的，重试不会产生副作用
5. 超时控制：设置合理的超时，避免工具调用卡住

13.3 测试策略

1. 使用 TestModel：不调用真实 LLM 测试 Agent 逻辑
2. Mock 依赖：通过 RunContext 注入 Mock 依赖
3. 校验边界：测试 Field 约束和自定义校验器
4. 集成测试：用真实的 LLM 进行端到端测试（定期运行，不是每次提交）
5. 监控重试率：跟踪校验失败率，优化 prompt 或 Schema

13.4 生产部署

1. 使用异步：生产环境用 agent.run() 而不是 agent.run_sync()
2. 连接池：数据库等依赖使用连接池
3. 重试策略：设置合理的重试次数和退避策略
4. 监控告警：监控 LLM 调用延迟、错误率、token 消耗
5. 日志记录：记录每次调用的输入输出，便于调试

结语：类型安全是 AI 工程化的基石

PydanticAI 的核心洞察是：AI Agent 的可靠性不取决于模型有多聪明，而取决于工程有多扎实。

当你的 Agent 能像调用 int() 一样可靠地获取结构化数据时，真正有价值的 AI 应用才刚刚开始。PydanticAI 用 Python 类型系统这座桥梁，连接了 LLM 的无限可能和生产系统的严格要求。

在这个 AI Agent 爆发的时代，选择一个正确的框架，不是选择一个最花哨的工具，而是选择一个最能帮你把 AI 能力转化为可靠产品的基础设施。PydanticAI，正是这样的基础设施。

结语：类型安全是 AI 工程化的基石

PydanticAI 的核心洞察是：AI Agent 的可靠性不取决于模型有多聪明，而取决于工程有多扎实。

当你的 Agent 能像调用 int() 一样可靠地获取结构化数据时，真正有价值的 AI 应用才刚刚开始。PydanticAI 用 Python 类型系统这座桥梁，连接了 LLM 的无限可能和生产系统的严格要求。

在这个 AI Agent 爆发的时代，选择一个正确的框架，不是选择一个最花哨的工具，而是选择一个最能帮你把 AI 能力转化为可靠产品的基础设施。PydanticAI，正是这样的基础设施。

十四、PydanticAI 与 Pydantic v2 的深度集成

14.1 利用 Pydantic v2 的新特性

PydanticAI 充分利用了 Pydantic v2 的新特性，包括更快的校验速度、更好的类型支持和更灵活的自定义：

from pydantic import BaseModel, Field, field_serializer, model_serializer
from datetime import datetime
from typing import Annotated

class TimestampedAnalysis(BaseModel):
    """带时间戳的分析结果"""
    analysis_id: str = Field(description="分析唯一ID")
    timestamp: datetime = Field(description="分析时间")
    result: dict = Field(description="分析结果")
    
    @field_serializer('timestamp')
    def serialize_timestamp(self, dt: datetime) -> str:
        """自定义时间序列化"""
        return dt.isoformat()
    
    @model_serializer
    def serialize_model(self) -> dict:
        """自定义模型序列化"""
        return {
            "id": self.analysis_id,
            "time": self.timestamp.isoformat(),
            "data": self.result,
            "version": "1.0",
        }

# PydanticAI 会自动使用这些序列化器
# LLM 看到的 Schema 会包含自定义的序列化逻辑

14.2 类型安全的工具参数

PydanticAI 支持复杂的类型注解，包括 Annotated、Literal、Union 等：

from typing import Annotated, Literal, Union
from pydantic import Field

@agent.tool
def complex_tool(
    ctx: RunContext[None],
    # 使用 Annotated 添加元数据
    query: Annotated[str, Field(min_length=1, max_length=1000, description="搜索查询")],
    # 使用 Literal 限制选项
    mode: Literal["fast", "accurate", "balanced"] = "balanced",
    # 使用 Union 支持多种类型
    filter: Union[str, list[str], None] = None,
    # 使用复杂嵌套类型
    options: dict[str, Union[int, str, bool]] | None = None,
) -> str:
    """复杂工具示例
    
    Args:
        query: 搜索查询，1-1000字符
        mode: 搜索模式：fast（快速）、accurate（精确）、balanced（平衡）
        filter: 过滤条件，可以是字符串、字符串列表或None
        options: 额外选项字典
    """
    # LLM 会看到完整的类型信息和约束
    # PydanticAI 会自动校验所有参数
    return f"Results for {query} in {mode} mode"

14.3 自定义 JSON Schema

对于需要精细控制 JSON Schema 的场景，PydanticAI 允许自定义：

from pydantic import BaseModel, ConfigDict

class CustomSchema(BaseModel):
    model_config = ConfigDict(
        json_schema_extra={
            "examples": [
                {
                    "total_revenue": 1000000.0,
                    "growth_rate": 0.15,
                    "top_products": ["产品A", "产品B"],
                    "confidence": 0.85,
                }
            ],
            "title": "销售分析结果",
            "description": "包含收入、增长率、TOP产品和置信度的分析结果",
        }
    )
    
    total_revenue: float = Field(ge=0)
    growth_rate: float = Field(ge=-1, le=10)
    top_products: list[str] = Field(min_length=1, max_length=5)
    confidence: float = Field(ge=0, le=1)

# PydanticAI 会将这些额外信息注入到 LLM 的提示中
# 帮助 LLM 更好地理解期望的输出格式

十五、PydanticAI 生态系统

15.1 官方工具

• pydantic-ai-slim：核心库，轻量级版本
• pydantic-ai-examples：官方示例集合
• pydantic-ai-docs：官方文档
• pydantic-ai-playground：在线实验环境

15.2 第三方集成

• Langfuse：可观测性平台集成
• OpenTelemetry：分布式追踪
• FastAPI：Web 框架集成
• Celery：异步任务队列集成
• Redis：缓存和会话管理

15.3 社区贡献

• pydantic-ai-mcp-adapters：MCP 服务器适配器集合
• pydantic-ai-tools：常用工具集合
• pydantic-ai-templates：Agent 模板库
• pydantic-ai-benchmarks：性能基准测试

十六、常见问题与解决方案

16.1 FAQ

Q: PydanticAI 和 LangChain 可以一起用吗？

A: 可以，但不推荐。PydanticAI 的设计理念是薄封装，如果你需要 LangChain 的生态，建议直接使用 LangChain。混合使用会增加复杂度。

Q: 如何处理 LLM 拒绝生成 JSON？

A: PydanticAI 会自动重试，并在错误反馈中明确要求 JSON 格式。如果模型持续拒绝，可以调整 system prompt 或换一个更支持 structured output 的模型。

Q: 支持流式输出吗？

A: 支持。使用 agent.run_stream() 可以获得流式输出，同时保持类型安全。

Q: 如何处理大模型的上下文窗口限制？

A: PydanticAI 本身不管理上下文窗口，但你可以通过工具和依赖注入来控制输入数据的大小。

16.2 调试技巧

# 1. 启用详细日志
import logging
logging.basicConfig(level=logging.DEBUG)

# 2. 使用 TestModel 查看 Schema
from pydantic_ai.models.test import TestModel
test_model = TestModel()
agent = Agent(test_model, result_type=SalesAnalysis)
# 查看 agent 的 system prompt 和工具定义

# 3. 检查重试历史
result = agent.run_sync("分析数据")
print(f"重试次数：{result.usage.requests - 1}")
print(f"总 token 消耗：{result.usage.total_tokens}")

# 4. 使用 Pydantic 的调试模式
from pydantic import TypeAdapter
adapter = TypeAdapter(SalesAnalysis)
# 验证 LLM 输出
try:
    validated = adapter.validate_python(llm_output)
except ValidationError as e:
    print(f"校验错误：{e}")