测试 PydanticAI 代理

TestModel（确定性测试）

使用 TestModel 进行无需 API 调用的测试：

python
import pytest
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

def testagentbasic():
agent = Agent(openai:gpt-4o)

# 使用 TestModel 覆盖进行测试
result = agent.run_sync(Hello, model=TestModel())

# TestModel 基于 output_type 生成确定性输出
assert isinstance(result.output, str)

TestModel 配置

python
from pydantic_ai.models.test import TestModel

自定义文本输出

自定义结构化输出（用于 output_type 代理）

class Response(BaseModel):
message: str
score: int

agent = Agent(openai:gpt-4o, output_type=Response)
model = TestModel(customoutputargs={message: 测试, score: 42})
result = agent.run_sync(Hello, model=model)
assert result.output.message == 测试

用于可重现随机输出的种子

强制工具调用

覆盖上下文管理器

python
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

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

def testwithoverride():
mock_deps = MyDeps(db=MockDB())

with agent.override(model=TestModel(), deps=mock_deps):
# 所有运行都使用 TestModel 和 mock_deps
result = agent.run_sync(Hello)
assert result.output

FunctionModel（自定义逻辑）

完全控制模型响应：

python
from pydantic_ai import Agent, ModelMessage, ModelResponse, TextPart
from pydantic_ai.models.function import AgentInfo, FunctionModel

def custom_model(
messages: list[ModelMessage],
info: AgentInfo
) -> ModelResponse:
检查消息并返回响应的自定义模型。
# 访问最后一条用户消息
last_msg = messages[-1]

# 返回自定义响应
return ModelResponse(parts=[TextPart(自定义响应)])

agent = Agent(FunctionModel(custom_model))
result = agent.run_sync(Hello)

带工具调用的 FunctionModel

python
from pydantic_ai import ToolCallPart, ModelResponse
from pydantic_ai.models.function import AgentInfo, FunctionModel

def modelwithtools(
messages: list[ModelMessage],
info: AgentInfo
) -> ModelResponse:
# 第一次请求：调用工具
if len(messages) == 1:
return ModelResponse(parts=[
ToolCallPart(
toolname=getdata,
args={id: 123}
)
])

# 工具响应后：返回最终结果
return ModelResponse(parts=[TextPart(使用工具结果完成)])

agent = Agent(FunctionModel(modelwithtools))

@agent.tool_plain
def get_data(id: int) -> str:
return fID {id} 的数据

result = agent.run_sync(获取数据)

VCR 磁带（录制的 API 调用）

录制和回放真实的 LLM API 交互：

python
import pytest

@pytest.mark.vcr
def testwithrecorded_response():
使用 tests/cassettes/ 中的录制磁带
agent = Agent(openai:gpt-4o)
result = agent.run_sync(Hello)
assert hello in result.output.lower()

录制/更新磁带：

uv run pytest --record-mode=rewrite tests/test_file.py

磁带文件以 YAML 格式存储在 tests/cassettes/ 中。

内联快照

使用自动更新的快照断言预期输出：

python
from inline_snapshot import snapshot

def testagentoutput():
result = agent.run_sync(Hello, model=TestModel())

# 首次运行：创建快照
# 后续运行：与快照进行断言
assert result.output == snapshot(此处为预期输出)

更新快照：

uv run pytest --inline-snapshot=fix

测试工具

python
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.test import TestModel

def testtoolis_called():
agent = Agent(openai:gpt-4o)
tool_called = False

@agent.tool_plain
def my_tool(x: int) -> str:
nonlocal tool_called
tool_called = True
return f结果: {x}

# 强制 TestModel 调用工具
result = agent.run_sync(
使用 my_tool,
model=TestModel(calltools=[mytool])
)

assert tool_called

带依赖项的测试

python
from dataclasses import dataclass
from unittest.mock import AsyncMock

@dataclass
class Deps:
api: ApiClient

def testtoolwith_deps():
# 创建模拟依赖
mock_api = AsyncMock()
mockapi.fetch.returnvalue = {data: 测试}

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

@agent.tool
async def fetch_data(ctx: RunContext[Deps]) -> dict:
return await ctx.deps.api.fetch()

with agent.override(
model=TestModel(calltools=[fetchdata]),
deps=Deps(api=mock_api)
):
result = agent.run_sync(获取数据)

mockapi.fetch.assertcalled_once()

捕获消息

检查运行中的所有消息：

python
from pydanticai import Agent, capturerun_messages

agent = Agent(openai:gpt-4o)

with capturerunmessages() as messages:
result = agent.run_sync(Hello, model=TestModel())

检查捕获的消息

测试模式总结

pytest 配置

典型的 pyproject.toml：

toml
[tool.pytest.ini_options]
testpaths = [tests]
asyncio_mode = auto # 用于异步测试

运行测试：
bash
uv run pytest tests/test_agent.py -v
uv run pytest --inline-snapshot=fix # 更新快照