为什么 Agent 需要回放测试
Agent 系统的测试比传统软件复杂得多。传统软件的函数调用是确定性的——同样的输入产生同样的输出。但 Agent 涉及 LLM 推理、工具调用、环境交互，每一步都可能引入非确定性。当你修改了 Prompt、升级了模型、或调整了工具参数，如何确保 Agent 的行为没有退化？
回放测试（Replay Testing）是解决这个问题的核心方法：录制 Agent 的真实执行轨迹，在变更后回放这些轨迹，比较行为差异。
回放测试的核心价值：
- 回归保护：确保修改不会破坏已有的正确行为
- 行为可追溯：每次变更后的行为差异可量化、可审查
- 非确定性管理：在不确定的系统中建立确定性的验证基线
- 成本控制：无需重新执行真实环境操作，降低测试成本
回放测试架构
┌──────────────────────────────────────────────────────┐
│ 回放测试系统 │
│ │
│ ┌──────────┐ 录制 ┌──────────┐ 存储 ┌────────┐│
│ │ Agent │ ──────> │ Recorder │ ─────> │ Trace ││
│ │ 执行环境 │ │ 记录器 │ │ Store ││
│ └──────────┘ └──────────┘ └───┬────┘│
│ │ │
│ ┌──────────┐ 回放 ┌──────────┐ 比对 ┌───v────┐│
│ │ Agent │ <────── │ Replayer │ <───── │ Trace ││
│ │ 测试环境 │ │ 回放器 │ │ Store ││
│ └──────────┘ └──────────┘ └────────┘│
│ │ │
│ v │
│ ┌──────────┐ │
│ │ Diff │ │
│ │ 差异分析 │ │
│ └──────────┘ │
└──────────────────────────────────────────────────────┘
一、轨迹录制
轨迹数据结构
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
import json
@dataclass
class AgentStep:
"""Agent 执行的单步操作"""
step_id: str
step_type: str # "reasoning" | "tool_call" | "observation" | "final_answer"
timestamp: str
# 推理内容
reasoning: Optional[str] = None
# 工具调用
tool_name: Optional[str] = None
tool_input: Optional[dict] = None
tool_output: Optional[Any] = None
tool_duration_ms: Optional[int] = None
# 环境状态快照
env_state_before: Optional[dict] = None
env_state_after: Optional[dict] = None
# LLM 调用详情
model_name: Optional[str] = None
prompt_tokens: Optional[int] = None
completion_tokens: Optional[int] = None
temperature: Optional[float] = None
@dataclass
class AgentTrace:
"""完整的 Agent 执行轨迹"""
trace_id: str
task: str # 用户任务描述
steps: list[AgentStep] = field(default_factory=list)
metadata: dict = field(default_factory=dict)
# 环境信息
agent_version: str = ""
model_version: str = ""
tool_versions: dict = field(default_factory=dict)
environment: str = "" # "production" | "staging" | "test"
# 结果
success: bool = False
final_answer: str = ""
total_duration_ms: int = 0
total_tokens: int = 0
def to_dict(self) -> dict:
return {
"trace_id": self.trace_id,
"task": self.task,
"steps": [
{k: v for k, v in step.__dict__.items() if v is not None}
for step in self.steps
],
"metadata": self.metadata,
"agent_version": self.agent_version,
"model_version": self.model_version,
"tool_versions": self.tool_versions,
"success": self.success,
"final_answer": self.final_answer,
"total_duration_ms": self.total_duration_ms,
"total_tokens": self.total_tokens,
}
def save(self, path: str):
with open(path, "w", encoding="utf-8") as f:
json.dump(self.to_dict(), f, ensure_ascii=False, indent=2)
@classmethod
def load(cls, path: str) -> "AgentTrace":
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
steps = [
AgentStep(
step_id=s["step_id"],
step_type=s["step_type"],
timestamp=s["timestamp"],
reasoning=s.get("reasoning"),
tool_name=s.get("tool_name"),
tool_input=s.get("tool_input"),
tool_output=s.get("tool_output"),
tool_duration_ms=s.get("tool_duration_ms"),
env_state_before=s.get("env_state_before"),
env_state_after=s.get("env_state_after"),
model_name=s.get("model_name"),
prompt_tokens=s.get("prompt_tokens"),
completion_tokens=s.get("completion_tokens"),
temperature=s.get("temperature"),
)
for s in data["steps"]
]
return cls(
trace_id=data["trace_id"],
task=data["task"],
steps=steps,
metadata=data.get("metadata", {}),
agent_version=data.get("agent_version", ""),
model_version=data.get("model_version", ""),
tool_versions=data.get("tool_versions", {}),
environment=data.get("environment", ""),
success=data.get("success", False),
final_answer=data.get("final_answer", ""),
total_duration_ms=data.get("total_duration_ms", 0),
total_tokens=data.get("total_tokens", 0),
)
录制器实现
class TraceRecorder:
"""Agent 执行轨迹录制器"""
def __init__(self, agent_version: str, model_version: str):
self.agent_version = agent_version
self.model_version = model_version
self.traces: list[AgentTrace] = []
def record_execution(self, agent, task: str, **kwargs) -> AgentTrace:
"""录制 Agent 的一次完整执行"""
import uuid
trace = AgentTrace(
trace_id=str(uuid.uuid4()),
task=task,
agent_version=self.agent_version,
model_version=self.model_version,
environment=kwargs.get("environment", "test"),
)
# 包装 agent 的方法以录制每一步
original_methods = self._wrap_agent_methods(agent, trace)
try:
# 执行 Agent
result = agent.run(task)
trace.success = result.get("success", False)
trace.final_answer = result.get("answer", "")
trace.total_duration_ms = sum(
s.tool_duration_ms or 0 for s in trace.steps
)
trace.total_tokens = sum(
(s.prompt_tokens or 0) + (s.completion_tokens or 0)
for s in trace.steps
)
finally:
# 恢复原始方法
self._unwrap_agent_methods(agent, original_methods)
self.traces.append(trace)
return trace
def _wrap_agent_methods(self, agent, trace: AgentTrace) -> dict:
"""包装 Agent 的关键方法以实现录制"""
original = {}
# 包装 LLM 调用
if hasattr(agent, "llm_call"):
original["llm_call"] = agent.llm_call
def wrapped_llm_call(prompt, *args, **kwargs):
import time
step = AgentStep(
step_id=f"step_{len(trace.steps)}",
step_type="reasoning",
timestamp=datetime.now().isoformat(),
model_name=trace.model_version,
temperature=kwargs.get("temperature", 0),
)
start = time.perf_counter()
result = original["llm_call"](prompt, *args, **kwargs)
step.tool_duration_ms = int((time.perf_counter() - start) * 1000)
step.reasoning = result
trace.steps.append(step)
return result
agent.llm_call = wrapped_llm_call
# 包装工具调用
if hasattr(agent, "call_tool"):
original["call_tool"] = agent.call_tool
def wrapped_call_tool(tool_name, tool_input, *args, **kwargs):
import time
step = AgentStep(
step_id=f"step_{len(trace.steps)}",
step_type="tool_call",
timestamp=datetime.now().isoformat(),
tool_name=tool_name,
tool_input=tool_input,
)
# 记录调用前的环境状态
if hasattr(agent, "get_env_state"):
step.env_state_before = agent.get_env_state()
start = time.perf_counter()
result = original["call_tool"](tool_name, tool_input, *args, **kwargs)
step.tool_duration_ms = int((time.perf_counter() - start) * 1000)
step.tool_output = result
# 记录调用后的环境状态
if hasattr(agent, "get_env_state"):
step.env_state_after = agent.get_env_state()
trace.steps.append(step)
return result
agent.call_tool = wrapped_call_tool
return original
def _unwrap_agent_methods(self, agent, original: dict):
"""恢复原始方法"""
for name, method in original.items():
setattr(agent, name, method)
def save_all(self, directory: str):
"""保存所有轨迹"""
from pathlib import Path
dir_path = Path(directory)
dir_path.mkdir(parents=True, exist_ok=True)
for trace in self.traces:
trace.save(dir_path / f"{trace.trace_id}.json")
二、回放器与确定性验证
回放器设计
class TraceReplayer:
"""轨迹回放器：重新执行录制的 Agent 轨迹"""
def __init__(self, agent, config: dict = None):
self.agent = agent
self.config = config or {}
# 是否模拟工具输出（不实际调用工具）
self.mock_tools = self.config.get("mock_tools", False)
# 是否比较推理步骤
self.compare_reasoning = self.config.get("compare_reasoning", True)
# 推理相似度阈值
self.reasoning_threshold = self.config.get("reasoning_threshold", 0.85)
def replay(self, trace: AgentTrace) -> dict:
"""回放一条轨迹并比较差异"""
results = {
"trace_id": trace.trace_id,
"task": trace.task,
"original_success": trace.success,
"replay_success": None,
"step_results": [],
"overall_diff": {},
}
# 如果 mock_tools，将工具输出预加载
mocked_outputs = {}
if self.mock_tools:
for step in trace.steps:
if step.step_type == "tool_call":
mocked_outputs[step.step_id] = step.tool_output
# 逐步回放
for i, original_step in enumerate(trace.steps):
replay_result = self._replay_step(original_step, i, mocked_outputs)
results["step_results"].append(replay_result)
# 比较最终结果
results["overall_diff"] = self._compute_overall_diff(
trace, results["step_results"]
)
return results
def _replay_step(self, original: AgentStep, index: int,
mocked_outputs: dict) -> dict:
"""回放单个步骤"""
result = {
"step_id": original.step_id,
"step_type": original.step_type,
"match": True,
"diff": {},
}
if original.step_type == "tool_call":
if self.mock_tools and original.step_id in mocked_outputs:
# 使用模拟输出
actual_output = mocked_outputs[original.step_id]
else:
# 实际调用工具
actual_output = self.agent.call_tool(
original.tool_name, original.tool_input
)
# 比较工具输出
output_match = self._compare_outputs(
original.tool_output, actual_output
)
result["match"] = output_match["exact_match"]
result["diff"]["output"] = output_match
# 比较环境状态变化
if original.env_state_after:
current_state = self.agent.get_env_state() if hasattr(self.agent, "get_env_state") else None
if current_state:
state_match = self._compare_states(
original.env_state_after, current_state
)
result["diff"]["env_state"] = state_match
if not state_match["match"]:
result["match"] = False
elif original.step_type == "reasoning" and self.compare_reasoning:
# 比较推理输出（使用语义相似度）
actual_reasoning = self.agent.llm_call(
self._reconstruct_prompt(original, index)
)
similarity = self._semantic_similarity(
original.reasoning, actual_reasoning
)
result["match"] = similarity >= self.reasoning_threshold
result["diff"]["reasoning"] = {
"similarity": similarity,
"original_length": len(original.reasoning or ""),
"actual_length": len(actual_reasoning or ""),
}
return result
def _compare_outputs(self, expected: Any, actual: Any) -> dict:
"""比较工具输出"""
if expected == actual:
return {"exact_match": True, "semantic_match": True}
# 对于字符串，尝试语义比较
if isinstance(expected, str) and isinstance(actual, str):
sim = self._semantic_similarity(expected, actual)
return {
"exact_match": False,
"semantic_match": sim > 0.9,
"similarity": sim,
}
# 对于字典，逐键比较
if isinstance(expected, dict) and isinstance(actual, dict):
diffs = {}
all_keys = set(expected.keys()) | set(actual.keys())
for key in all_keys:
if key not in expected:
diffs[key] = {"status": "added", "value": actual[key]}
elif key not in actual:
diffs[key] = {"status": "removed", "value": expected[key]}
elif expected[key] != actual[key]:
diffs[key] = {
"status": "changed",
"expected": expected[key],
"actual": actual[key],
}
return {
"exact_match": False,
"semantic_match": len(diffs) == 0,
"field_diffs": diffs,
}
return {"exact_match": False, "semantic_match": False}
def _compare_states(self, expected: dict, actual: dict) -> dict:
"""比较环境状态"""
diffs = {}
for key in set(expected.keys()) | set(actual.keys()):
if expected.get(key) != actual.get(key):
diffs[key] = {
"expected": expected.get(key),
"actual": actual.get(key),
}
return {"match": len(diffs) == 0, "diffs": diffs}
def _semantic_similarity(self, text1: str, text2: str) -> float:
"""计算语义相似度"""
from sentence_transformers import SentenceTransformer
import numpy as np
model = SentenceTransformer("all-MiniLM-L6-v2")
emb1 = model.encode([text1])
emb2 = model.encode([text2])
return float(np.dot(emb1[0], emb2[0]) / (
np.linalg.norm(emb1[0]) * np.linalg.norm(emb2[0])
))
def _reconstruct_prompt(self, step: AgentStep, index: int) -> str:
"""从轨迹步骤重建 LLM prompt"""
# 简化实现：实际需要根据 Agent 架构重建完整上下文
return f"Step {index}: {step.reasoning or ''}"
def _compute_overall_diff(self, trace: AgentTrace,
step_results: list[dict]) -> dict:
"""计算整体差异"""
total_steps = len(step_results)
matched_steps = sum(1 for r in step_results if r["match"])
return {
"total_steps": total_steps,
"matched_steps": matched_steps,
"mismatched_steps": total_steps - matched_steps,
"match_rate": matched_steps / total_steps if total_steps > 0 else 0,
"success_preserved": trace.success, # 简化
}
三、回归测试框架
测试套件管理
class RegressionTestSuite:
"""Agent 回归测试套件"""
def __init__(self, suite_name: str):
self.suite_name = suite_name
self.test_cases: list[dict] = []
self.baselines: dict = {} # trace_id -> baseline result
def add_trace_as_baseline(self, trace: AgentTrace,
expected_success: bool = True,
category: str = "general"):
"""将一条轨迹添加为回归测试基线"""
self.test_cases.append({
"trace_id": trace.trace_id,
"task": trace.task,
"category": category,
"expected_success": expected_success,
"baseline_trace": trace,
})
def load_from_directory(self, dir_path: str):
"""从目录加载所有轨迹作为基线"""
from pathlib import Path
for trace_file in Path(dir_path).glob("*.json"):
trace = AgentTrace.load(str(trace_file))
self.add_trace_as_baseline(trace, expected_success=trace.success)
def run_regression(self, agent, config: dict = None) -> dict:
"""运行完整回归测试"""
replayer = TraceReplayer(agent, config or {})
results = []
for tc in self.test_cases:
result = replayer.replay(tc["baseline_trace"])
result["category"] = tc["category"]
result["expected_success"] = tc["expected_success"]
result["passed"] = self._evaluate_pass(result, tc)
results.append(result)
return self._summarize(results)
def _evaluate_pass(self, result: dict, test_case: dict) -> bool:
"""判断是否通过回归"""
# 1. 成功状态保持
if result["overall_diff"].get("success_preserved") != test_case["expected_success"]:
return False
# 2. 步骤匹配率达到阈值
match_rate = result["overall_diff"].get("match_rate", 0)
if match_rate < 0.8:
return False
return True
def _summarize(self, results: list[dict]) -> dict:
from collections import defaultdict
total = len(results)
passed = sum(1 for r in results if r["passed"])
by_category = defaultdict(lambda: {"total": 0, "passed": 0})
for r in results:
by_category[r["category"]]["total"] += 1
if r["passed"]:
by_category[r["category"]]["passed"] += 1
return {
"suite_name": self.suite_name,
"total_tests": total,
"passed": passed,
"failed": total - passed,
"pass_rate": passed / total if total > 0 else 0,
"by_category": dict(by_category),
"failures": [
{
"trace_id": r["trace_id"],
"task": r["task"],
"category": r["category"],
"match_rate": r["overall_diff"].get("match_rate", 0),
"mismatched_steps": r["overall_diff"].get("mismatched_steps", 0),
}
for r in results if not r["passed"]
],
}
四、非确定性管理
确定性策略
class DeterminismManager:
"""管理 Agent 测试中的非确定性"""
STRATEGIES = {
"temperature_zero": "LLM 温度设为 0，最大化输出确定性",
"mock_llm": "模拟 LLM 输出，完全确定性",
"mock_tools": "模拟工具输出，消除环境非确定性",
"semantic_compare": "使用语义相似度替代精确匹配",
"n_run_consensus": "多次运行取共识",
}
@staticmethod
def n_run_consensus(agent, task: str, n: int = 5,
agreement_threshold: float = 0.8) -> dict:
"""多次运行取共识"""
results = []
for _ in range(n):
result = agent.run(task)
results.append(result)
# 计算答案一致性
from sentence_transformers import SentenceTransformer
import numpy as np
model = SentenceTransformer("all-MiniLM-L6-v2")
embeddings = model.encode([r.get("answer", "") for r in results])
sim_matrix = np.dot(embeddings, embeddings.T)
# 平均成对相似度
n_results = len(results)
upper_tri = sim_matrix[np.triu_indices(n_results, k=1)]
avg_agreement = np.mean(upper_tri)
return {
"n_runs": n,
"avg_agreement": float(avg_agreement),
"consensus_reached": avg_agreement >= agreement_threshold,
"results": results,
"strategy": "n_run_consensus",
}
快照测试
class SnapshotTester:
"""Agent 状态快照测试"""
def __init__(self, snapshot_dir: str = "snapshots"):
self.snapshot_dir = snapshot_dir
from pathlib import Path
Path(snapshot_dir).mkdir(parents=True, exist_ok=True)
def take_snapshot(self, agent, label: str) -> str:
"""拍摄 Agent 当前状态快照"""
import hashlib
snapshot = {
"label": label,
"timestamp": datetime.now().isoformat(),
"memory": getattr(agent, "memory", None),
"state": getattr(agent, "state", None),
"context": getattr(agent, "context", None),
"tool_registry": list(getattr(agent, "tools", {}).keys()),
}
# 计算快照哈希
snapshot_str = json.dumps(snapshot, sort_keys=True, ensure_ascii=False)
snapshot_hash = hashlib.sha256(snapshot_str.encode()).hexdigest()[:16]
# 保存快照
from pathlib import Path
filepath = Path(self.snapshot_dir) / f"{label}_{snapshot_hash}.json"
with open(filepath, "w", encoding="utf-8") as f:
json.dump(snapshot, f, ensure_ascii=False, indent=2)
return str(filepath)
def compare_snapshots(self, snapshot_path_a: str,
snapshot_path_b: str) -> dict:
"""比较两个快照"""
with open(snapshot_path_a, "r", encoding="utf-8") as f:
snap_a = json.load(f)
with open(snapshot_path_b, "r", encoding="utf-8") as f:
snap_b = json.load(f)
diffs = {}
all_keys = set(snap_a.keys()) | set(snap_b.keys())
for key in all_keys:
if snap_a.get(key) != snap_b.get(key):
diffs[key] = {
"snapshot_a": snap_a.get(key),
"snapshot_b": snap_b.get(key),
}
return {
"identical": len(diffs) == 0,
"diff_count": len(diffs),
"diffs": diffs,
}
五、CI/CD 集成
# .github/workflows/agent-regression.yml
name: Agent Regression Tests
on:
pull_request:
paths:
- "agent/**"
- "prompts/**"
- "tools/**"
push:
branches: [main]
jobs:
regression:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: "3.11"
- name: Install dependencies
run: pip install -r requirements.txt
- name: Load baseline traces
uses: actions/cache@v4
with:
path: tests/traces/baseline
key: baseline-traces-${{ hashFiles('tests/traces/baseline/**') }}
- name: Run regression tests
run: |
python -m agent_regression_test \
--suite "main" \
--baseline-dir tests/traces/baseline \
--config tests/regression_config.yaml \
--output reports/regression.json
- name: Check pass rate
run: |
PASS_RATE=$(python -c "import json; r=json.load(open('reports/regression.json')); print(r['pass_rate'])")
echo "Pass rate: $PASS_RATE"
if (( $(echo "$PASS_RATE < 0.9" | bc -l) )); then
echo "::error::Regression test pass rate below 90%"
exit 1
fi
- name: Upload regression report
uses: actions/upload-artifact@v4
with:
name: regression-report
path: reports/regression.json
测试策略对比
| 策略 | 确定性 | 覆盖率 | 维护成本 | 执行速度 | 适用阶段 |
|---|---|---|---|---|---|
| 精确回放 | 高 | 低 | 高 | 快 | 回归测试 |
| Mock 工具回放 | 高 | 中 | 中 | 快 | CI/CD |
| 语义比较回放 | 中 | 高 | 低 | 中 | 日常验证 |
| 多次运行共识 | 中 | 高 | 低 | 慢 | 发布前 |
| 快照测试 | 高 | 中 | 中 | 快 | 状态验证 |
| 全量重放 | 低 | 高 | 低 | 慢 | 深度验证 |
最佳实践
轨迹采样策略
class TraceSampler:
"""从生产环境采样轨迹用于回归测试"""
@staticmethod
def sample_diverse(traces: list[AgentTrace],
target_count: int = 100) -> list[AgentTrace]:
"""采样多样化的轨迹集"""
from collections import defaultdict
import random
# 按任务类型分组
by_type = defaultdict(list)
for t in traces:
task_type = t.metadata.get("task_type", "unknown")
by_type[task_type].append(t)
# 每个类型按比例采样
total = len(traces)
sampled = []
for task_type, type_traces in by_type.items():
n = max(1, int(target_count * len(type_traces) / total))
# 优先采样成功和失败的案例各一半
successes = [t for t in type_traces if t.success]
failures = [t for t in type_traces if not t.success]
n_success = min(n // 2, len(successes))
n_failure = min(n - n_success, len(failures))
sampled.extend(random.sample(successes, n_success) if n_success else [])
sampled.extend(random.sample(failures, n_failure) if n_failure else [])
return sampled[:target_count]
结语
Agent 回放测试是保障系统可靠性的关键基础设施。它通过录制真实执行轨迹、回放比较行为差异，在非确定性的 Agent 系统中建立可量化的质量基线。核心原则是：录制要全、回放要快、比较要智能。精确匹配在 LLM 时代往往过于严格，语义比较 + Mock 工具的组合能在保证检测力的同时控制测试的 flakiness。将回放测试集成到 CI/CD 流程中，才能确保每次 Agent 变更都有质量保障。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。