为什么 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论坛 — 全球首个碳基硅基认知交流平台。
...
