PydanticAI：类型安全的 AI Agent 框架¶

为什么要学 PydanticAI¶

构建 AI Agent 时，最常遇到的痛点是：LLM 的输出不可控、工具调用缺乏类型校验、调试困难、多模型切换成本高。LangChain 虽然功能全面但过度抽象，简单任务也需要大量样板代码。

PydanticAI 由 Pydantic 团队开发，核心设计哲学是用 Python 类型系统驯服 LLM 的不确定性。它借鉴了 FastAPI 的设计理念——通过类型注解自动完成验证、序列化和文档生成。如果你熟悉 FastAPI，PydanticAI 的 Agent 定义方式会让你立刻上手。

核心优势： - 类型安全的结构化输出（基于 Pydantic Model） - 模型无关设计（OpenAI、Anthropic、Gemini、Groq、Ollama 一行切换） - 依赖注入系统（类似 FastAPI 的 Depends） - 与 Logfire 深度集成（可观测性开箱即用） - 流式响应支持 - 原生 Python，不发明新概念

核心概念¶

白话解释¶

概念	白话说明
Agent	一个封装了系统提示、工具、输出类型的智能体定义
Tool	Agent 可调用的函数，框架自动从类型注解生成 schema
Structured Result	用 Pydantic Model 定义的输出格式，LLM 必须按格式返回
Dependencies	注入到工具/系统提示中的运行时上下文（如DB连接、用户信息）
System Prompt	可以是静态字符串，也可以是动态函数（基于依赖注入）
ModelRetry	当输出不符合 schema 时自动重试
RunContext	运行时上下文对象，携带依赖和重试信息
Usage	token 用量统计
Message History	对话历史，可持久化和恢复

与其他框架对比¶

特性	PydanticAI	LangChain	CrewAI
类型安全	原生 Pydantic	弱	无
学习曲线	低（类FastAPI）	高	中
抽象层级	薄（接近原生）	厚	中
多模型支持	好	好	好
可观测性	Logfire集成	LangSmith	无
流式输出	支持	支持	有限
适用场景	生产级Agent	快速原型	多Agent协作

安装配置¶

安装¶

# 基础安装
pip install pydantic-ai

# 安装特定模型支持
pip install pydantic-ai[openai]      # OpenAI
pip install pydantic-ai[anthropic]   # Anthropic/Claude
pip install pydantic-ai[groq]        # Groq
pip install pydantic-ai[google]      # Gemini

环境变量配置¶

# .env 文件
OPENAI_API_KEY=sk-xxx
ANTHROPIC_API_KEY=sk-ant-xxx
GEMINI_API_KEY=xxx
GROQ_API_KEY=gsk_xxx

验证安装¶

from pydantic_ai import Agent

agent = Agent("openai:gpt-4o-mini")
result = agent.run_sync("说一句话证明你在工作")
print(result.data)

快速上手¶

最简 Agent¶

from pydantic_ai import Agent

# 创建一个最简单的 Agent
agent = Agent(
    "openai:gpt-4o-mini",
    system_prompt="你是一个乐于助人的助手，回答要简洁。"
)

# 同步运行
result = agent.run_sync("Python 的 GIL 是什么？")
print(result.data)
print(f"Token 用量: {result.usage()}")

结构化输出¶

from pydantic import BaseModel
from pydantic_ai import Agent

class CityInfo(BaseModel):
    name: str
    country: str
    population: int
    famous_for: list[str]

agent = Agent(
    "openai:gpt-4o-mini",
    result_type=CityInfo,
    system_prompt="提取城市信息，population 用实际数字。"
)

result = agent.run_sync("告诉我关于东京的信息")
city = result.data  # 类型是 CityInfo
print(f"{city.name}, {city.country}")
print(f"人口: {city.population:,}")
print(f"闻名于: {', '.join(city.famous_for)}")

添加工具¶

from pydantic_ai import Agent, RunContext
import httpx

agent = Agent(
    "openai:gpt-4o-mini",
    system_prompt="你是天气助手，使用工具查询天气。"
)

@agent.tool_plain
async def get_weather(city: str) -> str:
    """获取指定城市的当前天气。"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://wttr.in/{city}?format=3"
        )
        return resp.text

# 运行
result = agent.run_sync("北京今天天气怎么样？")
print(result.data)

进阶用法¶

1. 依赖注入¶

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class UserContext:
    user_id: int
    user_name: str
    is_premium: bool

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

@agent.system_prompt
async def build_prompt(ctx: RunContext[UserContext]) -> str:
    tier = "高级会员" if ctx.deps.is_premium else "免费用户"
    return f"你正在为{tier} {ctx.deps.user_name} 提供服务。"

@agent.tool
async def get_user_orders(ctx: RunContext[UserContext], limit: int = 5) -> str:
    """查询用户最近的订单。"""
    # 这里可以访问 ctx.deps.user_id 来查数据库
    return f"用户 {ctx.deps.user_id} 最近有 {limit} 条订单"

# 使用时注入依赖
user = UserContext(user_id=42, user_name="张三", is_premium=True)
result = agent.run_sync("查看我的最近订单", deps=user)
print(result.data)

2. 多模型切换¶

from pydantic_ai import Agent

# 定义 Agent 时不绑定模型
agent = Agent(
    system_prompt="简洁回答问题。",
    result_type=str,
)

# 运行时指定模型
r1 = agent.run_sync("什么是量子计算？", model="openai:gpt-4o-mini")
r2 = agent.run_sync("什么是量子计算？", model="anthropic:claude-3-5-sonnet-20241022")
r3 = agent.run_sync("什么是量子计算？", model="groq:llama-3.1-70b-versatile")

# 比较输出和成本
for r in [r1, r2, r3]:
    print(f"Token: {r.usage().total_tokens}, 回答: {r.data[:50]}")

3. 流式输出¶

import asyncio
from pydantic_ai import Agent

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

async def main():
    async with agent.run_stream("用500字解释相对论") as response:
        async for chunk in response.stream_text():
            print(chunk, end="", flush=True)

    print(f"\n\n总 token: {response.usage().total_tokens}")

asyncio.run(main())

4. 结构化流式输出¶

import asyncio
from pydantic import BaseModel
from pydantic_ai import Agent

class StoryOutline(BaseModel):
    title: str
    genre: str
    characters: list[str]
    plot_points: list[str]

agent = Agent(
    "openai:gpt-4o-mini",
    result_type=StoryOutline,
    system_prompt="生成故事大纲"
)

async def main():
    async with agent.run_stream("写一个科幻故事大纲") as response:
        async for partial in response.stream_structured():
            # partial 是逐步填充的 StoryOutline
            print(f"当前状态: {partial}")

        # 最终完整结果
        final = await response.get_data()
        print(f"\n完整大纲: {final.title}")

asyncio.run(main())

5. 对话历史管理¶

from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage

agent = Agent("openai:gpt-4o-mini", system_prompt="你是一个记忆力很好的助手。")

# 第一轮对话
result1 = agent.run_sync("我叫小明，我喜欢Python。")
print(result1.data)

# 获取消息历史
history: list[ModelMessage] = result1.all_messages()

# 第二轮对话（带上历史）
result2 = agent.run_sync("我叫什么？我喜欢什么？", message_history=history)
print(result2.data)  # 应该记住 "小明" 和 "Python"

6. 错误处理与重试¶

from pydantic import BaseModel, field_validator
from pydantic_ai import Agent, ModelRetry

class SafeResponse(BaseModel):
    answer: str
    confidence: float

    @field_validator("confidence")
    @classmethod
    def check_confidence(cls, v):
        if not 0 <= v <= 1:
            raise ValueError("confidence 必须在 0-1 之间")
        return v

agent = Agent(
    "openai:gpt-4o-mini",
    result_type=SafeResponse,
    retries=3  # 最多重试3次
)

@agent.result_validator
async def validate_result(ctx, result: SafeResponse) -> SafeResponse:
    if len(result.answer) < 10:
        raise ModelRetry("回答太短了，请更详细一些")
    return result

result = agent.run_sync("解释什么是微服务架构")
print(f"答案: {result.data.answer}")
print(f"置信度: {result.data.confidence}")

7. 与 Logfire 集成实现可观测性¶

import logfire
from pydantic_ai import Agent

# 初始化 Logfire
logfire.configure()

# Agent 自动被 Logfire 追踪
agent = Agent("openai:gpt-4o-mini")

# 所有调用都会出现在 Logfire 仪表板
result = agent.run_sync("Hello!")
# 可以在 Logfire UI 看到：请求耗时、token用量、工具调用链路

常见问题¶

Q1: 结构化输出经常失败/重试次数用尽¶

解决方案： - 增加 retries 参数值 - 在 system_prompt 中明确说明输出格式要求 - 简化 Pydantic Model（减少嵌套层级） - 使用更强的模型（gpt-4o 比 gpt-4o-mini 格式遵循更好）

Q2: 如何使用本地模型（Ollama）¶

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

# Ollama 兼容 OpenAI API 格式
model = OpenAIModel(
    model_name="llama3.1",
    base_url="http://localhost:11434/v1",
    api_key="ollama"  # Ollama 不需要真实 key
)

agent = Agent(model)

Q3: Tool 定义的 docstring 有什么作用¶

PydanticAI 会将 tool 函数的 docstring 作为工具描述发送给 LLM。LLM 根据描述决定何时调用工具。所以 docstring 要写清楚工具的用途和参数含义。

Q4: 与 FastAPI 如何配合¶

from fastapi import FastAPI
from pydantic_ai import Agent

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

@app.post("/chat")
async def chat(message: str):
    result = await agent.run(message)
    return {"response": result.data, "usage": result.usage().model_dump()}

Q5: 如何处理超长对话导致的 token 超限¶

手动截断 message_history（只保留最近 N 轮）
使用摘要策略（让 LLM 总结历史再继续）
选择上下文窗口更大的模型

参考资源¶

资源	链接
GitHub 仓库	https://github.com/pydantic/pydantic-ai
官方文档	https://ai.pydantic.dev
PyPI	https://pypi.org/project/pydantic-ai
Pydantic 文档	https://docs.pydantic.dev
Logfire（可观测性）	https://pydantic.dev/logfire
示例代码	https://ai.pydantic.dev/examples

小结： PydanticAI 不追求大而全，而是在"类型安全+开发体验"这个切面做到极致。如果你用 Python 构建生产级 AI Agent，希望代码可维护、输出可预测、调试有迹可循，PydanticAI 是当前最值得投入的框架之一。