Pydantic-AI 是由 Pydantic 团队（FastAPI、Pydantic 背后的团队）推出的 Python Agent 框架，旨在将 FastAPI 的开发体验带入生成式 AI 应用开发领域。其核心目标是让开发者快速、自信且轻松地构建生产级 AI Agent，采用“FastAPI for agents”的设计理念。
一、特点
该框架具备类型安全、模型无关、结构化输出等特性，支持 OpenAI、Anthropic、Gemini 等主流模型提供商，并集成了依赖注入、流式响应、可观测性等功能。它将 Pydantic 的数据验证能力带入 LLM 系统，确保 AI 输出能够转换为经过验证的 Python 对象，使 Agent 更安全、可预测且易于扩展
二、安装
打开命令行，输入：
# 依赖注入或结果返回类型时用到
pip install pydantic
# 完整安装pydantic-ai
pip install pydantic-ai
# 技能加载和读取, 官方pydantic_ai_harness上扩展
pip install pydantic_ai_skills
三、基本使用
通过deepseek大模型提供服务
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.deepseek import DeepSeekProvider
# deepseek 密钥
api_key = "sk-xxxxxxxxxx"
# 也可通过环境变量中获取
api_key = os.environ["DEEPSEEK_API_KEY"]
provider = DeepSeekProvider(api_key=api_key)
deepseek_model = OpenAIChatModel(
"deepseek-chat",
provider=provider,
)
写一个函数追踪和量化一次 Agent 运行所消耗的 LLM 资源
def print_usage_info(usage):
print(f"输入token: {usage.input_tokens}")
print(f"输出token: {usage.output_tokens}")
print(f"请求次数: {usage.requests}")
同步，异步两种运行方式
也支持流式运行
下面是基础动用请求
agent = Agent(deepseek_model)
result = agent.run_sync("第一个次运行智能体")
print(result.output[:20])
print_usage_info(result.usage)
输出
你好！很高兴见到你！🎉
输入token: 9
输出token: 183
请求次数: 1
四、内部工具注册
Pydantic-ai可以通过装饰器内部工具注册到AI模型，让 AI 模型能够调用你自定义的 Python 函数，以获取额外信息或执行特定操作，从而扩展模型的能力边界。
4.1 注册纯函数
使用 @agent.tool_plain：纯函数，无需上下文
agent = Agent(
deepseek_model
)
@agent.tool_plain
def multiply(a: float, b: float) -> float:
"""计算两个数相乘。需要精确乘法时调用。"""
return a * b
result = agent.run_sync("计算 23 * 10 是多少?")
print(result.output)
print_usage_info(result.usage)输出,通过输出可以看到每次调用工具，AI模型都是请求2次
23 × 10 = **230**
输入token: 660
输出token: 61
请求次数: 2
4.2 注册上下文函数
使用 @agent.tool函数的第一个参数必须是 RunContext 类型。这个对象是你的工具与 Agent 运行时状态交互的桥梁。
- • ctx.deps：你在创建 Agent 时通过 deps_type 声明的依赖项。
- • ctx.usage：当前运行的 Token 消耗等用量统计。
- • ctx.messages：到目前为止的消息历史。
下面是依赖注入示例
from pydantic import BaseModel
from pydantic_ai import RunContext
# 使用pydantic定义依赖类
class Deps(BaseModel):
user_id: str
# 依赖注入类型为
agent = Agent(deepseek_model, deps_type=Deps)
# 自动注册
@agent.tool
def get_user(ctx: RunContext[Deps]) -> str:
# print(ctx.usage)
# print(ctx.agent)
# print(ctx.messages)
# print(ctx.retry)
# print(ctx.deps)
return ctx.deps.user_id
result = agent.run_sync("获取当前用户信息",
deps=Deps(user_id="u42"))
print(result.output)
print_usage_info(result.usage)
五、接MCP服务
MCP（Model Context Protocol，模型上下文协议） 是一个由 Anthropic 于 2024 年 11 月开源的标准化协议，旨在为 LLM 应用与外部数据源、工具之间提供统一的连接方式。你可以把它理解为 “AI 世界的 USB-C 接口”——无论背后是数据库、文件系统还是第三方 API，只要遵循 MCP 标准，AI 就能即插即用。
5.1 mcp服务定义
使用fastmcp定义mcp服务端的工具，资源，提示词等。
import sys
from pathlib import Path
from mcp.server.mcpserver import MCPServer
mcp = MCPServer(
"local-files",
instructions="提供本地目录扫描能力。相对路径按服务器的工作目录解析。",
)
@mcp.tool()
def list_dir(path: str = ".") -> str:
"""列出目录下的直接子项，目录名带 `/` 后缀。用于查看某个目录里有什么文件。"""
target = Path(path).expanduser().resolve()
if not target.is_dir():
return f"错误：{target} 不是一个目录"
entries = sorted(target.iterdir(), key=lambda item: (item.is_file(), item.name.lower()))
if not entries:
return f"{target} 是空目录"
lines = [f"{item.name}/" if item.is_dir() else item.name for item in entries]
return "\n".join([f"{target} 共 {len(entries)} 项：", *lines])
@mcp.prompt()
def ask_about_topic(topic: str) -> str:
"""生成一条询问某个主题解释的用户消息。"""
return f"你能解释一下‘{topic}’这个概念吗？"
@mcp.resource("resource://mcp_name")
def get_mcp_name() -> str:
"""提供一条简单的问候消息。"""
return "local-files"
@mcp.resource("resource://dir_size/{path}")
def get_dir_size(path: str) -> str:
"""报告目标目录下的条目数量。URI 里带 {path} 时会注册成资源模板。
注意：参数必须是「路径安全」的普通片段，不能塞 Windows 绝对路径 ——
URI 里的 `:` 和 `\\` 会让模板匹配失败，服务端会报 Unknown resource。
所以这里当相对目录名用。
"""
target = Path(path).expanduser().resolve()
if not target.is_dir():
return f"错误：{target} 不是一个目录"
return f"{target} 下有 {len(list(target.iterdir()))} 个条目"
if __name__ == "__main__":
transport = sys.argv[1] if len(sys.argv) > 1 else "stdio"
if transport in ("http", "streamable-http"):
# 客户端连接地址：http://127.0.0.1:/mcp
mcp.run("streamable-http", host="127.0.0.1", port=8000)
else:
mcp.run("stdio")
5.2 使用stdio方式调用
stdio方式启动mcp服务
async def mcp_stdio_demo() -> None:
# 原始 toolset：用来直接读提示词 / 资源
toolset = MCPToolset(
StdioTransport(sys.executable, ["server.py"], cwd=os.path.dirname(__file__))
)
# 加前缀的版本：只用来注册给 agent（工具名会变成 local_list_dir）。
# 注意 PrefixedToolset 只暴露工具，没有 list_prompts / read_resource 这些方法，
# 所以上面必须保留一份未加前缀的引用。
prefixed_toolset = toolset.prefixed("local")
# 注意：list_prompts / list_tools / list_resources / read_resource 都是协程，
# 必须 await，而且必须在 async with agent 里 —— 连接由 agent 管理，
# 不在上下文里直接调用只能拿到 coroutine 对象。
async with Agent(deepseek_model, toolsets=[prefixed_toolset])
as agent:
print("提示词:", await toolset.list_prompts())
print("工具:", await toolset.list_tools())
print("资源:", await toolset.list_resources())
# 模板资源和静态资源是两套 API：list_resources 只返回静态资源，
# 带 {占位符} 的模板要用 list_resource_templates 单独列。
print("资源模板:", await toolset.list_resource_templates())
print("静态资源内容:", await toolset.read_resource("resource://mcp_name"))
result = await agent.run("用 local_list_dir 工具列出当前目录")
print(result.output)
print_usage_info(result.usage)
# 模板资源：把占位符换成真实值再读。
# 路径参数必须是路径安全的片段，不能传 Windows 绝对路径（URI 里的 : 和 \ 匹配不上）。
print("模板资源内容:", await toolset.read_resource("resource://dir_size/agents"))
asyncio.run(mcp_stdio_demo())
5.3 使用streamable-http方式通信
启动mcp服务
python server.py http
这个和前面的stdio方式仅修改几行代码即可
...
toolset = MCPToolset(
StreamableHttpTransport("http://127.0.0.1:8000/mcp")
)
prefixed_toolset = toolset.prefixed("http")
...
result =
await agent.run("用 http_list_dir 工具列出当前目录")
...
同时工具可以同步调用
agent = Agent(deepseek_model, toolsets=[prefixed_toolset])
result = agent.run_sync("用 http_list_dir 工具列出当前目录")
print(result.output)
六、Capability 即插即用插件
一个 Capability 将 Agent 的指令、工具、生命周期钩子和模型设置捆绑成一个单一的、可组合的单元
6.1 提供工具
refunds = Capability(
id="refunds",
description="处理退款：查退款状态、判断是否可退、发起退款。",
instructions="处理退款前必须先确认订单号，再调用工具查询。"
)
@refunds.tool_plain
def refund_status(order_id: str) -> str:
"""查询某笔订单的退款状态。"""
return f"订单 {order_id}：退款已于 2026-05-01 发出。"
@refunds.tool_plain
def refund_eligible(order_id: str, days_since_purchase: int) -> str:
"""判断某笔订单是否还在可退款期内（30 天内）。"""
return f"订单 {order_id}：{'可退款' if days_since_purchase <= 30
else '已超过 30 天，不可退款'}"
调用插件的工具
agent = Agent(
deepseek_model,
instructions="你是一个客服助手。",
capabilities=[refunds],
)
result = agent.run_sync("帮我查一下订单 A123 能不能退款，已经买了 10 天")
print("回答:", result.output)
6.2 延迟加载
模型需要时调用内置的 load_capability 工具把这个包整体解包
只要设置defer_loading=True即可
refunds_deferred = Capability(
id="refunds",
description="处理退款：查退款状态、判断是否可退、发起退款。",
instructions="处理退款前必须先确认订单号，再调用工具查询。",
defer_loading=True, # ← 只差这一个参数
)
# 工具调用链，无能力包加载
result = agent.run_sync("你好，请介绍一下你能做什么。")
#问到退款：模型会先 load_capability 解包，再调用 refund_status
result2 = agent.run_sync("订单 A123 的退款状态？")
6.3 自定义 Capability
AbstractCapability[Deps] 钩子和方法里能拿到 RunContext[Deps]
@dataclass
class
WeatherCapability(AbstractCapability[None]):
"""提供天气查询工具的能力包。"""
city_default: str = "北京"
def get_toolset(self) -> FunctionToolset:
"""返回一个 toolset；工具在能力包内部定义，外部完全无感。"""
toolset = FunctionToolset()
@toolset.tool_plain
def get_weather(city: str) -> str:
"""查询指定城市的天气。"""
return f"{city}：晴，25°C"
return toolset
def get_instructions(self):
"""返回可调用对象 = 动态指令，每次运行实时计算。"""
def resolve(ctx: RunContext[None]) -> str:
return f"回答天气问题时，用户没指定城市就默认查 {self.city_default}。"
return resolve
@dataclass
class KnowsCurrentTime(AbstractCapability[None]):
"""把当前时间动态写进系统提示词。返回可调用对象即动态指令。"""
def get_instructions(self):
def resolve(ctx: RunContext[None]) -> str:
return f"当前时间是 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}。"
return resolve
调用方式
agent = Agent(
deepseek_model,
instructions="你是一个助手，回答尽量简短。",
capabilities=[
WeatherCapability(city_default="上海"),
KnowsCurrentTime()],
)
result = agent.run_sync("现在几点了？另外今天天气怎么样？")6.4 按步骤设置模型参数
from pydantic_ai.capabilities import AbstractCapability
@dataclass
class EffortOnRetry(AbstractCapability[None]):
"""首次尝试用默认参数，进入第 2 步之后提高温度（示例用 temperature 代替 thinking）。"""
def get_model_settings(self):
def resolve(ctx: RunContext[None]):
from pydantic_ai import ModelSettings
if ctx.run_step > 1:
print(f" [模型设置] 第 {ctx.run_step} 步 → temperature=0.9")
return ModelSettings(temperature=0.9)
print(f" [模型设置] 第 {ctx.run_step} 步 → 默认参数")
return ModelSettings()
return resolve
调用方式
agent = Agent(
deepseek_model,
instructions="你是一个助手。",
capabilities=[EffortOnRetry(), refunds],
)
# 触发工具调用 → 至少两步，能观察到参数随步骤变化
result = agent.run_sync("订单 A123 的退款状态？")
6.5 Hooks：生命周期钩子
- • Hooks 一种轻量形态：@hooks.on. 注册。
- • before_* 按声明顺序执行，after_* 按相反顺序执行。
from pydantic_ai.capabilities import Hooks
hooks = Hooks()
@hooks.on.before_model_request
async def log_request(ctx: RunContext[None], request_context):
"""每次请求模型前触发。必须把 request_context 返回回去（可以改了再返回）。"""
print(f" [钩子] before_model_request 第 {ctx.run_step} 步")
return request_context
@hooks.on.after_tool_execute
async def log_tool(ctx: RunContext[None], *, call, tool_def, args, result):
"""工具执行完触发。注意参数是关键字形式：call / tool_def / args / result。"""
print(f" [钩子] after_tool_execute {tool_def.name}({args}) → {result}")
return result
调用方式
agent = Agent(
deepseek_model,
instructions="你是一个客服助手。",
capabilities=[hooks, refunds],
)
result = agent.run_sync("订单 A123 的退款状态？")
6.6 按依赖动态选模型
选择器收到的是 ModelSelectionContext
def pick_model(ctx) -> str:
tier = (ctx.deps or {}).get("user_tier")
chosen = "deepseek:deepseek-reasoner" if tier == "premium" else "deepseek:deepseek-chat"
print(f" [选模型] 第 {ctx.run_step} 步 tier={tier} → {chosen}")
return chosen
调用方式
agent = Agent(
MODEL, # 兜底默认模型
deps_type=dict,
instructions="你是一个助手，回答一句话即可。",
capabilities=[SelectModel(pick_model)],
)
6.7 多 Capability 组合
顺序有先原则
step_a = Capability(
id="step-a",
instructions="你回答时，第一行必须输出：[流程A]",
)
step_b = Capability(
id="step-b",
instructions="你回答时，第二行必须输出：[流程B]",
)
step_c = Capability(
id="step-c",
instructions="你回答时，第三行必须输出：[流程C]",
)
agent = Agent(
deepseek-model,
instructions="严格按指令输出，不要多余解释。",
capabilities=[step_a, step_b, step_c],
)
七、技能SKILL加载使用
Skill（技能）是AI智能体领域的一个新兴概念，你可以把它理解为Agent的 “操作指南”或“专业课本” 。它是一套标准化的机制，用于将完成特定任务所需的指令、知识和资源打包成一个可复用的模块，让Agent在需要时按需加载，从而更稳定、更专业地完成工作。
7.1 技能介绍
一个Skill就是一个文件夹，其核心是一个名为 SKILL.md 的Markdown文件，其中包含YAML格式的元数据（如名称、描述）和详细的Markdown指令。典型的目录结构如下
my-skill/
├── SKILL.md # 核心：元数据 + 指令
├── scripts/ # 可选：可执行脚本（如Python、Bash）
├── references/ # 可选：参考文档、领域知识
└── assets/ # 可选：模板、数据文件等静态资源
7.2 定义和调用
可以使用pydantic_ai_skills或pydantic_ai_harness加载使用
安装技能调用库
pip install pydantic_ai_skills
流程
- • 从 agents/skills 发现 SKILL.md，只把 name+description 注入系统提示词，
- • 命中后再由 Agent 调 load_skill 读完整指令（渐进式披露）
agents/
└─skills
└─list-dir
└─ SKILL.md
SKILL.md定义
---
name: list-dir
description: 列出目录内容。当用户要求"列出目录""看看当前目录有什么文件"时使用。
---
# 目录扫描技能
## 步骤
1. 用 `list_dir` 工具列出目标目录。
2. 把结果按 文件 / 目录 分组返回给用户。
通过下面代码调用
from pydantic_ai_skills import SkillsCapability
capability_skills = SkillsCapability(
directories=[Path(__file__).parent / "agents" / "skills"]
)
agent = Agent(
deepseek_model,
capabilities=[capability_skills],
)
result = agent.run_sync("获取 list_dir 工具列出当前目录")
print(result.output)
from pydantic_ai
import ModelRequest
for message in result.all_messages():
if isinstance(message, ModelRequest):
print(message)
print_usage_info(result.usage)
上面介绍了pydantic-ai中工具，mcp、skill、capability定义及调用的全过程，在使用智能体中还有边界安全、短期记忆、长期记忆、多智能体排版各种功能，等待你去研究。
