引言 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推理慢、工具调用慢还是向量检索慢。持续基准测试确保每次代码变更都不会引起性能回归。
...
