引言

Agent系统的性能基准测试比传统Web应用复杂得多——响应延迟不仅取决于基础设施，还受LLM推理速度、工具调用延迟、Prompt长度等多重因素影响。没有经过充分基准测试的Agent系统，就像没有经过碰撞测试的自动驾驶汽车——上路后迟早会出事。

2026年，Agent性能测试已形成标准化的方法论。本文系统介绍如何对Agent系统进行全面、准确的性能基准测试。

性能测试维度

Agent性能测试矩阵
│
├── 吞吐量测试 (Throughput)
│   ├── 最大QPS
│   ├── 可持续QPS
│   └── QPS vs 延迟曲线
│
├── 延迟测试 (Latency)
│   ├── P50/P90/P99延迟
│   ├── 各阶段延迟分解
│   └── 长尾延迟分析
│
├── 并发测试 (Concurrency)
│   ├── 最大并发会话数
│   ├── 并发下质量保持
│   └── 资源竞争分析
│
└── 压力测试 (Stress)
    ├── 极限负载
    ├── 故障恢复时间
    └── 降级策略验证

测试环境搭建

# docker-compose-benchmark.yml
version: '3.8'
services:
  load-generator:
    image: agent/benchmark:latest
    environment:
      - TARGET_URL=http://agent-service:8080
      - CONCURRENT_USERS=100
      - TEST_DURATION=300s
    depends_on:
      - agent-service
      - llm-mock
  
  agent-service:
    image: agent/service:latest
    deploy:
      resources:
        limits:
          cpus: '4'
          memory: 8G
  
  llm-mock:
    image: agent/llm-mock:latest
    environment:
      - MOCK_LATENCY_MS=500  # 模拟LLM延迟
      - MOCK_ERROR_RATE=0.01
  
  prometheus:
    image: prom/prometheus:latest
  
  grafana:
    image: grafana/grafana:latest

吞吐量测试

import asyncio
import time
import statistics
from dataclasses import dataclass

@dataclass
class BenchmarkConfig:
    """基准测试配置"""
    target_qps: int
    duration_seconds: int
    concurrent_requests: int
    test_cases: list
    warmup_seconds: int = 30

@dataclass
class BenchmarkResult:
    """基准测试结果"""
    total_requests: int
    successful_requests: int
    failed_requests: int
    avg_latency_ms: float
    p50_latency_ms: float
    p90_latency_ms: float
    p99_latency_ms: float
    min_latency_ms: float
    max_latency_ms: float
    qps: float
    error_rate: float
    tokens_per_second: float

class ThroughputBenchmark:
    """吞吐量基准测试"""
    
    async def run(self, config: BenchmarkConfig) -> BenchmarkResult:
        """运行吞吐量测试"""
        
        # 预热
        await self._warmup(config.warmup_seconds)
        
        # 主测试
        latencies = []
        successes = 0
        failures = 0
        total_tokens = 0
        
        start_time = time.monotonic()
        end_time = start_time + config.duration_seconds
        
        # 创建并发请求
        tasks = []
        for i in range(config.concurrent_requests):
            task = asyncio.create_task(
                self._request_worker(
                    config, 
                    end_time,
                    latencies,
                    lambda s: nonlocal(successes) or successes++,
                    lambda f: nonlocal(failures) or failures++,
                    lambda t: nonlocal(total_tokens) or total_tokens += t
                )
            )
            tasks.append(task)
        
        # 等待完成
        await asyncio.gather(*tasks)
        
        actual_duration = time.monotonic() - start_time
        
        # 计算结果
        sorted_latencies = sorted(latencies)
        
        return BenchmarkResult(
            total_requests=len(latencies),
            successful_requests=successes,
            failed_requests=failures,
            avg_latency_ms=statistics.mean(latencies),
            p50_latency_ms=self._percentile(sorted_latencies, 0.5),
            p90_latency_ms=self._percentile(sorted_latencies, 0.9),
            p99_latency_ms=self._percentile(sorted_latencies, 0.99),
            min_latency_ms=min(latencies),
            max_latency_ms=max(latencies),
            qps=len(latencies) / actual_duration,
            error_rate=failures / len(latencies) if latencies else 0,
            tokens_per_second=total_tokens / actual_duration
        )
    
    async def _request_worker(
        self,
        config: BenchmarkConfig,
        end_time: float,
        latencies: list,
        on_success: callable,
        on_failure: callable,
        on_tokens: callable
    ):
        """请求工作线程"""
        while time.monotonic() < end_time:
            test_case = random.choice(config.test_cases)
            
            start = time.monotonic()
            try:
                response = await self.client.request(test_case["input"])
                latency = (time.monotonic() - start) * 1000
                
                latencies.append(latency)
                on_success()
                
                if "usage" in response:
                    on_tokens(response["usage"]["total_tokens"])
                    
            except Exception as e:
                latencies.append(30000)  # 超时记录为30s
                on_failure()

延迟分解测试

class LatencyBreakdownBenchmark:
    """延迟分解测试"""
    
    async def measure_latency_breakdown(
        self,
        session_id: str,
        user_input: str
    ) -> dict:
        """测量各阶段延迟"""
        
        breakdown = {
            "total_ms": 0,
            "stages": {}
        }
        
        # 使用链路追踪获取各阶段延迟
        trace = await self.tracer.get_trace_for_session(session_id)
        
        if trace:
            for span in trace.spans:
                stage_name = span.operation_name
                duration_ms = span.duration_ms
                
                breakdown["stages"][stage_name] = {
                    "duration_ms": duration_ms,
                    "percentage": 0,  # 稍后计算
                    "service": span.process.service_name,
                }
        
        # 计算百分比
        total = sum(s["duration_ms"] for s in breakdown["stages"].values())
        breakdown["total_ms"] = total
        
        for stage in breakdown["stages"].values():
            stage["percentage"] = stage["duration_ms"] / total if total > 0 else 0
        
        return breakdown
    
    def print_breakdown(self, breakdown: dict):
        """打印延迟分解"""
        print(f"\n{'='*60}")
        print(f"Total Latency: {breakdown['total_ms']:.1f}ms")
        print(f"{'='*60}")
        print(f"{'Stage':<30} {'Latency(ms)':<15} {'Percentage':<10}")
        print(f"{'-'*60}")
        
        for stage_name, data in sorted(
            breakdown["stages"].items(),
            key=lambda x: x[1]["duration_ms"],
            reverse=True
        ):
            print(
                f"{stage_name:<30} "
                f"{data['duration_ms']:<15.1f} "
                f"{data['percentage']*100:<10.1f}%"
            )

并发压力测试

class ConcurrencyBenchmark:
    """并发压力测试"""
    
    async def run_concurrency_test(
        self,
        max_concurrent: int,
        step: int = 10,
        hold_seconds: int = 60
    ) -> dict:
        """逐步增加并发数，测试系统极限"""
        
        results = []
        
        for concurrent in range(step, max_concurrent + 1, step):
            print(f"\nTesting with {concurrent} concurrent users...")
            
            config = BenchmarkConfig(
                target_qps=concurrent * 2,  # 每人2 QPS
                duration_seconds=hold_seconds,
                concurrent_requests=concurrent,
                test_cases=self._get_test_cases()
            )
            
            result = await self.benchmark.run(config)
            
            results.append({
                "concurrent_users": concurrent,
                "qps": result.qps,
                "avg_latency_ms": result.avg_latency_ms,
                "p99_latency_ms": result.p99_latency_ms,
                "error_rate": result.error_rate,
                "tokens_per_second": result.tokens_per_second,
            })
            
            # 如果错误率超过5%，停止测试
            if result.error_rate > 0.05:
                print(f"Stopping: error rate {result.error_rate:.1%} > 5%")
                break
        
        return {
            "test_results": results,
            "max_sustainable_concurrent": self._find_max_sustainable(results),
            "performance_curve": self._generate_curve(results),
        }
    
    def _find_max_sustainable(self, results: list) -> int:
        """找到可持续的最大并发数"""
        for r in results:
            if r["error_rate"] > 0.01 or r["p99_latency_ms"] > 5000:
                return r["concurrent_users"] - 10
        return results[-1]["concurrent_users"] if results else 0

测试结果解读

class BenchmarkReport:
    """基准测试报告生成器"""
    
    def generate_report(self, results: dict) -> str:
        """生成测试报告"""
        report = f"""
# Agent性能基准测试报告

## 测试概述
- 测试时间: {results["test_time"]}
- 测试版本: {results["version"]}
- 测试环境: {results["environment"]}

## 核心指标

### 吞吐量
- 最大QPS: {results["max_qps"]}
- 可持续QPS: {results["sustainable_qps"]}
- QPS vs 并发曲线: [见图表]

### 延迟
- P50延迟: {results["p50_latency_ms"]}ms
- P90延迟: {results["p90_latency_ms"]}ms
- P99延迟: {results["p99_latency_ms"]}ms
- 平均延迟: {results["avg_latency_ms"]}ms

### 并发
- 最大并发会话: {results["max_concurrent"]}
- 并发下平均质量评分: {results["quality_at_max_concurrent"]}

### 资源消耗
- 单请求平均Token消耗: {results["avg_tokens"]}
- GPU利用率: {results["gpu_utilization"]}
- CPU利用率: {results["cpu_utilization"]}

## 延迟分解
{self._format_latency_breakdown(results["latency_breakdown"])}

## 瓶颈分析
{results["bottleneck_analysis"]}

## 优化建议
{results["optimization_recommendations"]}
"""
        return report

持续基准测试

# .github/workflows/benchmark.yml
name: Performance Benchmark

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  benchmark:
    runs-on: self-hosted  # 需要稳定的硬件
    steps:
      - uses: actions/checkout@v3
      
      - name: Run Benchmark
        run: |
          docker-compose -f docker-compose-benchmark.yml up --abort-on-container-exit
      
      - name: Compare with Baseline
        run: |
          python scripts/compare_benchmark.py \
            --current results.json \
            --baseline baseline.json \
            --threshold 0.1  # 允许10%回归
      
      - name: Upload Results
        if: always()
        uses: actions/upload-artifact@v3
        with:
          name: benchmark-results
          path: results/

总结

Agent性能基准测试需要从吞吐量、延迟、并发、压力四个维度全面评估。延迟分解测试能够精准定位性能瓶颈——是LLM推理慢、工具调用慢还是向量检索慢。持续基准测试确保每次代码变更都不会引起性能回归。

核心原则：性能测试的价值不在于得到漂亮的数字，而在于建立性能基准线，并在每次变更后对比基准线。没有对比的性能数字是没有意义的。

加入讨论

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

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