Pydantic AI 教程 · 阶段六

高级特性

解锁 AI 的隐藏技能：让它思考、搜网页、执行代码、看图片、连外部服务。

1思考模式：让 AI "先想想再回答"

什么是思考模式？

你有没有遇到过 AI 胡说八道的情况？让它做数学题、写复杂逻辑时，经常翻车。

思考模式就是让 AI 在回答之前先"打个草稿" —— 它会把推理过程写出来，然后再给你最终答案。就像考试时先在草稿纸上演算，再把答案抄到答题卡上。

Anthropic（Claude）：扩展思考

Claude 的思考模式最成熟，通过 anthropic_thinking 参数开启：

from pydantic_ai import Agent
from pydantic_ai.models.anthropic import AnthropicModel, AnthropicModelSettings

model = AnthropicModel('claude-sonnet-4-5')
settings = AnthropicModelSettings(
    anthropic_thinking={
        'type': 'enabled',
        'budget_tokens': 1024,  # 给思考过程分配的 token 预算
    },
)

agent = Agent(model, model_settings=settings)
result = agent.run_sync('15 的阶乘是多少？')
print(result.output)
budget_tokens 是什么？ 就是你给 AI "打草稿"分配的额度。设大了思考得更充分但更贵更慢，设小了可能想不明白。通常 1024~10000 够用。

自适应思考（推荐）

从 Claude Opus 4.6 开始，支持让 AI 自己决定要不要思考、思考多久：

model = AnthropicModel('claude-opus-4-6')
settings = AnthropicModelSettings(
    anthropic_thinking={'type': 'adaptive'},  # AI 自己决定
    anthropic_effort='high',  # 高努力 = 思考更多
)
agent = Agent(model, model_settings=settings)
努力等级： 'low'（简单问题快速回答）、'medium'（默认）、'high'（复杂问题深度思考）

OpenAI：推理模型

OpenAI 的推理功能需要用 Responses API（注意不是普通的 Chat API）：

from pydantic_ai import Agent
from pydantic_ai.models.openai import (
    OpenAIResponsesModel,
    OpenAIResponsesModelSettings,
)

model = OpenAIResponsesModel('gpt-5.2')
settings = OpenAIResponsesModelSettings(
    openai_reasoning_effort='low',      # low / medium / high
    openai_reasoning_summary='detailed', # 思考摘要
)
agent = Agent(model, model_settings=settings)
result = agent.run_sync('解释量子纠缠')
print(result.output)

Google（Gemini）：思考配置

from pydantic_ai import Agent
from pydantic_ai.models.google import GoogleModel, GoogleModelSettings

model = GoogleModel('gemini-3-pro-preview')
settings = GoogleModelSettings(
    google_thinking_config={'include_thoughts': True}
)
agent = Agent(model, model_settings=settings)

Groq：推理格式

Groq 上的推理模型（如 QwQ）：

from pydantic_ai import Agent
from pydantic_ai.models.groq import GroqModel, GroqModelSettings

model = GroqModel('qwen-qwq-32b')
settings = GroqModelSettings(
    groq_reasoning_format='parsed',  # 'raw' | 'hidden' | 'parsed'
)
agent = Agent(model, model_settings=settings)

各提供商思考模式速查

提供商模型示例关键设置
Anthropicclaude-sonnet-4-5anthropic_thinking
OpenAIgpt-5.2（Responses）openai_reasoning_effort
Googlegemini-3-pro-previewgoogle_thinking_config
Groqqwen-qwq-32bgroq_reasoning_format
OpenRouter各种模型openrouter_reasoning
xAIgrok-4-fast-reasoningxai_include_encrypted_content

2内置工具：AI 的超能力

什么是内置工具？

之前我们写的工具都是自己实现的（@agent.tool）。但有些能力是 AI 提供商原生自带的 —— 搜网页、跑代码、生图片等。这些不需要你写代码，直接开启就行。

用法很简单，通过 builtin_tools 参数传给 Agent：

from pydantic_ai import Agent, WebSearchTool

agent = Agent(
    'anthropic:claude-sonnet-4-5',
    builtin_tools=[WebSearchTool()],  # 直接传进去
)

网络搜索：WebSearchTool

让 AI 能上网搜索最新信息，不再局限于训练数据的截止日期。

from pydantic_ai import Agent, WebSearchTool

# Anthropic
agent = Agent(
    'anthropic:claude-sonnet-4-5',
    builtin_tools=[WebSearchTool()],
)
result = agent.run_sync('本周 AI 领域最大的新闻是什么？')
print(result.output)

# OpenAI 需要用 Responses 模型
agent = Agent(
    'openai-responses:gpt-5.2',
    builtin_tools=[WebSearchTool()],
)

高级配置

from pydantic_ai import Agent, WebSearchTool, WebSearchUserLocation

agent = Agent(
    'anthropic:claude-sonnet-4-5',
    builtin_tools=[
        WebSearchTool(
            user_location=WebSearchUserLocation(
                city='Beijing',
                country='CN',
            ),
            blocked_domains=['spam-site.net'],  # 屏蔽的网站
            allowed_domains=None,   # 只允许的网站（与 blocked 二选一）
            max_uses=5,             # 最多搜索几次（仅 Anthropic）
        )
    ],
)
提供商差异： 各提供商支持的参数不同。OpenAI 支持 search_context_size 和 allowed_domains；Anthropic 支持 blocked_domains 和 max_uses；Google 不支持任何参数。

代码执行：CodeExecutionTool

让 AI 写完代码后直接在沙箱里运行，特别适合数学计算、数据处理：

from pydantic_ai import Agent, CodeExecutionTool

agent = Agent(
    'anthropic:claude-sonnet-4-5',
    builtin_tools=[CodeExecutionTool()],
)

result = agent.run_sync('计算 15 的阶乘')
print(result.output)
# 15 的阶乘是 1,307,674,368,000
代码在哪里运行？ 在提供商的沙箱环境里，不是你的电脑。所以很安全，AI 写的代码不会搞坏你的系统。

图片生成：ImageGenerationTool

让 AI 直接生成图片：

from pydantic_ai import Agent, BinaryImage, ImageGenerationTool

agent = Agent(
    'openai-responses:gpt-5.2',
    builtin_tools=[ImageGenerationTool()],
)

result = agent.run_sync('画一只穿西装的猫')
print(result.output)  # 文字描述

# 图片在这里
image = result.response.images[0]  # BinaryImage 对象
# image.data 就是图片的二进制数据，可以保存为文件

配置图片参数

agent = Agent(
    'openai-responses:gpt-5.2',
    builtin_tools=[
        ImageGenerationTool(
            size='1024x1024',   # 尺寸
            quality='high',     # 质量
            output_format='png', # 格式
            background='transparent', # 透明背景
        )
    ],
    output_type=BinaryImage,  # 直接输出图片
)

网页抓取：WebFetchTool

让 AI 直接访问指定 URL 获取内容：

from pydantic_ai import Agent, WebFetchTool

agent = Agent(
    'anthropic:claude-sonnet-4-5',
    builtin_tools=[WebFetchTool()],
)

result = agent.run_sync('总结一下 https://ai.pydantic.dev 的内容')
print(result.output)

远程 MCP 服务：MCPServerTool

不用自己搭 MCP 客户端，直接让提供商帮你连接：

from pydantic_ai import Agent, MCPServerTool

agent = Agent(
    'anthropic:claude-sonnet-4-5',
    builtin_tools=[
        MCPServerTool(
            id='deepwiki',
            url='https://mcp.deepwiki.com/mcp',
        )
    ],
)

result = agent.run_sync('介绍一下 pydantic/pydantic-ai 这个仓库')
print(result.output)

文件搜索（托管 RAG）：FileSearchTool

提供商托管的 RAG 系统，自动处理文件上传、分块、嵌入、搜索：

from pydantic_ai import Agent, FileSearchTool
from pydantic_ai.models.openai import OpenAIResponsesModel

model = OpenAIResponsesModel('gpt-5.2')

# 先上传文件到 OpenAI（只需做一次）
# file = await model.client.files.create(file=open('doc.txt','rb'), purpose='assistants')
# vector_store = await model.client.vector_stores.create(name='my-docs')
# await model.client.vector_stores.files.create(vector_store_id=vector_store.id, file_id=file.id)

agent = Agent(
    model,
    builtin_tools=[
        FileSearchTool(file_store_ids=['vs_xxx']),  # 向量存储 ID
    ],
)
result = agent.run_sync('文档里关于 Pydantic 说了什么？')

动态配置内置工具

有时候需要根据运行时条件决定是否启用某个工具：

from pydantic_ai import Agent, RunContext, WebSearchTool

async def maybe_search(ctx: RunContext[dict]) -> WebSearchTool | None:
    """只有传了位置信息时才启用搜索"""
    if not ctx.deps.get('location'):
        return None
    return WebSearchTool(
        user_location={'city': ctx.deps['location']},
    )

agent = Agent(
    'openai-responses:gpt-5.2',
    builtin_tools=[maybe_search],  # 传函数而不是实例
    deps_type=dict,
)

# 有位置 → 启用搜索
result = agent.run_sync('今天天气怎么样？', deps={'location': 'Beijing'})

# 没位置 → 不启用搜索
result = agent.run_sync('法国的首都是哪里？', deps={'location': None})

内置工具支持一览

工具OpenAIAnthropicGooglexAIGroq
WebSearchToolResponses完整无参数部分混合模型
CodeExecutionTool完整完整完整完整不支持
ImageGenerationToolResponses不支持图片模型不支持不支持
WebFetchTool不支持完整无参数不支持不支持
MCPServerToolResponses完整不支持完整不支持
FileSearchToolResponses不支持Gemini不支持不支持

3MCP 集成：连接万物

什么是 MCP？

MCP（Model Context Protocol） 是一个标准化协议，让 AI 应用能连接到外部工具和服务。就像 USB 是电脑连接外设的标准，MCP 是 AI 连接工具的标准。

有了 MCP，你的 AI Agent 可以：

三种连接方式

方式一：Streamable HTTP（推荐）

适合远程 MCP 服务器：

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

server = MCPServerStreamableHTTP('http://localhost:8000/mcp')
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    result = await agent.run('7 加 5 等于多少？')
    print(result.output)  # 答案是 12

方式二：Stdio（本地子进程）

MCP 服务器作为子进程运行，通过标准输入/输出通信：

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio(
    'python', args=['mcp_server.py'],
    timeout=10,
)
agent = Agent('openai:gpt-4o', toolsets=[server])
什么时候用 Stdio？ 当 MCP 服务器是一个本地脚本时。比如你自己写了一个 Python 脚本作为 MCP 工具服务器。

方式三：SSE（已废弃）

from pydantic_ai.mcp import MCPServerSSE

server = MCPServerSSE('http://localhost:3001/sse')
SSE 正在被 Streamable HTTP 替代，新项目建议用方式一。

写一个 MCP 服务器

用 FastMCP 几行代码就能写一个：

from mcp.server.fastmcp import FastMCP

app = FastMCP('我的工具服务器')

@app.tool()
def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

@app.tool()
def multiply(a: int, b: int) -> int:
    """两数相乘"""
    return a * b

if __name__ == '__main__':
    app.run(transport='streamable-http')

然后客户端连过去：

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP

server = MCPServerStreamableHTTP('http://localhost:8000/mcp')
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    result = await agent.run('3 乘以 7 再加 5 等于多少？')
    print(result.output)  # 26

从配置文件加载

对于多个 MCP 服务器，可以用 JSON 配置：

{
  "mcpServers": {
    "calculator": {
      "url": "http://localhost:8000/mcp"
    },
    "python-runner": {
      "command": "uv",
      "args": ["run", "mcp-run-python", "stdio"]
    },
    "weather": {
      "command": "python",
      "args": ["weather_server.py"],
      "env": {
        "API_KEY": "${WEATHER_API_KEY}"
      }
    }
  }
}
注意 ${WEATHER_API_KEY} —— 支持环境变量替换，敏感信息不用写死在配置里。

加载使用：

from pydantic_ai import Agent
from pydantic_ai.mcp import load_mcp_servers

servers = load_mcp_servers('mcp_config.json')
agent = Agent('openai:gpt-4o', toolsets=servers)

工具前缀：防止冲突

连接多个 MCP 服务器时，工具名可能冲突。加前缀解决：

weather_server = MCPServerStreamableHTTP(
    'http://localhost:3001/mcp',
    tool_prefix='weather',  # 工具名变成 weather_xxx
)
calc_server = MCPServerStreamableHTTP(
    'http://localhost:3002/mcp',
    tool_prefix='calc',     # 工具名变成 calc_xxx
)

agent = Agent('openai:gpt-4o', toolsets=[weather_server, calc_server])

MCP 采样：让服务器借用你的 AI

有时候 MCP 服务器自己也需要调用 AI（但它没有 API Key）。采样（Sampling） 让服务器可以借用客户端的 AI 模型：

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio('python', args=['generate_svg.py'])
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    # 告诉所有 MCP 服务器可以借用 Agent 的模型
    agent.set_mcp_sampling_model()
    result = await agent.run('生成一个机器人的 SVG 图片')
    print(result.output)
大白话： MCP 服务器说"我需要让 AI 帮我生成一段 SVG"，但它没有 API Key。采样机制让它通过客户端发起 AI 请求，由客户端来付费。

4多模态输入：让 AI 看图、听声音、读文档

图片输入

让 AI 看懂图片：

from pydantic_ai import Agent, ImageUrl

agent = Agent('openai:gpt-4o')

# 方式一：传 URL
result = agent.run_sync([
    '这个 logo 是什么公司的？',
    ImageUrl(url='https://iili.io/3Hs4FMg.png'),
])
print(result.output)
# 这是 Pydantic 的 logo

# 方式二：传本地文件
from pydantic_ai import BinaryContent

with open('photo.png', 'rb') as f:
    image_data = f.read()

result = agent.run_sync([
    '描述这张图片',
    BinaryContent(data=image_data, media_type='image/png'),
])
注意输入格式： 多模态输入时，run_sync() 的第一个参数变成了一个列表，里面混合文字和媒体内容。

音频和视频输入

from pydantic_ai import Agent, AudioUrl, VideoUrl

agent = Agent('google-gla:gemini-2.0-flash')

# 音频
result = agent.run_sync([
    '这段音频说了什么？',
    AudioUrl(url='https://example.com/audio.mp3'),
])

# 视频
result = agent.run_sync([
    '这个视频讲的是什么？',
    VideoUrl(url='https://example.com/video.mp4'),
])

文档输入（PDF 等）

from pydantic_ai import Agent, DocumentUrl

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

# 从 URL 读取 PDF
result = agent.run_sync([
    '总结这篇论文的主要内容',
    DocumentUrl(url='https://example.com/paper.pdf'),
])

# 从本地文件读取
from pathlib import Path
from pydantic_ai import BinaryContent

pdf_data = Path('report.pdf').read_bytes()
result = agent.run_sync([
    '这份报告的关键数据是什么？',
    BinaryContent(data=pdf_data, media_type='application/pdf'),
])

各模型支持情况

模型图片音频视频PDF
OpenAI ChatURL 直发需下载不支持需下载
OpenAI ResponsesURL 直发URL 直发不支持URL 直发
AnthropicURL 直发不支持不支持URL 直发
Google (Vertex)全部 URL全部 URL全部 URL全部 URL
Google (GLA)需下载需下载YouTube 直发需下载
"URL 直发" 表示直接把 URL 给模型，模型自己去下载。"需下载" 表示 Pydantic AI 会先下载内容再发给模型。可以用 force_download=True 强制先下载。

强制下载

某些场景下你想确保由 Pydantic AI 来下载（比如 URL 需要认证）：

from pydantic_ai import ImageUrl

ImageUrl(
    url='https://internal.company.com/image.png',
    force_download=True,  # 不管模型支不支持 URL，都先下载
)

5Direct API：绕过 Agent 直接调模型

什么时候用？

Agent 是个全能管家，但有时候你只是想简单调一次模型，不需要工具、不需要重试、不需要结构化输出。这时候用 Direct API 更轻量。

Agent API 完整框架 → 工具执行 → 重试 → 结构化输出 → ……
Direct API 发请求 → 拿响应。完事。

基础用法

from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_sync

# 一句话搞定
response = model_request_sync(
    'anthropic:claude-haiku-4-5',
    [ModelRequest.user_text_prompt('法国的首都是哪里？')],
)

print(response.parts[0].content)
# 法国的首都是巴黎。

print(response.usage)
# RequestUsage(input_tokens=56, output_tokens=7)

四种调用方式

from pydantic_ai.direct import (
    model_request,            # 异步
    model_request_sync,       # 同步
    model_request_stream,     # 异步流式
    model_request_stream_sync, # 同步流式
)

带工具定义的调用

Direct API 也能传工具定义，但需要你自己处理工具调用：

from pydantic import BaseModel
from pydantic_ai import ModelRequest, ToolDefinition
from pydantic_ai.direct import model_request
from pydantic_ai.models import ModelRequestParameters

class Divide(BaseModel):
    """两数相除"""
    numerator: float
    denominator: float

async def main():
    response = await model_request(
        'openai:gpt-4o',
        [ModelRequest.user_text_prompt('123 除以 456 等于多少？')],
        model_request_parameters=ModelRequestParameters(
            function_tools=[
                ToolDefinition(
                    name='divide',
                    description='两数相除',
                    parameters_json_schema=Divide.model_json_schema(),
                )
            ],
            allow_text_output=True,
        ),
    )
    # response 可能是文本回复，也可能是工具调用
    print(response)

加监控

import logfire
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_sync

logfire.configure()

# 方式一：全局启用
logfire.instrument_pydantic_ai()

# 方式二：单次请求启用
response = model_request_sync(
    'anthropic:claude-haiku-4-5',
    [ModelRequest.user_text_prompt('你好')],
    instrument=True,  # 只监控这一次
)

Agent vs Direct：怎么选？

需求选择
需要工具调用Agent
需要结构化输出Agent
需要重试和验证Agent
需要多轮对话Agent
只是简单问答Direct
想自己控制一切Direct
构建自定义抽象Direct

总结

特性一句话总结典型场景
思考模式让 AI 先想再答数学、逻辑、复杂推理
内置工具提供商原生能力搜网页、跑代码、生图片
MCP 集成连接外部服务数据库、文件系统、API
多模态输入看图听声读文档图片识别、PDF 分析
Direct API轻量级调用简单问答、自定义封装

这些特性怎么组合？

你的 AI 应用
Agent（核心引擎）
→ 思考模式 —— 让复杂推理更准确
→ 内置工具 —— 搜索 + 代码 + 图片
→ 自定义工具 —— 你自己的业务逻辑
→ MCP 工具集 —— 连接外部服务
→ 多模态输入 —— 图片/音频/视频/PDF
Direct API（辅助）
→ 简单的一次性调用

动手练习

练习 1：带思考的数学 Agent

from pydantic_ai import Agent
from pydantic_ai.models.anthropic import AnthropicModel, AnthropicModelSettings

model = AnthropicModel('claude-sonnet-4-5')
settings = AnthropicModelSettings(
    anthropic_thinking={
        'type': 'enabled',
        'budget_tokens': 2048,
    },
)

math_agent = Agent(
    model,
    model_settings=settings,
    instructions='你是一个数学老师，解题时要展示详细的推理过程。',
)

result = math_agent.run_sync('一个水池有两个进水管和一个排水管...')
print(result.output)

练习 2：多模态图片分析

from pydantic_ai import Agent, ImageUrl
from pydantic import BaseModel

class ImageAnalysis(BaseModel):
    description: str
    objects: list[str]
    mood: str

agent = Agent(
    'openai:gpt-4o',
    output_type=ImageAnalysis,
)

result = agent.run_sync([
    '分析这张图片',
    ImageUrl(url='https://example.com/photo.jpg'),
])
print(f'描述: {result.output.description}')
print(f'物体: {result.output.objects}')
print(f'氛围: {result.output.mood}')

练习 3：MCP 计算器

服务端（calc_server.py）：

from mcp.server.fastmcp import FastMCP

app = FastMCP('Calculator')

@app.tool()
def calculate(expression: str) -> str:
    """安全地计算数学表达式。"""
    try:
        result = eval(expression, {"__builtins__": {}})
        return str(result)
    except Exception as e:
        return f'计算错误: {e}'

if __name__ == '__main__':
    app.run()

客户端：

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

server = MCPServerStdio('python', args=['calc_server.py'])
agent = Agent('openai:gpt-4o', toolsets=[server])

async def main():
    result = await agent.run('(123 + 456) * 789 等于多少？')
    print(result.output)

下一步：阶段七 · 生产化

恭喜你完成了高级特性的学习！现在你已经掌握了 Pydantic AI 的全部核心能力。
接下来学习测试、评估、可观测性和持久化执行，为上线做准备。

Pydantic AI Tutorial · Phase 6: Advanced Features