推理评测:数据分层和延迟口径要同时公开
压测报告写的 15ms,线上实际 P99 飙到了 120ms
评测推理服务时,输入分层和延迟口径必须一起保留。本文的场景用于说明为什么不能只报一个平均值;实际结论要在锁定的模型、硬件、并发和脚本条件下复测。
换到预发环境并接入模拟或脱敏流量后,P99 往往会显著高于离线结果,吞吐也可能低于预期。先把输入长度、排队时间和显存使用拆开看,才能确认瓶颈究竟在哪里。
这类落差绝大多数源于基准测试 (Benchmark) 设计的失真。测试人员习惯于使用随机生成的固定形状 Tensor 作为输入,却忽略了真实业务场景中输入的变长 Token 特征、动态 Padding 开销,以及在高并发请求下 GPU 显存带宽争用造成的 Queueing Delay(队列等待延迟)。
+-----------------------------------------------------------------------+ | 失真的离线测试 (Synthetic Benchmark) | | - 纯随机 Tensor (Fixed Shape: 1x3x224x224 或 Fixed SeqLen: 128) | | - 单线程串行循环调用 (No Concurrency Queue) | | - 仅统计 Mean Latency (忽略 P99 / P999) | +-----------------------------------------------------------------------+ vs +-----------------------------------------------------------------------+ | 真实生产场景 (Production Baseline) | | - 真实业务 Token 长度分布 (长尾数据占比 15% 以上) | | - 动态 Batching + 并发并发 Queue 等待 | | - P99 / P999 延迟、GPU 显存带宽饱和度、首包时间 (TTFT) | +-----------------------------------------------------------------------+真实基准测试的数据准备:基于真实分布切片与真实 Token 长度分桶
要让压测数据具备指导生产的真实价值,首先必须改造基准测试数据集的构造逻辑。绝对不能全盘使用torch.randn生成的数据,而必须从历史线上 Request 日志中抽取真实的输入样本。
针对 NLP 或大语言模型,需要根据序列长度(Sequence Length)进行分桶(Bucketing)。例如,将测试集划分为 0-128、128-512、512-2048 等不同区间的长短文本混编流。针对 CV 类模型,则需要还原线上图像的真实尺寸比例与色彩通道分布,避免 TensorRT 优化器在优化 Dynamic Shape 策略时被错误的 Profile 数据误导。
推理性能压测流水线与指标收集链路
构建一条包含预热(Warmup)、并发梯度探测和百分位数(Percentile)统计的自动化测试流水线。
自动化 Benchmark 工具脚本:带 Warmup 与 Percentile 统计
下面是一段用 Python 编写的生产级推理性能 Benchmark 工具脚本。它集成了 CUDA 预热、精确到微秒级的计时器、并发请求模拟以及 P50/P90/P99 百分位数的计算。
import time import math import numpy as np import torch import torch.nn as nn from typing import List, Dict, Any # 模拟一个实际的 Transformer 编码器模型 class MockEncoderModel(nn.Module): def __init__(self, vocab_size: int = 30000, hidden_dim: int = 768, num_layers: int = 6): super().__init__() self.embedding = nn.Embedding(vocab_size, hidden_dim) encoder_layer = nn.TransformerEncoderLayer(d_model=hidden_dim, nhead=8, batch_first=True) self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers) self.fc = nn.Linear(hidden_dim, 2) def forward(self, input_ids): x = self.embedding(input_ids) x = self.encoder(x) return self.fc(x.mean(dim=1)) # 性能压测核心类 class InferenceBenchmarker: def __init__(self, model: nn.Module, device: str = "cuda"): self.device = torch.device(device if torch.cuda.is_available() else "cpu") self.model = model.to(self.device) self.model.eval() def warmup(self, sample_input: torch.Tensor, num_warmup: int = 50): """CUDA 预热:促使 GPU 提升到最高工作主频,并完成 CUDNN/TensorRT 算子初始化""" print(f"正在执行 GPU 预热 ({num_warmup} 次)...") sample_input = sample_input.to(self.device) with torch.no_grad(): for _ in range(num_warmup): _ = self.model(sample_input) if self.device.type == "cuda": torch.cuda.synchronize() print("预热完成。") def run_benchmark( self, test_buckets: Dict[str, torch.Tensor], iterations_per_bucket: int = 200 ) -> Dict[str, Any]: results = {} for bucket_name, input_data in test_buckets.items(): input_data = input_data.to(self.device) latencies_ms: List[float] = [] print(f"开始压测分桶 [{bucket_name}],Shape: {list(input_data.shape)}...") with torch.no_grad(): for _ in range(iterations_per_bucket): if self.device.type == "cuda": torch.cuda.synchronize() start_time = time.perf_counter() _ = self.model(input_data) if self.device.type == "cuda": torch.cuda.synchronize() elapsed_ms = (time.perf_counter() - start_time) * 1000.0 latencies_ms.append(elapsed_ms) # 计算百分位数与统计量 latencies_ms.sort() p50 = np.percentile(latencies_ms, 50) p90 = np.percentile(latencies_ms, 90) p99 = np.percentile(latencies_ms, 99) mean_lat = np.mean(latencies_ms) std_lat = np.std(latencies_ms) qps = 1000.0 / mean_lat * input_data.shape[0] results[bucket_name] = { "batch_size": input_data.shape[0], "seq_len": input_data.shape[1] if input_data.ndim > 1 else 1, "mean_ms": round(mean_lat, 2), "p50_ms": round(p50, 2), "p90_ms": round(p90, 2), "p99_ms": round(p99, 2), "std_ms": round(std_lat, 2), "estimated_qps": round(qps, 2) } return results # 执行基准测试的主流程 if __name__ == "__main__": model = MockEncoderModel() benchmarker = InferenceBenchmarker(model=model, device="cuda" if torch.cuda.is_available() else "cpu") # 1. 构造代表真实分布的测试分桶数据 # 模拟:短文本、中等文本、长文本 batch_size = 4 test_buckets = { "Short_Text_Seq64": torch.randint(0, 30000, (batch_size, 64)), "Medium_Text_Seq256": torch.randint(0, 30000, (batch_size, 256)), "Long_Text_Seq512": torch.randint(0, 30000, (batch_size, 512)), } # 2. 预热 GPU benchmarker.warmup(sample_input=test_buckets["Short_Text_Seq64"], num_warmup=30) # 3. 运行 Benchmark metrics = benchmarker.run_benchmark(test_buckets, iterations_per_bucket=100) # 4. 打印格式化指标报告 print("\n" + "="*60) print(f"{'分桶名称':<20} | {'Mean(ms)':<8} | {'P90(ms)':<8} | {'P99(ms)':<8} | {'QPS':<8}") print("-" * 60) for bucket, data in metrics.items(): print(f"{bucket:<20} | {data['mean_ms']:<8} | {data['p90_ms']:<8} | {data['p99_ms']:<8} | {data['estimated_qps']:<8}") print("="*60)脚本通过在每次推理前后调用torch.cuda.synchronize()避开了 CUDA 异步发射机制导致的微秒级计时失真,同时利用真实的分桶 shape 计算 P99 指标,能够直接爆露出由于序列拉长导致的注意力矩阵计算复杂度上升开销。
数据解读干货:显存带宽瓶颈还是 Compute Bound,看这个特征曲线
拿到了包含 P99 和 QPS 的压测数据报告后,如何判断模型的优化的方向?
一个核心指标是关注 Batch Size 增加时延迟上升的斜率曲线:
若 Batch Size 从 1 增加到 8,P50 延迟几乎没有上升(例如仅从 15ms 变到 16ms),但 QPS 翻了将近 8 倍。这说明当前模型部署处于Memory Bound(显存带宽瓶颈)阶段。此时 GPU 算子绝大部分时间在等待显存读写数据。针对这种场景,首选的优化手段是量化(FP16/INT8)或者 Kernel Fusion(算子融合),减少显存访问开销。
如果 Batch Size 刚从 1 增加到 2,延迟就呈现线性翻倍上升。这表明模型已经撞上了Compute Bound(算力瓶颈),算子已经把 GPU 的 Tensor Core 填满。此时再盲目增加 Dynamic Batching 策略不仅无法提升 QPS,反而会让 P99 延迟严重恶化。
结语:推理结果应连同硬件、批量和输入长度报告,单独报一个延迟没有解释力。
结果表旁边放上运行条件
延迟数据至少注明并发、预热次数和输入长度。单条样本的耗时可以用来排查回归,却不能替代批量请求的观察。评测脚本固定后,再比较不同版本,结论才不会被临时改动的参数带偏。