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_factory=list)
raw_results: list[SingleResult] = field(default_factory=list)
def render_summary(self) -> str:
"""渲染文本摘要"""
return (
f"========== Benchmark Report ==========\\n"
f"Duration: {self.duration_sec:.1f}s\\n"
f"Concurrency: {self.concurrency}\\n"
f"Total Tasks: {self.total_tasks}\\n"
f"Success: {self.success_count} ({self.success_rate:.1%})\\n"
f"Failure: {self.failure_count}\\n"
f"Timeout: {self.timeout_count}\\n"
f"— Latency —\\n"
f"Avg: {self.latency_avg_ms:.1f}ms\\n"
f"P50: {self.latency_p50_ms:.1f}ms\\n"
f"P95: {self.latency_p95_ms:.1f}ms\\n"
f"P99: {self.latency_p99_ms:.1f}ms\\n"
f"Min/Max: {self.latency_min_ms:.1f}/{self.latency_max_ms:.1f}ms\\n"
f"— Throughput —\\n"
f"QPS: {self.qps:.1f}\\n"
f"— Token —\\n"
f"Total: {self.total_tokens}\\n"
f"Avg/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, record=False)
for t in warmup
], return_exceptions=True)
# 正式 benchmark
semaphore = asyncio.Semaphore(concurrency)
results: list[SingleResult] = []
async def bounded_task(task: BenchmarkTask) -> None:
async with semaphore:
result = await self._execute_single(task, record=True)
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),
timeout=task.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_id=task.task_id,
success=success,
latency_ms=latency_ms,
tokens_used=tokens,
status="success" if success else "failure",
)
except asyncio.TimeoutError:
latency_ms = task.timeout_ms
return SingleResult(
task_id=task.task_id,
success=False,
latency_ms=latency_ms,
status="timeout",
error="timeout",
)
except Exception as e:
latency_ms = (time.perf_counter() – t_start) * 1000
return SingleResult(
task_id=task.task_id,
success=False,
latency_ms=latency_ms,
status="failure",
error=str(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_time=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(start_time)),
end_time=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
duration_sec=exec_time,
concurrency=concurrency,
total_tasks=len(results),
latency_p50_ms=float(np.percentile(lat_array, 50)),
latency_p95_ms=float(np.percentile(lat_array, 95)),
latency_p99_ms=float(np.percentile(lat_array, 99)),
latency_avg_ms=float(np.mean(lat_array)) if lat_array.size > 0 else 0,
latency_min_ms=float(np.min(lat_array)) if lat_array.size > 0 else 0,
latency_max_ms=float(np.max(lat_array)) if lat_array.size > 0 else 0,
qps=len(success_results) / exec_time if exec_time > 0 else 0,
success_count=len(success_results),
failure_count=sum(1 for r in results if r.status == "failure"),
timeout_count=sum(1 for r in results if r.status == "timeout"),
success_rate=len(success_results) / len(results) if results else 0,
total_tokens=sum(all_tokens),
avg_tokens_per_task=sum(all_tokens) / len(all_tokens) if all_tokens else 0,
raw_latencies=latencies,
raw_results=results,
)
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": f'Response 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_fn=mock_agent,
validator_fn=mock_validator,
timeout_ms=10_000,
)
tasks = [
BenchmarkTask(
task_id=f"task_{i}",
input_data={"query": f"测试问题 {i}"},
)
for i in range(100)
]
report = await benchmark.run(tasks=tasks, concurrency=20)
print(report.render_summary())
if __name__ == "__main__":
logging.basicConfig(level=logging.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 不是为了出漂亮的报告，而是为了让你在每次修改后都能精确回答：这一次，我们是在优化还是在加熵？
