性能测试

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

引言 “你的 Agent 快吗？"——这个问题无法简单回答。Agent 的性能不是单一数字，而是延迟、吞吐量、成本、质量的四维空间。2026年，随着 AgentBench、SWE-bench 等标准化评测框架成熟，我们终于有了科学的 Agent 性能基准测试方法论。 一、四维性能模型 ┌──────────────────────────────────────────────┐ │ Agent 性能四维空间 │ ├──────────────┬───────────────────────────────┤ │ 延迟 (Latency) │ 首 Token 延迟 │ │ │ 完整响应延迟 │ │ │ P50/P95/P99 分布 │ ├──────────────┼───────────────────────────────┤ │ 吞吐 (Throughput)│ 请求/秒 │ │ │ 并发用户数 │ │ │ Token/秒 │ ├──────────────┼─────────────────────────────── │ 成本 (Cost) │ 单次请求成本 │ │ │ Token 效率 │ │ │ 月度总成本 │ ├──────────────┼───────────────────────────────┤ │ 质量 (Quality) │ 任务完成率 │ │ │ 输出准确率 │ │ │ 用户满意度 │ └──────────────┴───────────────────────────────┘ 关键洞察：四维之间存在 tradeoff - 提高质量通常增加延迟和成本 - 降低成本通常降低质量 - 提高吞吐通常增加延迟 二、延迟基准测试 2.1 延迟分解 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%，是首要优化目标 """ 2.2 延迟测试框架 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!" 八、基准测试 Checklist □ 四维基准测试覆盖（延迟/吞吐/成本/质量） □ 测试场景分级（简单/中等/复杂） □ 延迟测试包含首 Token 延迟 □ 吞吐测试找到最大并发数 □ 成本测试计算成本效率比 □ 质量测试使用标准化评测集 □ 持续基准测试（每周自动运行） □ 基线对比检测性能回归 □ SLA 定义明确（P95 延迟、错误率） □ 性能评分模型用于横向对比 结语 基准测试不是一次性的活动，而是持续的过程。Agent 的性能会随着 Prompt 修改、模型升级、工具变更而变化。建立持续的基准测试体系，让性能回归在 CI 阶段就被发现，而不是等到用户投诉。记住：没有测量就没有优化。在你开始优化 Agent 性能之前，先确保你能准确测量它。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...