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-aiPydantic AI 是 Pydantic 团队出品的 Python AI SDK核心是一个类型化、可扩展的 Agent 运行循环agent loop任何模型都只是一串字符串的差异同一个 Agent 定义可以运行在 Web 前端、终端、实时语音会话、持久化后台队列或普通函数调用等多种界面之下。读完本文你将掌握 Pydantic AI 的五大典型落地场景编码代理、结构化数据抽取、持久化工作流、实时语音、图像生成、核心的 Capability 扩展机制并能用依赖注入、函数工具、结构化输出和按需加载能力组合出一个生产可用的完整 Agent。项目概览How Python does AIPydantic AI 定位为Python 的 AI SDKAgents, 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-harnessfrom 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不是黑盒它就是一个常规的组合型 Capabilitycomposite capability。整用或拆用完全等价capabilities [ FileSystem(.), Shell(cwd.), RepoContext(), Planning(), SubAgents(...), ClearToolResults(), WarnNearLimits(), ToolOutputLimits(), ]想在不写任何代码的情况下先体验可以用claiPydantic AI 的 CLI直接运行导出的coder_agentuvx --with pydantic-ai-harness clai -a pydantic_ai_harness.coder:coder_agent -m anthropic:claude-fable-5场景二类型化数据抽取Data extraction给 Agent 一个输出类型和若干工具每一次运行都会返回经过校验、带类型的结构化结果uv add pydantic-aifrom 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, le1) agent Agent(openai:gpt-5.6-sol, output_typeSentiment) 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) # labelpositive score0.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, instructionsResearch the topic and write a structured brief., nameresearcher, 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(fWrite a brief on: {topic}) return result.outputDBOS 与 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( instructionsYou 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 fOrder {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-aifrom 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_typeBinaryImage的 provider 原生图像生成工具docs/native-tools.md、类型化的图像输出docs/output.md以及带 fallback 的ImageGenerationCapability用于处理本身不产图的模型。为什么选择 Pydantic AI六大设计取向1. 任何模型一套 Python API几乎所有主流模型与 ProviderOpenAI、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 提供其余部分完整的 AgentCoder、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_loadingTrue时必填供模型的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(descriptionAdvice returned to the customer) block_card: bool Field(descriptionWhether to block the customers card) risk: int Field(descriptionRisk level of query, ge0, le10) customer_context CapabilitySupportDependencies customer_context.instructions async def add_customer_name(ctx: RunContext[SupportDependencies]) - str: customer_name await ctx.deps.db.customer_name(idctx.deps.customer_id) return fThe customers name is {customer_name!r} customer_context.tool # 签名与 docstring 成为 LLM 看到的工具 Schema async def customer_balance( ctx: RunContext[SupportDependencies], include_pending: bool ) - float: Returns the customers current account balance. return await ctx.deps.db.customer_balance( idctx.deps.customer_id, include_pendinginclude_pending, ) refunds CapabilitySupportDependencies refunds.tool async def refund_status(ctx: RunContext[SupportDependencies]) - str: Look up the refund status for the customers most recent charge. return await ctx.deps.db.refund_status(idctx.deps.customer_id) support_agent Agent( openai:gpt-5.6-sol, deps_typeSupportDependencies, output_typeSupportOutput, # 运行返回经过校验的 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_id123, dbDatabaseConn()) result await support_agent.run(What is my balance?, depsdeps) print(result.output) support_adviceHello John, your current account balance, including pending transactions, is $123.45. block_cardFalse risk1 result await support_agent.run(I just lost my card!, depsdeps) print(result.output) support_adviceIm sorry to hear that, John. We are temporarily blocking your card to prevent unauthorized transactions. block_cardTrue risk8 result await support_agent.run( # 模型按需加载 refunds然后作答 Was I refunded for the duplicate charge on my last statement?, depsdeps ) print(result.output) support_adviceGood news, John: the duplicate charge on your last statement was refunded on 2026-05-01. block_cardFalse risk1 对这个示例逐块拆解对应文档注释见 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_loadingTrue让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.mdrun()异步、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 基于 sqlite3logfire.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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考