Pydantic AI 完整指南 2026：类型安全的 AI Agent 开发框架

在 2026 年的 AI Agent 开发生态中，Pydantic AI 以其独特的类型安全理念和简洁的 API 设计脱颖而出。作为 Pydantic 团队的最新力作，它将数据验证的严谨性与 LLM 的灵活性完美结合，成为构建生产级 AI 应用的首选框架。

为什么选择 Pydantic AI？

核心优势

类型安全是第一公民。与 LangGraph、CrewAI 等框架不同，Pydantic AI 从底层就建立在 Pydantic 的类型系统之上。这意味着：

• ✅ 函数参数和返回值自动验证
• ✅ LLM 输出结构化，避免解析错误
• ✅ IDE 自动补全和类型检查
• ✅ 运行时错误大幅减少

简洁的 API 设计。无需学习复杂的状态机或图论概念，用纯 Python 代码即可构建强大的 Agent。

原生流式支持。内置 SSE 和流式响应，轻松实现打字机效果。

多模型兼容。支持 OpenAI、Anthropic、Google、Ollama 以及任何兼容 OpenAI API 的模型。

快速开始

安装

pip install pydantic-ai

基础示例：第一个 Agent

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel

# 初始化模型
model = OpenAIModel('gpt-4o')

# 创建 Agent
agent = Agent(model)

# 运行对话
result = agent.run_sync('解释量子纠缠，用高中生能理解的语言')
print(result.data)

结构化输出

这是 Pydantic AI 最强大的功能之一：

from pydantic import BaseModel
from pydantic_ai import Agent

class WeatherReport(BaseModel):
    temperature: float
    condition: str
    humidity: int
    recommendation: str

agent = Agent('openai:gpt-4o', result_type=WeatherReport)
result = agent.run_sync('北京今天的天气如何？')

# 直接获得 Pydantic 模型实例
weather: WeatherReport = result.data
print(f"温度：{weather.temperature}°C")
print(f"建议：{weather.recommendation}")

核心概念详解

1. 依赖注入系统

Pydantic AI 提供优雅的依赖注入，让 Agent 可以访问外部工具和服务：

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class AppDeps:
    user_id: str
    api_key: str

agent = Agent('openai:gpt-4o', deps_type=AppDeps)

@agent.tool
async def get_user_profile(ctx: RunContext[AppDeps]) -> str:
    """获取当前用户资料"""
    # ctx.deps.user_id 和 ctx.deps.api_key 可用
    return f"用户 {ctx.deps.user_id} 的资料..."

# 运行时传入依赖
result = await agent.run(
    '显示我的个人资料',
    deps=AppDeps(user_id='12345', api_key='secret')
)

2. 多步骤对话

from pydantic_ai import Agent

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

# 第一轮
result1 = agent.run_sync('我想学习 Python，给我制定一个学习计划')
print(result1.data)

# 继续对话（保持上下文）
result2 = agent.run_sync('第二周应该重点学什么？', message_history=result1.history)
print(result2.data)

3. 流式响应

from pydantic_ai import Agent

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

async def stream_response():
    async with agent.run_stream('写一首关于 AI 的短诗') as result:
        async for message in result.stream_text():
            print(message, end='', flush=True)

# 或使用 run_stream_sync
for message in agent.run_stream_sync('讲个程序员笑话'):
    print(message, end='')

实战案例：智能客服 Agent

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from typing import Literal

class SupportTicket(BaseModel):
    category: Literal['技术', '账单', '产品咨询', '其他']
    priority: Literal['低', '中', '高', '紧急']
    summary: str = Field(description='问题摘要，50 字以内')
    suggested_action: str

class SupportDeps:
    def __init__(self, customer_id: str):
        self.customer_id = customer_id
        self.tier = 'premium'  # 从数据库获取

agent = Agent(
    'anthropic:claude-3-7-sonnet',
    result_type=SupportTicket,
    deps_type=SupportDeps
)

@agent.tool
async def check_customer_history(ctx: RunContext[SupportDeps]) -> str:
    """查询客户历史工单"""
    return f"客户 {ctx.deps.customer_id} 历史：3 个已解决工单"

@agent.tool
async def escalate_ticket(ctx: RunContext[SupportDeps], reason: str) -> str:
    """升级工单到人工客服"""
    return f"工单已升级，原因：{reason}"

# 使用
deps = SupportDeps(customer_id='CUST-789')
result = agent.run_sync(
    '我的账户被错误扣费了，这已经是第三次发生！',
    deps=deps
)

ticket: SupportTicket = result.data
print(f"分类：{ticket.category}")
print(f"优先级：{ticket.priority}")
print(f"建议操作：{ticket.suggested_action}")

高级特性

1. 结果验证器

from pydantic_ai import Agent, ResultValidator

def validate_length(result):
    if len(result.data) < 10:
        raise ValueError('回复太短，请详细说明')
    return result

agent = Agent('openai:gpt-4o')
agent.result_validator(validate_length)

2. 自定义模型提供者

from pydantic_ai.models import Model
from pydantic_ai.messages import ModelRequest, ModelResponse

class CustomModel(Model):
    async def request(self, request: ModelRequest) -> ModelResponse:
        # 实现自定义推理逻辑
        pass

agent = Agent(CustomModel())

3. 批量处理

from pydantic_ai import Agent

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

prompts = [
    '总结这篇文章',
    '提取关键词',
    '生成标题',
]

results = agent.run_sync_multiple(prompts)
for i, result in enumerate(results):
    print(f"任务 {i+1}: {result.data}")

性能对比

根据 2026 年 Q1 的基准测试：

框架	首次 Token 延迟	类型安全	学习曲线	生产就绪度
Pydantic AI	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐
LangGraph	⭐⭐⭐	⭐⭐⭐	⭐⭐	⭐⭐⭐⭐⭐
CrewAI	⭐⭐⭐⭐	⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐
AutoGen	⭐⭐	⭐⭐⭐	⭐⭐	⭐⭐⭐

最佳实践

1. 始终定义结果类型

# ❌ 不推荐
agent = Agent('openai:gpt-4o')

# ✅ 推荐
class Response(BaseModel):
    answer: str
    sources: list[str]

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

2. 合理使用工具

# ❌ 避免过多工具
@agent.tool
@agent.tool
@agent.tool
@agent.tool  # 太多工具会降低性能

# ✅ 聚焦核心功能
@agent.tool
@agent.tool  # 2-3 个关键工具

3. 错误处理

from pydantic_ai.exceptions import ModelRetry

@agent.tool
async def search_database(query: str) -> str:
    try:
        # 搜索逻辑
        pass
    except Exception as e:
        raise ModelRetry(f'搜索失败：{str(e)}')

生态整合

与 FastAPI 集成

from fastapi import FastAPI
from pydantic_ai import Agent

app = FastAPI()
agent = Agent('openai:gpt-4o')

@app.post('/chat')
async def chat(prompt: str):
    result = await agent.run(prompt)
    return {'response': result.data}

与 LangChain 互操作

from langchain.pydantic_ai import PydanticAIWrapper

wrapper = PydanticAIWrapper(agent)
chain = wrapper | output_parser

总结

Pydantic AI 代表了 AI Agent 开发的未来方向：类型安全、简洁优雅、生产就绪。如果你正在寻找：

• 🎯 减少运行时错误
• 🎯 提高代码可维护性
• 🎯 快速迭代原型
• 🎯 无缝集成现有 Python 项目

那么 Pydantic AI 值得你投入时间学习。

参考资源

• Pydantic AI 官方文档
• GitHub 仓库
• Pydantic 官方文档
• 示例项目集合
• 社区讨论

---

最后更新：2026-03-28