Pydantic AI 完全入门：用类型安全的方式在 Python 中构建 Agent、实时语音与图像生成

Pydantic AI 完全入门：用类型安全的方式在 Python 中构建 Agent、实时语音与图像生成

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

Pydantic AI 是 Pydantic 团队出品的 Python AI SDK，核心是一个类型化、可扩展的 Agent 运行循环（agent loop）：任何模型都只是一串字符串的差异，同一个 Agent 定义可以运行在 Web 前端、终端、实时语音会话、持久化后台队列或普通函数调用等多种界面之下。读完本文，你将掌握 Pydantic AI 的五大典型落地场景（编码代理、结构化数据抽取、持久化工作流、实时语音、图像生成）、核心的 Capability 扩展机制，并能用依赖注入、函数工具、结构化输出和按需加载能力组合出一个生产可用的完整 Agent。

项目概览：How Python does AI

Pydantic AI 定位为Python 的 AI SDK：Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.（Agent、实时语音、图像生成、嵌入向量，覆盖每个模型、每种界面，全程类型化）。

• 类型化的 Agent 循环：Agent 在依赖类型和输出类型上是泛型的，例如一个接受Foobar依赖、返回list[str]的 Agent，其类型就是Agent[Foobar, list[str]]，IDE、类型检查器与 LLM 对返回值达成一致，把整类错误从运行时提前到写代码时（参见 docs/agent.md）。
• 模型即字符串：实例化Agent时只需传入<provider>:<model>形式的名称（如openai:gpt-5.2、anthropic:claude-fable-5），框架会自动选择对应的模型类、Provider 与 Profile（详见 docs/models/overview.md）。
• 同一 Agent 处处可跑：一个 Agent 定义可以运行在 Web 前端、终端 CLI、实时语音、持久化后台队列，或者就是一个你直接调用run()的普通对象。图像生成 与 嵌入向量 也在同一个包里。
• Capability 扩展机制：框架用一个原语——Capability——把工具、指令、钩子与模型设置打包成可复用单元。核心包内置 MCP、Web 搜索等基础能力，完整的编码代理（Coder）、研究员（Researcher）等本身就是由 Capability 组合而成，可拆可合（见 docs/capabilities/overview.md）。

从五大场景开始：直接可跑的代码

README 以"What are you building?"切入，覆盖从简单类型化数据抽取到复杂多 Agent 协作的典型场景。以下五个场景均可直接复制运行。

场景一：终端里的完整编码代理（Coding agent）

一个开箱即用的编码代理，具备以工作区为根的文件访问、白名单 Shell、仓库导向、规划与跨长会话的上下文管理。下面还叠加了 Web 搜索与"第二意见"顾问：

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', capabilities=[ Coder(), # 文件、Shell、仓库上下文、规划、子代理、上下文管理 WebSearch(), # 在网络上查阅文档与错误信息 Advisor('openai:gpt-5.6-sol'), # 卡住时由另一个模型提供第二意见 ], ) agent.to_cli_sync()

关键点：Coder不是黑盒，它就是一个常规的组合型 Capability（composite capability）。整用或拆用完全等价：

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

想在不写任何代码的情况下先体验，可以用clai（Pydantic AI 的 CLI）直接运行导出的coder_agent：

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

场景二：类型化数据抽取（Data extraction）

给 Agent 一个输出类型和若干工具，每一次运行都会返回经过校验、带类型的结构化结果：

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-5.6-sol', output_type=Sentiment) @agent.tool def recent_reviews(ctx: RunContext[None], 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，函数签名的其余部分与 docstring 会生成工具 Schema，参数在代码运行前完成校验，且运行结果保证是Sentiment——IDE、类型检查器与 LLM 对返回类型三方一致。

场景三：持久化工作流（Durable workflow）

挂上TemporalDurability，同一个 Agent 就能运行在 Temporal 工作流里：每一次模型调用与工具调用都成为持久化活动（durable activity），因此运行在后台队列中的任务可以扛住重启、失败与长时间等待：

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-5.6-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、Kitaru 与 Airflow。按 docs/durable_execution/overview.md 的说明，Pydantic AI 官方共支持五种持久化方案（Temporal、DBOS、Prefect、Restate、AWS Lambda durable functions），均支持流式输出与 MCP，并内置容错与人工审批（human-in-the-loop）能力。

场景四：实时语音（Realtime voice）

把同一个 Agent 放到实时语音会话上，工具与 Capability 原样可用：

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')], # 语音场景下 Capability 同样生效 ) @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())) # 你的麦克风 → 模型 speaker = asyncio.create_task(play_audio(session.stream_audio())) # 模型音频 → 你的扬声器 async for part in session.stream_transcripts(): print(f'{part.speaker}: {part.transcript}')

模型可以在持续说话的过程中中途调用你的工具，且每个会话都会被埋点观测（instrumented）。语音只是另一种前端，支持 OpenAI Realtime、Gemini Live、Azure 与 xAI Grok Voice。

场景五：图像生成（Image generation）

用专用图像模型生成图片，无需经过 Agent 运行：

uv add pydantic-ai

from pathlib import Path from pydantic_ai import ImageGenerator generator = ImageGenerator('openai:gpt-image-2') result = generator.generate_sync('A minimalist logo for a coffee shop called Extract.') Path('logo.png').write_bytes(result.image.data)

独立的图像 API 用于"由你的应用决定"的场景；而"由 Agent 运行决定"时，还有三条路径：output_type=BinaryImage的 provider 原生图像生成工具（docs/native-tools.md）、类型化的图像输出（docs/output.md），以及带 fallback 的ImageGenerationCapability，用于处理本身不产图的模型。

为什么选择 Pydantic AI：六大设计取向

1. 任何模型，一套 Python API

几乎所有主流模型与 Provider（OpenAI、Anthropic、Google、Bedrock、Azure AI Foundry、Groq、Mistral、xAI、Ollama 等数十家）都可以用一串字符串切换，或者通过 Pydantic AI Gateway 一把密钥统一接入，内置故障切换与成本监控。没有哪个旗舰特性被锁定在单一厂商上。仓库中 docs/models/overview.md 列出了全部内建 Provider（含 OpenAI 兼容的 Alibaba DashScope、DeepSeek、Fireworks、LiteLLM、vLLM 等）以及测试用的TestModel与FunctionModel。

2. 端到端类型化

结构化输出、类型化依赖注入、类型化工具：你的 IDE、类型检查器乃至编码代理都知道 Agent 会返回什么。当普通控制流不够用时，Pydantic Graph 把同样的类型化带到基于图的流程中。

3. 可观测，而非玄学

Pydantic AI 是 OpenTelemetry 原生的：InstrumentationCapability 为每一次模型调用与工具调用发出标准 OTel span，任何 OTLP 后端都可用（见 docs/logfire.md）。一行配置即可点亮 Pydantic Logfire，配合 genai-prices 做实时调试、链路追踪与成本统计；Pydantic Evals 则像 pytest 测试代码一样测试 Agent 行为。

4. 电池可组合（Batteries, composably）

唯一原语 Capability 把工具、指令、钩子与模型设置打包成可复用单元。核心包提供 MCP、Web 搜索等基础能力，Harness 提供其余部分，完整的 Agent（Coder、Researcher）不过是 Capability 的组合——能怎么装起来就能怎么拆开。甚至可以直接跳过代码，用 YAML/JSON 的 Agent Spec 声明 Agent。

从源码看，Capability的便利实现位于 pydantic_ai_slim/pydantic_ai/capabilities/capability.py：其构造函数接收instructions、toolsets、tools、id、description、defer_loading六个参数，内部把函数工具封装进FunctionToolset，多个 toolset 通过CombinedToolset合并；id是稳定标识（defer_loading=True时必填，供模型的load_capability调用引用），description在延迟加载时展示给模型以决定是否加载。这也解释了为何Capability是"无需子类化即可声明式定义技能"的首选方式。

5. 每一种界面

同一份 Agent 定义可以作为 CLI、内建 Web 聊天或实时语音（OpenAI Realtime、Gemini Live、Azure、xAI Grok Voice）运行；AG-UI、Vercel AI 等 UI 事件流把它接到你自己的前端；ACP（实验性）还能把它作为编辑器代理提供服务（见 docs/interfaces.md）。

6. 持久化执行

在 Temporal、DBOS、Prefect、Restate 上提供官方维护的持久化执行，另有 Kitaru 与 Airflow 的外部 SDK 集成。Agent 能扛住重启、连续运行数天，内置人工审批（docs/deferred-tools.md）。

综合实战：一个银行客服代理

README 用一个银行客服代理把多个特性串起来：依赖注入、函数工具、结构化输出、打包客户上下文的可复用 Capability，以及模型仅在对话需要时按需加载的 on-demand Capability。该示例的完整可运行版本位于 examples/pydantic_ai_examples/bank_support.py，仓库文档中的详细版见 docs/examples/bank-support.md。

from dataclasses import dataclass from pydantic import BaseModel, Field from pydantic_ai import Agent, Capability, RunContext from bank_database import DatabaseConn @dataclass class SupportDependencies: # 注入任意客户端：数据库连接池、HTTP API、用户信息 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 = CapabilitySupportDependencies @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 # 签名与 docstring 成为 LLM 看到的工具 Schema 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 = CapabilitySupportDependencies @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-5.6-sol', deps_type=SupportDependencies, output_type=SupportOutput, # 运行返回经过校验的 SupportOutput，且类型如此声明 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], ) ... # 真实场景：更多工具、更长的指令 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( # 模型按需加载 refunds，然后作答 '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 """

对这个示例逐块拆解（对应文档注释见 docs/index.md 与 docs/agent.md）：

• 依赖注入：SupportDependenciesdataclass 把数据、连接与逻辑传进指令和工具函数，RunContext[SupportDependencies]携带依赖且类型安全。在单元测试与 evals 中替换成测试替身（test double），同一个 Agent 即可运行。
• 结构化输出：SupportOutput是 Pydantic 模型，从中自动生成 JSON Schema 告诉 LLM 如何返回数据，并在运行结束时校验数据正确性；该 Agent 的类型是Agent[SupportDependencies, SupportOutput]，校验失败时会被提示重试（反射与自我纠正机制）。
• Capability 打包：customer_context把客户上下文相关的指令与工具打包成一个可复用单元，可以直接原样放进任意其他 Agent 的capabilities列表；它还能整体放入语音 Agent 或 Web 应用而不变。
• 按需加载：defer_loading=True让refunds成为 on-demand capability：在提示词里它坍缩为一行目录条目，工具保持隐藏，直到模型判断相关并调用框架托管的load_capability工具加载它（详见 docs/capabilities/on-demand.md）。
• 工具 Schema 生成：工具 docstring 会传给 LLM 作为工具描述，参数描述从 docstring 中提取并加入参数 Schema；参数由 Pydantic 校验，出错信息回传给 LLM 以便重试。

没有 API Key 也能起步：内置'test'模型

任何场景都可以先用内置的'test'模型（Agent('test')）离线试跑——它完全不调用 LLM，让你先验证 Agent、工具与输出的正确性（见 docs/testing.md）。准备好真实模型后，参考 docs/models/overview.md 选择 Provider 并配置 API Key。

此外，框架提供了五种运行 Agent 的方式（docs/agent.md）：run()（异步）、run_sync()（同步）、run_stream()（流式文本与结构化输出）、run_stream_events()（遍历完整事件流）、iter()（底层图节点迭代）。消息历史与多轮对话可通过 docs/message-history.md 的说明继续传递。

可观测性：用 Logfire 点亮每一次模型与工具调用

即使只有几个工具的简单 Agent，也会与 LLM 产生大量往返，单靠读代码几乎无法确证发生了什么。配置 Logfire 后，在代码中加入三行即可：

import logfire logfire.configure() # 配置 Logfire SDK（项目未初始化时会失败） logfire.instrument_pydantic_ai() # 埋点此后所有 Pydantic AI Agent logfire.instrument_sqlite3() # 记录数据库查询（示例中 DatabaseConn 基于 sqlite3）

logfire.instrument_pydantic_ai()会为之后创建的所有 Agent 开启观测；若只想观测特定 Agent，可在其capabilities=[...]中加入一个Instrumentation条目（见 docs/capabilities/instrumentation.md）。底层是标准 OpenTelemetry：任何 OTLP 后端都可用，Pydantic Logfire 只是最省事的查看方式。

下一步与更多资源

• 立即运行：执行uvx --with pydantic-ai-harness clai -a pydantic_ai_harness.coder:coder_agent -m anthropic:claude-fable-5即可在终端启动完整编码代理；或参考 docs/install.md 安装后自建 Agent。
• 深入核心：Agent 核心指南见 docs/agent.md，完整接口见 docs/api/agent.md；工具与 Toolset 见 docs/tools.md 与 docs/toolsets.md。
• 扩展电池：官方 Capability 索引与自定义方法见 docs/capabilities/overview.md 和 docs/capabilities/custom.md。
• 示例代码：仓库 examples/pydantic_ai_examples 提供 bank_support、flight_booking、rag、sql_gen、weather_agent、realtime_voice 等可直接运行的示例；文档版示例从 docs/examples/setup.md 起步。

Pydantic AI 的核心理念可以概括为一句话：一个类型安全的 Agent 定义，任意模型、任意界面、全程可观测。从本文的五段可运行代码出发，配合 Capability 的组合机制，你可以在一个统一抽象之上逐步搭建从数据抽取到多 Agent 协作的生产级应用。

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai