引言 LLM的本质是非确定性的——同样的输入可能产生不同的输出。这给Agent系统的测试带来了根本性挑战：如何验证Agent的"正确性"？如何检测"回归"？回放测试（Replay Testing）通过将真实请求重新执行并与历史结果对比，为Agent系统提供了一种实用且有效的验证手段。
2026年，回放测试已成为Agent系统质量保障的核心手段。本文系统介绍如何设计和实施Agent回放测试体系。
回放测试原理 ┌──────────────────────────────────────────────────────┐ │ 回放测试流程 │ │ │ │ 生产流量 ──▶ 录制 (Record) │ │ │ │ │ ▼ │ │ 测试数据集 (Test Set) │ │ │ │ │ ▼ │ │ 新版本 ──▶ 回放 (Replay) ──▶ 结果对比 ──▶ 报告 │ │ │ │ │ │ ▼ ▼ │ │ 差异分析 质量门禁 │ │ │ └──────────────────────────────────────────────────────┘ 录制生产流量 class ProductionRecorder: """生产流量录制器""" def __init__(self, storage_client): self.storage = storage_client self.sampling_rate = 0.01 # 1%采样 async def record_request( self, session_id: str, request: dict, response: dict, metadata: dict ): """录制请求和响应""" import random if random.random() > self.sampling_rate: return recording = { "session_id": session_id, "timestamp": datetime.now().isoformat(), "request": { "input": request["input"], "context": request.get("context", {}), "config": request.get("config", {}), }, "response": { "output": response["output"], "tool_calls": response.get("tool_calls", []), "tokens_used": response.get("usage", {}), "latency_ms": response.get("latency_ms"), }, "metadata": { "model": metadata.get("model"), "prompt_version": metadata.get("prompt_version"), "quality_score": metadata.get("quality_score"), "user_feedback": metadata.get("user_feedback"), }, "recording_version": "1.0" } # 存储到对象存储 key = f"recordings/{datetime.now().strftime('%Y%m%d')}/{session_id}.json" await self.storage.put(key, json.dumps(recording, ensure_ascii=False)) 回放执行 class ReplayExecutor: """回放执行器""" async def replay_test_set( self, test_set: list, config: dict ) -> dict: """回放测试集""" results = [] for i, test_case in enumerate(test_set): logger.info(f"Replaying test case {i+1}/{len(test_set)}") try: result = await self._replay_single(test_case, config) results.append(result) except Exception as e: logger.error(f"Replay failed for case {i+1}: {e}") results.append({ "test_case_id": test_case.get("id"), "status": "error", "error": str(e) }) # 汇总分析 return self._analyze_results(results) async def _replay_single( self, test_case: dict, config: dict ) -> dict: """回放单个测试用例""" # 使用录制时的配置（或覆盖） replay_config = {**test_case["request"]["config"], **config} # 执行请求 start = time.monotonic() response = await self.agent.process( test_case["request"]["input"], context=test_case["request"]["context"], config=replay_config ) latency_ms = (time.monotonic() - start) * 1000 # 对比结果 comparison = self._compare_responses( expected=test_case["response"], actual=response ) return { "test_case_id": test_case.get("id"), "status": "pass" if comparison["passed"] else "fail", "comparison": comparison, "actual_response": response, "latency_ms": latency_ms, } def _compare_responses(self, expected: dict, actual: dict) -> dict: """对比预期和实际响应""" comparison = { "passed": True, "checks": [] } # 检查1：工具调用是否相同 expected_tools = [t["name"] for t in expected.get("tool_calls", [])] actual_tools = [t["name"] for t in actual.get("tool_calls", [])] tools_match = set(expected_tools) == set(actual_tools) comparison["checks"].append({ "name": "tool_calls_match", "passed": tools_match, "expected": expected_tools, "actual": actual_tools }) if not tools_match: comparison["passed"] = False # 检查2：响应质量是否相似（语义相似度） similarity = self._compute_similarity( expected["output"], actual["output"] ) quality_ok = similarity > 0.85 # 85%相似度阈值 comparison["checks"].append({ "name": "response_quality", "passed": quality_ok, "similarity": similarity, "threshold": 0.85 }) if not quality_ok: comparison["passed"] = False # 检查3：Token消耗是否在合理范围 token_ratio = actual.get("usage", {}).get("total_tokens", 0) / \ max(expected.get("tokens_used", {}).get("total_tokens", 1), 1) token_ok = 0.5 < token_ratio < 2.0 # Token消耗在0.5x-2x之间 comparison["checks"].append({ "name": "token_consumption", "passed": token_ok, "ratio": token_ratio, "threshold": "0.5-2.0" }) return comparison 快照测试 class SnapshotTesting: """快照测试——保存首次运行结果为快照，后续运行对比快照""" def __init__(self, snapshot_dir: str): self.snapshot_dir = snapshot_dir os.makedirs(snapshot_dir, exist_ok=True) async def run_with_snapshot( self, test_name: str, test_fn: callable, update_snapshot: bool = False ) -> dict: """运行快照测试""" snapshot_file = os.path.join( self.snapshot_dir, f"{test_name}.snapshot.json" ) # 执行测试 actual = await test_fn() if update_snapshot: # 更新快照 with open(snapshot_file, "w") as f: json.dump(actual, f, indent=2, ensure_ascii=False) return {"status": "snapshot_updated", "result": actual} # 对比快照 if not os.path.exists(snapshot_file): raise FileNotFoundError( f"Snapshot not found: {snapshot_file}. " f"Run with update_snapshot=True to create." ) with open(snapshot_file) as f: expected = json.load(f) diff = self._deep_diff(expected, actual) if diff: return { "status": "fail", "diff": diff, "expected": expected, "actual": actual } else: return {"status": "pass", "result": actual} def _deep_diff(self, expected, actual, path: str = "") -> list: """深度对比，返回差异列表""" diffs = [] if isinstance(expected, dict) and isinstance(actual, dict): for key in set(list(expected.keys()) + list(actual.keys())): new_path = f"{path}.{key}" if path else key if key not in actual: diffs.append(f"Missing in actual: {new_path}") elif key not in expected: diffs.append(f"Extra in actual: {new_path}") else: diffs.extend( self._deep_diff(expected[key], actual[key], new_path) ) elif isinstance(expected, list) and isinstance(actual, list): if len(expected) != len(actual): diffs.append( f"List length mismatch at {path}: " f"expected {len(expected)}, actual {len(actual)}" ) else: for i, (e, a) in enumerate(zip(expected, actual)): diffs.extend(self._deep_diff(e, a, f"{path}[{i}]")) else: # 标量对比（对LLM输出用模糊匹配） if self._is_llm_output(path): similarity = self._compute_similarity(str(expected), str(actual)) if similarity < 0.9: diffs.append( f"Semantic difference at {path}: " f"similarity={similarity:.2f}" ) else: if expected != actual: diffs.append( f"Value mismatch at {path}: " f"expected={expected}, actual={actual}" ) return diffs 回归测试框架 class RegressionTestSuite: """回归测试套件""" def __init__(self): self.test_cases = [] self.quality_thresholds = { "response_similarity": 0.85, "tool_call_accuracy": 0.95, "latency_increase_max": 1.2, # 延迟最多增加20% "token_increase_max": 1.3, # Token最多增加30% } def add_test_case(self, test_case: dict): """添加测试用例""" self.test_cases.append(test_case) async def run_regression_test(self, version: str) -> dict: """运行回归测试""" results = { "version": version, "total_cases": len(self.test_cases), "passed": 0, "failed": 0, "regressions": [], "improvements": [], } for test_case in self.test_cases: # 获取基线结果 baseline = await self._get_baseline(test_case["id"]) # 执行测试 actual = await self._execute_test(test_case) # 对比 comparison = self._compare_with_baseline(baseline, actual) if comparison["is_regression"]: results["failed"] += 1 results["regressions"].append({ "test_case": test_case["id"], "regression_type": comparison["regression_type"], "details": comparison["details"] }) elif comparison["is_improvement"]: results["improvements"].append({ "test_case": test_case["id"], "improvement_type": comparison["improvement_type"], }) else: results["passed"] += 1 return results def _compare_with_baseline(self, baseline: dict, actual: dict) -> dict: """与基线对比""" issues = [] # 质量回归 if actual["quality_score"] < baseline["quality_score"] * 0.95: issues.append({ "type": "quality_regression", "baseline": baseline["quality_score"], "actual": actual["quality_score"], }) # 延迟回归 latency_ratio = actual["latency_ms"] / baseline["latency_ms"] if latency_ratio > self.quality_thresholds["latency_increase_max"]: issues.append({ "type": "latency_regression", "baseline": baseline["latency_ms"], "actual": actual["latency_ms"], "ratio": latency_ratio, }) # Token回归 token_ratio = actual["tokens_used"] / baseline["tokens_used"] if token_ratio > self.quality_thresholds["token_increase_max"]: issues.append({ "type": "token_regression", "baseline": baseline["tokens_used"], "actual": actual["tokens_used"], "ratio": token_ratio, }) return { "is_regression": len(issues) > 0, "is_improvement": actual["quality_score"] > baseline["quality_score"] * 1.05, "issues": issues, } CI/CD集成 # .github/workflows/regression-test.yml name: Agent Regression Test on: pull_request: branches: [main] jobs: regression-test: runs-on: self-hosted steps: - uses: actions/checkout@v3 - name: Setup Test Environment run: | docker-compose -f docker-compose-test.yml up -d sleep 30 # 等待服务就绪 - name: Run Regression Tests id: regression run: | python -m pytest tests/regression/ \ --baseline=baseline.json \ --output=regression-report.json \ --junitxml=junit.xml - name: Check Quality Gate run: | python scripts/check_quality_gate.py \ --report=regression-report.json \ --max-regressions=5 \ --min-quality-score=0.85 - name: Upload Test Results if: always() uses: actions/upload-artifact@v3 with: name: regression-test-results path: | regression-report.json junit.xml - name: Comment PR if: always() uses: actions/github-script@v6 with: script: | const report = require('./regression-report.json'); const comment = ` ## Regression Test Results - Total: ${report.total_cases} - Passed: ${report.passed} - Failed: ${report.failed} - Regressions: ${report.regressions.length} ${report.regressions.length > 0 ? '⚠️ Regressions detected!' : '✅ No regressions'} `; github.rest.issues.createComment({ issue_number: context.issue.number, body: comment }); 总结 回放测试为Agent系统提供了一种实用的验证手段——通过录制生产流量并在新版本上回放，可以有效检测功能回归和质量下降。快照测试通过保存首次运行结果作为基准，简化了测试用例的创建。回归测试套件则系统性地验证新版本在质量、延迟、Token消耗等维度上是否出现退化。
...
