Agent 基础 · system prompt 与 provider 切换

一、先把构造函数拆开看

Agent 是 Pydantic AI 的唯一主角。你和它打交道的方式就两步——构造(Agent(...)),然后调用(run / run_sync / run_stream)。构造函数吃的参数不算少,但每个都有明确职责:

from pydantic_ai import Agent

agent = Agent(
    model="openai:gpt-4o",           # 1. 绑定哪个模型
    output_type=str,                   # 2. 输出类型:str / BaseModel / TypedDict / Union
    system_prompt="你是一位...",       # 3. 静态系统提示
    deps_type=None,                    # 4. 依赖注入类型(第 5 章专讲)
    tools=[],                          # 5. 初始工具列表(也可以用 @agent.tool)
    name="assistant",                 # 6. 可读名字(trace/日志里用)
    model_settings={"temperature": 0}, # 7. 模型参数
    retries=2,                         # 8. 校验失败最多重试几次
    instrument=True,                   # 9. 是否开 Logfire 自动追踪
)

二、System Prompt:静态 vs 动态

静态写法:构造时直接传

agent = Agent(
    "openai:gpt-4o",
    system_prompt="你是一位 Python 技术答疑助手。",
)

想要多段拼接?传元组就行,Pydantic AI 会自动用空行拼接成一段:

agent = Agent(
    "openai:gpt-4o",
    system_prompt=(
        "你是一位 Python 技术答疑助手。",
        "回答不超过 3 句话。",
        "不确定的部分必须明说'我不确定',不要编造。",
    ),
)

动态写法:调用时现算

需要把当前时间、当前用户、当前 A/B 分桶塞进 system prompt?用 @agent.system_prompt 装饰器,每次 run 时它会被调用一次,结果作为 system prompt 的一部分:

from datetime import datetime
from pydantic_ai import Agent, RunContext
from dataclasses import dataclass

@dataclass
class UserDeps:
    name: str
    vip_level: int

agent = Agent(
    "openai:gpt-4o",
    deps_type=UserDeps,
    system_prompt="你是一位电商客服。",   # 静态部分
)

@agent.system_prompt
def add_user_context(ctx: RunContext[UserDeps]) -> str:
    return f"当前用户: {ctx.deps.name}, VIP 等级: {ctx.deps.vip_level}。"

@agent.system_prompt
def add_time() -> str:
    return f"当前北京时间: {datetime.now():%Y-%m-%d %H:%M}。"

result = agent.run_sync("我的 VIP 折扣还剩多少?", deps=UserDeps(name="张三", vip_level=3))

最终 LLM 看到的 system prompt 是三段拼接:

你是一位电商客服。

当前用户: 张三, VIP 等级: 3。

当前北京时间: 2026-05-07 14:22。

动态 prompt 函数签名的玩法

• 不需要 context:写 def fn() -> str
• 需要 context:写 def fn(ctx: RunContext[DepsType]) -> str
• 异步也可以:async def fn(...) -> str,Agent 会自动 await
• 可以注册任意多个,按注册顺序拼接

三、Provider 切换:一根字符串走天下

Pydantic AI 目前第一方支持的 provider(截至教程编写时):

最常见切换方式:换字符串

agent = Agent("openai:gpt-4o")
agent = Agent("anthropic:claude-sonnet-4-6")
agent = Agent("groq:llama-3.3-70b")
agent = Agent("ollama:qwen2.5")

需要自定义 base_url / API key 时:传 Model 实例

from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider

# 例:接入 DeepSeek(OpenAI 兼容接口)
model = OpenAIModel(
    "deepseek-chat",
    provider=OpenAIProvider(
        base_url="https://api.deepseek.com/v1",
        api_key="sk-...",
    ),
)
agent = Agent(model)

这条思路可以接入 所有 OpenAI 兼容网关——DeepSeek、Moonshot、智谱、OpenRouter、LiteLLM Proxy、vLLM。这是最大的自由度来源。

Ollama 本地跑

本地跑不花钱,开发调试首选:

# 先起 Ollama
ollama serve
ollama pull qwen2.5

agent = Agent("ollama:qwen2.5")

注意:小模型(7B / 14B)的 function calling 和 structured output 能力良莠不齐——调试逻辑可以用,生产还是得上大模型或服务化 provider。

运行时动态切换模型

每次 run 都可以覆盖构造时的 model:

agent = Agent("openai:gpt-4o")  # 默认

# 这一次临时换 Claude
result = agent.run_sync("...", model="anthropic:claude-sonnet-4-6")

用途:AB 测试、降级路由、只在重要场景用高级模型。

四、三条 API:run_sync · run · run_stream

run_sync:同步版,最简单

result = agent.run_sync("...")
print(result.output)

阻塞当前线程,适合脚本、CLI、Jupyter。内部其实是启动事件循环跑一遍 run()。

run:异步版,生产首选

import asyncio

async def main():
    result = await agent.run("...")
    print(result.output)

asyncio.run(main())

在 FastAPI / Starlette / Aiohttp 里直接 await agent.run(...)。这是生产代码的唯一正确选择——LLM 调用是 IO 密集,阻塞的代价太高。

run_stream:流式

async with agent.run_stream("写一首关于 Python GIL 的短诗") as response:
    async for chunk in response.stream_text(delta=True):
        print(chunk, end="", flush=True)
    final = await response.get_output()

• stream_text() 默认增量模式(delta=True),和 OpenAI SDK 的流式体验一致
• stream_structured() 用于结构化流(第 6 章详讲)
• 必须用 async with 上下文:离开块时底层连接会被释放
• response.get_output() 会等待全部流完,然后校验成 output_type

run_stream 的常见用途

• 前端 SSE 返回给浏览器,用户看到打字机效果
• 长回答的感知延迟降低(首字节时间 < 1s)
• 提前取消——用户看一半不满意可以中断

五、model_settings:温度、超时、并行工具全在这

model_settings 是一个 TypedDict,字段全部可选。Pydantic AI 会把它适配到各家 SDK:

agent = Agent(
    "openai:gpt-4o",
    model_settings={
        "temperature": 0.2,            # 输出确定性
        "max_tokens": 1024,            # 生成上限
        "top_p": 0.95,
        "timeout": 30.0,               # 秒
        "parallel_tool_calls": True,   # 允许一次调多工具
        "presence_penalty": 0.0,
        "frequency_penalty": 0.0,
        "seed": 42,                     # 可重现
    },
)

调用时也能覆盖:

result = await agent.run(
    "...",
    model_settings={"temperature": 0.9},  # 这次调用临时放宽温度
)

六、retries:校验失败怎么办

Agent 的 retries 参数和网络重试无关。它控制的是:

• LLM 输出不满足 output_type(Pydantic 校验失败)
• LLM 调工具时给了错参数(工具参数校验失败)
• 你在工具里主动抛 ModelRetry("理由")(第 4 章讲)

此时 Pydantic AI 不直接抛——它把错误原文作为 ToolReturn 发回去,让模型看着错误再改一次。retries=2 意味着最多给模型改两次机会。

from pydantic import BaseModel

class Answer(BaseModel):
    answer: int

agent = Agent("openai:gpt-4o-mini", output_type=Answer, retries=3)
result = agent.run_sync("二乘三等于几?把数字用中文回答")
# 模型第一次可能返回 "六",Pydantic 校验失败(不是 int)
# Pydantic AI 把"int_parsing: Input should be a valid integer"发回去
# 模型第二次大概率会修正为 6

网络重试是另一回事——那是 HTTP 层的事,由底层 SDK(openai/anthropic/httpx)的重试中间件负责。

七、result 里都装了什么

所有 run 返回的都是 AgentRunResult(流式是 StreamedRunResult)。常用属性:

result = agent.run_sync("...")

result.output                  # 最终输出,类型 = output_type
result.all_messages()          # 本次 run 完整消息列表(含 system/user/assistant/tool)
result.new_messages()          # 本次 run 新增的消息(不含 message_history 传入的)
result.usage()                 # Usage 对象:input_tokens / output_tokens / total_tokens
str(result.usage())         # '{"requests":1,"request_tokens":87,"response_tokens":42,...}'

result.all_messages() 和 new_messages() 的返回值可以直接作为下一次 run 的 message_history,实现多轮对话——第 6 章细讲。

八、完整示例:一个简易"模型选择器"

把本章所有要点串起来——按任务难度选模型,带动态 prompt 和超时保护:

import asyncio
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class TaskDeps:
    task_type: str      # "code" | "math" | "chat"
    user_tier: str      # "free" | "pro"

agent = Agent(
    "openai:gpt-4o-mini",
    deps_type=TaskDeps,
    system_prompt="你是一位专业的 AI 助手。",
    model_settings={"temperature": 0.3, "timeout": 20.0},
    retries=2,
)

@agent.system_prompt
def tune_by_task(ctx: RunContext[TaskDeps]) -> str:
    if ctx.deps.task_type == "code":
        return "注意:代码要能直接运行,错误要指出。"
    if ctx.deps.task_type == "math":
        return "注意:数学推导要一步一步来。"
    return "自然对话即可。"

async def ask(question: str, deps: TaskDeps):
    # Pro 用户用 gpt-4o,Free 用户用 mini(便宜)
    model = "openai:gpt-4o" if deps.user_tier == "pro" else "openai:gpt-4o-mini"
    result = await agent.run(question, deps=deps, model=model)
    print("回答:", result.output)
    print("用量:", result.usage())

asyncio.run(ask("写个 Python 的快排", TaskDeps(task_type="code", user_tier="pro")))

九、八个常见坑

1. 忘记设 API Key:不是 Pydantic AI 的报错,是底层 SDK 的 401。先 echo $OPENAI_API_KEY 确认。
2. Ollama 本地模型不支持结构化输出:很多 7B 模型的 function calling 不靠谱,结构化需求请换到服务化模型。
3. system_prompt 传了 list(不是 tuple/str):只接受 str | Sequence[str],传 ["a", "b"] 也行,但传字典会报错。
4. 把慢查询写在动态 system_prompt 里:每次 run 都会执行,很容易变成性能瓶颈。用缓存或 deps 预查。
5. 混用 run_sync 和 async 场景:已经在 async 函数里还 run_sync() 会报 "cannot be called from a running event loop"。统一用 await agent.run(...)。
6. run_stream 没 async with:底层连接不会释放,并发跑几次就 leak。务必用 async with。
7. retries 跟超时混淆:超时是 model_settings["timeout"],重试是 retries,两者独立。
8. 切换 provider 后以为行为完全一致:temperature=0 在 OpenAI 和 Claude 的"确定性程度"体感不同;tool calling 的 schema 严格度也有差别。生产前跑一遍 eval(第 9 章)。

十、本章小结