2026年5月10日 3 分钟阅读

Pydantic AI 实战指南:用类型安全构建生产级 AI Agent

tinyash 0 条评论

在 2026 年的 AI Agent 框架混战中,大多数框架在追求”快速搭建”的路上越走越远——启动一个 Demo 只需要 5 行代码,但把它推向生产环境却要踩无数坑。Pydantic AI 选择了另一条路:它从一开始就为生产环境而设计。

截至 2026 年 5 月,Pydantic AI 已迭代至 v1.92.0,GitHub 星标突破 20K,在一项 2026 年 Agent 框架基准测试中,它的开发者体验评分达到 8/10,远超 LangChain(5/10)和 CrewAI(6/10)。更重要的是,它的类型安全机制在开发阶段就拦截了 23 个潜在 Bug——这些 Bug 在其他框架中很可能直接冲进生产环境。

为什么类型安全对 AI Agent 如此重要?

LLM 的输出本质上是不可靠的。你让模型返回 JSON,它可能在末尾多打一个逗号;你让它返回一个城市名的列表,它可能给你一段解释文字。传统做法是写一堆正则表达式和 try-catch 去解析这些输出——脆、丑、而且难以维护。

Pydantic AI 的解决方案很优雅:让 Pydantic 模型成为 LLM 输出的”契约”

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class CityInfo(BaseModel):
    name: str
    country: str
    population: int
    fun_fact: str = Field(max_length=200)

agent = Agent(
    "google-gla:gemini-2.5-flash",
    system_prompt="你是一个地理信息助手。",
    output_type=CityInfo,
)

result = agent.run_sync("Tell me about Tokyo")
city = result.output
print(f"{city.name}: population {city.population:,}")

背后发生的事情:Pydantic AI 自动将 CityInfo 转换为 JSON Schema 发送给 LLM,LLM 返回符合 Schema 的数据,Pydantic 进行校验。如果校验失败,框架会将错误信息反馈给模型并自动重试——模型能看到”刚才哪里出了问题”并自我纠正。

你拿到的 result.output 是一个强类型的 CityInfo 实例,IDE 会给你完整的字段补全,类型检查器会在编译期发现问题。不再有字符串解析,不再有”这个字段到底叫什么来着”。

工具注册:让 Agent 真正干活

光有结构化输出还不够。AI Agent 需要能调用外部 API、查询数据库、操作文件。Pydantic AI 通过 @agent.tool 装饰器注册类型安全的工具函数:

import httpx
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class DeploymentDeps:
    http_client: httpx.AsyncClient
    api_base_url: str

agent = Agent(
    "openai:gpt-4o",
    deps_type=DeploymentDeps,
    system_prompt="你帮助用户检查部署状态。",
)

@agent.tool
async def get_deployment_status(
    ctx: RunContext[DeploymentDeps], deployment_id: str
) -> str:
    """通过 ID 获取部署的当前状态。"""
    response = await ctx.deps.http_client.get(
        f"{ctx.deps.api_base_url}/deployments/{deployment_id}"
    )
    response.raise_for_status()
    data = response.json()
    return f"部署 {deployment_id}: {data['status']}"

@agent.tool
async def list_recent_deployments(
    ctx: RunContext[DeploymentDeps], limit: int = 5
) -> str:
    """列出最近的部署记录。"""
    response = await ctx.deps.http_client.get(
        f"{ctx.deps.api_base_url}/deployments",
        params={"limit": limit, "sort": "desc"}
    )
    response.raise_for_status()
    data = response.json()
    return "\n".join(
        f"- {d['id']}: {d['status']}" for d in data[:limit]
    )

工具函数的参数类型由 Python 类型注解自动推导,文档字符串会成为模型的工具描述。调用时,Agent 会根据用户意图自动选择和组合工具。

依赖注入:告别全局状态

生产环境中,Agent 通常需要访问数据库连接、HTTP 客户端、配置对象等运行时上下文。全局变量在这里是灾难——测试困难、并发不安全、配置难以切换。

Pydantic AI 的 deps_type 提供了类型安全的依赖注入:

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
import asyncpg

@dataclass
class DatabaseDeps:
    db_pool: asyncpg.Pool

agent = Agent(
    "openai:gpt-4o",
    deps_type=DatabaseDeps,
)

@agent.tool
async def query_user_by_email(
    ctx: RunContext[DatabaseDeps], email: str
) -> str:
    async with ctx.deps.db_pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, name, role FROM users WHERE email = $1", email
        )
    if row is None:
        return f"未找到邮箱为 {email} 的用户。"
    return f"用户 {row['name']}(ID: {row['id']}),角色: {row['role']}"

在测试中,你只需要注入 Mock 依赖即可,无需修改任何 Agent 代码。

多模型支持:不锁定供应商

Pydantic AI 支持 15+ 模型提供商,切换模型只需改一行字符串:

# OpenAI
agent = Agent("openai:gpt-4o")

# Anthropic
agent = Agent("anthropic:claude-opus-4-7")

# Google
agent = Agent("google-gla:gemini-2.5-pro")

# DeepSeek(兼容 OpenAI API)
agent = Agent("openai:deepseek-chat",
    model_settings={"base_url": "https://api.deepseek.com/v1"}
)

# 本地 Ollama
agent = Agent("openai:llama3.1",
    model_settings={"base_url": "http://localhost:11434/v1"}
)

你的 Schema 定义、工具函数、依赖注入逻辑全部保持不变。这对于需要多模型对比测试、或者不想被单一供应商锁定的团队来说,是巨大的成本节约。

生产级特性:日志、流式与持久化

Pydantic AI 内置了对 Logfire 的支持(OpenTelemetry 兼容),一行代码即可接入可观测性:

import logfire
logfire.configure()
logfire.instrument_pydantic_ai()

Agent 的每一步推理、每次工具调用、每条 LLM 请求都会自动生成结构化日志,方便在 Logfire 或 Grafana 中进行追踪和调试。

流式输出同样类型安全:

async with agent.run_stream("Generate 5 city info records") as stream:
    async for chunk in stream.stream_structured():
        if chunk.is_partial:
            print(f"构建中... {chunk.partial}")
        else:
            city = chunk.output  # 类型安全的 CityInfo
            print(f"✓ {city.name}")

对于长时间运行的 Agent 任务,Pydantic AI 支持通过 Temporal、Prefect 或 DBOS 实现持久化执行——Agent 意外崩溃后可以从断点恢复,而不是从头开始。

何时选择 Pydantic AI?

场景Pydantic AI替代方案
需要 LLM 返回结构化、可校验的数据✅ 最佳选择
快速原型或单 Agent 应用✅ 适用
已使用 Pydantic / FastAPI 技术栈✅ 完美契合
需要大量预置集成(向量库、检索器等)LangChain、LlamaIndex
需要极致精细的 Prompt 控制直接调用 API

快速上手

pip install "pydantic-ai[all]"
export OPENAI_API_KEY="sk-..."
from pydantic_ai import Agent

agent = Agent("openai:gpt-4o")
result = agent.run_sync("用一句话解释什么是 Python GIL")
print(result.output)

总结

Pydantic AI 没有试图重新发明 AI Agent——它只是用 Pydantic 团队在数据验证领域十几年的积累,把 Agent 开发中最容易出错的环节(输出解析、类型转换、依赖管理)处理得干净利落。

如果你厌倦了在 try-catch 中解析 LLM 返回的”半成品 JSON”,如果你想让 AI Agent 的输出像 FastAPI 的请求体一样类型安全——Pydantic AI 值得一试。


参考资源:

发表评论

你的邮箱地址不会被公开,带 * 的为必填项。