Agent 性能基准测试方法论 2026

┌──────────────────────────────────────────────┐
│            Agent 性能四维空间                  │
├──────────────┬───────────────────────────────┤
│  延迟 (Latency)  │ 首 Token 延迟              │
│                  │ 完整响应延迟                │
│                  │ P50/P95/P99 分布            │
├──────────────┼───────────────────────────────┤
│  吞吐 (Throughput)│ 请求/秒                    │
│                  │ 并发用户数                  │
│                  │ Token/秒                    │
├──────────────┼───────────────────────────────
│  成本 (Cost)     │ 单次请求成本                │
│                  │ Token 效率                  │
│                  │ 月度总成本                  │
├──────────────┼───────────────────────────────┤
│  质量 (Quality)  │ 任务完成率                  │
│                  │ 输出准确率                  │
│                  │ 用户满意度                  │
└──────────────┴───────────────────────────────┘

关键洞察：四维之间存在 tradeoff
- 提高质量通常增加延迟和成本
- 降低成本通常降低质量
- 提高吞吐通常增加延迟

class LatencyBreakdown:
    """Agent 延迟分解模型"""
    
    COMPONENTS = {
        "network_ingress": "API Gateway 到达延迟",
        "auth": "认证授权延迟",
        "queue": "排队等待延迟",
        "context_preparation": "上下文准备（历史压缩等）",
        "llm_first_token": "LLM 首 Token 延迟",
        "llm_streaming": "LLM 流式输出延迟",
        "tool_execution": "工具执行延迟",
        "tool_overhead": "工具调度开销",
        "state_persistence": "状态持久化延迟",
        "network_egress": "响应返回延迟",
    }
    
    @dataclass
    class LatencyMeasurement:
        component: str
        duration_ms: float
        percentage: float  # 占总延迟百分比
    
    def analyze(self, trace: list[dict]) -> list[LatencyMeasurement]:
        """从执行 trace 分析延迟分布"""
        total = sum(t["duration_ms"] for t in trace)
        
        return [
            LatencyMeasurement(
                component=t["component"],
                duration_ms=t["duration_ms"],
                percentage=t["duration_ms"] / total * 100
            )
            for t in sorted(trace, key=lambda x: -x["duration_ms"])
        ]

# 典型 Agent 延迟分布
TYPICAL_BREAKDOWN = """
组件                    延迟(ms)    占比
─────────────────────────────────────────
llm_first_token         1200      40%
llm_streaming           800       27%
tool_execution          450       15%
context_preparation     200       7%
queue                   150       5%
state_persistence       100       3%
auth                    50        2%
network                 40        1%
─────────────────────────────────────────
总计                    2990      100%

优化优先级：LLM 延迟占 67%，是首要优化目标
"""

class AgentLatencyBenchmark:
    """Agent 延迟基准测试"""
    
    TEST_SCENARIOS = [
        BenchmarkScenario(
            name="simple_qa",
            description="简单问答（无工具）",
            query="What is 2+2?",
            expected_max_latency_ms=3000,
            tools=[],
        ),
        BenchmarkScenario(
            name="single_tool",
            description="单工具调用",
            query="Search for latest AI news",
            expected_max_latency_ms=8000,
            tools=["web_search"],
        ),
        BenchmarkScenario(
            name="multi_tool",
            description="多工具串联（3步）",
            query="Research and summarize quantum computing breakthroughs in 2026",
            expected_max_latency_ms=30000,
            tools=["web_search", "summarizer", "write_file"],
        ),
        BenchmarkScenario(
            name="complex_reasoning",
            description="复杂推理（5+步）",
            query="Analyze the competitive landscape of AI chip market",
            expected_max_latency_ms=60000,
            tools=["web_search", "data_analyzer", "chart_gen", "write_file"],
        ),
    ]
    
    async def run_benchmark(
        self,
        agent: Agent,
        scenarios: list[BenchmarkScenario] | None = None,
        iterations: int = 100
    ) -> BenchmarkReport:
        
        scenarios = scenarios or self.TEST_SCENARIOS
        results = {}
        
        for scenario in scenarios:
            latencies = []
            first_token_latencies = []
            
            for _ in range(iterations):
                start = time.time()
                first_token_time = None
                
                async for chunk in agent.run_stream(scenario.query):
                    if first_token_time is None:
                        first_token_time = time.time()
                
                end = time.time()
                
                total_latency = (end - start) * 1000
                first_token_latency = (first_token_time - start) * 1000
                
                latencies.append(total_latency)
                first_token_latencies.append(first_token_latency)
            
            results[scenario.name] = LatencyResult(
                scenario=scenario.name,
                p50=np.percentile(latencies, 50),
                p95=np.percentile(latencies, 95),
                p99=np.percentile(latencies, 99),
                mean=np.mean(latencies),
                std=np.std(latencies),
                first_token_p50=np.percentile(first_token_latencies, 50),
                first_token_p95=np.percentile(first_token_latencies, 95),
                passed_p95=np.percentile(latencies, 95) < scenario.expected_max_latency_ms,
            )
        
        return BenchmarkReport(results=results)

class ThroughputBenchmark:
    """吞吐量基准测试"""
    
    async def test_concurrent_users(
        self,
        agent: Agent,
        query: str,
        concurrent_users: list[int] = [1, 10, 50, 100, 200, 500]
    ) -> list[ThroughputResult]:
        
        results = []
        
        for n_users in concurrent_users:
            print(f"Testing with {n_users} concurrent users...")
            
            # 创建并发请求
            tasks = [
                self._timed_request(agent, query, user_id=i)
                for i in range(n_users)
            ]
            
            start = time.time()
            responses = await asyncio.gather(*tasks, return_exceptions=True)
            total_time = time.time() - start
            
            # 统计
            success_count = sum(1 for r in responses if not isinstance(r, Exception))
            error_count = sum(1 for r in responses if isinstance(r, Exception))
            
            result = ThroughputResult(
                concurrent_users=n_users,
                total_requests=n_users,
                successful_requests=success_count,
                failed_requests=error_count,
                total_time_s=total_time,
                requests_per_second=success_count / total_time,
                avg_latency_ms=np.mean([
                    r["latency_ms"] for r in responses
                    if isinstance(r, dict)
                ]),
                p95_latency_ms=np.percentile([
                    r["latency_ms"] for r in responses
                    if isinstance(r, dict)
                ], 95),
                error_rate=error_count / n_users,
            )
            
            results.append(result)
            
            # 如果错误率 > 20%，停止加压
            if result.error_rate > 0.2:
                print(f"Error rate {result.error_rate:.0%} > 20%, stopping")
                break
        
        return results
    
    async def find_max_throughput(
        self,
        agent: Agent,
        query: str,
        target_latency_p95_ms: float = 10000,
        target_error_rate: float = 0.01
    ) -> int:
        """找到满足 SLA 的最大并发数"""
        
        # 二分搜索
        low, high = 1, 1000
        best = 1
        
        while low <= high:
            mid = (low + high) // 2
            results = await self.test_concurrent_users(
                agent, query, [mid]
            )
            
            result = results[0]
            
            if (result.p95_latency_ms <= target_latency_p95_ms and
                result.error_rate <= target_error_rate):
                best = mid
                low = mid + 1
            else:
                high = mid - 1
        
        return best

class CostEfficiencyBenchmark:
    """成本效率基准测试"""
    
    async def benchmark(
        self,
        agent: Agent,
        test_cases: list[TestCase]
    ) -> CostReport:
        
        results = []
        
        for case in test_cases:
            start_cost = agent.total_cost
            
            response = await agent.run(case.input)
            
            cost = agent.total_cost - start_cost
            
            # 评估输出质量
            quality = await self.judge.evaluate(
                case.input, response, case.criteria
            )
            
            results.append(CostResult(
                test_id=case.id,
                input_tokens=agent.last_input_tokens,
                output_tokens=agent.last_output_tokens,
                total_tokens=agent.last_total_tokens,
                cost_usd=cost,
                quality_score=quality.score,
                cost_per_quality=cost / max(quality.score, 0.01),  # 成本效率比
                iterations=agent.iteration_count,
            ))
        
        return CostReport(
            results=results,
            avg_cost=np.mean([r.cost_usd for r in results]),
            avg_quality=np.mean([r.quality_score for r in results]),
            avg_cost_per_quality=np.mean([r.cost_per_quality for r in results]),
            total_cost=sum(r.cost_usd for r in results),
            cost_distribution=self._analyze_distribution(
                [r.cost_usd for r in results]
            ),
        )
    
    def compare_models(
        self,
        models: list[str],
        test_cases: list[TestCase]
    ) -> ComparisonReport:
        """对比不同模型的成本效率"""
        
        model_results = {}
        for model in models:
            agent = Agent(llm=LLM(model=model))
            report = self.benchmark(agent, test_cases)
            model_results[model] = report
        
        # 生成对比表
        return ComparisonReport(
            models=model_results,
            best_cost=min(model_results.items(), key=lambda x: x[1].avg_cost),
            best_quality=max(model_results.items(), key=lambda x: x[1].avg_quality),
            best_efficiency=min(
                model_results.items(),
                key=lambda x: x[1].avg_cost_per_quality
            ),
        )

class QualityBenchmark:
    """Agent 输出质量基准测试"""
    
    BENCHMARK_SUITES = {
        "reasoning": ReasoningSuite(),      # 推理能力
        "coding": CodingSuite(),            # 代码生成
        "tool_use": ToolUseSuite(),         # 工具使用
        "safety": SafetySuite(),            # 安全性
        "instruction_follow": InstructionSuite(),  # 指令遵循
        "multilingual": MultilingualSuite(),  # 多语言
    }
    
    async def run_full_benchmark(
        self,
        agent: Agent,
        suites: list[str] | None = None
    ) -> FullBenchmarkReport:
        
        suites = suites or list(self.BENCHMARK_SUITES.keys())
        results = {}
        
        for suite_name in suites:
            suite = self.BENCHMARK_SUITES[suite_name]
            suite_results = []
            
            for test_case in suite.get_cases():
                # 运行 Agent
                output = await agent.run(test_case.input)
                
                # 自动化评估
                auto_score = await suite.evaluate(
                    test_case, output
                )
                
                # LLM-as-Judge 评估
                judge_score = await self.judge.evaluate(
                    test_case.input, output, test_case.criteria
                )
                
                # 统计
                suite_results.append(QualityResult(
                    test_id=test_case.id,
                    category=test_case.category,
                    output_preview=output[:200],
                    auto_score=auto_score,
                    judge_score=judge_score.score,
                    passed=judge_score.score >= test_case.min_score,
                    duration_ms=test_case.duration_ms,
                ))
            
            results[suite_name] = SuiteResult(
                total=len(suite_results),
                passed=sum(1 for r in suite_results if r.passed),
                pass_rate=sum(1 for r in suite_results if r.passed) / len(suite_results),
                avg_score=np.mean([r.judge_score for r in suite_results]),
                results=suite_results,
            )
        
        return FullBenchmarkReport(
            suites=results,
            overall_pass_rate=np.mean([
                r.pass_rate for r in results.values()
            ]),
            timestamp=datetime.now(),
        )

class AgentPerformanceScore:
    """Agent 综合性能评分"""
    
    def calculate(
        self,
        latency: LatencyResult,
        throughput: ThroughputResult,
        cost: CostReport,
        quality: FullBenchmarkReport
    ) -> PerformanceScore:
        
        # 归一化评分（0-100）
        
        # 延迟分（越低越好，基准 30s = 0分, 1s = 100分）
        latency_score = max(0, min(100, 
            100 * (30 - latency.p95 / 1000) / 29
        ))
        
        # 吞吐分（越高越好，基准 1 RPS = 0分, 100 RPS = 100分）
        throughput_score = max(0, min(100,
            100 * throughput.requests_per_second / 100
        ))
        
        # 成本分（越低越好，基准 $0.1/请求 = 0分, $0.001/请求 = 100分）
        cost_score = max(0, min(100,
            100 * (0.1 - cost.avg_cost) / 0.099
        ))
        
        # 质量分（越高越好）
        quality_score = quality.overall_pass_rate * 100
        
        # 加权综合
        weights = {
            "latency": 0.20,
            "throughput": 0.15,
            "cost": 0.25,
            "quality": 0.40,
        }
        
        overall = sum(score * weights[key] for key, score in [
            ("latency", latency_score),
            ("throughput", throughput_score),
            ("cost", cost_score),
            ("quality", quality_score),
        ])
        
        return PerformanceScore(
            overall=overall,
            latency=latency_score,
            throughput=throughput_score,
            cost=cost_score,
            quality=quality_score,
            grade=self._grade(overall),
            tradeoffs=self._analyze_tradeoffs(
                latency_score, throughput_score,
                cost_score, quality_score
            ),
        )
    
    def _grade(self, score: float) -> str:
        if score >= 90: return "A+"
        if score >= 80: return "A"
        if score >= 70: return "B"
        if score >= 60: return "C"
        if score >= 50: return "D"
        return "F"

# .github/workflows/agent-benchmark.yml
name: Agent Performance Benchmark

on:
  schedule:
    - cron: "0 2 * * 1"  # 每周一凌晨2点
  workflow_dispatch:      # 手动触发

jobs:
  benchmark:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Run latency benchmark
        run: python benchmarks/latency_benchmark.py --output results/latency.json
      
      - name: Run throughput benchmark
        run: python benchmarks/throughput_benchmark.py --output results/throughput.json
      
      - name: Run cost benchmark
        run: python benchmarks/cost_benchmark.py --output results/cost.json
      
      - name: Run quality benchmark
        run: python benchmarks/quality_benchmark.py --output results/quality.json
      
      - name: Generate report
        run: python benchmarks/generate_report.py --input results/ --output report.md
      
      - name: Compare with baseline
        run: |
          python benchmarks/compare_baseline.py \
            --current results/ \
            --baseline benchmarks/baseline/ \
            --threshold-latency 10 \
            --threshold-cost 5 \
            --threshold-quality 2
      
      - name: Upload results
        uses: actions/upload-artifact@v4
        with:
          name: benchmark-results
          path: results/
      
      - name: Notify on regression
        if: failure()
        uses: ./.github/actions/slack-notify
        with:
          message: "Agent performance regression detected!"

□ 四维基准测试覆盖（延迟/吞吐/成本/质量）
□ 测试场景分级（简单/中等/复杂）
□ 延迟测试包含首 Token 延迟
□ 吞吐测试找到最大并发数
□ 成本测试计算成本效率比
□ 质量测试使用标准化评测集
□ 持续基准测试（每周自动运行）
□ 基线对比检测性能回归
□ SLA 定义明确（P95 延迟、错误率）
□ 性能评分模型用于横向对比