PydanticAI — Python中的类型化AI Agent

PydanticAI是Pydantic团队开发的Python Agent框架，将为LLM应用带来与Pydantic相同的类型安全和验证保证。它支持结构化输出（使用Pydantic模型验证）、用于可测试性的依赖注入、流式响应、多轮对话和工具使用——跨OpenAI、Anthropic、Google Gemini、Groq、Mistral和Ollama。在构建生产级AI Agent、聊天机器人或LLM管道时使用此技能，其中正确性和可测试性至关重要。

pip install pydantic-ai

# 安装特定提供商的扩展
pip install 'pydantic-ai[openai]'       # OpenAI / Azure OpenAI
pip install 'pydantic-ai[anthropic]'    # Anthropic Claude
pip install 'pydantic-ai[gemini]'       # Google Gemini
pip install 'pydantic-ai[groq]'         # Groq
pip install 'pydantic-ai[vertexai]'     # Google Vertex AI

from pydantic_ai import Agent

# 简单Agent——返回纯文本字符串
agent = Agent(
    'anthropic:claude-sonnet-4-6',
    system_prompt='你是一个有帮助的助手。请简洁回答。',
)

result = agent.run_sync('日本的首都是什么？')
print(result.data)  # "东京"
print(result.usage())  # Usage(requests=1, request_tokens=..., response_tokens=...)

步骤3：使用Pydantic模型的结构化输出

from pydantic import BaseModel
from pydantic_ai import Agent

class MovieReview(BaseModel):
    title: str
    year: int
    rating: float  # 0.0 to 10.0
    summary: str
    recommended: bool

agent = Agent(
    'openai:gpt-4o',
    result_type=MovieReview,
    system_prompt='你是一名影评人。返回结构化的评论。',
)

result = agent.run_sync('评论《盗梦空间》(2010)')
review = result.data  # 完全类型化的MovieReview实例
print(f"{review.title} ({review.year}): {review.rating}/10")
print(f"推荐: {review.recommended}")

使用@agent.tool注册工具——LLM可以在运行期间调用它们：

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

class WeatherReport(BaseModel):
    city: str
    temperature_c: float
    condition: str

weather_agent = Agent(
    'anthropic:claude-sonnet-4-6',
    result_type=WeatherReport,
    system_prompt='获取请求城市的当前天气。',
)

@weather_agent.tool
async def get_temperature(ctx: RunContext, city: str) -> dict:
    """从天气API获取城市的当前温度。"""
    async with httpx.AsyncClient() as client:
        r = await client.get(f'https://wttr.in/{city}?format=j1')
        data = r.json()
        return {
            'temp_c': float(data['current_condition'][0]['temp_C']),
            'description': data['current_condition'][0]['weatherDesc'][0]['value'],
        }

import asyncio
result = asyncio.run(weather_agent.run('东京的天气如何？'))
print(result.data)

将服务（数据库、HTTP客户端、配置）注入Agent以实现可测试性：

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel

@dataclass
class Deps:
    db: Database
    user_id: str

class SupportResponse(BaseModel):
    message: str
    escalate: bool

support_agent = Agent(
    'openai:gpt-4o-mini',
    deps_type=Deps,
    result_type=SupportResponse,
    system_prompt='你是一名支持Agent。使用工具帮助客户。',
)

@support_agent.tool
async def get_order_history(ctx: RunContext[Deps]) -> list[dict]:
    """获取当前用户的近期订单。"""
    return await ctx.deps.db.get_orders(ctx.deps.user_id, limit=5)

@support_agent.tool
async def create_refund(ctx: RunContext[Deps], order_id: str, reason: str) -> dict:
    """为特定订单发起退款。"""
    return await ctx.deps.db.create_refund(order_id, reason, ctx.deps.user_id)

# 使用示例
async def handle_support(user_id: str, message: str):
    deps = Deps(db=get_db(), user_id=user_id)
    result = await support_agent.run(message, deps=deps)
    return result.data

编写无需真实LLM调用的单元测试：

from pydantic_ai.models.test import TestModel

def test_support_agent_escalates():
    with support_agent.override(model=TestModel()):
        # TestModel返回匹配result_type的最小有效响应
        result = support_agent.run_sync(
            '我想取消我的账户',
            deps=Deps(db=FakeDb(), user_id='user-123'),
        )
    # 测试结构，而非LLM的准确措辞
    assert isinstance(result.data, SupportResponse)
    assert isinstance(result.data.escalate, bool)

FunctionModel用于确定性测试响应：

from pydantic_ai.models.function import FunctionModel, ModelContext

def my_model(messages, info):
    return ModelResponse(parts=[TextPart('Always this response')])

with agent.override(model=FunctionModel(my_model)):
    result = agent.run_sync('anything')

import asyncio
from pydantic_ai import Agent

agent = Agent('anthropic:claude-sonnet-4-6')

async def stream_response():
    async with agent.run_stream('写一首关于Python的俳句') as result:
        async for chunk in result.stream_text():
            print(chunk, end='', flush=True)
    print()  # 换行
    print(f"总令牌数: {result.usage()}")

asyncio.run(stream_response())

from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessagesTypeAdapter

agent = Agent('openai:gpt-4o', system_prompt='你是一个有帮助的助手。')

# 第一轮
result1 = agent.run_sync('我叫Alice。')
history = result1.all_messages()

# 第二轮——传递对话历史
result2 = agent.run_sync('我叫什么名字？', message_history=history)
print(result2.data)  # "你的名字是Alice。"

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

class CodeReview(BaseModel):
    quality: Literal['excellent', 'good', 'needs_work', 'poor']
    issues: list[str] = Field(default_factory=list)
    suggestions: list[str] = Field(default_factory=list)
    approved: bool

code_review_agent = Agent(
    'anthropic:claude-sonnet-4-6',
    result_type=CodeReview,
    system_prompt="""
    你是一名执行代码审查的高级工程师。
    评估代码质量，识别问题，并提供可操作的建议。
    仅对良好或优秀且无安全问题的代码设置approved=True。
    """,
)

def review_code(diff: str) -> CodeReview:
    result = code_review_agent.run_sync(f"审查这段代码：\n\n{diff}")
    return result.data

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

class StrictJson(BaseModel):
    value: int

    @field_validator('value')
    def must_be_positive(cls, v):
        if v <= 0:
            raise ValueError('value必须是正数')
        return v

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

@agent.result_validator
async def validate_result(ctx, result: StrictJson) -> StrictJson:
    if result.value > 1000:
        raise ModelRetry('值必须小于1000。请用较小的数字重试。')
    return result

from pydantic_ai import Agent
from pydantic import BaseModel

class ResearchSummary(BaseModel):
    key_points: list[str]
    conclusion: str

class BlogPost(BaseModel):
    title: str
    body: str
    meta_description: str

researcher = Agent('openai:gpt-4o', result_type=ResearchSummary)
writer = Agent('anthropic:claude-sonnet-4-6', result_type=BlogPost)

async def research_and_write(topic: str) -> BlogPost:
    # 阶段1：研究
    research = await researcher.run(f'研究主题：{topic}')

    # 阶段2：基于研究结果写作
    post = await writer.run(
        f'写一篇关于以下内容的博客文章：{topic}\n\n研究结果：\n' +
        '\n'.join(f'- {p}' for p in research.data.key_points) +
        f'\n\n结论：{research.data.conclusion}'
    )
    return post.data

解决方案： 简化result_type字段。在适当的地方使用Optional和default。模型可能难以处理过于严格的模式。

解决方案： 为工具函数编写清晰、具体的文档字符串——PydanticAI将文档字符串作为工具描述发送给LLM。

解决方案： 在调用agent.run()或agent.run_sync()时传递deps=。依赖项不是全局设置的。

解决方案： 在异步FastAPI路由处理程序中直接使用await agent.run()——不要包装在asyncio.run()中。

来源说明：本文内容整理自开源项目 agentic-awesome-skills（中文翻译版），遵循 CC BY 4.0（知识共享署名 4.0 国际）许可协议，此处作为中文资料整理分享，版权归原作者所有。

Pydantic Ai 实用指南

使用PydanticAI构建生产级AI Agent——类型安全的工具使用、结构化输出、依赖注入和多模型支持。

人工智能

编程语言

开发工具

PydanticAI实用指南：Python类型化AI Agent框架，提供结构化输出、工具使用、依赖注入、流式响应与多轮对话支持，跨OpenAI、Anthropic等主流模型。文中含代码示例与最佳实践，助你构建生产级可测试AI应用。

聊天机器人

结构化数据

单元测试

数据库

AI Agent

路由

Python

Agent

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

Pydantic Ai 实用指南

Pydantic Ai 实用指南

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐