迭代式代理测试,具有会话恢复功能。执行、分析、修复、从检查点恢复。适用于测试新构建的代理、迭代调试失败的代理或在不从头开始的情况下验证修复。
代理测试
迭代测试代理:执行、分析失败、修复、从检查点恢复,重复。
使用场景
- 针对其目标测试新构建的代理
- 迭代调试失败的代理
- 在不重新运行昂贵的早期节点的情况下验证修复
- 在部署前运行最终回归测试
前提条件
- 代理包在
exports/{agent_name}/(用/hive-create构建) - 配置了凭据(
/hive-credentials) - 设置了
ANTHROPIC_API_KEY(或适当的 LLM 提供商密钥)
路径区分(关键 —— 不要混淆这些):
exports/{agent_name}/— 代理源代码(在此编辑)~/.hive/agents/{agent_name}/— 运行时数据:会话、检查点、日志(在此读取)
迭代测试循环
这是核心工作流程。当一个晚期节点失败时,不要重新运行整个代理 —— 分析、修复,并从最后一个干净的检查点恢复。
┌──────────────────────────────────────┐
│ 第1阶段:生成测试场景 │
│ 目标 → 合成测试输入 + 测试 │
└──────────────┬───────────────────────┘
↓
┌──────────────────────────────────────┐
│ 第2阶段:执行 │◄────────────────┐
│ 运行代理(CLI 或 pytest) │ │
└──────────────┼───────────────────────┘ │
↓ │
通过? ──是──► 第6阶段:最终验证 │
│ │
不 │
↓ │
┌──────────────────────────────────────┐ │
│ 第3阶段:分析 │ │
│ 会话 + 运行时日志 + 检查点 │ │
└──────────────┼───────────────────────┘ │
↓ │
┌──────────────────────────────────────┐ │
│ 第4阶段:修复 │ │
│ 提示 / 代码 / 图 / 目标 │ │
└──────────────┼───────────────────────┘ │
↓ │
┌──────────────────────────────────────┐ │
│ 第5阶段:恢复 & 恢复 │─────────────────┘
│ 检查点恢复 OR 新鲜重运行 │
└──────────────────────────────────────┘
第1阶段:生成测试场景
根据代理的目标、约束和成功标准创建合成测试。
第1a步:阅读目标
# 从 agent.py 读取目标
Read(file_path="exports/{agent_name}/agent.py")
# 提取目标定义并转换为 JSON 字符串
第1b步:获取测试指南
# 获取约束测试指南
generate_constraint_tests(
goal_id="your-goal-id",
goal_json='{"id": "...", "constraints": [...]}',
agent_path="exports/{agent_name}"
)
# 获取成功标准测试指南
generate_success_tests(
goal_id="your-goal-id",
goal_json='{"id": "...", "success_criteria": [...]}',
node_names="intake,research,review,report",
tool_names="web_search,web_scrape",
agent_path="exports/{agent_name}"
)
这些返回 file_header, test_template, constraints_formatted/success_criteria_formatted, 和 test_guidelines。它们不生成测试代码 —— 你编写测试。
第1c步:编写测试
Write(
file_path=result["output_file"],
content=result["file_header"] + "
" + your_test_code
)
测试编写规则
- 每个测试必须是
async并接受@pytest.mark.asyncio - 每个测试必须接受
runner, auto_responder, mock_mode固定装置 - 使用
await auto_responder.start()在运行前,await auto_responder.stop()在finally中 - 使用
await runner.run(input_dict)—— 这通过 AgentRunner → AgentRuntime → ExecutionStream - 通过
result.output.get("key")访问输出 —— 永远不要result.output["key"] result.success=True意味着没有异常,而不是目标实现 —— 总是检查输出- 总共编写8-15个测试,而不是30+
- 每个真实测试成本约3秒 + LLM 令牌
- 永远不要使用
default_agent.run()—— 它绕过运行时(没有会话,没有日志,客户端面向节点挂起)
第1d步:检查现有测试
在生成之前,检查是否已经有测试:
list_tests(
goal_id="your-goal-id",
agent_path="exports/{agent_name}"
)
第2阶段:执行
两条执行路径,根据你的情况使用正确的一条。
迭代调试(针对复杂代理)
通过 CLI 运行代理。这会在 ~/.hive/agents/{agent_name}/sessions/ 创建带有检查点的会话:
uv run hive run exports/{agent_name} --input '{"query": "test topic"}'
会话和检查点会自动保存。
客户端面向节点:具有 client_facing=True 节点(交互式对话)的代理在真实终端中以无头模式运行 —— 代理将输出流式传输到 stdout,并通过 >>> 提示从 stdin 读取用户输入。在非交互式 shell(如 Claude Code 的 Bash 工具)中,客户端面向节点会挂起,因为没有 stdin。对于从 Claude Code 测试交互式代理,使用 run_tests 与模拟模式或让用户在他们的终端手动运行代理。
自动化回归(针对 CI 或最终验证)
使用 run_tests MCP 工具运行所有 pytest 测试:
run_tests(
goal_id="your-goal-id",
agent_path="exports/{agent_name}"
)
返回结构化结果:
{
"overall_passed": false,
"summary": {"total": 12, "passed": 10, "failed": 2, "pass_rate": "83.3%"},
"test_results": [{"test_name": "test_success_source_diversity", "status": "failed"}],
"failures": [{"test_name": "test_success_source_diversity", "details": "..."}]
}
选项:
# 只运行约束测试
run_tests(goal_id, agent_path, test_types='["constraint"]')
# 遇到第一个失败就停止
run_tests(goal_id, agent_path, fail_fast=True)
# 并行执行
run_tests(goal_id, agent_path, parallel=4)
注意: run_tests 使用 AgentRunner 与 tmp_path 存储,因此会话是每个测试运行隔离的。对于基于检查点的恢复与持久会话,使用 CLI 执行。对于快速回归检查和最终验证,使用 run_tests。
第3阶段:分析失败
当测试失败时,系统地深入研究。不要猜测 —— 使用工具。
第3a步:获取错误类别
debug_test(
goal_id="your-goal-id",
test_name="test_success_source_diversity",
agent_path="exports/{agent_name}"
)
返回错误类别(IMPLEMENTATION_ERROR, ASSERTION_FAILURE, TIMEOUT, IMPORT_ERROR, API_ERROR)加上完整的 traceback 和建议。
第3b步:找到失败的会话
list_agent_sessions(
agent_work_dir="~/.hive/agents/{agent_name}",
status="failed",
limit=5
)
返回会话列表,包括 ID、时间戳、当前节点(失败位置)、执行质量。
第3c步:检查会话状态
get_agent_session_state(
agent_work_dir="~/.hive/agents/{agent_name}",
session_id="session_20260209_143022_abc12345"
)
返回执行路径,当前节点是哪个,步骤计数,时间戳 —— 但不包括内存值(为了避免上下文膨胀)。显示 memory_keys 和 memory_size 代替。
第3d步:检查运行时日志(L2/L3)
# L2:每个节点的成功/失败,重试计数
query_runtime_log_details(
agent_work_dir="~/.hive/agents/{agent_name}",
run_id="session_20260209_143022_abc12345",
needs_attention_only=True
)
# L3:确切的 LLM 响应,工具调用输入/输出
query_runtime_log_raw(
agent_work_dir="~/.hive/agents/{agent_name}",
run_id="session_20260209_143022_abc12345",
node_id="research"
)
第3e步:检查内存数据
# 查看节点实际产生的数据
get_agent_session_memory(
agent_work_dir="~/.hive/agents/{agent_name}",
session_id="session_20260209_143022_abc12345",
key="research_results"
)
第3f步:找到恢复点
list_agent_checkpoints(
agent_work_dir="~/.hive/agents/{agent_name}",
session_id="session_20260209_143022_abc12345",
is_clean="true"
)
返回检查点摘要,包括 ID、类型(node_start, node_complete),哪个节点,和 is_clean 标志。干净的检查点是安全的恢复点。
第3g步:比较检查点(可选)
要了解执行中两个点之间发生了什么变化:
compare_agent_checkpoints(
agent_work_dir="~/.hive/agents/{agent_name}",
session_id="session_20260209_143022_abc12345",
checkpoint_id_before="cp_node_complete_research_143030",
checkpoint_id_after="cp_node_complete_review_143115"
)
返回内存差异(添加/删除/更改的键)和执行路径差异。
第4阶段:根据根本原因修复
使用第3阶段的分析来确定要修复的内容和位置。
| 根本原因 | 修复什么 | 编辑哪里 |
|---|---|---|
| 提示问题 — LLM 产生错误的输出格式,错过指令 | 节点 system_prompt |
exports/{agent}/nodes/__init__.py |
| 代码错误 — TypeError, KeyError, Python 中的逻辑错误 | 代理代码 | exports/{agent}/agent.py, nodes/__init__.py |
| 图问题 — 错误的路由,缺少边,bad condition_expr | 边,节点配置 | exports/{agent}/agent.py |
| 工具问题 — MCP 工具失败,配置错误,缺少凭据 | 工具配置 | exports/{agent}/mcp_servers.json, /hive-credentials |
| 目标问题 — 成功标准太严格/模糊,错误的约束 | 目标定义 | exports/{agent}/agent.py(目标部分) |
| 测试问题 — 测试期望与实际代理行为不匹配 | 测试代码 | exports/{agent}/tests/test_*.py |
根据错误类别的修复策略
IMPLEMENTATION_ERROR (TypeError, AttributeError, KeyError):
# 阅读失败的代码
Read(file_path="exports/{agent_name}/nodes/__init__.py")
# 修复错误
Edit(
file_path="exports/{agent_name}/nodes/__init__.py",
old_string="results.get('videos')",
new_string="(results or {}).get('videos', [])"
)
ASSERTION_FAILURE (测试断言失败但代理成功运行):
- 检查代理的输出是否真的错了 → 修复提示
- 检查测试的期望是否不现实 → 修复测试
- 使用
get_agent_session_memory查看代理实际产生了什么
TIMEOUT / STALL (代理运行时间太长):
- 检查
node_visit_counts是否有反馈循环达到 max_node_visits - 检查 L3 日志是否有挂起的工具调用
- 在 loop_config 中减少
max_iterations或修复提示以更快收敛
API_ERROR (连接,速率限制,认证):
- 验证
/hive-credentials中的凭据 - 检查 MCP 服务器配置
第5阶段:恢复 & 恢复
修复代理后,决定是恢复还是重新运行。
何时从检查点恢复
当所有这些条件都为真时恢复:
- 修复是对现有干净检查点之后的节点
- 存在干净的检查点(来自带有检查点的 CLI 执行)
- 早期节点是昂贵的(网络抓取,API 调用,长 LLM 链)
# 在失败节点之前的最后一个干净检查点恢复
uv run hive run exports/{agent_name} \
--resume-session session_20260209_143022_abc12345 \
--checkpoint cp_node_complete_research_143030
这将跳过检查点之前的所有节点,只重新运行修复节点之后的部分。
何时从头开始重新运行
当任何这些条件为真时重新运行:
- 修复是对入口节点或早期节点
- 没有检查点存在(例如,代理是通过
run_tests运行的) - 代理很快(2-3个节点,几秒钟内完成)
- 你更改了图结构(添加/删除节点/边)
uv run hive run exports/{agent_name} --input '{"query": "test topic"}'
恢复前检查检查点
get_agent_checkpoint(
agent_work_dir="~/.hive/agents/{agent_name}",
session_id="session_20260209_143022_abc12345",
checkpoint_id="cp_node_complete_research_143030"
)
返回完整的检查点:共享内存快照,执行路径,当前节点,下一个节点,是否干净。
返回第2阶段
恢复或重新运行后,检查修复是否有效。如果没有,返回第3阶段。
第6阶段:最终验证
一旦迭代修复循环收敛(代理产生正确的输出),运行完整的自动化测试套件:
run_tests(
goal_id="your-goal-id",
agent_path="exports/{agent_name}"
)
所有测试应该通过。如果没有,重复剩余失败的循环。
凭据要求
关键:测试需要代理依赖的所有凭据。 这包括 LLM API 密钥和任何特定工具的凭据(HubSpot, Brave Search 等)。
前提条件
在运行代理测试之前,你必须从用户那里收集所有所需的凭据。
第1步:LLM API 密钥(始终需要)
export ANTHROPIC_API_KEY="your-key-here"
第2步:特定工具的凭据(取决于代理的工具)
检查代理的 mcp_servers.json 和工具配置,以确定代理使用的工具,然后检查所有所需的凭据:
from aden_tools.credentials import CredentialManager, CREDENTIAL_SPECS
creds = CredentialManager()
# 确定代理使用的工具(从 agent.json 或 mcp_servers.json)
agent_tools = [...] # 例如,["hubspot_search_contacts", "web_search", ...]
# 查找这些工具的所有缺失凭据
missing = creds.get_missing_for_tools(agent_tools)
常用工具凭据:
| 工具 | 环境变量 | 帮助 URL |
|---|---|---|
| HubSpot CRM | HUBSPOT_ACCESS_TOKEN |
https://developers.hubspot.com/docs/api/private-apps |
| Brave Search | BRAVE_SEARCH_API_KEY |
https://brave.com/search/api/ |
| Google Search | GOOGLE_SEARCH_API_KEY + GOOGLE_SEARCH_CX |
https://developers.google.com/custom-search |
为什么需要所有凭据:
- 测试需要执行代理的 LLM 节点以验证行为
- 缺少凭据的工具将返回错误字典而不是真实数据
- 模拟模式绕过一切,无法对现实世界性能有信心
模拟模式限制
模拟模式(--mock 标志或 MOCK_MODE=1)仅用于结构验证:
- 验证图结构(节点,边,连接)
- 验证
AgentRunner.load()成功且代理可导入 - 不执行事件循环代理 — MockLLMProvider 从不调用
set_output,因此事件循环节点永远循环 - 不测试 LLM 推理,内容质量或约束验证
- 不测试实际 API 集成或工具使用
底线: 如果你正在测试代理是否实现了其目标,你必须使用真实凭据。
在测试中强制执行凭据
编写测试时,总是包括凭据检查:
import os
import pytest
from aden_tools.credentials import CredentialManager
pytestmark = pytest.mark.skipif(
not CredentialManager().is_available("anthropic") and not os.environ.get("MOCK_MODE"),
reason="API key required for real testing. Set ANTHROPIC_API_KEY or use MOCK_MODE=1."
)
@pytest.fixture(scope="session", autouse=True)
def check_credentials():
"""Ensure ALL required credentials are set for real testing."""
creds = CredentialManager()
mock_mode = os.environ.get("MOCK_MODE")
if not creds.is_available("anthropic"):
if mock_mode:
print("
Running in MOCK MODE - structure validation only")
else:
pytest.fail(
"
ANTHROPIC_API_KEY not set!
"
"Set API key: export ANTHROPIC_API_KEY='your-key-here'
"
"Or run structure validation: MOCK_MODE=1 pytest exports/{agent}/tests/"
)
if not mock_mode:
agent_tools = [] # Update per agent
missing = creds.get_missing_for_tools(agent_tools)
if missing:
lines = ["
Missing tool credentials!"]
for name in missing:
spec = creds.specs.get(name)
if spec:
lines.append(f" {spec.env_var} - {spec.description}")
pytest.fail("
".join(lines))
用户沟通
当用户要求测试代理时,总是首先检查所有凭据:
- 从
mcp_servers.json确定代理的工具 - 使用
CredentialManager检查所有必需的凭据 - 在继续之前要求用户提供任何缺失的凭据
- 在一个提示中收集所有缺失的凭据 —— 而不是一个接一个
安全测试模式
OutputCleaner
框架在边遍历时自动验证和清理节点输出,使用快速 LLM。测试仍然应该使用安全模式,因为 OutputCleaner 可能无法捕获所有问题。
安全访问(必需)
# 不安全 - 会在缺少键时崩溃
approval = result.output["approval_decision"]
category = result.output["analysis"]["category"]
# 安全 - 使用 .get() 带默认值
output = result.output or {}
approval = output.get("approval_decision", "UNKNOWN")
# 安全 - 在操作前类型检查
analysis = output.get("analysis", {})
if isinstance(analysis, dict):
category = analysis.get("category", "unknown")
# 安全 - 处理 JSON 解析陷阱(LLM 响应作为字符串)
import json
recommendation = output.get("recommendation", "{}")
if isinstance(recommendation, str):
try:
parsed = json.loads(recommendation)
if isinstance(parsed, dict):
approval = parsed.get("approval_decision", "UNKNOWN")
except json.JSONDecodeError:
approval = "UNKNOWN"
elif isinstance(recommendation, dict):
approval = recommendation.get("approval_decision", "UNKNOWN")
# 安全 - 在迭代前类型检查
items = output.get("items", [])
if isinstance(items, list):
for item in items:
...
conftest.py 的辅助函数
import json
import re
def _parse_json_from_output(result, key):
"""从代理输出中解析 JSON(框架可能将完整的 LLM 响应存储为字符串)。"""
response_text = result.output.get(key, "")
json_text = re.sub(r'```json\s*|\s*```', '', response_text).strip()
try:
return json.loads(json_text)
except (json.JSONDecodeError, AttributeError, TypeError):
return result.output.get(key)
def safe_get_nested(result, key_path, default=None):
"""安全地从 result.output 中获取嵌套值。"""
output = result.output or {}
current = output
for key in key_path:
if isinstance(current, dict):
current = current.get(key)
elif isinstance(current, str):
try:
json_text = re.sub(r'```json\s*|\s*```', '', current).strip()
parsed = json.loads(json_text)
if isinstance(parsed, dict):
current = parsed.get(key)
else:
return default
except json.JSONDecodeError:
return default
else:
return default
return current if current is not None else default
# 在测试中可用
pytest.parse_json_from_output = _parse_json_from_output
pytest.safe_get_nested = safe_get_nested
ExecutionResult 字段
result.success=True 意味着没有异常,而不是目标实现
# 错误
assert result.success
# 正确
assert result.success, f"Agent failed: {result.error}"
output = result.output or {}
approval = output.get("approval_decision")
assert approval == "APPROVED", f"Expected APPROVED, got {approval}"
所有字段:
success: bool— 完成没有异常(不是目标实现!)output: dict— 完整的内存快照(可能包含原始字符串)error: str | None— 失败时的错误消息steps_executed: int— 执行的节点数total_tokens: int— 累积的令牌使用量total_latency_ms: int— 总执行时间path: list[str]— 遍历的节点 ID(在反馈循环中可能重复)paused_at: str | None— 如果暂停,节点 IDsession_state: dict— 恢复的状态node_visit_counts: dict[str, int]— 每个节点的访问计数(反馈循环测试)execution_quality: str— “clean”, “degraded”, 或 “failed”
测试计数指导
编写8-15个测试,而不是30+
- 每个成功标准2-3个测试
- 1个快乐路径测试
- 1个边界/边缘情况测试
- 1个错误处理测试(可选)
每个真实测试成本约3秒 + LLM 令牌。12个测试 = 约36秒,$0.12。
测试模式
快乐路径
@pytest.mark.asyncio
async def test_happy_path(runner, auto_responder, mock_mode):
"""测试正常成功的执行。"""
await auto_responder.start()
try:
result = await runner.run({"query": "python tutorials"})
finally:
await auto_responder.stop()
assert result.success, f"Agent failed: {result.error}"
output = result.output or {}
assert output.get("report"), "No report produced"
边界条件
@pytest.mark.asyncio
async def test_minimum_sources(runner, auto_responder, mock_mode):
"""测试在最小源阈值时。"""
await auto_responder.start()
try:
result = await runner.run({"query": "niche topic"})
finally:
await auto_responder.stop()
assert result.success, f"Agent failed: {result.error}"
output = result.output or {}
sources = output.get("sources", [])
if isinstance(sources, list):
assert len(sources) >= 3, f"Expected >= 3 sources, got {len(sources)}"
错误处理
@pytest.mark.asyncio
async def test_empty_input(runner, auto_responder, mock_mode):
"""测试对空输入的优雅处理。"""
await auto_responder.start()
try:
result = await runner.run({"query": ""})
finally:
await auto_responder.stop()
# 代理应该要么优雅地失败,要么产生错误消息
output = result.output or {}
assert not result.success or output.get("error"), "Should handle empty input"
反馈循环
@pytest.mark.asyncio
async def test_feedback_loop_terminates(runner, auto_responder, mock_mode):
"""测试反馈循环不会永远运行。"""
await auto_responder.start()
try:
result = await runner.run({"query": "test"})
finally:
await auto_responder.stop()
visits = result.node_visit_counts or {}
for node_id, count in visits.items():
assert count <= 5, f"Node {node_id} visited {count} times — possible infinite loop"
MCP 工具参考
第1阶段:测试生成
# 检查现有测试
list_tests(goal_id, agent_path)
# 获取约束测试指南(返回模板,不是生成的测试)
generate_constraint_tests(goal_id, goal_json, agent_path)
# 返回:output_file, file_header, test_template, constraints_formatted, test_guidelines
# 获取成功标准测试指南
generate_success_tests(goal_id, goal_json, node_names, tool_names, agent_path)
# 返回:output_file, file_header, test_template, success_criteria_formatted, test_guidelines
第2阶段:执行
# 自动化回归(无检查点,新运行)
run_tests(goal_id, agent_path, test_types='["all"]', parallel=-1, fail_fast=False)
# 只运行特定测试类型
run_tests(goal_id, agent_path, test_types='["constraint"]')
run_tests(goal_id, agent_path, test_types='["success"]')
# 通过 CLI 进行迭代调试(带有检查点)
uv run hive run exports/{agent_name} --input '{"query": "test"}'
第3阶段:分析
# 调试特定的失败测试
debug_test(goal_id, test_name, agent_path)
# 查找失败的会话
list_agent_sessions(agent_work_dir, status="failed", limit=5)
# 检查会话状态(不包括内存值)
get_agent_session_state(agent_work_dir, session_id)
# 检查内存数据
get_agent_session_memory(agent_work_dir, session_id, key="research_results")
# 运行时日志:L1 摘要
query_runtime_logs(agent_work_dir, status="needs_attention")
# 运行时日志:L2 每个节点的详细信息
query_runtime_log_details(agent_work_dir, run_id, needs_attention_only=True)
# 运行时日志:L3 工具/LLM 原始数据
query_runtime_log_raw(agent_work_dir, run_id, node_id="research")
# 查找干净的检查点
list_agent_checkpoints(agent_work_dir, session_id, is_clean="true")
# 比较检查点(内存差异)
compare_agent_checkpoints(agent_work_dir, session_id, cp_before, cp_after)
第5阶段:恢复
# 在恢复前检查检查点
get_agent_checkpoint(agent_work_dir, session_id, checkpoint_id)
# 空的 checkpoint_id = 最新的检查点
# 通过 CLI 从检查点恢复(无头)
uv run hive run exports/{agent_name} \
--resume-session {session_id} --checkpoint {checkpoint_id}
反模式
| 不要 | 而是 |
|---|---|
在测试中使用 default_agent.run() |
使用 runner.run() 与 auto_responder 固定装置(通过 AgentRuntime) |
| 当晚期节点失败时重新运行整个代理 | 从上一个干净的检查点恢复 |
将 result.success 视为目标实现 |
检查 result.output 是否符合实际标准 |
直接访问 result.output["key"] |
使用 result.output.get("key") |
| 随机修复希望测试通过 | 首先分析 L2/L3 日志以找到根本原因 |
| 编写30+测试 | 编写8-15个专注的测试 |
| 跳过凭据检查 | 在测试前使用 /hive-credentials |
混淆 exports/ 与 ~/.hive/agents/ |
在 exports/ 中编写代码,在 ~/.hive/ 中读取运行时数据 |
使用 run_tests 进行迭代调试 |
使用带有检查点的无头 CLI 进行迭代调试 |
| 使用无头 CLI 进行最终回归 | 使用 run_tests 进行自动化回归 |
使用 Claude Code 的 --tui |
使用无头 run 命令 — TUI 在非交互式 shell 中挂起 |
| 从 Claude Code 测试客户端面向节点 | 使用模拟模式,或让用户在他们的终端中运行代理 |
| 不读目标就开始测试 | 在编写测试前始终理解目标 |
| 跳过第3阶段分析并猜测 | 使用会话 + 日志工具识别根本原因 |
示例演示:深度研究代理
一个完整的迭代,展示了一个具有节点 intake → research → review → report 的代理的测试循环。
第1阶段:生成测试
# 读取目标
Read(file_path="exports/deep_research_agent/agent.py")
# 获取成功标准测试指南
result = generate_success_tests(
goal_id="rigorous-interactive-research",
goal_json='{"id": "rigorous-interactive-research", "success_criteria": [{"id": "source-diversity", "target": ">=5"}, {"id": "citation-coverage", "target": "100%"}, {"id": "report-completeness", "target": "90%"}]}',
node_names="intake,research,review,report",
tool_names="web_search,web_scrape",
agent_path="exports/deep_research_agent"
)
# 编写测试
Write(
file_path=result["output_file"],
content=result["file_header"] + "
" + test_code
)
第2阶段:首次执行
run_tests(
goal_id="rigorous-interactive-research",
agent_path="exports/deep_research_agent",
fail_fast=True
)
结果:test_success_source_diversity 失败 — 代理只找到了2个源而不是5个。
第3阶段:分析
# 调试失败的测试
debug_test(
goal_id="rigorous-interactive-research",
test_name="test_success_source_diversity",
agent_path="exports/deep_research_agent"
)
# → ASSERTION_FAILURE: 预期 >= 5 个源,得到 2
# 查找会话
list_agent_sessions(
agent_work_dir="~/.hive/agents/deep_research_agent",
status="completed",
limit=1
)
# → session_20260209_150000_abc12345
# 查看研究节点产生了什么
get_agent_session_memory(
agent_work_dir="~/.hive/agents/deep_research_agent",
session_id="session_20260209_150000_abc12345",
key="research_results"
)
# → 只进行了2次 web_search 调用,每次返回1个源
# 检查研究节点中 LLM 的行为
query_runtime_log_raw(
agent_work_dir="~/.hive/agents/deep_research_agent",
run_id="session_20260209_150000_abc12345",
node_id="research"
)
# → LLM 只调用了两次 web_search,然后调用 set_output
根本原因:研究节点的提示没有告诉 LLM 搜索至少5个不同的权威来源。它在前几次搜索后就停止了。
第4阶段:修复提示
Read(file_path="exports/deep_research_agent/nodes/__init__.py")
Edit(
file_path="exports/deep_research_agent/nodes/__init__.py",
old_string='system_prompt="Search for information on the user\'s topic."',
new_string='system_prompt="Search for information on the user\'s topic. You MUST find at least 5 diverse, authoritative sources. Use multiple different search queries to ensure source diversity. Do not stop searching until you have at least 5 distinct sources."'
)
第5阶段:从检查点恢复
对于这个例子,修复是对 research 节点的。如果我们通过 CLI 带有检查点运行,我们可以从 intake 之后的检查点恢复,跳过重新运行 intake:
# 检查是否存在干净的检查点
list_agent_checkpoints(
agent_work_dir="~/.hive/agents/deep_research_agent",
session_id="session_20260209_150000_abc12345",
is_clean="true"
)
# → cp_node_complete_intake_150005
# 从 intake 之后恢复,重新运行带有修复提示的研究
uv run hive run exports/deep_research_agent \
--resume-session session_20260209_150000_abc12345 \
--checkpoint cp_node_complete_intake_150005
或者对于这个简单案例(intake 很快),只需重新运行:
uv run hive run exports/deep_research_agent --input '{"topic": "test"}'
第6阶段:最终验证
run_tests(
goal_id="rigorous-interactive-research",
agent_path="exports/deep_research_agent"
)
# → 所有12个测试通过
测试文件结构
exports/{agent_name}/
├── agent.py ← 要测试的代理(目标,节点,边)
├── nodes/__init__.py ← 节点实现(提示,配置)
├── config.py ← 代理配置
├── mcp_servers.json ← 工具服务器配置
└── tests/
├── conftest.py ← 共享固定装置 + 安全访问辅助函数
├── test_constraints.py ← 约束测试
├── test_success_criteria.py ← 成功标准测试
└── test_edge_cases.py ← 边缘情况测试
与其他技能的集成
| 场景 | 从 | 到 | 动作 |
|---|---|---|---|
| 构建代理,准备测试 | /hive-create |
/hive-test |
生成测试,开始循环 |
| 需要修复提示 | /hive-test 第4阶段 |
直接编辑 | 编辑 nodes/__init__.py,恢复 |
| 目标定义错误 | /hive-test 第4阶段 |
/hive-create |
更新目标,可能需要重建 |
| 缺少凭据 | /hive-test 第3阶段 |
/hive-credentials |
设置凭据 |
| 复杂的运行时失败 | /hive-test 第3阶段 |
/hive-debugger |
深入 L1/L2/L3 分析 |
| 所有测试通过 | /hive-test 第6阶段 |
完成 | 代理验证 |