Agent 性能 benchmark定义吞吐量、延迟和成功率的标准化测试框架一、深度引言与场景痛点Agent 系统上线一段时间后团队内部会出现一种奇怪的分歧。后端工程师说平均响应时间 800ms挺好的产品经理说用户反馈慢得要死等 3 秒才出结果。两边都没错——后端看的是 P50中位数延迟产品经理感受到的是 P99长尾延迟。度量标准不统一讨论就变成了鸡同鸭讲。大到行业层面这个问题更加严重。A 公司的 Agent benchmark 宣称90% 的任务在 2 秒内完成B 公司说吞吐量达到 100 QPS。但你没法比较——因为不知道 A 的任务是啥回答简单问题还是做多步骤推理也不知道 B 的 QPS 是在什么硬件配置下跑出来的、延迟分布长什么样。标准化 benchmark 的目的不是排出谁第一谁第二而是让团队能回答三个基本问题我的 Agent 性能如何绝对值这次改动是变快还是变慢了对比值哪里是瓶颈诊断值要做到这些benchmark 需要定义清楚四个核心指标和一个标准化的执行框架。二、底层机制与原理深度剖析Agent 性能评估的核心指标只有四个但每个都需要精确定义吞吐量每秒完成的请求数QPS但这个定义需要加上限定条件——在给定的并发数下、在延迟不超过某个 SLO 上限的前提下。裸 QPS 没有意义一个用 1000 并发撑出来的 500 QPS每个请求等 2 秒和一个用 50 并发跑出来的 500 QPS每个请求等 0.1 秒用户体验完全不同。延迟端到端的响应时间分布。必须报告 P50、P95、P99 三个分位数而不是平均值因为 Agent 的延迟分布通常呈长尾状——少数复杂任务会把平均值拉高掩盖大多数简单任务的真实体验。成功率在超时窗口内返回了有效结果的比例。这里有效的定义很关键——不是 HTTP 200 就算成功而是响应内容通过了结构校验JSON schema 合法、包含必要字段、没有截断。Token 效率完成一个任务消耗的 Token 数包括输入和输出。这个指标直接和成本相关但很多 benchmark 忽略它——同样的成功率一个消耗 5000 Token 的 Agent 和一个消耗 1500 Token 的 Agent成本差 3 倍。Benchmark 的执行 pipleline 分四步。任务准备阶段最关键的是定义成功判定条件——这不是简单的状态码而是一个判定函数可以包含结构校验、LLM-as-Judge 评分、关键字段匹配。并发执行阶段asyncio.Semaphore精确控制并发度模拟真实负载。指标聚合阶段只统计在超时窗口内完成的请求超时请求单独记录为 timeouts。报告生成阶段把原始数据转化为可对比的图表和摘要。三、生产级代码实现import asyncio import json import logging import statistics import time from dataclasses import dataclass, field from typing import Any, Callable, Awaitable import numpy as np logger logging.getLogger(__name__) dataclass class BenchmarkTask: 单个 benchmark 任务 task_id: str input_data: dict[str, Any] expected_schema: dict[str, Any] | None None # JSON schema 校验 timeout_ms: int 30_000 dataclass class SingleResult: 单次执行结果 task_id: str success: bool latency_ms: float tokens_used: int 0 error: str status: str success # success / failure / timeout dataclass class BenchmarkReport: Benchmark 报告 start_time: str end_time: str duration_sec: float concurrency: int total_tasks: int # 延迟指标 latency_p50_ms: float 0 latency_p95_ms: float 0 latency_p99_ms: float 0 latency_avg_ms: float 0 latency_min_ms: float 0 latency_max_ms: float 0 # 吞吐与成功率 qps: float 0 success_count: int 0 failure_count: int 0 timeout_count: int 0 success_rate: float 0 # Token total_tokens: int 0 avg_tokens_per_task: float 0 # 原始数据用于绘图 raw_latencies: list[float] field(default_factorylist) raw_results: list[SingleResult] field(default_factorylist) def render_summary(self) - str: 渲染文本摘要 return ( f Benchmark Report \n fDuration: {self.duration_sec:.1f}s\n fConcurrency: {self.concurrency}\n fTotal Tasks: {self.total_tasks}\n fSuccess: {self.success_count} ({self.success_rate:.1%})\n fFailure: {self.failure_count}\n fTimeout: {self.timeout_count}\n f--- Latency ---\n fAvg: {self.latency_avg_ms:.1f}ms\n fP50: {self.latency_p50_ms:.1f}ms\n fP95: {self.latency_p95_ms:.1f}ms\n fP99: {self.latency_p99_ms:.1f}ms\n fMin/Max: {self.latency_min_ms:.1f}/{self.latency_max_ms:.1f}ms\n f--- Throughput ---\n fQPS: {self.qps:.1f}\n f--- Token ---\n fTotal: {self.total_tokens}\n fAvg/Task: {self.avg_tokens_per_task:.1f}\n f ) class AgentBenchmark: Agent 性能 Benchmark 标准化框架 def __init__( self, agent_fn: Callable[[dict[str, Any]], Awaitable[dict[str, Any]]], validator_fn: Callable[[dict[str, Any], dict[str, Any] | None], bool], timeout_ms: int 30_000, ) - None: agent_fn: 异步 Agent 调用函数(input_data) - response_dict validator_fn: 结果验证函数(response_dict, expected_schema) - bool self._agent_fn agent_fn self._validator_fn validator_fn self._default_timeout timeout_ms async def run( self, tasks: list[BenchmarkTask], concurrency: int 10, warmup_tasks: int 3, ) - BenchmarkReport: 运行 benchmark start_time time.time() # 预热让模型/连接池进入稳定状态 if warmup_tasks 0 and tasks: warmup tasks[:min(warmup_tasks, len(tasks))] await asyncio.gather(*[ self._execute_single(t, recordFalse) for t in warmup ], return_exceptionsTrue) # 正式 benchmark semaphore asyncio.Semaphore(concurrency) results: list[SingleResult] [] async def bounded_task(task: BenchmarkTask) - None: async with semaphore: result await self._execute_single(task, recordTrue) results.append(result) exec_start time.perf_counter() await asyncio.gather(*[bounded_task(t) for t in tasks]) exec_end time.perf_counter() return self._build_report( results, concurrency, exec_end - exec_start, start_time, ) async def _execute_single( self, task: BenchmarkTask, record: bool True ) - SingleResult: 执行单个任务并计时 t_start time.perf_counter() try: response await asyncio.wait_for( self._agent_fn(task.input_data), timeouttask.timeout_ms / 1000.0, ) latency_ms (time.perf_counter() - t_start) * 1000 success self._validator_fn(response, task.expected_schema) tokens response.get(usage, {}).get(total_tokens, 0) if record: logger.debug( Task %s: %s (%.0fms, %d tokens), task.task_id, success if success else validation_failed, latency_ms, tokens, ) return SingleResult( task_idtask.task_id, successsuccess, latency_mslatency_ms, tokens_usedtokens, statussuccess if success else failure, ) except asyncio.TimeoutError: latency_ms task.timeout_ms return SingleResult( task_idtask.task_id, successFalse, latency_mslatency_ms, statustimeout, errortimeout, ) except Exception as e: latency_ms (time.perf_counter() - t_start) * 1000 return SingleResult( task_idtask.task_id, successFalse, latency_mslatency_ms, statusfailure, errorstr(e), ) def _build_report( self, results: list[SingleResult], concurrency: int, exec_time: float, start_time: float, ) - BenchmarkReport: 从原始结果构建报告 success_results [r for r in results if r.status success] latencies [r.latency_ms for r in success_results] all_tokens [r.tokens_used for r in success_results] lat_array np.array(latencies) if latencies else np.array([0]) return BenchmarkReport( start_timetime.strftime(%Y-%m-%dT%H:%M:%SZ, time.gmtime(start_time)), end_timetime.strftime(%Y-%m-%dT%H:%M:%SZ, time.gmtime()), duration_secexec_time, concurrencyconcurrency, total_taskslen(results), latency_p50_msfloat(np.percentile(lat_array, 50)), latency_p95_msfloat(np.percentile(lat_array, 95)), latency_p99_msfloat(np.percentile(lat_array, 99)), latency_avg_msfloat(np.mean(lat_array)) if lat_array.size 0 else 0, latency_min_msfloat(np.min(lat_array)) if lat_array.size 0 else 0, latency_max_msfloat(np.max(lat_array)) if lat_array.size 0 else 0, qpslen(success_results) / exec_time if exec_time 0 else 0, success_countlen(success_results), failure_countsum(1 for r in results if r.status failure), timeout_countsum(1 for r in results if r.status timeout), success_ratelen(success_results) / len(results) if results else 0, total_tokenssum(all_tokens), avg_tokens_per_tasksum(all_tokens) / len(all_tokens) if all_tokens else 0, raw_latencieslatencies, raw_resultsresults, ) async def main() - None: 使用示例 # 模拟 Agent 函数 async def mock_agent(input_data: dict[str, Any]) - dict[str, Any]: await asyncio.sleep(0.1 0.3 * (hash(input_data.get(query, )) % 3) / 3) return { answer: fResponse to: {input_data.get(query, )}, usage: {total_tokens: 150 hash(input_data.get(query, )) % 100}, } def mock_validator(response: dict[str, Any], _schema: Any) - bool: return answer in response and len(response[answer]) 0 benchmark AgentBenchmark( agent_fnmock_agent, validator_fnmock_validator, timeout_ms10_000, ) tasks [ BenchmarkTask( task_idftask_{i}, input_data{query: f测试问题 {i}}, ) for i in range(100) ] report await benchmark.run(taskstasks, concurrency20) print(report.render_summary()) if __name__ __main__: logging.basicConfig(levellogging.WARNING) asyncio.run(main())框架设计了预热阶段——前几个请求不计入统计目的是让模型连接池、JIT 编译等冷启动效应不影响真实性能数据。超时时间按每个任务独立设置不是全局超时——复杂任务可能需要更长超时。_validator_fn作为外部注入函数让不同的 benchmark 可以定义不同的成功标准。报告里的 P95/P99 用 numpy 的分位数计算比手动排序更精确。四、边界分析与架构权衡Benchmark 最大的问题是实验室性能和线上性能的巨大落差。用 100 个预定义的测试用例跑 benchmark延迟分布可能非常漂亮上线后面对真实用户的千奇百怪的输入延迟立刻飙到三倍。差距来自两个原因一是测试用例不具代表性覆盖不到长尾查询二是负载模型不真实固定的并发数 vs 线上突发流量。解决方向一是用线上真实流量的重放recording replay替代手工编写的测试用例。从生产日志中抽取真实用户请求脱敏后作为 test case。二是用波动的并发模型——不是固定 10 并发跑到底而是模拟峰谷变化如 0-100 秒 50 并发100-150 秒 200 并发150-200 秒 10 并发观察 Agent 在压力突变下的行为是否堆积、是否熔断、恢复时间。Token 效率指标容易被忽视但对成本敏感的场景至关重要。同一个查询订单任务优化前的 Prompt 消耗 800 Token 做上下文优化后只用 300 Token——吞吐量和延迟可能没变化但单次请求成本降了 60%。benchmark 应该把 Token 效率作为和延迟同等重要的指标来报告。五、总结标准化的 Agent benchmark 框架围绕四个核心指标展开吞吐量有并发约束的 QPS、延迟P50/P95/P99 分位数分布、成功率含结构校验的有效完成率、Token 效率单任务 Token 消耗。执行流程分任务准备、预热执行、指标聚合、报告生成四步。生产级部署的关键在两点用线上真实流量替代手工测试用例用波动负载模型替代固定并发。benchmark 不是为了出漂亮的报告而是为了让你在每次修改后都能精确回答这一次我们是在优化还是在加熵