引言

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消耗等维度上是否出现退化。

核心原则：Agent测试的目标不是"完全相同"（因为LLM是非确定性的），而是"合理相似"。建立合理的相似度阈值，比追求完全确定性更有实际价值。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。