（十四）32天GPU测试从入门到精通-ResNet50 推理测试day12
·
目录
引言
在 AI 应用中，推理性能直接决定用户体验和服务器成本。与训练不同，推理场景更关注延迟、吞吐量和资源效率。
ResNet50 推理测试是评估 GPU 推理性能的标准方法：
- 为什么需要专用推理引擎？ PyTorch 原生推理不是最优选择
- TensorRT 能提升多少性能？ 通常 2-5 倍延迟降低
- 量化对精度影响多大？ INT8 通常<1% 精度损失
- 如何选择最优 batch size？ 延迟 vs 吞吐量的权衡
这些问题都指向一个核心主题：ResNet50 推理测试。
推理 vs 训练
┌─────────────────────────────────────────────────┐
│ 推理与训练的差异 │
├─────────────────────────────────────────────────┤
│ │
│ 训练: │
│ ├── 目标：最小化损失函数 │
│ ├── 计算：前向 + 反向传播 │
│ ├── 精度：FP32/FP16/BF16 │
│ ├── 指标：训练速度 (img/s) │
│ ├── 显存：存储梯度、优化器状态 │
│ └── 时长：数小时至数周 │
│ │
│ 推理: │
│ ├── 目标：快速准确预测 │
│ ├── 计算：仅前向传播 │
│ ├── 精度：FP32/FP16/INT8 │
│ ├── 指标：延迟 (ms)、吞吐量 (QPS) │
│ ├── 显存：仅模型和中间结果 │
│ └── 时长：毫秒级响应 │
│ │
│ 优化重点: │
│ ├── 训练：梯度同步效率、数据加载 │
│ └── 推理：延迟优化、吞吐量优化、量化 │
│ │
└─────────────────────────────────────────────────┘
推理性能指标
┌─────────────────────────────────────────────────┐
│ 推理性能核心指标 │
├─────────────────────────────────────────────────┤
│ │
│ 延迟 (Latency): │
│ ├── 定义：单次推理所需时间 │
│ ├── 单位：毫秒 (ms) │
│ ├── 测量：P50/P90/P99 延迟 │
│ └── 场景：实时应用、交互式服务 │
│ │
│ 吞吐量 (Throughput): │
│ ├── 定义：单位时间处理请求数 │
│ ├── 单位：QPS (Queries Per Second) │
│ ├── 测量：并发请求下的总吞吐 │
│ └── 场景：批量处理、离线推理 │
│ │
│ 资源效率: │
│ ├── 显存占用：模型 + 中间结果 │
│ ├── 功耗：瓦特/推理 │
│ └── 成本：$/1000 次推理 │
│ │
│ 精度: │
│ ├── Top-1 Accuracy: 正确率 │
│ ├── Top-5 Accuracy: 前 5 包含正确答案 │
│ └── 量化损失：与 FP32 对比 │
│ │
└─────────────────────────────────────────────────┘
性能参考值
┌────────────────────────────────────────────────────────────────────┐
│ ResNet50 推理性能参考 (batch=1, FP32) │
├──────────────┬─────────────┬─────────────┬─────────────┬──────────┤
│ 引擎 │ 延迟 (ms) │ 吞吐量 │ 显存 (MB) │ 备注 │
│ │ P50/P99 │ (QPS) │ │ │
├──────────────┼─────────────┼─────────────┼─────────────┼──────────┤
│ PyTorch │ 3.5/5.2 │ 280 │ 500 │ 原生 │
│ TensorRT │ 1.2/2.1 │ 830 │ 300 │ FP32 │
│ TensorRT │ 0.8/1.5 │ 1250 │ 250 │ FP16 │
│ TensorRT │ 0.5/1.0 │ 2000 │ 200 │ INT8 │
│ ONNX Runtime │ 1.5/2.5 │ 670 │ 350 │ FP32 │
│ ONNX Runtime │ 1.0/1.8 │ 1000 │ 300 │ FP16 │
└──────────────┴─────────────┴─────────────┴─────────────┴──────────┘
注：测试硬件 A100 SXM，实际性能受硬件和配置影响
推理引擎概览
主流推理引擎对比
┌────────────────────────────────────────────────────────────────────┐
│ 主流推理引擎对比 │
├──────────────┬─────────────┬─────────────┬─────────────┬──────────┤
│ 引擎 │ 优势 │ 劣势 │ 适用场景 │ 难度 │
├──────────────┼─────────────┼─────────────┼─────────────┼──────────┤
│ PyTorch │ 易用、灵活 │ 性能一般 │ 原型开发 │ 低 │
│ Native │ 无需转换 │ 无优化 │ 研究实验 │ │
├──────────────┼─────────────┼─────────────┼─────────────┼──────────┤
│ TensorRT │ 性能最优 │ 仅 NVIDIA │ 生产部署 │ 中 │
│ │ 深度优化 │ 编译时间长 │ 高性能场景 │ │
├──────────────┼─────────────┼─────────────┼─────────────┼──────────┤
│ ONNX │ 跨平台 │ 性能次优 │ 多平台部署 │ 低 │
│ Runtime │ 易转换 │ 优化有限 │ 快速验证 │ │
├──────────────┼─────────────┼─────────────┼─────────────┼──────────┤
│ OpenVINO │ Intel 优化 │ 仅 Intel │ Intel 硬件 │ 中 │
├──────────────┼─────────────┼─────────────┼─────────────┼──────────┤
│ TVM │ 自动优化 │ 配置复杂 │ 研究探索 │ 高 │
└──────────────┴─────────────┴─────────────┴─────────────┴──────────┘
引擎选择指南
┌─────────────────────────────────────────────────┐
│ 推理引擎选择指南 │
├─────────────────────────────────────────────────┤
│ │
│ 选择 NVIDIA GPU + 追求极致性能: │
│ └── TensorRT (首选) │
│ │
│ 需要跨平台部署: │
│ └── ONNX Runtime │
│ │
│ 快速原型验证: │
│ └── PyTorch Native │
│ │
│ Intel 硬件: │
│ └── OpenVINO │
│ │
│ 研究/自定义优化: │
│ └── TVM │
│ │
└─────────────────────────────────────────────────┘
TensorRT 推理
TensorRT 简介
┌─────────────────────────────────────────────────┐
│ TensorRT 核心特性 │
├─────────────────────────────────────────────────┤
│ │
│ 定位： │
│ ├── NVIDIA 高性能推理推理引擎 │
│ ├── 针对 NVIDIA GPU 深度优化 │
│ ├── 支持 FP32/FP16/INT8 精度 │
│ └── 集成于 NVIDIA AI 企业软件栈 │
│ │
│ 优化技术: │
│ ├── 层融合 (Layer Fusion) │
│ ├── 内核自动调优 (Auto-tuning) │
│ ├── 多流执行 (Multi-stream) │
│ ├── 动态张量内存 (Memory Optimization) │
│ ├── 精度校准 (INT8 Calibration) │
│ └── 动态形状 (Dynamic Shapes) │
│ │
│ 性能提升: │
│ ├── FP16 vs FP32: 2-3x │
│ ├── INT8 vs FP32: 4-8x │
│ └── 延迟降低：50-80% │
│ │
└─────────────────────────────────────────────────┘
TensorRT 模型转换
#!/usr/bin/env python3
# tensorrt_convert.py - TensorRT 模型转换
import torch
import tensorrt as trt
import onnx
from pathlib import Path
def export_onnx(model, output_path, input_shape=(1, 3, 224, 224)):
"""导出 ONNX 模型"""
model.eval()
dummy_input = torch.randn(input_shape)
torch.onnx.export(
model,
dummy_input,
output_path,
export_params=True,
opset_version=13,
do_constant_folding=True,
input_names=['input'],
output_names=['output'],
dynamic_axes={
'input': {0: 'batch_size'},
'output': {0: 'batch_size'}
}
)
print(f"ONNX 模型已导出：{output_path}")
# 验证 ONNX 模型
onnx_model = onnx.load(output_path)
onnx.checker.check_model(onnx_model)
print("ONNX 模型验证通过")
def build_trt_engine(onnx_path, engine_path, precision='fp16', batch_size=1):
"""构建 TensorRT 引擎"""
TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
builder = trt.Builder(TRT_LOGGER)
network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH))
parser = trt.OnnxParser(network, TRT_LOGGER)
config = builder.create_builder_config()
# 解析 ONNX
with open(onnx_path, 'rb') as f:
if not parser.parse(f.read()):
for error in range(parser.num_errors):
print(parser.get_error(error))
return False
# 配置
config.max_workspace_size = 1 << 30 # 1GB
if precision == 'fp16':
config.set_flag(trt.BuilderFlag.FP16)
elif precision == 'int8':
config.set_flag(trt.BuilderFlag.INT8)
# 需要校准器 (略)
# 设置 batch size
builder.max_batch_size = batch_size
# 构建引擎
engine = builder.build_engine(network, config)
# 保存引擎
with open(engine_path, 'wb') as f:
f.write(engine.serialize())
print(f"TensorRT 引擎已保存：{engine_path}")
return True
def main():
from torchvision.models import resnet50
# 1. 导出 ONNX
model = resnet50(pretrained=True)
export_onnx(model, 'resnet50.onnx')
# 2. 构建 TensorRT 引擎 (FP32)
build_trt_engine('resnet50.onnx', 'resnet50_fp32.engine', precision='fp32')
# 3. 构建 TensorRT 引擎 (FP16)
build_trt_engine('resnet50.onnx', 'resnet50_fp16.engine', precision='fp16')
print("模型转换完成")
if __name__ == "__main__":
main()
TensorRT 推理
#!/usr/bin/env python3
# tensorrt_inference.py - TensorRT 推理
import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np
import time
from pathlib import Path
class TensorRTInferencer:
"""TensorRT 推理器"""
def __init__(self, engine_path):
self.logger = trt.Logger(trt.Logger.WARNING)
# 加载引擎
with open(engine_path, 'rb') as f:
engine_data = f.read()
runtime = trt.Runtime(self.logger)
self.engine = runtime.deserialize_cuda_engine(engine_data)
self.context = self.engine.create_execution_context()
# 分配内存
self.inputs = []
self.outputs = []
self.bindings = []
for binding in self.engine:
shape = self.engine.get_binding_shape(binding)
dtype = trt.nptype(self.engine.get_binding_dtype(binding))
if self.engine.binding_is_input(binding):
self.input_shape = shape
else:
self.output_shape = shape
# 分配设备内存
size = trt.volume(shape) * np.dtype(dtype).itemsize
device_mem = cuda.mem_alloc(size)
# 创建主机内存
host_mem = np.empty(shape, dtype=dtype)
self.bindings.append(int(device_mem))
if self.engine.binding_is_input(binding):
self.inputs.append({'host': host_mem, 'device': device_mem})
else:
self.outputs.append({'host': host_mem, 'device': device_mem})
self.stream = cuda.Stream()
def infer(self, input_data):
"""执行推理"""
# 复制输入到设备
cuda.memcpy_htod_async(self.inputs[0]['device'], input_data, self.stream)
# 执行推理
self.context.execute_async_v2(
bindings=self.bindings,
stream_handle=self.stream.handle
)
# 复制输出到主机
cuda.memcpy_dtoh_async(self.outputs[0]['host'], self.outputs[0]['device'], self.stream)
self.stream.synchronize()
return self.outputs[0]['host']
def benchmark(self, input_shape=(1, 3, 224, 224), iterations=1000):
"""性能基准测试"""
# 创建随机输入
input_data = np.random.randn(*input_shape).astype(np.float32)
# 预热
for _ in range(100):
_ = self.infer(input_data)
# 正式测试
latencies = []
start_time = time.time()
for _ in range(iterations):
start = time.perf_counter()
_ = self.infer(input_data)
end = time.perf_counter()
latencies.append(end - start)
total_time = time.time() - start_time
latencies = np.array(latencies) * 1000 # ms
results = {
'mean_ms': float(np.mean(latencies)),
'median_ms': float(np.median(latencies)),
'p95_ms': float(np.percentile(latencies, 95)),
'p99_ms': float(np.percentile(latencies, 99)),
'min_ms': float(np.min(latencies)),
'max_ms': float(np.max(latencies)),
'qps': float(iterations / total_time)
}
return results
def main():
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('--engine', type=str, default='resnet50_fp16.engine')
parser.add_argument('--iterations', type=int, default=1000)
args = parser.parse_args()
print(f"加载引擎：{args.engine}")
inferencer = TensorRTInferencer(args.engine)
print(f"\n性能基准测试 ({args.iterations} 次迭代)...")
results = inferencer.benchmark(iterations=args.iterations)
print("\n" + "="*60)
print("TensorRT 推理性能")
print("="*60)
print(f"平均延迟：{results['mean_ms']:.2f} ms")
print(f"P50 延迟：{results['median_ms']:.2f} ms")
print(f"P95 延迟：{results['p95_ms']:.2f} ms")
print(f"P99 延迟：{results['p99_ms']:.2f} ms")
print(f"吞吐量：{results['qps']:.0f} QPS")
print("="*60)
if __name__ == "__main__":
main()
ONNX Runtime 推理
ONNX Runtime 简介
ONNX Runtime 是一个用于机器学习模型推理的高性能开源引擎，专门为部署和生产环境设计。它支持以 ONNX 格式保存的模型，实现了训练框架与部署环境之间的高效衔接。
核心特点
-
跨平台与多硬件支持：可在 Windows、Linux、macOS 上运行，并支持 CPU、GPU（CUDA、DirectML、ROCm）、NPU 等多种硬件加速后端。
-
高性能推理：通过图优化、内核融合、量化等技术大幅提升模型执行速度。
-
语言绑定丰富：提供 Python、C++、C#、Java、JavaScript 等语言的 API，便于不同技术栈集成。
-
框架互通性：作为“中间件”，允许来自 PyTorch、TensorFlow、Scikit-learn 等框架的模型（导出为 ONNX 格式）在同一运行时上执行。
主要用途
-
生产环境部署：将训练好的模型高效部署到服务器、云或边缘设备。
-
多框架模型统一运行时：解决团队使用不同训练框架带来的部署碎片化问题。
-
移动端与边缘推理：提供轻量级版本，支持在资源受限的设备上运行模型。
┌─────────────────────────────────────────────────┐
│ ONNX Runtime 核心特性 │
├─────────────────────────────────────────────────┤
│ │
│ 定位： │
│ ├── 微软开发的跨平台推理引擎 │
│ ├── 支持 ONNX 标准格式 │
│ ├── 多后端：CPU/GPU/TensorRT/OpenVINO │
│ └── 易用性高，API 简洁 │
│ │
│ 优化技术: │
│ ├── 图优化 (Graph Optimization) │
│ ├── 算子融合 │
│ ├── 内存优化 │
│ ├── 多线程执行 │
│ └── TensorRT 集成 (可选) │
│ │
│ 优势: │
│ ├── 一次导出，多平台运行 │
│ ├── 无需编译，快速部署 │
│ └── 活跃的社区支持 │
│ │
└─────────────────────────────────────────────────┘
ONNX Runtime 推理
#!/usr/bin/env python3
# onnx_runtime_inference.py - ONNX Runtime 推理
import onnxruntime as ort
import numpy as np
import time
class ONNXInferencer:
"""ONNX Runtime 推理器"""
def __init__(self, model_path, use_gpu=True, use_tensorrt=False):
# 配置 Session 选项
sess_options = ort.SessionOptions()
sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
if use_tensorrt:
# 启用 TensorRT
sess_options.add_session_config_entry('session.intra_op.allow_spinning', '1')
providers = [('TensorrtExecutionProvider', {
'trt_engine_cache_enable': True,
'trt_engine_cache_path': './trt_cache',
}), 'CUDAExecutionProvider', 'CPUExecutionProvider']
elif use_gpu:
providers = ['CUDAExecutionProvider', 'CPUExecutionProvider']
else:
providers = ['CPUExecutionProvider']
self.session = ort.InferenceSession(
model_path,
sess_options=sess_options,
providers=providers
)
self.input_name = self.session.get_inputs()[0].name
self.output_name = self.session.get_outputs()[0].name
print(f"加载模型：{model_path}")
print(f"执行提供者：{self.session.get_providers()}")
def infer(self, input_data):
"""执行推理"""
outputs = self.session.run([self.output_name], {self.input_name: input_data})
return outputs[0]
def benchmark(self, input_shape=(1, 3, 224, 224), iterations=1000):
"""性能基准测试"""
# 创建随机输入
input_data = np.random.randn(*input_shape).astype(np.float32)
# 预热
for _ in range(100):
_ = self.infer(input_data)
# 正式测试
latencies = []
start_time = time.time()
for _ in range(iterations):
start = time.perf_counter()
_ = self.infer(input_data)
end = time.perf_counter()
latencies.append(end - start)
total_time = time.time() - start_time
latencies = np.array(latencies) * 1000 # ms
results = {
'mean_ms': float(np.mean(latencies)),
'median_ms': float(np.median(latencies)),
'p95_ms': float(np.percentile(latencies, 95)),
'p99_ms': float(np.percentile(latencies, 99)),
'min_ms': float(np.min(latencies)),
'max_ms': float(np.max(latencies)),
'qps': float(iterations / total_time)
}
return results
def main():
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('--model', type=str, default='resnet50.onnx')
parser.add_argument('--gpu', action='store_true', default=True)
parser.add_argument('--tensorrt', action='store_true', default=False)
parser.add_argument('--iterations', type=int, default=1000)
args = parser.parse_args()
print(f"GPU: {args.gpu}, TensorRT: {args.tensorrt}")
inferencer = ONNXInferencer(
args.model,
use_gpu=args.gpu,
use_tensorrt=args.tensorrt
)
print(f"\n性能基准测试 ({args.iterations} 次迭代)...")
results = inferencer.benchmark(iterations=args.iterations)
print("\n" + "="*60)
print("ONNX Runtime 推理性能")
print("="*60)
print(f"平均延迟：{results['mean_ms']:.2f} ms")
print(f"P50 延迟：{results['median_ms']:.2f} ms")
print(f"P95 延迟：{results['p95_ms']:.2f} ms")
print(f"P99 延迟：{results['p99_ms']:.2f} ms")
print(f"吞吐量：{results['qps']:.0f} QPS")
print("="*60)
if __name__ == "__main__":
main()
批处理与延迟优化
Batch Size 对性能的影响
在ResNet50推理测试中，Batch Size（批处理大小）对性能的影响主要体现在吞吐量、延迟和资源利用率三个方面，其关系并非线性，需要权衡。
核心影响分析
|
性能指标 |
Batch Size 增大的影响 |
说明 |
|---|---|---|
|
吞吐量 |
显著提升 |
单位时间（如每秒）内能处理的图片总数增加。GPU等硬件能并行计算更多数据，利用率高。 |
|
单张图片延迟 |
可能降低或持平 |
处理一批图片的总时间增加，但平均到每张图片的时间可能减少（因并行计算分摊了开销）。 |
|
单次推理延迟 |
明显增加 |
处理整批图片所需的时间（从输入到输出）变长。这对实时应用是关键指标。 |
|
内存/显存占用 |
线性增长 |
需要同时加载更多图片和中间计算结果，可能触发OOM（内存不足）错误。 |
典型性能曲线与权衡
-
吞吐量 vs. Batch Size：
-
曲线通常先快速上升，后趋于平缓。当Batch Size达到硬件（如GPU SM单元）的并行计算极限后，吞吐量增长会放缓甚至饱和。
-
-
延迟 vs. Batch Size：
-
单次推理延迟几乎随Batch Size线性增长。
-
单张图片平均延迟则会先下降后上升。存在一个最优Batch Size，使平均延迟最低或吞吐量成本比最佳。
-
实践建议
-
追求高吞吐的场景（如离线批量处理图片）：
-
在不超出内存限制的前提下，尽可能使用较大的Batch Size，以最大化硬件利用率和吞吐量。
-
-
追求低延迟的场景（如在线API、实时检测）：
-
通常使用较小的Batch Size（如1, 2, 4），以保证每次响应的速度。有时甚至为每个请求单独推理。
-
-
如何找到最佳点：
-
必须在你的目标硬件（CPU/具体型号GPU）和推理软件环境（如ONNX Runtime、TensorRT）上进行实际基准测试。
-
绘制 “吞吐量-Batch Size” 和 “延迟-Batch Size” 曲线，根据你的业务需求（例如：要求吞吐量 > 100 张/秒，且延迟 < 50ms）来确定最佳点。
-
#!/usr/bin/env python3
# batch_size_benchmark.py - Batch Size 性能测试
import torch
import time
import numpy as np
from torchvision.models import resnet50
def benchmark_inference(model, batch_size, iterations=100):
"""基准推理测试"""
model.eval()
device = next(model.parameters()).device
# 创建输入
input_tensor = torch.randn(batch_size, 3, 224, 224, device=device)
# 预热
with torch.inference_mode():
for _ in range(20):
_ = model(input_tensor)
torch.cuda.synchronize()
# 正式测试
latencies = []
start_time = time.time()
with torch.inference_mode():
for _ in range(iterations):
start = time.perf_counter()
_ = model(input_tensor)
torch.cuda.synchronize()
end = time.perf_counter()
latencies.append(end - start)
total_time = time.time() - start_time
latencies = np.array(latencies) * 1000 # ms
# 计算指标
latency_per_request = np.mean(latencies) / batch_size # 每请求延迟
throughput = (batch_size * iterations) / total_time # QPS
return {
'batch_size': batch_size,
'batch_latency_ms': float(np.mean(latencies)),
'per_request_latency_ms': float(latency_per_request),
'p99_latency_ms': float(np.percentile(latencies, 99)),
'throughput_qps': float(throughput)
}
def main():
device = torch.device('cuda')
model = resnet50(pretrained=True).to(device)
model.eval()
print(f"GPU: {torch.cuda.get_device_name(device)}")
print()
batch_sizes = [1, 2, 4, 8, 16, 32, 64, 128, 256]
results = []
print("="*80)
print(f"{'Batch':<8} {'Batch 延迟':<12} {'每请求延迟':<14} {'P99 延迟':<12} {'吞吐量':<12}")
print(f"{'Size':<8} {'(ms)':<12} {'(ms)':<14} {'(ms)':<12} {'(QPS)':<12}")
print("="*80)
for bs in batch_sizes:
result = benchmark_inference(model, bs, iterations=100)
results.append(result)
print(f"{result['batch_size']:<8} "
f"{result['batch_latency_ms']:<12.2f} "
f"{result['per_request_latency_ms']:<14.2f} "
f"{result['p99_latency_ms']:<12.2f} "
f"{result['throughput_qps']:<12.0f}")
print("="*80)
# 找到最优 batch size (延迟<10ms 的最大吞吐量)
optimal = max([r for r in results if r['per_request_latency_ms'] < 10],
key=lambda x: x['throughput_qps'], default=None)
if optimal:
print(f"\n最优配置 (延迟<10ms): Batch Size = {optimal['batch_size']}")
print(f" 吞吐量：{optimal['throughput_qps']:.0f} QPS")
print(f" 每请求延迟：{optimal['per_request_latency_ms']:.2f} ms")
if __name__ == "__main__":
main()
延迟优化技巧
┌─────────────────────────────────────────────────┐
│ 推理延迟优化技巧 │
├─────────────────────────────────────────────────┤
│ │
│ 模型层面: │
│ ├── 使用更小的模型 (ResNet18 vs ResNet50) │
│ ├── 模型剪枝 (Pruning) │
│ ├── 知识蒸馏 (Distillation) │
│ └── 神经架构搜索 (NAS) │
│ │
│ 精度层面: │
│ ├── FP16 推理 (2x 加速) │
│ ├── INT8 量化 (4x 加速) │
│ └── 混合精度 │
│ │
│ 引擎层面: │
│ ├── TensorRT 优化 │
│ ├── 算子融合 │
│ ├── 内核自动调优 │
│ └── 内存优化 │
│ │
│ 系统层面: │
│ ├── GPU 性能模式 │
│ ├── CPU 亲和性绑定 │
│ ├── 减少上下文切换 │
│ └── 使用 CUDA Graphs │
│ │
│ 服务层面: │
│ ├── 动态批处理 (Dynamic Batching) │
│ ├── 请求队列优化 │
│ ├── 模型并行/流水线 │
│ └── 缓存热点结果 │
│ │
└─────────────────────────────────────────────────┘
量化测试
量化原理
┌─────────────────────────────────────────────────┐
│ 量化技术原理 │
├─────────────────────────────────────────────────┤
│ │
│ 量化类型: │
│ ├── 训练后量化 (PTQ): 无需重新训练 │
│ │ ├── 动态量化：运行时计算 scale │
│ │ └── 静态量化：使用校准集确定 scale │
│ │ │
│ └── 量化感知训练 (QAT): 训练中模拟量化 │
│ └── 精度更高，需要训练 │
│ │
│ 精度格式: │
│ ├── INT8: 8 位整数，4x 压缩，4x 加速 │
│ ├── FP16: 16 位浮点，2x 压缩，2x 加速 │
│ └── BF16: Brain Float，训练友好 │
│ │
│ 量化影响: │
│ ├── 精度损失：通常<1% (ResNet50) │
│ ├── 模型大小：减少 75% (INT8) │
│ └── 推理速度：提升 2-4x │
│ │
└─────────────────────────────────────────────────┘
TensorRT INT8 量化
#!/usr/bin/env python3
# tensorrt_int8_quantization.py - TensorRT INT8 量化
import tensorrt as trt
import torch
import numpy as np
from torchvision.models import resnet50
from torchvision import transforms
from PIL import Image
from pathlib import Path
class Int8Calibrator(trt.IInt8EntropyCalibrator2):
"""INT8 校准器"""
def __init__(self, data_dir, batch_size, cache_file):
super().__init__()
self.batch_size = batch_size
self.cache_file = cache_file
# 准备校准数据
self.data_dir = Path(data_dir)
self.image_paths = list(self.data_dir.glob('*.jpg'))[:1000] # 使用 1000 张图
self.current_index = 0
# 预处理
self.preprocess = transforms.Compose([
transforms.Resize(256),
transforms.CenterCrop(224),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]),
])
# 分配内存
self.input_shape = (batch_size, 3, 224, 224)
self.input_size = int(np.prod(self.input_shape) * 4) # float32
self.device_input = cuda.mem_alloc(self.input_size)
def get_batch_size(self):
return self.batch_size
def get_batch(self, names, p_str=None):
"""获取一个 batch 的校准数据"""
if self.current_index + self.batch_size > len(self.image_paths):
return None
batch_images = []
for i in range(self.batch_size):
img_path = self.image_paths[self.current_index + i]
img = Image.open(img_path).convert('RGB')
img_tensor = self.preprocess(img)
batch_images.append(img_tensor)
self.current_index += self.batch_size
# 转换为 numpy 数组
batch_array = torch.stack(batch_images).numpy().astype(np.float32)
# 复制到 GPU
cuda.memcpy_htod(self.device_input, batch_array)
return [int(self.device_input)]
def read_calibration_cache(self):
"""读取校准缓存"""
if Path(self.cache_file).exists():
with open(self.cache_file, 'rb') as f:
return f.read()
return None
def write_calibration_cache(self, cache):
"""写入校准缓存"""
with open(self.cache_file, 'wb') as f:
f.write(cache)
def build_int8_engine(onnx_path, engine_path, calibrator, cache_file='int8_cache.bin'):
"""构建 INT8 TensorRT 引擎"""
TRT_LOGGER = trt.Logger(trt.Logger.WARNING)
builder = trt.Builder(TRT_LOGGER)
network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH))
parser = trt.OnnxParser(network, TRT_LOGGER)
config = builder.create_builder_config()
# 解析 ONNX
with open(onnx_path, 'rb') as f:
parser.parse(f.read())
# INT8 配置
config.set_flag(trt.BuilderFlag.INT8)
config.int8_calibrator = calibrator
config.max_workspace_size = 1 << 30
# 构建引擎
engine = builder.build_engine(network, config)
# 保存
with open(engine_path, 'wb') as f:
f.write(engine.serialize())
print(f"INT8 引擎已保存：{engine_path}")
def main():
# 1. 导出 ONNX
model = resnet50(pretrained=True)
dummy_input = torch.randn(1, 3, 224, 224)
torch.onnx.export(model, dummy_input, 'resnet50.onnx')
# 2. INT8 校准
import pycuda.autoinit
import pycuda.driver as cuda
calibrator = Int8Calibrator(
data_dir='/data/calibration_images',
batch_size=32,
cache_file='int8_cache.bin'
)
# 3. 构建 INT8 引擎
build_int8_engine(
'resnet50.onnx',
'resnet50_int8.engine',
calibrator
)
print("INT8 量化完成")
if __name__ == "__main__":
main()
量化精度验证
#!/usr/bin/env python3
# quantization_accuracy_test.py - 量化精度测试
import torch
import numpy as np
from torchvision.models import resnet50
from torchvision import datasets, transforms
from torch.utils.data import DataLoader
def evaluate_model(model, data_loader, device):
"""评估模型精度"""
model.eval()
correct = 0
total = 0
with torch.no_grad():
for images, targets in data_loader:
images = images.to(device)
targets = targets.to(device)
outputs = model(images)
_, predicted = outputs.max(1)
total += targets.size(0)
correct += predicted.eq(targets).sum().item()
accuracy = 100.0 * correct / total
return accuracy
def main():
device = torch.device('cuda')
# 加载验证集
val_transform = transforms.Compose([
transforms.Resize(256),
transforms.CenterCrop(224),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]),
])
val_dataset = datasets.ImageFolder('/data/imagenet/val', transform=val_transform)
val_loader = DataLoader(val_dataset, batch_size=64, shuffle=False, num_workers=4)
# FP32 模型
model_fp32 = resnet50(pretrained=True).to(device)
acc_fp32 = evaluate_model(model_fp32, val_loader, device)
print(f"FP32 精度：{acc_fp32:.2f}%")
# 模拟 INT8 量化 (简化版)
model_int8 = resnet50(pretrained=True).to(device)
model_int8.qconfig = torch.quantization.get_default_qconfig('fbgemm')
torch.quantization.prepare(model_int8, inplace=True)
torch.quantization.convert(model_int8, inplace=True)
acc_int8 = evaluate_model(model_int8, val_loader, device)
print(f"INT8 精度：{acc_int8:.2f}%")
# 精度损失
accuracy_drop = acc_fp32 - acc_int8
print(f"精度损失：{accuracy_drop:.2f}%")
print("\n" + "="*60)
if accuracy_drop < 1.0:
print("✓ 量化精度损失在可接受范围内 (<1%)")
else:
print("⚠️ 量化精度损失较大，建议使用 QAT")
print("="*60)
if __name__ == "__main__":
main()
吞吐量 Benchmark
并发推理测试
#!/usr/bin/env python3
# throughput_benchmark.py - 吞吐量基准测试
import torch
import time
import threading
import numpy as np
from torchvision.models import resnet50
from concurrent.futures import ThreadPoolExecutor
class ThroughputBenchmark:
"""吞吐量基准测试"""
def __init__(self, model, batch_size=1):
self.model = model
self.model.eval()
self.batch_size = batch_size
self.device = next(model.parameters()).device
# 创建输入
self.input_tensor = torch.randn(
batch_size, 3, 224, 224, device=self.device
)
def infer_single(self):
"""单次推理"""
with torch.inference_mode():
_ = self.model(self.input_tensor)
def benchmark_concurrent(self, num_threads=1, duration=30):
"""并发推理测试"""
self.model.eval()
# 预热
for _ in range(10):
self.infer_single()
torch.cuda.synchronize()
# 并发测试
request_count = [0]
stop_flag = [False]
def worker():
while not stop_flag[0]:
self.infer_single()
request_count[0] += 1
# 启动工作线程
threads = []
for _ in range(num_threads):
t = threading.Thread(target=worker)
t.start()
threads.append(t)
# 运行指定时长
time.sleep(duration)
stop_flag[0] = True
# 等待所有线程完成
for t in threads:
t.join()
# 计算指标
total_requests = request_count[0]
total_images = total_requests * self.batch_size
qps = total_requests / duration
ips = total_images / duration
return {
'num_threads': num_threads,
'batch_size': self.batch_size,
'duration_s': duration,
'total_requests': total_requests,
'total_images': total_images,
'qps': qps,
'images_per_sec': ips
}
def main():
device = torch.device('cuda')
model = resnet50(pretrained=True).to(device)
print(f"GPU: {torch.cuda.get_device_name(device)}")
print()
batch_size = 1
benchmark = ThroughputBenchmark(model, batch_size=batch_size)
print("="*70)
print(f"吞吐量基准测试 (Batch Size={batch_size}, Duration=30s)")
print("="*70)
print()
thread_counts = [1, 2, 4, 8, 16, 32, 64]
results = []
print(f"{'并发数':<10} {'QPS':<15} {'Images/s':<15}")
print("-"*70)
for num_threads in thread_counts:
result = benchmark.benchmark_concurrent(num_threads=num_threads, duration=30)
results.append(result)
print(f"{result['num_threads']:<10} "
f"{result['qps']:<15.0f} "
f"{result['images_per_sec']:<15.0f}")
print("="*70)
# 找到最优并发数
optimal = max(results, key=lambda x: x['images_per_sec'])
print(f"\n最优配置：并发数 = {optimal['num_threads']}")
print(f" 最大吞吐量：{optimal['images_per_sec']:.0f} images/s")
if __name__ == "__main__":
main()
实战：推理性能对比
完整对比测试
#!/bin/bash
# inference_comparison.sh - 推理引擎性能对比测试
echo "=========================================="
echo " 推理引擎性能对比测试"
echo "=========================================="
RESULTS_DIR="results/inference_$(date +%Y%m%d_%H%M%S)"
mkdir -p $RESULTS_DIR
echo "结果目录：$RESULTS_DIR"
echo ""
# 1. 准备模型
echo "[1/5] 准备模型..."
python3 -c "
from torchvision.models import resnet50
import torch
model = resnet50(pretrained=True)
dummy_input = torch.randn(1, 3, 224, 224)
torch.onnx.export(model, dummy_input, '$RESULTS_DIR/resnet50.onnx')
print('ONNX 模型已导出')
"
# 2. PyTorch 基准测试
echo ""
echo "[2/5] PyTorch 基准测试..."
python3 -c "
import torch
from torchvision.models import resnet50
import time
model = resnet50(pretrained=True).cuda()
model.eval()
input_tensor = torch.randn(1, 3, 224, 224).cuda()
# 预热
for _ in range(100):
with torch.inference_mode():
_ = model(input_tensor)
torch.cuda.synchronize()
# 测试
latencies = []
for _ in range(1000):
start = time.perf_counter()
with torch.inference_mode():
_ = model(input_tensor)
torch.cuda.synchronize()
latencies.append(time.perf_counter() - start)
import numpy as np
latencies = np.array(latencies) * 1000
print(f'PyTorch P50: {np.median(latencies):.2f} ms')
print(f'PyTorch P99: {np.percentile(latencies, 99):.2f} ms')
" | tee $RESULTS_DIR/pytorch_result.txt
# 3. ONNX Runtime 基准测试
echo ""
echo "[3/5] ONNX Runtime 基准测试..."
python3 onnx_runtime_inference.py --model $RESULTS_DIR/resnet50.onnx --iterations 1000 | tee $RESULTS_DIR/onnx_result.txt
# 4. TensorRT 基准测试
echo ""
echo "[4/5] TensorRT 基准测试..."
python3 tensorrt_inference.py --engine resnet50_fp16.engine --iterations 1000 | tee $RESULTS_DIR/trt_result.txt
# 5. 生成对比报告
echo ""
echo "[5/5] 生成对比报告..."
cat << EOF > $RESULTS_DIR/comparison_report.md
# 推理引擎性能对比报告
**测试日期:** $(date)
**GPU:** $(nvidia-smi --query-gpu=name --format=csv,noheader)
## PyTorch Native
\`\`\`
$(cat $RESULTS_DIR/pytorch_result.txt)
\`\`\`
## ONNX Runtime
\`\`\`
$(cat $RESULTS_DIR/onnx_result.txt)
\`\`\`
## TensorRT
\`\`\`
$(cat $RESULTS_DIR/trt_result.txt)
\`\`\`
## 性能对比
| 引擎 | P50 延迟 (ms) | P99 延迟 (ms) | 相对提升 |
|------|---------------|---------------|----------|
| PyTorch | - | - | 基准 |
| ONNX Runtime | - | - | 计算中... |
| TensorRT | - | - | 计算中... |
## 建议
根据延迟和吞吐量需求选择合适的推理引擎
EOF
echo "报告已生成：$RESULTS_DIR/comparison_report.md"
echo ""
echo "=========================================="
echo " 对比测试完成"
echo "=========================================="
echo ""
echo "结果目录：$RESULTS_DIR"
ls -la $RESULTS_DIR
总结
今天学到的内容
- ✅ 推理引擎概览：TensorRT、ONNX Runtime 对比
- ✅ TensorRT 推理：模型转换、推理、优化
- ✅ ONNX Runtime 推理：跨平台推理
- ✅ 批处理与延迟优化：Batch Size 选择、优化技巧
- ✅ 量化测试：INT8 量化原理、实现、精度验证
- ✅ 吞吐量 Benchmark：并发推理测试
- ✅ 实战演练：推理引擎性能对比
下一步
明天我们将学习 Day 13 - 图像分类模型性能对比，了解：
- 主流模型对比（ResNet、EfficientNet、ViT）
- 精度 - 速度权衡
- 不同 GPU 型号性能对比
- 选型建议
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
9
0- 0
已为社区贡献13条内容
所有评论(0)