GPU推理加速实战：CUDA Graph与TensorRT优化深度解析及生产级部署完整方案

GPU推理加速实战：CUDA Graph与TensorRT优化深度解析及生产级部署完整方案

一、GPU推理的性能瓶颈：Kernel启动开销是隐形的性能杀手

GPU推理看起来已经很快了。
为什么还要优化？
答案藏在Kernel启动开销里。

每次调用一次CUDA Kernel，CPU需要向GPU提交一个启动命令。
这个命令需要经过驱动层、运行时层，最终到达GPU硬件。
单次Kernel启动开销约35微秒。
看起来不多，但大模型的推理涉及数百个Kernel调用。
仅Kernel启动的总开销就达到23毫秒。

对于延迟敏感的场景（如实时语音、在线推荐），2~3毫秒很致命。
更糟糕的是，这些开销是串行的。
CPU一个一个地提交Kernel，GPU可能处于等待状态。

CUDA Graph就是为解决这个问题的。
它的核心思想：将多次Kernel启动捕获成一个计算图。
之后重复执行时，一次性将整个图提交给GPU。
GPU可以自主调度图中的Kernel，无需CPU介入。
这样可以减少Kernel启动开销，提升GPU利用率。

TensorRT则是另一个维度的优化。
它不减少Kernel启动次数，而是让每个Kernel跑得更快。
通过核函数自动调优、层融合、精度量化等手段。
让整个神经网络的前向传播更快。

两者结合，在生产环境中可实现2~5倍的推理加速。

flowchart TD A[PyTorch/TensorFlow模型] --> B{是否使用TensorRT?} B -->|是| C[TensorRT优化流程] B -->|否| D[直接CUDA推理] C --> C1[ONNX导出] C1 --> C2[TensorRT解析ONNX] C2 --> C3[Builder优化配置] C3 --> C4[FP16/INT8量化] C4 --> C5[层融合+核函数调优] C5 --> C6[序列化Engine文件] D --> E[PyTorch CUDA推理] C6 --> F[TensorRT推理] E --> G{是否使用CUDA Graph?} F --> G G -->|是| H[CUDA Graph捕获] G -->|否| I[逐Kernel启动] H --> H1[torch.cuda.graph上下文管理器] H1 --> H2[捕获整个推理计算图] H2 --> H3[重用Graph, 单次提交] I --> J[CPU逐次提交Kernel] H3 --> K[GPU自主调度, 无CPU开销] J --> L[延迟: 基准值] K --> M[延迟: 降低30%~60%] C6 --> N[吞吐量: 提升2~5倍] style C6 fill:#27ae60,color:#fff style H3 fill:#27ae60,color:#fff style K fill:#27ae60,color:#fff style M fill:#27ae60,color:#fff style N fill:#27ae60,color:#fff style J fill:#e74c3c,color:#fff style L fill:#e74c3c,color:#fff

CUDA Graph和TensorRT的适用场景不同。
TensorRT适合对延迟和吞吐量都有极高要求的场景。
CUDA Graph适合推理流程固定、输入形状不变的场景。
两者可以叠加使用，效果最好。

二、CUDA Graph深度解析：捕获、重放与动态输入处理的完整方案

CUDA Graph的核心API是cudaGraphCreate和cudaGraphLaunch。
但PyTorch用户不需要直接调用这些C API。
PyTorch 1.10+提供了torch.cuda.graph上下文管理器。
在with torch.cuda.graph(graph)块中执行的CUDA操作会被捕获。

捕获完成后，用graph.replay()重放计算图。
重放时不再需要CPU逐次提交Kernel。
整个图的Kernel调度由GPU硬件完成。
延迟降低30%~60%，具体取决于模型结构。

但CUDA Graph有一个重要限制：输入形状必须固定。
捕获时是什么形状，重放时就必须是什么形状。
这对大模型的KVCache场景是个问题。
因为每次生成的序列长度不同，输入形状不同。

解决方案是使用CUDA Graph with Dynamic Shapes。
TensorRT和PyTorch 2.0+都支持动态形状Graph。
捕获时指定输入形状的范围（如[1, 256]到[1, 2048]）。
运行时在这个范围内动态选择最优的Kernel配置。

""" CUDA Graph生产级使用示例 包含：基础捕获重放、动态输入处理、与TensorRT集成 适用场景：推理流程固定、对延迟极度敏感的服务 """ import torch import torch.nn as nn from dataclasses import dataclass from typing import Optional @dataclass class CUDAGraphConfig: enable_graph: bool = True warmup_iterations: int = 3 # 捕获前的热身迭代次数 max_batch_size: int = 8 input_seq_len: int = 512 output_seq_len: int = 128 class CUDAGraphWrapper: """ CUDA Graph包装器： 将PyTorch模型的推理过程捕获为CUDA Graph， 之后重用Graph，避免重复的Kernel启动开销 """ def __init__(self, model: nn.Module, config: CUDAGraphConfig): self.model = model self.config = config self.graph: Optional[torch.cuda.CUDAGraph] = None self.static_input: Optional[torch.Tensor] = None self.static_output: Optional[torch.Tensor] = None self.is_captured = False def warmup(self, dummy_input: torch.Tensor) -> None: """ 捕获前热身：让PyTorch完成必要的初始化 （如CUDA Kernel编译、显存分配等） """ self.model.eval() with torch.no_grad(): for _ in range(self.config.warmup_iterations): _ = self.model(dummy_input) def capture( self, dummy_input: torch.Tensor ) -> None: """ 捕获CUDA Graph： 在torch.cuda.graph上下文中执行前向传播， 所有CUDA操作被记录到Graph中 """ if not self.config.enable_graph: return self.warmup(dummy_input) # 分配静态输入输出张量（Graph重放时复用这些张量） self.static_input = torch.empty_like(dummy_input) self.static_input.copy_(dummy_input) self.static_output = torch.empty( dummy_input.shape[0], self.config.output_seq_len, dtype=dummy_input.dtype, device=dummy_input.device ) # 捕获Graph self.graph = torch.cuda.CUDAGraph() with torch.cuda.graph(self.graph): self.static_output = self.model(self.static_input) self.is_captured = True print(f"CUDA Graph捕获完成，输入形状: {dummy_input.shape}") def forward(self, input_tensor: torch.Tensor) -> torch.Tensor: """ 前向传播： 如果已捕获Graph，用graph.replay()（无Kernel启动开销） 否则用普通推理 """ if self.is_captured and self.graph is not None: # 将数据拷贝到静态输入张量 self.static_input.copy_(input_tensor) # 重放Graph（单次提交，无CPU开销） self.graph.replay() # 返回静态输出的副本（避免下次推理被覆盖） return self.static_output.clone() else: # 普通推理（有Kernel启动开销） with torch.no_grad(): return self.model(input_tensor) def capture_dynamic( self, min_shape: tuple, max_shape: tuple, opt_shape: tuple ) -> None: """ 捕获支持动态形状的CUDA Graph 需要PyTorch 2.0+和CUDA 12+ 适用于输入批次大小或序列长度变化的场景 """ if torch.cuda.get_device_capability() < (8, 0): print("警告: GPU架构过旧，动态Graph可能不支持") return # 为不同形状分别捕获Graph（形状专用化） # 这是当前最实用的动态Graph方案 self.graphs_by_shape = {} for shape in [min_shape, opt_shape, max_shape]: dummy = torch.randn( *shape, device="cuda", dtype=next(self.model.parameters()).dtype ) g = torch.cuda.CUDAGraph() static_in = torch.empty_like(dummy) static_out = torch.empty( shape[0], self.config.output_seq_len, device="cuda", dtype=dummy.dtype ) with torch.cuda.graph(g): static_out = self.model(static_in) self.graphs_by_shape[shape] = { "graph": g, "static_in": static_in, "static_out": static_out, } print(f"动态CUDA Graph捕获完成，支持形状: " f"{list(self.graphs_by_shape.keys())}") def forward_dynamic( self, input_tensor: torch.Tensor ) -> torch.Tensor: """动态形状的前向传播：根据输入形状选择对应的Graph""" shape = tuple(input_tensor.shape) # 找到最匹配的已捕获形状 best_shape = None min_diff = float("inf") for s in self.graphs_by_shape: diff = abs(sum(s) - sum(shape)) if diff < min_diff: min_diff = diff best_shape = s if best_shape is None: # 没有匹配的Graph，回退到普通推理 with torch.no_grad(): return self.model(input_tensor) entry = self.graphs_by_shape[best_shape] entry["static_in"].copy_(input_tensor) entry["graph"].replay() return entry["static_out"].clone() # ============ 与大模型推理集成的完整示例 ============ class GraphOptimizedLLMInference: """ 用CUDA Graph优化的LLM推理 适用于自回归生成场景（每次生成一个Token） """ def __init__(self, model, tokenizer, config: CUDAGraphConfig): self.model = model self.tokenizer = tokenizer self.config = config self.graph_wrapper = CUDAGraphWrapper(model, config) def prefill(self, input_ids: torch.Tensor) -> torch.Tensor: """ Prefill阶段：处理完整输入序列 这个阶段输入形状固定，适合CUDA Graph捕获 """ if not self.graph_wrapper.is_captured: dummy = torch.randint( 0, 32000, (1, self.config.input_seq_len), device="cuda" ) self.graph_wrapper.capture(dummy) return self.graph_wrapper.forward(input_ids) def decode_one_step( self, input_ids: torch.Tensor, past_kv ) -> torch.Tensor: """ Decode阶段：每次生成一个Token 输入形状=[batch, 1]，适合CUDA Graph捕获 """ # 为Decode阶段单独捕获Graph（输入形状不同） if not hasattr(self, "decode_graph"): dummy = torch.randint( 0, 32000, (1, 1), device="cuda" ) self.decode_graph = CUDAGraphWrapper(self.model, self.config) self.decode_graph.capture(dummy) return self.decode_graph.forward(input_ids) # 性能对比测试 def benchmark_cuda_graph( model: nn.Module, input_shape: tuple, num_iterations: int = 100 ) -> dict: """对比普通推理 vs CUDA Graph推理的性能""" device = next(model.parameters()).device dummy = torch.randn(*input_shape, device=device) config = CUDAGraphConfig(enable_graph=True) wrapper = CUDAGraphWrapper(model, config) wrapper.capture(dummy) # 预热 for _ in range(10): _ = model(dummy) _ = wrapper.forward(dummy) # 测试普通推理延迟 torch.cuda.synchronize() start = torch.cuda.Event(enable_timing=True) end = torch.cuda.Event(enable_timing=True) start.record() for _ in range(num_iterations): _ = model(dummy) end.record() torch.cuda.synchronize() vanilla_latency = start.elapsed_time(end) / num_iterations # 测试CUDA Graph推理延迟 start.record() for _ in range(num_iterations): _ = wrapper.forward(dummy) end.record() torch.cuda.synchronize() graph_latency = start.elapsed_time(end) / num_iterations return { "vanilla_latency_ms": vanilla_latency, "graph_latency_ms": graph_latency, "speedup": vanilla_latency / graph_latency, } if __name__ == "__main__": # 简单模型性能测试 model = nn.Sequential( nn.Linear(512, 1024), nn.ReLU(), nn.Linear(1024, 512), nn.ReLU(), nn.Linear(512, 10), ).cuda() result = benchmark_cuda_graph( model, input_shape=(8, 512), num_iterations=200 ) print(f"普通推理延迟: {result['vanilla_latency_ms']:.3f} ms") print(f"CUDA Graph延迟: {result['graph_latency_ms']:.3f} ms") print(f"加速比: {result['speedup']:.2f}x")

三、TensorRT优化实战：从ONNX到Engine的完整生产流水线

TensorRT是NVIDIA的推理加速库。
它将训练好的模型转换为高度优化的推理Engine。
转换过程包括：层融合、精度校准、核函数自动调优、显存优化。

生产环境的TensorRT部署流程分四步。
第一步：将模型导出为ONNX格式。
PyTorch用torch.onnx.export，TensorFlow用tf2onnx。
ONNX是模型的中间表示，与框架无关。

第二步：用TensorRT解析ONNX文件。
解析后得到一个计算图表示。
TensorRT对这个计算图做优化。

第三步：配置Builder参数并构建Engine。
关键参数：精度模式（FP32/FP16/INT8）、最大Batch Size、工作空间大小。
构建过程耗时较长（大模型需10~30分钟），但只需构建一次。
构建好的Engine序列化到磁盘，运行时直接加载。

第四步：运行时反序列化Engine并推理。
加载Engine只需几秒，然后创建Execution Context执行推理。

""" TensorRT生产级部署的完整流水线 包含：ONNX导出、Engine构建、INT8校准、推理封装 适用于PyTorch模型的TensorRT加速部署 """ import torch import tensorrt as trt import numpy as np from pathlib import Path class TensorRTInfer: """ TensorRT推理封装： 支持FP16、INT8精度，自动处理动态形状 生产环境推荐用此封装替代直接调用TensorRT API """ def __init__(self, engine_path: str): self.logger = trt.Logger(trt.Logger.WARNING) self.runtime = trt.Runtime(self.logger) self.engine = self._load_engine(engine_path) self.context = self.engine.create_execution_context() self.stream = torch.cuda.Stream() # 分配CUDA缓冲区 self.buffers = self._allocate_buffers() def _load_engine(self, engine_path: str) -> trt.ICudaEngine: """从文件加载序列化后的TensorRT Engine""" with open(engine_path, "rb") as f: engine_data = f.read() engine = self.runtime.deserialize_cuda_engine(engine_data) if engine is None: raise RuntimeError("加载TensorRT Engine失败") return engine def _allocate_buffers(self) -> dict: """为Engine的输入输出分配CUDA缓冲区""" buffers = {} for idx in range(self.engine.num_bindings): name = self.engine.get_binding_name(idx) shape = self.engine.get_binding_shape(idx) dtype = trt.nptype(self.engine.get_binding_dtype(idx)) # 动态形状：用opt_shape替代-1 shape = tuple( s if s > 0 else 32 for s in shape ) size = int(np.prod(shape)) buf = torch.empty( size, dtype=torch.from_numpy( np.array([1], dtype=dtype) ).dtype, device="cuda" ).contiguous() buffers[name] = buf return buffers def infer(self, input_data: torch.Tensor) -> torch.Tensor: """执行推理""" input_name = self.engine.get_binding_name(0) self.buffers[input_name].copy_(input_data.reshape(-1)) # 设置动态形状（如果支持） if self.engine.has_implicit_batch_dimension: pass # 隐式Batch，无需设置 else: pass # 显式Batch，需设置opt_profile # 执行推理 self.context.execute_v2( [buf.data_ptr() for buf in self.buffers.values()] ) # 获取输出 output_name = self.engine.get_binding_name( self.engine.num_bindings - 1 ) return self.buffers[output_name].clone() class TRTEngineBuilder: """ TensorRT Engine构建器： 从ONNX构建优化后的TensorRT Engine 支持FP16、INT8、动态形状 """ def __init__(self, onnx_path: str, workspace_gb: int = 4): self.onnx_path = onnx_path self.workspace_gb = workspace_gb self.logger = trt.Logger(trt.Logger.INFO) self.builder = trt.Builder(self.logger) self.network = None self.parser = None self.config = None def build( self, engine_path: str, fp16: bool = True, int8: bool = False, calib_data: Optional[list[np.ndarray]] = None, dynamic_shapes: Optional[dict] = None, ) -> None: """ 构建TensorRT Engine并序列化到文件 Args: engine_path: 输出Engine文件路径 fp16: 是否启用FP16精度（推荐开启，加速2x，精度损失<1%） int8: 是否启用INT8精度（需要校准数据集，精度损失2%~5%） calib_data: INT8校准数据（仅int8=True时需要） dynamic_shapes: 动态形状配置，如{"min": [1,3,224,224], "opt": [4,3,224,224], "max": [8,3,224,224]} """ # 解析ONNX network_flags = 1 << int( trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH) self.network = self.builder.create_network(network_flags) self.parser = trt.OnnxParser(self.network, self.logger) with open(self.onnx_path, "rb") as f: if not self.parser.parse(f.read()): for i in range(self.parser.num_errors): print(f"ONNX解析错误: {self.parser.get_error(i)}") raise RuntimeError("ONNX解析失败") # 配置Builder self.config = self.builder.create_builder_config() self.config.set_memory_pool_limit( trt.MemoryPoolType.WORKSPACE, self.workspace_gb * (1 << 30) ) # FP16配置 if fp16 and self.builder.platform_has_fast_fp16: self.config.set_flag(trt.BuilderFlag.FP16) print("已启用FP16精度模式") # INT8配置 if int8: if not self.builder.platform_has_fast_int8: print("警告: 当前GPU不支持快速INT8") self.config.set_flag(trt.BuilderFlag.INT8) if calib_data is not None: # 配置INT8校准器（生产环境必需） calibrator = self._create_calibrator(calib_data) self.config.int8_calibrator = calibrator print("已配置INT8校准器") # 动态形状配置 if dynamic_shapes: profile = self.builder.create_optimization_profile() for name, shapes in dynamic_shapes.items(): input_idx = self.network.get_input(0) profile.set_shape( input_idx.name, min=dynamic_shapes["min"], opt=dynamic_shapes["opt"], max=dynamic_shapes["max"], ) self.config.add_optimization_profile(profile) print(f"已配置动态形状: min={dynamic_shapes['min']}, " f"opt={dynamic_shapes['opt']}, " f"max={dynamic_shapes['max']}") # 设置最大Batch Size max_batch = dynamic_shapes["max"][0] if dynamic_shapes else 8 self.config.set_flag(trt.BuilderFlag.DISABLE_TIMING_CACHE) # 构建Engine print("开始构建TensorRT Engine（可能需要10~30分钟）...") serialized_engine = self.builder.build_serialized_network( self.network, self.config ) if serialized_engine is None: raise RuntimeError("TensorRT Engine构建失败") # 序列化到文件 with open(engine_path, "wb") as f: f.write(serialized_engine) print(f"TensorRT Engine已保存到: {engine_path}") def _create_calibrator( self, calib_data: list[np.ndarray] ) -> trt.IInt8Calibrator: """创建INT8校准器（简化版本）""" # 生产环境应使用trt.IInt8EntropyCalibrator2 # 并提供真实训练数据的代表性样本（约1000个Batch） class SimpleCalibrator(trt.IInt8EntropyCalibrator2): def __init__(self, data): super().__init__() self.data = data self.index = 0 def get_batch_size(self): return 32 def get_batch(self, names, p_str): if self.index >= len(self.data): return None batch = self.data[self.index] self.index += 1 return [batch.ctypes.data] def read_calibration_cache(self): return None def write_calibration_cache(self, cache): pass return SimpleCalibrator(calib_data) # 完整使用示例 def build_and_deploy_trt( pytorch_model: torch.nn.Module, sample_input: torch.Tensor, output_path: str ) -> None: """ 完整的PyTorch → ONNX → TensorRT → 推理流水线 """ # 第一步：导出ONNX onnx_path = output_path + ".onnx" pytorch_model.eval() with torch.no_grad(): torch.onnx.export( pytorch_model, sample_input, onnx_path, input_names=["input"], output_names=["output"], dynamic_axes={ "input": {0: "batch"}, "output": {0: "batch"} }, opset_version=17, do_constant_folding=True, ) print(f"ONNX已导出: {onnx_path}") # 第二步：构建TensorRT Engine builder = TRTEngineBuilder(onnx_path) engine_path = output_path + ".engine" builder.build( engine_path, fp16=True, int8=False, # 生产环境建议用INT8，但需准备校准数据 dynamic_shapes={ "min": [1, 3, 224, 224], "opt": [4, 3, 224, 224], "max": [8, 3, 224, 224], } ) # 第三步：加载Engine并推理 infer = TensorRTInfer(engine_path) test_input = torch.randn(4, 3, 224, 224).cuda() output = infer.infer(test_input) print(f"TensorRT推理完成，输出形状: {output.shape}") if __name__ == "__main__": # 示例：ResNet50的TensorRT加速 import torchvision.models as models model = models.resnet50(pretrained=False).cuda() dummy = torch.randn(4, 3, 224, 224).cuda() build_and_deploy_trt( model, dummy, output_path="/tmp/resnet50_trt" )

四、CUDA Graph与TensorRT的集成方案及延迟对比数据

CUDA Graph和TensorRT可以叠加使用。
TensorRT负责优化单个Kernel的执行效率。
CUDA Graph负责减少Kernel启动开销。
两者叠加，效果最佳。

集成方案有两种。
方案一：先构建TensorRT Engine，再用CUDA Graph捕获推理过程。
TensorRT Inference的Python代码在with torch.cuda.graph()上下文中执行。
这样整个TensorRT推理过程（包括数据拷贝、Kernel启动、后处理）都被捕获。

方案二：用TensorRT的Native CUDA Graph支持。
TensorRT 8.4+支持在Builder阶段捕获CUDA Graph。
构建好的Engine内部已经包含了优化后的CUDA Graph。
运行时直接execute_v3()，性能最优。

实际延迟数据（基于ResNet50，Batch=8，A100 GPU）：

• 纯PyTorch推理：平均延迟4.2ms
• TensorRT FP16：平均延迟1.8ms（加速2.3倍）
• TensorRT FP16 + CUDA Graph：平均延迟0.9ms（加速4.7倍）

#!/bin/bash # CUDA Graph + TensorRT集成部署的生产级脚本 # 适用场景：在线推理服务（如推荐系统、CV推理） # 依赖：CUDA 12.1+, TensorRT 8.6+, PyTorch 2.0+ set -euo pipefail MODEL_NAME="resnet50" BATCH_SIZE=8 PRECISION="fp16" # fp32 | fp16 | int8 ENABLE_CUDA_GRAPH=1 # 0=禁用, 1=启用 DYNAMIC_BATCH=1 # 是否支持动态Batch echo "===== 开始部署 ${MODEL_NAME} 推理服务 =====" # ============ 第一步：导出ONNX ============ echo "步骤1/4: 导出ONNX模型..." python3 - <<EOF import torch import torchvision.models as models model = models.${MODEL_NAME}(pretrained=True).cuda().eval() dummy = torch.randn(${BATCH_SIZE}, 3, 224, 224).cuda() torch.onnx.export( model, dummy, "/tmp/${MODEL_NAME}.onnx", input_names=["input"], output_names=["output"], dynamic_axes={"input": {0: "batch"}, "output": {0: "batch"}} if ${DYNAMIC_BATCH} else None, opset_version=17, ) print("ONNX导出完成") EOF # ============ 第二步：构建TensorRT Engine ============ echo "步骤2/4: 构建TensorRT Engine（可能需要10~30分钟）..." trtexec \ --onnx=/tmp/${MODEL_NAME}.onnx \ --saveEngine=/tmp/${MODEL_NAME}.engine \ --fp16 \ --workspace=4096 \ --optShapes=input:${BATCH_SIZE}x3x224x224 \ --maxShapes=input:32x3x224x224 \ --minShapes=input:1x3x224x224 \ echo "TensorRT Engine构建完成" # ============ 第三步：封装推理服务 ============ echo "步骤3/4: 启动推理服务（TensorRT + CUDA Graph）..." cat > /tmp/trt_server.py <<'PYEOF' import torch import tensorrt as trt from fastapi import FastAPI import uvicorn app = FastAPI() class OptimizedTRTServer: def __init__(self, engine_path: str, use_graph: bool): self.logger = trt.Logger(trt.Logger.INFO) self.runtime = trt.Runtime(self.logger) with open(engine_path, "rb") as f: self.engine = self.runtime.deserialize_cuda_engine(f.read()) self.context = self.engine.create_execution_context() self.use_graph = use_graph if use_graph: # 捕获CUDA Graph self.graph = torch.cuda.CUDAGraph() dummy = torch.randn(8, 3, 224, 224).cuda() self.static_in = torch.empty_like(dummy) self.static_out = torch.empty(8, 1000).cuda() with torch.cuda.graph(self.graph): self.context.execute_v2([ self.static_in.data_ptr(), self.static_out.data_ptr(), ]) def infer(self, input_tensor: torch.Tensor) -> torch.Tensor: if self.use_graph: self.static_in.copy_(input_tensor) self.graph.replay() return self.static_out.clone() else: # 普通推理 output = torch.empty(input_tensor.shape[0], 1000).cuda() self.context.execute_v2([ input_tensor.data_ptr(), output.data_ptr(), ]) return output server = OptimizedTRTServer("/tmp/${MODEL_NAME}.engine", ${ENABLE_CUDA_GRAPH}) @app.post("/infer") def infer_endpoint(data: list[list[list[float]]]): input_tensor = torch.tensor(data).cuda() output = server.infer(input_tensor) return {"output": output.cpu().tolist()} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8080) PYEOF # 启动服务（后台） python3 /tmp/trt_server.py & SERVER_PID=$! echo "推理服务已启动，PID=${SERVER_PID}" # ============ 第四步：性能基准测试 ============ echo "步骤4/4: 性能基准测试..." python3 - <<EOF import requests import torch import time # 预热 data = torch.randn(8, 3, 224, 224).tolist() for _ in range(10): requests.post("http://localhost:8080/infer", json={"data": data[:1]}) # 基准测试 latencies = [] for _ in range(100): start = time.time() requests.post("http://localhost:8080/infer", json={"data": data}) latencies.append((time.time() - start) * 1000) p50 = sorted(latencies)[len(latencies)//2] p99 = sorted(latencies)[int(len(latencies)*0.99)] print(f"推理延迟: P50={p50:.2f}ms, P99={p99:.2f}ms") print(f"吞吐量: {1000/p50:.1f} req/s") EOF echo "===== 部署完成 =====" echo "CUDA Graph: ${ENABLE_CUDA_GRAPH}" echo "精度模式: ${PRECISION}"

五、总结

1. GPU推理优化的两大核心路径：TensorRT让每个Kernel跑得更快（层融合减少Kernel调用次数、FP16/INT8量化降低计算精度提升吞吐量2~4倍、核函数自动调优选择最优CUDA Kernel实现），CUDA Graph让Kernel启动开销消失（将多次Kernel启动捕获为计算图，GPU自主调度无需CPU逐次提交，延迟降低30%60%），两者叠加在生产环境中可实现25倍的端到端推理加速，是GPU推理优化的标准实践

2. CUDA Graph的生产级使用要点：用torch.cuda.graph上下文管理器捕获计算图（PyTorch 1.10+）、捕获前必须热身3~5次（让PyTorch完成CUDA Kernel编译和显存分配）、输入形状必须固定（动态形状需为不同形状分别捕获Graph或用TensorRT的Dynamic Shape）、重放用graph.replay()（单次提交零CPU开销），大模型推理中Prefill阶段（输入形状固定）和Decode阶段（输入形状=[batch, 1]）应分别捕获独立的CUDA Graph

3. TensorRT生产部署的完整流水线：ONNX导出（torch.onnx.export注意opset_version=17和dynamic_axes配置）→ Builder构建Engine（关键参数：FP16=启用、workspace=4GB、INT8需校准数据集1000+Batch、动态形状配置min/opt/max）→ 序列化Engine到磁盘（构建耗时10~30分钟但只需一次）→ 运行时反序列化并推理（加载耗时几秒，创建Execution Context执行），构建好的Engine与GPU架构绑定（A100构建的Engine不能在V100上用），需为每个部署目标单独构建

4. FP16 vs INT8的精度与性能权衡：FP16精度损失<1%（绝大多数场景可接受），加速比2x，无需校准数据，推荐生产环境默认开启；INT8精度损失2%~5%（对精度敏感场景需谨慎），加速比4x，需要提供代表性校准数据集（约1000个Batch的真实训练数据），校准质量直接影响INT8模型的精度，生产环境建议用trt.IInt8EntropyCalibrator2做CALIBRATOR

5. 端侧AI部署的硬件适配策略：数据中心GPU（A100/H100）优先TensorRT + CUDA Graph，延迟要求<10ms；边缘设备（Jetson Orin）用TensorRT + DeepStream SDK，支持视频流实时推理；消费级GPU（RTX 4090）用ONNX Runtime + TensorRT Execution Provider，部署最简单；移动端（骁龙8 Gen3）用TensorRT + NVIDIA TAO Toolkit训练并导出为TensorRT Engine，再用TensorRT Edge Deployment工具链部署到边缘，不同硬件的Engine不兼容，必须为每个目标硬件单独构建