使用 Pydantic AI 在 Python 中构建 AI 智能体

引言

本文将带你学习如何使用 Pydantic AI 在 Python 中构建生产就绪的 AI 智能体，涵盖结构化输出、自定义工具以及依赖注入。

主要内容包括：

• Pydantic AI 如何通过类型安全的验证输出改进原始 LLM API 调用
• 如何将 Pydantic 模型定义为结构化输出模式
• 如何将 Python 函数注册为智能体工具
• 如何使用依赖注入实现简洁可测试的智能体设计
• 部署 Pydantic AI 智能体的最佳实践

什么是 Pydantic AI？

Pydantic AI 是由 Pydantic 团队构建的 Python 智能体框架——这个团队正是 FastAPI 及数百个其他项目所使用的 Pydantic 数据验证库的创造者。

其核心理念与 FastAPI 一脉相承：将 FastAPI 为 REST API 开发带来的那种符合人体工程学、类型安全的开发体验移植到 AI 智能体开发中。

主要特性：

• 结构化、验证后的 LLM 输出
  ——通过 Pydantic 模型实现
• 函数工具注册
  ——自动生成 JSON Schema
• 依赖注入
  ——清晰的关注点分离
• 模型无关
  ——支持 OpenAI、Google Gemini、Anthropic 及本地模型
• 原生异步
  ——同步封装方便脚本使用

安装

pip install pydantic-ai

# 或按模型提供商安装：
pip install "pydantic-ai-slim[google]"# Google Gemini
pip install "pydantic-ai-slim[openai]"# OpenAI

# 设置 API 密钥
export GOOGLE_API_KEY="your-api-key-here"
# Windows PowerShell：
$ENV:GOOGLE_API_KEY = "your-api-key-here"

1. 创建基础智能体

最简单的 Pydantic AI 智能体只需要一个模型名称和一个指令字符串：

from pydantic_ai import Agent

agent = Agent(
"google-gla:gemini-2.5-flash",
    instructions="你是一位 Python 专家。请用一句话回答。",
)

result = agent.run_sync("什么是 Pydantic AI？")
print(result.output)

输出示例：

Pydantic AI 是一个 Python 智能体框架，用于使用 Pydantic 模型驱动的经过验证的 
LLM 输出构建类型安全的生产级 AI 应用程序。

注意： 在脚本和笔记本中使用 run_sync()；在生产异步代码中使用 await agent.run()。

2. 使用 Pydantic 模型实现结构化输出

Pydantic AI 最有价值的特性之一是强制 LLM 返回结构化、类型验证的数据：

from pydantic import BaseModel
from pydantic_ai import Agent

classCityInfo(BaseModel):
    name: str
    country: str
    population: int
    fun_fact: str

agent = Agent(
"google-gla:gemini-2.5-flash",
    output_type=CityInfo
)

result = agent.run_sync("告诉我关于东京的信息")
print(result.output)
print(f"{result.output.name}, {result.output.country}")
print(f"人口：{result.output.population:,}")
print(f"趣闻：{result.output.fun_fact}")

工作原理：

1. Pydantic AI 将模型类转换为 JSON Schema
2. Schema 作为响应格式约束发送给 LLM
3. LLM 返回符合 Schema 的 JSON
4. Pydantic 验证并将响应解析为你的模型
5. 验证失败时自动重试（通过 output_retries 参数配置）

3. 注册自定义工具

工具允许你的智能体调用 Python 函数——从 API 获取数据、查询数据库、执行计算：

使用 @agent.tool_plain 装饰器

对于不需要访问智能体运行上下文的工具使用 tool_plain：

import requests
from pydantic_ai import Agent

agent = Agent(
"google-gla:gemini-2.5-flash",
    instructions="帮助用户了解猫的品种信息。请简洁回答。",
)

@agent.tool_plain
deffind_breed_info(breed_name: str) -> dict:
"""查找猫品种的相关信息。"""
    response = requests.get("https://api.thecatapi.com/v1/breeds")
    response.raise_for_status()
for breed in response.json():
if breed["name"] == breed_name:
return breed
return {"error": "未找到该品种"}

result = agent.run_sync("告诉我暹罗猫的信息。")
print(result.output)

在初始化时传入工具列表

适合在多个智能体间复用工具：

agent = Agent(
"google-gla:gemini-2.5-flash",
    instructions="帮助用户了解猫的品种信息。请简洁回答。",
    tools=[find_breed_info]  # 直接传入工具函数列表
)

4. 依赖注入

依赖注入是 Pydantic AI 最强大的设计模式之一。它允许你将外部资源（数据库连接、API 客户端、配置参数）传递给智能体工具，而无需硬编码——使你的智能体更整洁、可测试，更适合生产环境。

定义依赖项

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

classUserDatabase:
"""模拟使用外部 API 的用户数据库。"""
    _base_url = "https://jsonplaceholder.typicode.com"

defget_user_info(self, user_id: int) -> dict:
        response = requests.get(f"{self._base_url}/users/{user_id}")
        response.raise_for_status()
return response.json()

classUserSummary(BaseModel):
    name: str
    email: str
    company: str

agent = Agent(
"google-gla:gemini-2.5-flash",
    output_type=UserSummary,
    deps_type=UserDatabase,  # 声明依赖类型
    instructions=(
"你从外部数据库获取用户信息。"
"使用可用工具收集用户信息，"
"然后返回结构化摘要。"
    ),
)

@agent.tool  # 需要访问运行上下文时使用 @agent.tool（而非 tool_plain）
deffetch_user(ctx: RunContext[UserDatabase], user_id: int) -> str:
"""从服务中获取用户档案。"""
try:
        user = ctx.deps.get_user_info(user_id)
returnstr(user)
except requests.HTTPError:
returnf"未找到 ID 为 {user_id} 的用户"

# 在运行时注入依赖
db = UserDatabase()
result = agent.run_sync("获取用户 7 的摘要", deps=db)

print(f"姓名：{result.output.name}")
print(f"邮箱：{result.output.email}")
print(f"公司：{result.output.company}")

使用依赖注入进行测试

依赖注入模式使测试变得简单——无需修改智能体定义即可替换依赖：

classMockDatabase:
defget_user_info(self, user_id: int) -> dict:
return {"name": "测试用户", "email": "test@example.com", 
"company": {"name": "测试公司"}}

with agent.override(deps=MockDatabase()):
    result = agent.run_sync("获取用户 7 的摘要")
# 使用模拟数据——无需真实 API 调用

5. 综合示例

以下是结合结构化输出、工具和依赖注入的完整示例：

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

classWeatherDeps:
def__init__(self, api_key: str):
self.api_key = api_key

classWeatherReport(BaseModel):
    city: str
    temperature_celsius: float
    conditions: str
    recommendation: str

agent = Agent(
"google-gla:gemini-2.5-flash",
    output_type=WeatherReport,
    deps_type=WeatherDeps,
    instructions="你是一个天气助手。获取天气数据并提供带建议的报告。",
)

@agent.tool
defget_weather(ctx: RunContext[WeatherDeps], city: str) -> dict:
"""获取某城市的当前天气。"""
    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {"q": city, "appid": ctx.deps.api_key, "units": "metric"}
    response = requests.get(url, params=params)
return response.json()

deps = WeatherDeps(api_key="your-openweather-api-key")
result = agent.run_sync("伦敦现在天气怎么样？", deps=deps)
print(result.output)

最佳实践

总结

Pydantic AI 将 Pydantic 的类型安全、验证数据理念带入了 AI 智能体开发：

• 结构化输出
  确保智能体返回可在程序中直接使用的数据
• 自定义工具
  赋予智能体访问真实世界数据和执行操作的能力
• 依赖注入
  使智能体代码整洁、模块化且易于测试
• 模型无关设计
  意味着你只需修改极少的代码即可在 OpenAI、Gemini、Anthropic 或本地模型之间切换

对于可靠性和可维护性至关重要的生产 AI 系统，Pydantic AI 是目前最实用的选择之一。