推理性能基准测试方法
推理性能基准测试是评估和优化推理系统的关键环节。本文介绍系统化的推理性能评测方法,涵盖延迟、吞吐量、首Token延迟等核心指标以及标准化的测试流程。
核心指标定义
| 指标 | 定义 | 单位 | 重要性 |
|---|---|---|---|
| TTFT | 首 Token 延迟 (Time To First Token) | ms | 交互体验 |
| TPOT | 每 Token 输出延迟 (Time Per Output Token) | ms | 生成速度 |
| E2E Latency | 端到端延迟 | ms | 总体响应 |
| Throughput | 吞吐量 | tok/s | 系统容量 |
| Concurrency | 最大并发请求数 | 个 | 负载能力 |
标准测试流程
python
# 推理基准测试框架
import time
import statistics
from dataclasses import dataclass
@dataclass
class BenchmarkResult:
ttft_ms: list # 首 Token 延迟
tpot_ms: list # 每 Token 延迟
e2e_ms: list # 端到端延迟
output_tokens: list # 输出 Token 数
def run_benchmark(engine, prompts, max_tokens=256, num_rounds=3):
"""执行推理基准测试"""
results = BenchmarkResult([], [], [], [])
for _ in range(num_rounds):
for prompt in prompts:
start = time.perf_counter()
first_token_time = None
for token in engine.stream_generate(prompt, max_tokens=max_tokens):
if first_token_time is None:
first_token_time = time.perf_counter()
results.ttft_ms.append(
(first_token_time - start) * 1000
)
end = time.perf_counter()
results.e2e_ms.append((end - start) * 1000)
# 计算统计量
print(f"TTFT P50: {statistics.median(results.ttft_ms):.1f}ms")
print(f"TTFT P99: {sorted(results.ttft_ms)[int(len(results.ttft_ms)*0.99)]:.1f}ms")
print(f"Throughput: {sum(results.output_tokens) / sum(results.e2e_ms) * 1000:.1f} tok/s")
return results测试数据集设计
基准测试需要多样化的输入:
- 短输入短输出:10-50 token 输入,50-100 token 输出
- 长输入短输出:2000-4000 token 输入,50-100 token 输出
- 短输入长输出:10-50 token 输入,500-1000 token 输出
- 长输入长输出:2000-4000 token 输入,500-1000 token 输出
测试建议
每种输入/输出组合至少运行 50 次取统计值。避免仅报告平均值,应包含 P50、P90、P99 分位数。
压力测试
python
# 并发压力测试
import asyncio
import aiohttp
async def stress_test(url, num_concurrent=100, total_requests=1000):
"""并发压力测试"""
semaphore = asyncio.Semaphore(num_concurrent)
latencies = []
async def single_request(session):
async with semaphore:
start = time.time()
async with session.post(url, json={
"prompt": "测试请求",
"max_tokens": 128,
}) as resp:
await resp.json()
latencies.append((time.time() - start) * 1000)
async with aiohttp.ClientSession() as session:
tasks = [single_request(session) for _ in range(total_requests)]
await asyncio.gather(*tasks)
print(f"并发数: {num_concurrent}")
print(f"完成请求: {len(latencies)}")
print(f"成功率: {len(latencies)/total_requests*100:.1f}%")
print(f"P50 延迟: {statistics.median(latencies):.1f}ms")
print(f"P99 延迟: {sorted(latencies)[int(len(latencies)*0.99)]:.1f}ms")报告规范
基准测试报告必须注明:GPU 型号、驱动版本、CUDA 版本、模型版本、量化方式、批处理配置、输入/输出长度分布。缺少环境信息的基准数据没有参考价值。