测试你的智能体#
用确定性的模型测试运行时行为，再用少量按需启用的在线评估测试模型质量。它们回答的是不同的问题：“我的工具、路由和清理是否正常工作？”不需要网络请求；“这个模型能否可靠地解决任务？”则需要。
把工具当作 Python 操作来测试#
装饰器会返回一个 AgentTool，因此可以用参数映射来调用它。这样无需让模型选择它，就能检验校验和处理函数：
import pytest
from pydantic import ValidationError
from zett_agent.tools.base import tool
@tool(guidelines="Use for exact integer addition.")
def add(left: int, right: int) -> int:
"""Add two integers exactly."""
return left + right
@pytest.mark.asyncio
async def test_add() -> None:
assert await add({"left": 20, "right": 22}) == 42
with pytest.raises(ValidationError):
await add({"left": "not-an-integer", "right": 22})
在测试环境中安装 pytest 和 pytest-asyncio。直接调用工具会向外传播异常；而在智能体运行内部，普通的工具异常会变成失败的工具结果。对重要的集成，请同时测试这两种边界情况。
用脚本预设模型的决策#
实现公开的 AgentModel 协议，就能在测试中精确决定模型返回什么。这里没有专门的 mock 智能体 API：真实的运行时仍会恢复历史记录、校验工具、分发事件并进行清理。
class MathModel:
"""Script a model/tool/model round trip without making a network request."""
retry = RetryOptions(max_retries=0)
async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
if isinstance(request.messages[-1], ToolMessage):
answer = f"Answer: {request.messages[-1].content}"
yield ModelEvent.text(answer)
yield ModelEvent.completed(ModelResponse(AssistantMessage(content=answer)))
else:
assert any(definition.name == "add" for definition in request.tools)
yield ModelEvent.reasoning("Use exact arithmetic.")
yield ModelEvent.completed(
ModelResponse(AssistantMessage(tool_calls=(ToolCall("sum-1", "add", {"left": 20, "right": 22}),)))
)
这个适配器先请求 add，然后使用返回的工具结果作答。它的导入、工具、分发器和断言都在 streaming_tools.py 中。成功的适配器必须返回一个包含完整消息的终止 ModelResponse。仅有文本增量并不构成最终答案。关于流的约束条件，参见 实现模型适配器。
断言行为，而非偶然的时序#
对于工具调用工作流，请验证请求中包含预期的工具、处理函数收到正确的参数、模型收到其结果，以及客户端返回预期的答案。对于并发工具，应通过工具调用 ID 进行关联，而不是假设完成顺序。
扩展同一个测试，以覆盖以下情况：
无效参数，以及模型能够从中恢复的处理函数错误。
两个相互隔离的会话，以及重新打开存储后恢复的历史记录。
流的提前关闭、请求超时，以及请求本地的清理。
被拒绝或重复的审批响应，而不只是审批成功。
预算耗尽，以及提供商流在部分输出之后失败。
使用事件、队列或显式的测试屏障来同步并发。避免依赖时序的 sleep。许多这类边界都在 示例 的示例中包含断言。
在没有真实端点的情况下测试真实适配器#
注入 httpx.MockTransport，即可测试内置提供商的请求编码和流式响应处理：
"""Run a real provider adapter against an in-memory HTTP transport."""
import asyncio
import json
import httpx
from zett_agent.client import create_agent
from zett_agent.model import ReasoningEffort, RetryOptions
from zett_agent.providers.openai import OpenAIProvider
def respond(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/v1/chat/completions"
payload = json.loads(request.content)
assert payload["messages"][-1]["content"] == "Hello"
assert payload["stream"] is True
chunks = [
{"choices": [{"delta": {"content": "Hello from a mock endpoint."}}]},
{"choices": [{"delta": {}, "finish_reason": "stop"}]},
]
body = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks)
return httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=body + "data: [DONE]\n\n",
)
async def main() -> None:
model = OpenAIProvider(
model="fixture-model",
api_key="offline-test-key",
transport=httpx.MockTransport(respond),
retry=RetryOptions(max_retries=0),
)
try:
client = await create_agent(model, reasoning_effort=ReasoningEffort.OFF)
reply = await client.run("Hello")
assert reply.content == "Hello from a mock endpoint."
print(f"Mock provider reply: {reply.content}")
finally:
await model.aclose()
if __name__ == "__main__":
asyncio.run(main())
这个程序使用真实的 OpenAI 适配器，搭配本地预设的 SSE 响应。它会断言请求路径和用户消息，因此不需要凭据或网络访问。在检出目录中运行 uv run python docs/_examples/provider_mock.py：
Mock provider reply: Hello from a mock endpoint.
下载 provider_mock.py。其他提供商暴露相同的 transport 注入点，但它们的传输格式不同。通过的 mock 测试只能证明你的协议夹具可用，而不能证明线上端点支持所请求的每一项模型特性。
让测试安全且可复现#
文件和 SQLite 数据库请使用临时目录。绝不要让测试构造默认的用户数据存储，或发现个人的 MCP/技能文件。请传入显式的路径、显式的服务器列表和显式的技能根目录。离线测试中不要出现凭据，并单独选择启用在线调用。
可下载的文档程序由 make docs-examples 在禁用网络连接并使用临时工作目录的情况下运行。make check 还会额外检查格式、运行时覆盖率门槛，以及一次无警告的隔离文档构建。