核心概念

1深入理解 Agent

阶段一里我们已经会用 Agent 了，现在来搞懂它的全貌。

Agent 就是一个容器

你可以把 Agent 想象成一个工具箱，里面装着 AI 完成任务所需的一切：

Agent 工具箱
├── instructions（指令）   → 告诉 AI 你是谁、该怎么做
├── tools（工具）          → AI 能调用的函数
├── output_type（输出类型） → 规定 AI 必须返回什么格式
├── deps_type（依赖类型）   → 运行时需要传入的数据
├── model（模型）           → 用哪个大语言模型
└── model_settings（设置）  → 温度、最大 token 数等参数

三种指令写法

1. 静态指令 —— 直接写在构造函数里

最常用，适合指令内容不会变的情况：

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-4o',
    instructions='你是一个专业的 Python 程序员，用中文回答问题。'
)

2. 动态指令 —— 用装饰器，可以访问依赖

指令内容需要根据运行时数据变化时用这个：

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class UserInfo:
    name: str
    vip: bool

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

@agent.instructions
def add_user_info(ctx: RunContext[UserInfo]) -> str:
    level = 'VIP 客户' if ctx.deps.vip else '普通客户'
    return f'当前用户是 {ctx.deps.name}，{level}，请提供相应等级的服务。'

# 不同用户运行时，指令会自动变化
result = agent.run_sync('帮我查询余额', deps=UserInfo('张三', vip=True))

3. 运行时指令 —— 调用 run 时临时追加

result = agent.run_sync(
    '帮我写个函数',
    instructions='这次请用英文回答，代码注释也用英文。'
)

三种指令可以叠加使用！ 静态指令先发，动态指令跟上，运行时指令最后追加。AI 会看到所有指令。

instructions vs system_prompt

你可能在其他教程里见过 system_prompt，这里解释一下两者的区别：

	instructions	system_prompt
多轮对话时	每次运行重新发送，不保留在历史里	保留在消息历史中
推荐场景	大部分场景（默认选这个）	需要让后续对话也能看到系统提示时

运行 Agent 的三种常用方式

from pydantic_ai import Agent

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

# 方式 1：同步运行（最简单，适合脚本和测试）
result = agent.run_sync('你好')
print(result.output)

# 方式 2：异步运行（适合 Web 应用）
import asyncio

async def main():
    result = await agent.run('你好')
    print(result.output)

asyncio.run(main())

# 方式 3：流式运行（打字机效果，体验好）
async def stream_demo():
    async with agent.run_stream('给我讲个笑话') as response:
        async for text in response.stream_text():
            print(text)  # 一段一段打印出来

asyncio.run(stream_demo())

怎么选？

• 写脚本、做测试 → run_sync()
• Web 应用、异步代码 → await agent.run()
• 需要实时展示 → run_stream()

调节模型参数

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-4o',
    model_settings={
        'temperature': 0.0,   # 0=最确定，1=最随机
        'max_tokens': 500,     # 最多返回多少 token
    }
)

# 也可以在运行时临时覆盖
result = agent.run_sync(
    '写一首诗',
    model_settings={'temperature': 0.9}  # 写诗时调高随机性
)

2依赖注入：给 AI 传递外部数据

什么是依赖注入？

听着高大上，其实就是一句话：把工具函数需要的外部数据，在运行时传进去。

为什么不直接用全局变量？因为：

• 全局变量不好测试（测试时想换个假数据库就很麻烦）
• 全局变量不安全（不同请求可能互相污染）
• 全局变量不灵活（不同用户传不同数据）

三步走

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

# 第一步：定义依赖（用 dataclass 最方便）
@dataclass
class MyDeps:
    user_id: int
    db_connection: object  # 数据库连接
    api_key: str            # 外部 API 密钥

# 第二步：创建 Agent 时声明依赖类型
agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)

# 第三步：在工具和指令里通过 ctx.deps 访问
@agent.tool
def get_user_orders(ctx: RunContext[MyDeps], limit: int = 10) -> str:
    """查询用户最近的订单。

    Args:
        limit: 返回的订单数量
    """
    # ctx.deps 就是你传进来的 MyDeps 对象
    orders = ctx.deps.db_connection.query(
        f'SELECT * FROM orders WHERE user_id = {ctx.deps.user_id} LIMIT {limit}'
    )
    return str(orders)

运行时把真实数据传进去：

deps = MyDeps(user_id=42, db_connection=db, api_key='sk-xxx')
result = agent.run_sync('我最近买了什么？', deps=deps)

依赖能用在三个地方

@dataclass
class Deps:
    username: str

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

# 1. 动态指令里
@agent.instructions
def personalize(ctx: RunContext[Deps]) -> str:
    return f'用户名是 {ctx.deps.username}，请用他的名字称呼他。'

# 2. 工具函数里
@agent.tool
def get_profile(ctx: RunContext[Deps]) -> str:
    """获取用户资料。"""
    return f'{ctx.deps.username} 的资料信息...'

# 3. 输出验证器里（后面会讲）
@agent.output_validator
def check_output(ctx: RunContext[Deps], output: str) -> str:
    if ctx.deps.username not in output:
        raise ModelRetry('请在回复中包含用户名')
    return output

测试时替换依赖

这是依赖注入最大的好处——测试时可以用假数据：

# 生产代码
real_deps = MyDeps(user_id=42, db_connection=real_db, api_key='sk-real')
result = await agent.run('查询余额', deps=real_deps)

# 测试代码 —— 用 override 替换
fake_deps = MyDeps(user_id=0, db_connection=fake_db, api_key='sk-test')
with agent.override(deps=fake_deps):
    result = await agent.run('查询余额', deps=fake_deps)

3结构化输出：让 AI 返回你想要的格式

阶段一讲了基本用法，现在来看更多高级玩法。

支持的输出类型一览

from pydantic import BaseModel
from dataclasses import dataclass
from typing import TypedDict

# 1. 简单类型
Agent('openai:gpt-4o', output_type=bool)    # True / False
Agent('openai:gpt-4o', output_type=int)     # 整数
Agent('openai:gpt-4o', output_type=str)     # 字符串（默认）

# 2. Pydantic 模型（最常用）
class UserInfo(BaseModel):
    name: str
    age: int

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

# 3. dataclass
@dataclass
class Point:
    x: float
    y: float

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

# 4. TypedDict
class Config(TypedDict):
    debug: bool
    version: str

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

联合类型：AI 自己选返回哪种格式

有时候 AI 需要根据情况返回不同的数据结构。比如 SQL 生成器，成功时返回 SQL，失败时返回错误原因：

from pydantic import BaseModel
from pydantic_ai import Agent

class SQLResult(BaseModel):
    sql_query: str

class ErrorResult(BaseModel):
    error_message: str
    suggestion: str

# 用 list 传入多个输出类型
agent = Agent(
    'openai:gpt-4o',
    output_type=[SQLResult, ErrorResult],
    instructions='将用户的自然语言转换为 SQL 查询。如果无法转换，返回错误信息。'
)

result = agent.run_sync('查询所有年龄大于 18 的用户')
if isinstance(result.output, SQLResult):
    print(f'SQL: {result.output.sql_query}')
else:
    print(f'错误: {result.output.error_message}')
    print(f'建议: {result.output.suggestion}')

输出验证器：对 AI 的回答做质量检查

AI 返回的数据格式对了，但内容可能不对。输出验证器就是在 AI 回答之后、返回给你之前，做一次质量检查：

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext, ModelRetry

class CityInfo(BaseModel):
    city: str
    country: str
    population: int

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

@agent.output_validator
def check_population(ctx: RunContext, output: CityInfo) -> CityInfo:
    # 人口不可能是负数
    if output.population < 0:
        raise ModelRetry('人口数量不能是负数，请重新回答')
    # 人口不可能超过 20 亿（单个城市）
    if output.population > 2_000_000_000:
        raise ModelRetry('单个城市人口不可能超过 20 亿，请检查')
    return output

result = agent.run_sync('上海的信息')
print(result.output)
# 如果 AI 返回的人口数据不合理，它会自动重试

执行流程

三种输出模式

from pydantic_ai import Agent
from pydantic_ai.output import ToolOutput, NativeOutput, PromptedOutput

class MyModel(BaseModel):
    name: str
    value: int

# 1. Tool Output（默认）—— 用工具调用来返回结构化数据
Agent('openai:gpt-4o', output_type=MyModel)
# 等同于
Agent('openai:gpt-4o', output_type=ToolOutput(MyModel))

# 2. Native Output —— 用模型自带的结构化输出功能（更快，但不是所有模型都支持）
Agent('openai:gpt-4o', output_type=NativeOutput(MyModel))

# 3. Prompted Output —— 在提示词里要求 AI 返回 JSON（兼容性最好）
Agent('openai:gpt-4o', output_type=PromptedOutput(MyModel))

4工具进阶

阶段一学了基本的工具注册，现在来看更高级的用法。

工具的 docstring 就是给 AI 的说明书

这一点怎么强调都不过分。框架会从你的 docstring 里提取两样东西：

@agent.tool_plain
def search_flights(
    origin: str,
    destination: str,
    date: str,
    max_price: float = 1000.0,
) -> str:
    """搜索两个城市之间的航班。          ← 这一行变成工具描述

    Args:
        origin: 出发城市，如 "北京"       ← 这些变成参数描述
        destination: 到达城市，如 "上海"
        date: 出发日期，格式 YYYY-MM-DD
        max_price: 最高票价（元），默认 1000
    """
    return f'从 {origin} 到 {destination}，{date} 出发，{max_price} 元以内的航班...'

AI 看到的是这样的：

工具：search_flights
描述：搜索两个城市之间的航班。
参数：
  - origin (string, 必填): 出发城市，如 "北京"
  - destination (string, 必填): 到达城市，如 "上海"
  - date (string, 必填): 出发日期，格式 YYYY-MM-DD
  - max_price (number, 可选, 默认 1000): 最高票价（元），默认 1000

工具重试：AI 用错了让它再来

工具执行可能失败（比如 AI 传了个不存在的用户名）。用 ModelRetry 可以把错误信息反馈给 AI，让它换个方式重试：

from pydantic_ai import Agent, RunContext, ModelRetry

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

@agent.tool_plain(retries=3)  # 最多让 AI 重试 3 次
def get_user_id(username: str) -> int:
    """通过用户名查找用户 ID。

    Args:
        username: 用户的完整用户名（名+姓）
    """
    # 模拟数据库查询
    users = {'张三': 1, '李四': 2, '王五': 3}
    if username not in users:
        # 抛出 ModelRetry，AI 会看到这条消息然后重新调用
        raise ModelRetry(
            f'找不到用户 "{username}"，请提供完整的姓名。'
            f'可用的用户有：{", ".join(users.keys())}'
        )
    return users[username]

重试流程

动态工具：根据条件决定给不给 AI 用

有时候你不想让 AI 在所有情况下都能用某个工具。用 prepare 参数可以动态控制：

from pydantic_ai import Agent, RunContext, ToolDefinition

@dataclass
class UserDeps:
    is_admin: bool

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

# prepare 函数：返回 tool_def 表示可用，返回 None 表示隐藏
async def admin_only(
    ctx: RunContext[UserDeps], tool_def: ToolDefinition
) -> ToolDefinition | None:
    if ctx.deps.is_admin:
        return tool_def   # 管理员可以使用
    return None            # 普通用户看不到这个工具

@agent.tool(prepare=admin_only)
def delete_user(ctx: RunContext[UserDeps], user_id: int) -> str:
    """删除用户（仅管理员可用）。

    Args:
        user_id: 要删除的用户 ID
    """
    return f'用户 {user_id} 已删除'

# 管理员 —— AI 可以调用 delete_user
result = agent.run_sync('删除用户 42', deps=UserDeps(is_admin=True))

# 普通用户 —— AI 根本不知道有这个工具
result = agent.run_sync('删除用户 42', deps=UserDeps(is_admin=False))

第三种注册方式：tools 参数

除了装饰器，你还可以把工具函数直接传给 Agent：

from pydantic_ai import Agent, Tool

def my_tool(query: str) -> str:
    """搜索知识库。

    Args:
        query: 搜索关键词
    """
    return f'搜索结果：{query}'

# 直接传函数（自动判断不需要 context）
agent = Agent('openai:gpt-4o', tools=[my_tool])

# 或者用 Tool 对象获得更多控制
agent = Agent('openai:gpt-4o', tools=[
    Tool(my_tool, takes_ctx=False, retries=2)
])

5消息历史：让 AI 记住之前说了什么

为什么需要消息历史？

默认情况下，每次调用 agent.run_sync() 都是一次全新的对话，AI 不记得之前说过什么。

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

result1 = agent.run_sync('我叫张三')
result2 = agent.run_sync('我叫什么名字？')
print(result2.output)
# AI 会说 "我不知道你的名字" —— 因为它忘了！

传递消息历史实现多轮对话

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

# 第一轮
result1 = agent.run_sync('我叫张三，今年 25 岁')
print(result1.output)

# 第二轮 —— 把第一轮的消息传过去
result2 = agent.run_sync(
    '我叫什么名字？今年多大？',
    message_history=result1.all_messages()
)
print(result2.output)
# AI 会说 "你叫张三，今年 25 岁" —— 它记住了！

all_messages() vs new_messages()

result1 = agent.run_sync('你好')
result2 = agent.run_sync('再见', message_history=result1.all_messages())

# all_messages() —— 包含所有历史 + 本次对话
print(len(result2.all_messages()))    # 4 条（两轮各 2 条）

# new_messages() —— 只有本次对话的消息
print(len(result2.new_messages()))    # 2 条（本次的请求 + 回复）

消息持久化：保存和恢复对话

对话不能只在内存里，你得能存到数据库、文件里，下次接着聊：

from pydantic_core import to_json
from pydantic_ai import Agent, ModelMessagesTypeAdapter

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

# 第一轮对话
result = agent.run_sync('帮我记住：明天下午 3 点开会')
messages = result.all_messages()

# ===== 保存到文件 =====
json_bytes = to_json(messages)
with open('chat_history.json', 'wb') as f:
    f.write(json_bytes)

# ===== 从文件恢复 =====
with open('chat_history.json', 'rb') as f:
    json_bytes = f.read()

restored = ModelMessagesTypeAdapter.validate_json(json_bytes)

# 接着聊
result2 = agent.run_sync('我让你记住什么了？', message_history=restored)
print(result2.output)  # AI 会说 "明天下午 3 点开会"

消息处理器：控制历史消息的长度

对话太长了怎么办？token 会爆，费用会涨。用消息处理器来裁剪：

from pydantic_ai import Agent, ModelMessage

# 只保留最近 N 条消息
def keep_last_n(messages: list[ModelMessage]) -> list[ModelMessage]:
    return messages[-10:]  # 只保留最近 10 条

agent = Agent(
    'openai:gpt-4o',
    history_processors=[keep_last_n]
)

# 每次运行前，框架会自动用处理器裁剪历史消息

6重试机制：AI 出错了怎么办

两种重试

Pydantic AI 有两层重试机制：

1. 输出格式重试（自动的）

当 AI 返回的数据不符合你定义的 output_type 时，框架会自动把验证错误发给 AI，让它修正：

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class Score(BaseModel):
    value: int = Field(ge=0, le=100)  # 必须在 0-100 之间

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

# 如果 AI 返回 value=150，Pydantic 验证失败
# 框架自动告诉 AI："value 必须 <= 100，请修正"
# AI 重新回答 value=95 ✔

2. 工具重试（你控制的）

在工具函数里抛出 ModelRetry，框架会把你的错误消息发给 AI：

from pydantic_ai import Agent, ModelRetry

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

@agent.tool_plain(retries=2)
def divide(a: float, b: float) -> float:
    """两个数相除。

    Args:
        a: 被除数
        b: 除数
    """
    if b == 0:
        raise ModelRetry('除数不能为 0，请换一个数')
    return a / b

控制重试次数

# 全局设置：所有工具默认最多重试 3 次
agent = Agent('openai:gpt-4o', retries=3)

# 单个工具覆盖：这个工具最多重试 5 次
@agent.tool_plain(retries=5)
def picky_tool(x: int) -> str:
    """一个很挑剔的工具。"""
    if x < 0:
        raise ModelRetry('x 必须是正数')
    return str(x)

实战示例：SQL 验证重试

这是一个综合了结构化输出、输出验证器和重试的完整例子：

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext, ModelRetry

class SQLQuery(BaseModel):
    query: str
    explanation: str

agent = Agent(
    'openai:gpt-4o',
    output_type=SQLQuery,
    retries=3,
    instructions=(
        '你是一个 SQL 专家。根据用户的自然语言描述生成 PostgreSQL 查询。'
        '只生成 SELECT 查询，不允许 INSERT、UPDATE、DELETE。'
    ),
)

@agent.output_validator
def validate_sql(ctx: RunContext, output: SQLQuery) -> SQLQuery:
    # 检查是不是 SELECT 语句
    query_upper = output.query.strip().upper()
    if not query_upper.startswith('SELECT'):
        raise ModelRetry('只允许 SELECT 查询，请修改你的 SQL')

    # 检查有没有危险操作
    dangerous = ['DROP', 'DELETE', 'INSERT', 'UPDATE', 'ALTER', 'TRUNCATE']
    for keyword in dangerous:
        if keyword in query_upper:
            raise ModelRetry(f'SQL 中包含危险关键字 {keyword}，请移除')

    return output

result = agent.run_sync('查询所有 2024 年注册的用户，按注册日期排序')
print(result.output.query)
print(result.output.explanation)