PydanticAI实战教程 — 用FastAPI的感觉构建类型安全的AI智能体

PydanticAI实战教程 — 用FastAPI的感觉构建类型安全的AI智能体

我实际安装了PydanticAI 1.88.0,直接测试了TestModel、output_type、@agent.tool和多提供商切换。包括result_type→output_type变更等真实陷阱,以及完整的FunctionModel测试策略。

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

agent = Agent('test', system_prompt='Python专家')
result = agent.run_sync('f-string和.format()哪个更快?', model=TestModel())
print(result.output)  # → success (no tool calls)

第一次看到这段代码不需要API密钥就能运行时,我略感惊讶。就像第一次使用FastAPI那样——结构如此直观,反而让人产生怀疑。PydanticAI就是这样一个库。

坦白说,我最初的印象是”这不就是给Instructor加了个包装器吗?“实际使用后想法变了。这是一个以类型系统为核心的框架,把FastAPI带给Web API的那套哲学直接搬到了AI智能体上。今天分享的是我实际安装和运行后的结果,包括失败的测试。

为什么选PydanticAI — 与比较文章不同的视角

我之前写过一篇Python AI智能体库对比文章,涵盖PydanticAI、Instructor和Smolagents。那篇文章回答”选哪个”,这篇文章回答”用PydanticAI怎么实际构建”。实现方法才是目的。

各库定位快速对比:

核心职责智能体循环类型安全
InstructorLLM输出解析仅结构化输出
PydanticAI智能体框架完整支持输入+输出+工具全覆盖
LangGraph编排图结构较弱
CrewAI多智能体团队角色制较弱

第二行才是实际使用时的真正差别。在LLM调用工具、接收结果、处理的整个循环中,类型始终保持一致。运行时错误提前变成开发阶段的IDE错误。

安装与前置条件

pip install pydantic-ai

截至今日(2026年4月29日)最新版本为1.88.0。

python3 -m venv venv
source venv/bin/activate
pip install pydantic-ai

各提供商专用包按需安装:

pip install pydantic-ai[anthropic]   # Claude
pip install pydantic-ai[openai]       # GPT-4o
pip install pydantic-ai[google]       # Gemini

要求: Python 3.9+、pydantic v2(不支持v1)。TestModel无需API密钥即可工作。

第一个智能体:用TestModel验证结构

这是我构建新智能体时的第一步。在接入真实API之前先验证结构是否正确。

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

agent = Agent(
    'test',
    system_prompt='你是Python代码审查员。请简洁回答。',
)

result = agent.run_sync(
    'f-string和.format()哪个更快?',
    model=TestModel()  # 无需API密钥
)

print(result.output)    # → "success (no tool calls)"
print(result.usage())   # → RunUsage(input_tokens=64, output_tokens=4, requests=1)

TestModel不调用任何API。它是专为测试设计的模型,用于验证智能体结构、工具配置和依赖注入是否正确。在CI流水线中零成本验证智能体逻辑。

切换到真实Claude只需改一行:

import os
os.environ['ANTHROPIC_API_KEY'] = 'sk-ant-...'

# 开发阶段: TestModel
result = agent.run_sync('代码审查', model=TestModel())

# 生产环境: 真实Claude
result = agent.run_sync('代码审查', model='anthropic:claude-sonnet-4-6')

智能体代码不变,只换模型。

结构化输出:用output_type返回Pydantic模型

这是PydanticAI的核心价值。强制LLM返回Pydantic模型实例,而不是自由文本。

重要 — v1.88.0破坏性变更: result_type参数已重命名为output_type。使用旧文档或教程会报错:

pydantic_ai.exceptions.UserError: Unknown keyword arguments: `result_type`

我亲自踩坑了。通过inspect.signature(Agent.__init__)确认output_type才是正确的参数名。官方文档在某些地方仍引用旧API,需要注意。

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class CodeReview(BaseModel):
    severity: str = Field(description="'low'、'medium'或'high'之一")
    issues: list[str] = Field(description="发现的问题列表")
    suggestions: list[str] = Field(description="改进建议列表")
    score: int = Field(ge=0, le=100, description="代码质量分数0-100")

review_agent = Agent(
    'anthropic:claude-sonnet-4-6',
    output_type=CodeReview,  # ← v1.88.0: 不是result_type
    system_prompt='审查Python代码并提供结构化反馈。',
)

result = review_agent.run_sync('''
def get_user(id):
    db = connect()
    return db.query(f"SELECT * FROM users WHERE id={id}")
''')

print(type(result.output))       # → <class '__main__.CodeReview'>
print(result.output.severity)    # → 'high'
print(result.output.score)       # → 25
print(result.output.issues[0])   # → 'SQL注入漏洞'

返回值是Pydantic模型实例,不是dict或字符串。result.output.后IDE会自动补全所有字段。

自动重试机制

当LLM返回不符合schema的输出时,PydanticAI会将ValidationError反馈给LLM并请求重试:

review_agent = Agent(
    'anthropic:claude-sonnet-4-6',
    output_type=CodeReview,
    retries=3,        # 验证失败时最多重试3次
    output_retries=2  # 输出专用重试次数
)

3次全部失败后抛出UnexpectedModelBehavior异常。

@agent.tool与依赖注入 — 与FastAPI的Depends()相同的模式

from pydantic_ai import Agent, RunContext

class AppDeps:
    def __init__(self, db_url: str, user_id: int):
        self.db_url = db_url
        self.user_id = user_id

agent = Agent(
    'anthropic:claude-sonnet-4-6',
    deps_type=AppDeps,
    system_prompt='任务管理智能体。',
)

# 异步工具 — 适用于I/O密集操作
@agent.tool
async def get_pending_tasks(ctx: RunContext[AppDeps], limit: int = 5) -> list[dict]:
    """获取待处理任务列表"""
    return [
        {"id": f"task_{i}", "title": f"任务 {i}", "priority": "high"}
        for i in range(limit)
    ]

# 同步工具 — 适用于快速计算
@agent.tool
def calculate_priority_score(
    ctx: RunContext[AppDeps],
    urgency: int,
    importance: int
) -> float:
    """计算任务优先级分数"""
    return urgency * 0.6 + importance * 0.4

deps = AppDeps(db_url="postgresql://localhost/taskdb", user_id=42)
result = agent.run_sync("从紧急任务中选出最高优先级的", deps=deps)

@agent.tool读取函数签名(参数类型、默认值)和docstring,自动生成传给LLM的JSON Schema。

消息流可通过result.all_messages()追踪,共4个阶段:

1. ModelRequest  → system_prompt + user_prompt
2. ModelResponse → ToolCallPart(tool_name='get_pending_tasks', ...)
3. ModelRequest  → ToolReturnPart(content=[{...}])
4. ModelResponse → 最终回答

PydanticAI执行日志 — 沙箱测试结果

TestModel vs FunctionModel — 测试策略

在沙箱测试中发现了TestModel的重要限制。

TestModel的限制

TestModel为str字段返回'a',为int字段返回0等最小值。对于严格的自定义validator会失败:

from pydantic_ai.exceptions import UnexpectedModelBehavior

class UserProfile(BaseModel):
    email: str

    @field_validator('email')
    @classmethod
    def valid_email(cls, v):
        if '@' not in v:  # TestModel返回'a',始终失败
            raise ValueError('邮箱必须包含@')
        return v

agent = Agent('test', output_type=UserProfile, retries=3)
try:
    result = agent.run_sync('...', model=TestModel())
except UnexpectedModelBehavior as e:
    print(e)  # → Exceeded maximum retries (3) for output validation

这不是bug。TestModel用于结构验证,而非业务逻辑validator测试。

用FunctionModel精确控制

有validator或需要测试工具响应时使用FunctionModel:

from pydantic_ai.models.function import FunctionModel
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart
from pydantic_ai.settings import ModelSettings
import json

def mock_valid_response(messages: list[ModelMessage], settings: ModelSettings) -> ModelResponse:
    """直接提供测试时要返回的精确响应"""
    data = {"email": "test@example.com", "name": "测试用户"}
    return ModelResponse(parts=[TextPart(content=json.dumps(data))])

agent = Agent(FunctionModel(mock_valid_response), output_type=UserProfile)
result = agent.run_sync("...")
assert result.output.email == "test@example.com"

完整测试策略

class TestMyAgent:
    def test_structure(self):
        """智能体结构验证 — TestModel"""
        result = my_agent.run_sync("测试", model=TestModel())
        assert result is not None

    def test_tool_called(self):
        """工具调用确认 — TestModel + call_tools"""
        result = my_agent.run_sync(
            "从数据库获取数据",
            deps=test_deps,
            model=TestModel(call_tools=['query_database'])
        )
        assert 'query_database' in result.output

    def test_response_processing(self):
        """响应处理逻辑 — FunctionModel"""
        def mock_fn(messages, settings):
            return ModelResponse(parts=[TextPart(content='{"email": "t@t.com"}')])

        result = my_agent.run_sync("...", model=FunctionModel(mock_fn))
        assert result.output.email == "t@t.com"

多提供商切换

不修改智能体代码,只换模型字符串即可切换提供商:

review_agent = Agent(
    system_prompt='作为高级Python开发者进行代码审查。',
    output_type=CodeReview,
)

# 运行时指定模型
result_claude = review_agent.run_sync(code, model='anthropic:claude-sonnet-4-6')
result_gpt    = review_agent.run_sync(code, model='openai:gpt-4o')
result_gemini = review_agent.run_sync(code, model='google-gla:gemini-2.5-flash')
result_local  = review_agent.run_sync(code, model='ollama:llama3.3')

上下文工程的角度看,system_promptoutput_type schema是上下文的核心,在此基础上让模型可替换才是好的架构设计。

实际使用感受 — 直言不讳

优点:

  • 类型安全切实产生差异。修改output_type schema,IDE立即捕捉所有相关错误
  • @agent.tool自动生成JSON Schema,无需手动重写工具规范
  • TestModel + FunctionModel组合,无需任何API即可完整进行单元测试
  • deps_type使依赖注入显式化,测试中的mock替换更加简洁

不足之处:

  • 直到v1.88.0仍有result_type → output_type这样的不兼容变更。库还未进入稳定阶段。需要通过inspect.signature(Agent.__init__)确认实际参数名的情况时有发生
  • 流式结构化输出仍处于beta阶段。在LLM生成部分响应时解析为Pydantic模型比较棘手,当前实现不够稳定
  • 强依赖Pydantic v2。如果是v1遗留代码库,需要考虑迁移成本
  • Logfire集成(Pydantic团队的监控服务)是官方可观测性路径,但需付费。直接连接OpenTelemetry虽然可行,但官方文档支持不足

结合生产级AI智能体设计原则阅读,可以更清楚地了解选择智能体框架时哪些标准最重要。

核心模式快速总结

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.test import TestModel
from pydantic_ai.models.function import FunctionModel
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart
from pydantic_ai.settings import ModelSettings
import json

# 1. 结构化输出模型
class ReviewResult(BaseModel):
    score: int = Field(ge=0, le=100)
    verdict: str  # 'approve', 'request_changes'
    summary: str

# 2. 依赖类型
class AgentDeps:
    def __init__(self, repo_name: str, author: str):
        self.repo_name = repo_name
        self.author = author

# 3. 智能体定义
agent = Agent(
    system_prompt='代码审查智能体。',
    output_type=ReviewResult,  # v1.88.0: output_type(不是result_type)
    deps_type=AgentDeps,
    retries=3,
)

# 4. 工具注册
@agent.tool
def get_context(ctx: RunContext[AgentDeps]) -> dict:
    return {"repo": ctx.deps.repo_name, "author": ctx.deps.author}

# 5. 测试
result = agent.run_sync("审查这段代码",
    deps=AgentDeps("my-repo", "jangwook"),
    model=TestModel())

def mock(messages, settings):
    return ModelResponse(parts=[TextPart(content=json.dumps({
        "score": 85, "verdict": "approve", "summary": "代码整洁"
    }))])

result = agent.run_sync("审查这段代码",
    deps=AgentDeps("my-repo", "jangwook"),
    model=FunctionModel(mock))
assert result.output.verdict == "approve"

# 6. 生产环境:只换模型
result = agent.run_sync("审查这段代码",
    deps=AgentDeps("my-repo", "jangwook"),
    model='anthropic:claude-sonnet-4-6')

下一步

TypeScript技术栈可参考使用Vercel AI SDK构建Claude流式智能体

将PydanticAI投入生产的推荐步骤:

  1. 先用output_type定义返回schema
  2. deps_type管理DB连接、HTTP客户端等依赖
  3. 通过@agent.tool添加外部API集成
  4. 按TestModel → FunctionModel → 真实模型的顺序逐步测试
  5. retries=3output_retries=2配置重试策略
  6. 固定版本(pydantic-ai==1.88.0)。这个库变化频繁

PydanticAI GitHub仓库更新很快。先看CHANGELOG再看官方文档能节省真实的调试时间。根据亲身经历——这个库还没到1.0,但对于Python智能体技术栈,它目前是最一致的类型安全选择。

阅读其他语言版本

这篇文章有帮助吗?

您的支持能帮助我创作更好的内容。请我喝杯咖啡吧。

关于作者

jw

Kim Jangwook

AI/LLM专业全栈开发者

凭借10年以上的Web开发经验,构建AI代理系统、LLM应用程序和自动化解决方案。分享Claude Code、MCP和RAG系统的实践经验。

返回博客列表