📦 Pydantic Ai Common Pitfalls — PydanticAI 避坑指南

工具装饰器错误

错误：在 tool_plain 中使用 RunContext

# 错误：tool_plain 中不允许使用 RunContext
@agent.tool_plain
async def bad_tool(ctx: RunContext[MyDeps]) -> str:
    return "oops"
# UserError: RunContext annotations can only be used with tools that take context

修复：如果需要上下文，请使用 @agent.tool：

@agent.tool
async def good_tool(ctx: RunContext[MyDeps]) -> str:
    return "works"

错误：tool 中缺少 RunContext

# 错误：第一个参数必须是 RunContext
@agent.tool
def bad_tool(user_id: int) -> str:
    return "oops"
# UserError: First parameter of tools that take context must be annotated with RunContext[...]

修复：添加 RunContext 作为第一个参数：

@agent.tool
def good_tool(ctx: RunContext[MyDeps], user_id: int) -> str:
    return "works"

错误：RunContext 不是第一个参数

# 错误：RunContext 必须是第一个参数
@agent.tool
def bad_tool(user_id: int, ctx: RunContext[MyDeps]) -> str:
    return "oops"

修复：RunContext 必须始终是第一个参数。

有效模式（不是错误）

原始函数工具注册

以下模式是有效的，并且受 pydantic-ai 支持：

from pydantic_ai import Agent, RunContext
async def search_db(ctx: RunContext[MyDeps], query: str) -> list[dict]:
    """Search the database."""
    return await ctx.deps.db.search(query)
async def get_user(ctx: RunContext[MyDeps], user_id: int) -> dict:
    """Get user by ID."""
    return await ctx.deps.db.get_user(user_id)
# 有效：将原始函数传递给 Agent(tools=[...])
agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
    tools=[search_db, get_user]  # RunContext 从签名中检测
)

为什么这样有效： PydanticAI 检查函数签名。如果第一个参数是 RunContext[T]，它被视为上下文感知工具。不需要装饰器。

参考： https://ai.pydantic.dev/agents/#registering-tools-via-the-tools-argument

不要标记 将带有 RunContext 签名的函数传递给 Agent(tools=[...]) 的代码。这等同于使用 @agent.tool，并且有明确文档说明。

依赖类型不匹配

错误：运行时缺少 deps

agent = Agent('openai:gpt-4o', deps_type=MyDeps)
# 错误：需要 deps 但未提供
result = agent.run_sync('Hello')  # 缺少 deps！

修复：设置 deps_type 时始终提供 deps：

result = agent.run_sync('Hello', deps=MyDeps(...))

错误的 deps 类型

@dataclass
class AppDeps:
    db: Database
@dataclass
class WrongDeps:
    api: ApiClient
agent = Agent('openai:gpt-4o', deps_type=AppDeps)
# 类型错误：WrongDeps != AppDeps
result = agent.run_sync('Hello', deps=WrongDeps(...))

输出类型问题

Pydantic 验证失败

class Response(BaseModel):
    count: int
    items: list[str]
agent = Agent('openai:gpt-4o', output_type=Response)
result = agent.run_sync('List items')
# 如果 LLM 返回错误结构可能会失败

修复：增加重试次数或改进提示：

agent = Agent(
    'openai:gpt-4o',
    output_type=Response,
    retries=3,  # 更多尝试
    instructions='Return JSON with count (int) and items (list of strings).'
)

复杂的嵌套类型

# 可能会导致某些模型的 schema 问题
class Complex(BaseModel):
    nested: dict[str, list[tuple[int, str]]]

修复：简化或使用中间模型：

class Item(BaseModel):
    id: int
    name: str
class Simple(BaseModel):
    items: list[Item]

Async vs Sync 错误

错误：在同步上下文中调用 async

# 错误：无法在同步函数中 await
def handler():
    result = await agent.run('Hello')  # SyntaxError!

修复：使用 run_sync 或使 handler 变为 async：

def handler():
    result = agent.run_sync('Hello')
# 或
async def handler():
    result = await agent.run('Hello')

错误：在 async 工具中阻塞

@agent.tool
async def slow_tool(ctx: RunContext[Deps]) -> str:
    time.sleep(5)  # 错误：阻塞事件循环！
    return "done"

修复：使用 async I/O：

@agent.tool
async def slow_tool(ctx: RunContext[Deps]) -> str:
    await asyncio.sleep(5)  # 正确
    return "done"

模型配置错误

缺少 API 密钥

# 错误：未设置 OPENAI_API_KEY
agent = Agent('openai:gpt-4o')
result = agent.run_sync('Hello')
# ModelAPIError: Authentication failed

修复：设置环境变量或使用 defer_model_check：

# 用于测试
agent = Agent('openai:gpt-4o', defer_model_check=True)
with agent.override(model=TestModel()):
    result = agent.run_sync('Hello')

无效的模型字符串

# 错误：未知提供者
agent = Agent('unknown:model')
# ValueError: Unknown model provider

修复：使用有效的 provider:model 格式。

流式传输问题

错误：在流完成前使用 result

async with agent.run_stream('Hello') as response:
    # 不要在流完成前访问 .output
    print(response.output)  # 可能不完整！
    # 正确：在上下文管理器后访问
    print(response.output)  # 完整结果

错误：不迭代流

async with agent.run_stream('Hello') as response:
    pass  # 从未使用！
    # 流从未被读取 - 输出可能不完整

修复：始终消费流：

async with agent.run_stream('Hello') as response:
    async for chunk in response.stream_output():
        print(chunk, end='')

工具返回问题

错误：返回不可序列化对象

@agent.tool_plain
def bad_return() -> object:
    return CustomObject()  # 无法序列化！

修复：返回可序列化类型（str、dict、Pydantic 模型）：

@agent.tool_plain
def good_return() -> dict:
    return {"key": "value"}

调试技巧

启用追踪

import logfire
logfire.configure()
logfire.instrument_pydantic_ai()
# 或按 agent
agent = Agent('openai:gpt-4o', instrument=True)

捕获消息

from pydantic_ai import capture_run_messages
with capture_run_messages() as messages:
    result = agent.run_sync('Hello')
    for msg in messages:
        print(type(msg).__name__, msg)

检查模型响应

result = agent.run_sync('Hello')
print(result.all_messages())  # 完整消息历史
print(result.response)        # 最后模型响应
print(result.usage())         # token 使用量

常见错误消息