pydantic-ai

英文原文正文为项目原始 README（英文），本站后续会翻译为中文，当前仅剔除图片与无关章节并统一排版。

代理、实时语音、图像生成、嵌入。每一个模型，每一个接口，端到端输入输出。

Pydantic AI is the Python AI SDK: a typed, extensible agent loop with every model a string swap away. The same agent runs everywhere you need it: behind a web frontend, in the terminal, on a voice call, on a durable background queue, in GitHub Actions, or as a plain object you call run() on. Image generation and embeddings come in the same box.

Pydantic AI Harness 拥有一个代理在复杂、长时间运行任务中所需的所有功能，这些功能作为能力附加，其中包括记忆、防护措施、子代理、规划、上下文管理和持久化，甚至完整编程代理。

Pydantic Evals 用于评估代理的最终输出和工具调用轨迹，可在数据集或实时生产流量上进行评测，Pydantic Graph 在单个代理循环不足时提供类型化的控制流程。这两个都是独立的开源包，genai-prices 也是独立包，用于保持模型价格最新，Monty 是运行模型生成代码的沙盒 Python 解释器。

Pydantic Logfire 是一个 AI 工程平台，可以看见你的整个应用，而不仅仅是 LLM 调用，Pydantic AI Gateway 是一个可以访问所有模型的关键平台，提供实时成本监控和预算控制；我们的监控工具使用的是纯 OpenTelemetry，因此可以兼容你已经运行的任何后台服务。

完整文档请访问 pydantic.dev/docs/ai。

你在建造什么？

从简单的类型化数据提取到复杂的、长时间运行的多代理协作，Pydantic AI 和 Pydantic AI Harness 都能满足您的需求。

编码代理

在您的终端中拥有完整的编码代理：以工作区为根的 文件访问、白名单 Shell、仓库导向、规划 以及可持续长时间会话的 上下文管理。此外，还配备了 网页搜索 和第二意见 顾问。

uv add pydantic-ai pydantic-ai-harness

from pydantic_ai import Agent
from pydantic_ai.capabilities import WebSearch
from pydantic_ai_harness import Advisor, Coder

agent = Agent(
    'anthropic:claude-fable-5-1',
    capabilities=[
        Coder(),  # files, shell, repo context, sub-agents, context management
        WebSearch(),  # look up docs and error messages on the web
        Advisor('openai:gpt-6.1-sol'),  # a second opinion from another model when stuck
    ],
)
agent.to_cli_sync()

Coder 是一个常规的 组合能力，而不是一个黑箱：可以整体使用它，也可以直接使用它捆绑的模块；两者是等效的：

capabilities = [
    FileSystem('.'), Shell(cwd='.'), RepoContext(), SubAgents(...),
    ClearToolResults(), ToolOutputLimits(), RepairToolArguments(),
]

运行该文件，你就可以在终端与代理聊天。要在编写任何代码之前进行尝试，可以通过 uvx 使用 clai（Pydantic AI CLI）运行导出的 coder_agent：

uvx --with pydantic-ai-harness clai -a pydantic_ai_harness.coder:coder_agent -m anthropic:claude-fable-5-1

构建这个 → Coder，来自 Harness

在 GitHub 上运行 → GitHub Agentic Workflows，可在问题、拉取请求或计划任务上运行

数据提取

为代理提供 输出类型 和 工具，每次运行都会返回经过验证和类型化的结果：

uv add pydantic-ai

from typing import Literal

from pydantic import BaseModel, Field

from pydantic_ai import Agent, RunContext

class Sentiment(BaseModel):
    label: Literal['positive', 'negative', 'neutral']
    score: float = Field(ge=-1, le=1)

agent = Agent('openai:gpt-6.1-sol', output_type=Sentiment)

@agent.tool
def recent_reviews(ctx: RunContext, product: str) -> list[str]:
    """Fetch recent review snippets for a product."""
    return ['The new release fixed everything I complained about!']

result = agent.run_sync('How are people feeling about the Extract app?')
print(result.output)
#> label='positive' score=0.9

@agent.tool 函数接收一个 RunContext，它携带你的依赖；其余的签名和文档字符串成为工具模式，参数在你的代码运行前会被验证，且运行保证返回 Sentiment，因此你的 IDE、类型检查器和 LLM 都会对返回类型达成一致。

构建这个 → Agents、Function Tools 和 Structured Output

持久工作流

附加 TemporalDurability，同一个 agent 在 Temporal 工作流下运行 durable execution：每一个模型和工具调用都成为持久活动，因此通过后台队列进行的运行可以在重启、故障和长时间等待中存活。

uv add "pydantic-ai[temporal]"

from temporalio import workflow

from pydantic_ai import Agent
from pydantic_ai.capabilities import WebFetch, WebSearch
from pydantic_ai.durable_exec.temporal import PydanticAIWorkflow, TemporalDurability

agent = Agent(
    'openai:gpt-6.1-sol',
    instructions='Research the topic and write a structured brief.',
    name='researcher',
    capabilities=[WebSearch(), WebFetch(), TemporalDurability()],
)

@workflow.defn
class ResearchWorkflow(PydanticAIWorkflow):
    __pydantic_ai_agents__ = [agent]

    @workflow.run
    async def run(self, topic: str) -> str:
        result = await agent.run(f'Write a brief on: {topic}')
        return result.output

DBOS 和 Prefect 的附加方式相同，都是第一方和共同维护的，除此之外还与 Restate、AWS Lambda、Airflow、Absurd 和 pgtask 集成。

保存、恢复和分叉运行不需要引擎：Pydantic AI Harness 的 StepPersistence 会将每个已完成的步骤快照到你选择的存储中，就像检查点一样（参见 Persistence）。引擎用于恢复在步骤中途崩溃的运行。

构建这个 → Durable Execution

实时语音

在实时语音会话中放置相同的代理，包含 工具 和 能力:

uv add "pydantic-ai[openai-realtime]"

import asyncio

from pydantic_ai import Agent
from pydantic_ai.capabilities import MCP

agent = Agent(
    instructions='You are a helpful voice assistant.',
    capabilities=[MCP('https://internal.example.com/mcp')],  # capabilities work in voice too
)

@agent.tool_plain
def order_status(order_id: str) -> str:
    """Look up the status of an order."""
    return f'Order {order_id}: shipped, arriving Thursday.'

async with agent.realtime('openai:gpt-realtime-2.1').session() as session:
    microphone = asyncio.create_task(session.send_audio(microphone_chunks()))  # your microphone → the model
    speaker = asyncio.create_task(play_audio(session.stream_audio()))  # model audio → your speaker
    async for part in session.stream_transcripts():
        print(f'{part.speaker}: {part.transcript}')

模型在对话过程中会调用你的工具，同时继续讲话，每个会话都是可检测的；语音只是另一个前端，支持 OpenAI Realtime、Gemini Live、Azure 和 xAI Grok Voice。

构建这个 → 实时语音

图像生成

使用专用图像模型生成图像，无需运行代理：

uv add pydantic-ai

from pathlib import Path

from pydantic_ai import ImageGenerator

generator = ImageGenerator('openai:gpt-image-2.5-flare')
result = generator.generate_sync('A minimalist logo for a coffee shop called Extract.')
Path('logo.png').write_bytes(result.image.data)

那个独立图像 API是由你的应用程序决定使用的；当代理运行决定时，有提供者原生生成功能，并使用 output_type=BinaryImage 来生成类型化图像输出，以及ImageGeneration能力，它可以在模型自身不生成图像时使用回退方案。

构建这个 → 图像生成

在 Logfire 中查看你的首次运行

提示： 在任何这些代理运行之前添加两行代码，每次模型调用和工具调用都会显示在Pydantic Logfire中。Logfire 有一个免费套餐，无需信用卡，你只需使用 GitHub 账号即可注册。首次运行时执行 uvx logfire auth 和 uvx logfire projects new，或者将你的编码代理指向Logfire 设置技能来为你完成。详细信息请参阅Logfire 指南，同时也可以使用任何 OpenTelemetry 后端代替。

logfire.configure()
logfire.instrument_pydantic_ai()

为什么选择 Pydantic AI

• 任何模型，一个 Python API。 几乎所有模型和提供商（OpenAI、Anthropic、Google、Bedrock、Azure AI Foundry、Groq、Mistral、xAI、Ollama 及数十个其他）都可以通过字符串或 Pydantic AI Gateway 进行切换：一个密钥即可通用，内置故障转移和成本监控。没有任何旗舰功能被锁定在某一个供应商。

• 端到端类型化。 结构化输出、类型化的 依赖注入、类型化工具：你的 IDE、类型检查器和编码代理都能知道你的代理返回什么，将整类错误从运行时移到写时。当普通控制流不够时，Pydantic Graph 将相同的类型系统带入基于图的工作流。

• 以测量为准，而非感觉。 OpenTelemetry 原生 监测 适用于任何 OTel 后端；一行代码即可启用 Pydantic Logfire 进行实时调试、追踪和成本跟踪，由 genai-prices 提供支持。Pydantic Evals 对最终输出及 工具调用轨迹 进行评分，可在代码中的数据集上执行，也可通过 在线评估 针对生产流量样本进行。

• 可组合的组件。 一个原语，功能能力 将 工具、指令、钩子 及 模型设置 打包成可复用单元。核心模块提供基础功能，如 MCP 和 网页搜索；Harness 提供其余所有功能；完整代理如 Coder 和 Researcher 只是功能能力的组合：可以像组合时一样拆开。或者完全跳过代码，使用 YAML/JSON 代理规范。

• 每个接口。 一个代理定义可以作为CLI、内置网页聊天或实时语音（OpenAI Realtime、Gemini Live、Azure、xAI Grok Voice）运行；UI 事件流（AG-UI、Vercel AI）将其连接到你自己的前端或其他任何地方；ACP 将其作为编辑器代理提供；GitHub Agentic 工作流 可在问题、拉取请求或计划上以无头方式运行。

• 持久执行。 在八个引擎上支持持久执行：Temporal、DBOS、Prefect、Restate、AWS Lambda、Airflow、Absurd 和 pgtask，其中前五个与供应商团队共同维护。代理可在你已运营的引擎上重启后存续并运行数天，并内置人工审核。

• 来自其他框架？ 对比显示 Pydantic AI 与 LangChain、Google ADK、Claude Agent SDK 及其他七种框架的区别，而迁移技能可让你的编码代理将现有应用迁移过来。

由Pydantic 团队构建：Pydantic 验证 是 OpenAI SDK、Anthropic SDK、Google ADK、LangChain 以及大多数 AI 生态系统的验证层（也是 FastAPI 构建的基础）。Pydantic AI 将这种体验带入到代理中。

整合：一个银行支持代理

一个类型化的支持代理展示了多个功能协同工作：依赖注入、函数工具、结构化输出、一个可复用的能力打包了客户上下文，以及一个按需能力，模型仅在对话需要时加载它：

from dataclasses import dataclass

from pydantic import BaseModel, Field

from pydantic_ai import Agent, Capability, RunContext

from bank_database import DatabaseConn

@dataclass
class SupportDependencies:  # inject any client: DB pools, HTTP APIs, user info
    customer_id: int
    db: DatabaseConn

class SupportOutput(BaseModel):
    support_advice: str = Field(description='Advice returned to the customer')
    block_card: bool = Field(description="Whether to block the customer's card")
    risk: int = Field(description='Risk level of query', ge=0, le=10)

customer_context = Capability[SupportDependencies](  # a reusable unit of tools + instructions
    id='customer-context',
    description="Who the customer is and what's on their account.",
)

@customer_context.instructions
async def add_customer_name(ctx: RunContext[SupportDependencies]) -> str:
    customer_name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id)
    return f"The customer's name is {customer_name!r}"

@customer_context.tool  # signature and docstring become the tool schema the LLM sees
async def customer_balance(
    ctx: RunContext[SupportDependencies], include_pending: bool
) -> float:
    """Returns the customer's current account balance."""
    return await ctx.deps.db.customer_balance(
        id=ctx.deps.customer_id,
        include_pending=include_pending,
    )

refunds = Capability[SupportDependencies](  # deferred: loads on demand, like a skill
    id='refunds',
    description='Refund eligibility and refund status.',
    defer_loading=True,
)

@refunds.tool
async def refund_status(ctx: RunContext[SupportDependencies]) -> str:
    """Look up the refund status for the customer's most recent charge."""
    return await ctx.deps.db.refund_status(id=ctx.deps.customer_id)

support_agent = Agent(
    'openai:gpt-6.1-sol',
    deps_type=SupportDependencies,
    output_type=SupportOutput,  # the run returns a validated SupportOutput, typed as such
    instructions=(
        'You are a support agent in our bank, give the '
        'customer support and judge the risk level of their query.'
    ),
    capabilities=[customer_context, refunds],
)

...  # in a real use case: more tools, longer instructions

async def main():
    deps = SupportDependencies(customer_id=123, db=DatabaseConn())
    result = await support_agent.run('What is my balance?', deps=deps)
    print(result.output)
    """
    support_advice='Hello John, your current account balance, including pending transactions, is $123.45.' block_card=False risk=1
    """

    result = await support_agent.run('I just lost my card!', deps=deps)
    print(result.output)
    """
    support_advice="I'm sorry to hear that, John. We are temporarily blocking your card to prevent unauthorized transactions." block_card=True risk=8
    """

    result = await support_agent.run(  # the model loads `refunds` on demand, then answers
        'Was I refunded for the duplicate charge on my last statement?', deps=deps
    )
    print(result.output)
    """
    support_advice='Good news, John: the duplicate charge on your last statement was refunded on 2026-05-01.' block_card=False risk=1
    """

有关带注释的演练和 Logfire 跟踪，请参见文档中的相同示例。

下一步

• 安装 Pydantic AI 并让你自己的编程代理开始工作：安装 Pydantic AI 技能，指向 示例 和 Harness 索引，并告诉它你想构建什么。开始无需 API 密钥（内置了 'test' 模型），当准备好时，Pydantic AI Gateway 为每个模型提供一个密钥。
• 查看你的代理做了什么：用一行设置对其进行监控，每次模型调用和工具调用都会显示出来。这是标准的 OpenTelemetry：Pydantic Logfire，提供 免费等级（无需信用卡，只需 GitHub 账户注册），是最简单的查看方式，任何 OTLP 后端都适用。
• 交付它：投入生产 回答团队在发布代理前会问的问题，从崩溃恢复和人工审批到成本限制、追踪和部署。
• 阅读 文档 和 API 参考。
• 给你的代理补充能量：Pydantic AI Harness。
• 加入 Slack 或在 GitHub 提交问题。

Pydantic 堆栈的一部分

你交付生产级 AI 代理所需的一切：

• Pydantic AI：类型安全的 AI SDK
• Pydantic AI Harness：官方功能库和 Harness，从单一功能到完整代理
• Pydantic Logfire：AI 优先的全栈可观测性
• Pydantic AI Gateway：每个模型一个密钥，带有成本监控和支出限制
• Pydantic Evals：评估任意 Python 函数，包括代理，从最终输出到 工具调用轨迹，在数据集或 实时流量 上（使用 Logfire 上的生产评估）
• Pydantic Graph：类型化图形控制流
• genai-prices：模型定价数据，保持最新
• Monty：一个为模型编写代码提供的沙箱 Python 解释器