模型部署与推理优化

# PyTorch 模型导出为 ONNX 格式 import torch import torch.onnx import torchvision.models as models # 1. 加载预训练模型 model = models.resnet50(pretrained=True) model.eval() # 2. 创建示例输入 (batch_size=1, 3通道, 224x224) dummy_input = torch.randn(1, 3, 224, 224) # 3. 导出 ONNX 模型 torch.onnx.export( model, # 模型实例 dummy_input, # 示例输入 "resnet50.onnx", # 输出路径 export_params=True, # 导出训练好的参数 opset_version=17, # ONNX opset 版本 do_constant_folding=True, # 常量折叠优化 input_names=["input"], # 输入名称 output_names=["output"], # 输出名称 dynamic_axes={ # 动态轴配置 "input": {0: "batch_size"}, "output": {0: "batch_size"} } ) print("ONNX 模型导出成功：resnet50.onnx")

# NLP 模型动态轴配置示例（BERT 类模型） import torch # 假设我们有一个 BERT 分类模型 class BertClassifier(torch.nn.Module): def __init__(self, bert_model): super().__init__() self.bert = bert_model self.classifier = torch.nn.Linear(768, 2) def forward(self, input_ids, attention_mask): output = self.bert(input_ids, attention_mask=attention_mask) return self.classifier(output.pooler_output) model = BertClassifier(bert_model) model.eval() # 导出时配置动态轴——batch_size 和序列长度均为动态 torch.onnx.export( model, (torch.randint(0, 1000, (1, 128)), # input_ids: (batch, seq_len) torch.ones(1, 128, dtype=torch.long)), # attention_mask "bert_classifier.onnx", opset_version=17, input_names=["input_ids", "attention_mask"], output_names=["logits"], dynamic_axes={ "input_ids": {0: "batch_size", 1: "seq_len"}, "attention_mask": {0: "batch_size", 1: "seq_len"}, "logits": {0: "batch_size"} } )

# TensorFlow 模型转 ONNX # pip install tf2onnx import tensorflow as tf import tf2onnx # 加载 Keras 模型 model = tf.keras.models.load_model("my_model.h5") # 指定输入签名 spec = (tf.TensorSpec((None, 224, 224, 3), tf.float32, name="input"),) # 转换并保存 ONNX output_path = model.output_names[0] + ".onnx" model_proto, _ = tf2onnx.convert.from_keras( model, input_signature=spec, opset=17, output_path=output_path ) print(f"转换完成: {output_path}")

# 验证 ONNX 模型完整性 import onnx import onnx.checker import onnx.helper # 加载模型 model = onnx.load("resnet50.onnx") # 检查模型格式完整性 onnx.checker.check_model(model) print("模型格式检查通过") # 查看模型信息 print(f"IR 版本: {model.ir_version}") print(f"Producer: {model.producer_name}") print(f"Opset 版本: {model.opset_import[0].version}") # 查看输入输出信息 for inp in model.graph.input: print(f"输入: {inp.name}, 形状: {[d.dim_value for d in inp.type.tensor_type.shape.dim]}") for out in model.graph.output: print(f"输出: {out.name}, 形状: {[d.dim_value for d in out.type.tensor_type.shape.dim]}")

# ONNX 模型优化 # pip install onnxoptimizer import onnx import onnxoptimizer # 加载原始模型 model = onnx.load("resnet50.onnx") # 应用所有可用优化pass passes = onnxoptimizer.get_available_passes() print(f"可用优化pass数: {len(passes)}") # 执行优化 optimized_model = onnxoptimizer.optimize( model, passes=[ "fuse_consecutive_transposes", "fuse_transpose_into_gemm", "eliminate_identity", "eliminate_nop_transpose", "eliminate_nop_pad", "fuse_matmul_add_bias_into_gemm", ] ) # 保存优化后的模型 onnx.save(optimized_model, "resnet50_optimized.onnx") print("模型优化完成，已保存: resnet50_optimized.onnx") # 比较优化前后的图规模 print(f"原始节点数: {len(model.graph.node)}") print(f"优化后节点数: {len(optimized_model.graph.node)}")

# ONNX Runtime 基础推理 # pip install onnxruntime-gpu # GPU 版本 # pip install onnxruntime # CPU 版本 import onnxruntime as ort import numpy as np # 创建推理会话 session = ort.InferenceSession("resnet50.onnx") # 查看输入输出信息 for inp in session.get_inputs(): print(f"输入: {inp.name}, 形状: {inp.shape}, 类型: {inp.type}") for out in session.get_outputs(): print(f"输出: {out.name}, 形状: {out.shape}, 类型: {out.type}") # 准备输入数据 input_data = np.random.randn(1, 3, 224, 224).astype(np.float32) # 执行推理 outputs = session.run( output_names=["output"], input_feed={"input": input_data} ) print(f"推理结果形状: {outputs[0].shape}") print(f"推理结果: {outputs[0]}")

# 配置不同的执行提供者 import onnxruntime as ort # CPU 推理 session_cpu = ort.InferenceSession( "resnet50.onnx", providers=["CPUExecutionProvider"] ) # CUDA GPU 推理 session_cuda = ort.InferenceSession( "resnet50.onnx", providers=[ ("CUDAExecutionProvider", { "device_id": 0, "arena_extend_strategy": "kNextPowerOfTwo", "gpu_mem_limit": 4 * 1024 * 1024 * 1024, # 4GB "cudnn_conv_algo_search": "EXHAUSTIVE", }), "CPUExecutionProvider" # 回退方案 ] ) # TensorRT 加速推理 session_trt = ort.InferenceSession( "resnet50.onnx", providers=[ ("TensorrtExecutionProvider", { "device_id": 0, "trt_max_workspace_size": 1073741824, # 1GB "trt_fp16_enable": True, "trt_int8_enable": False, "trt_engine_cache_enable": True, "trt_engine_cache_path": "./trt_cache", }), "CUDAExecutionProvider", "CPUExecutionProvider" ] ) # OpenVINO 推理（Intel CPU/GPU） session_ov = ort.InferenceSession( "resnet50.onnx", providers=[ ("OpenVINOExecutionProvider", { "device_type": "CPU_FP32", }), "CPUExecutionProvider" ] ) print("所有会话创建成功")

# ONNX Runtime 会话配置与性能优化 import onnxruntime as ort import psutil # 配置会话选项 sess_options = ort.SessionOptions() # 设置线程数 sess_options.intra_op_num_threads = psutil.cpu_count(logical=True) sess_options.inter_op_num_threads = 2 # 启用图优化级别 sess_options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL # 启用内存优化 sess_options.enable_mem_pattern = True sess_options.enable_cpu_mem_arena = True # 启用序列化优化（缓存优化后的图） sess_options.optimized_model_filepath = "./optimized_resnet50.onnx" # 设置日志级别 sess_options.log_severity_level = 3 # 0:Verbose, 1:Info, 2:Warning, 3:Error # 创建会话 session = ort.InferenceSession( "resnet50.onnx", sess_options=sess_options, providers=["CUDAExecutionProvider", "CPUExecutionProvider"] ) print(f"会话已创建, 使用提供者: {session.get_providers()}")

# ONNX 模型量化（动态量化 + 静态量化） import onnx from onnxruntime.quantization import quantize_dynamic, QuantType from onnxruntime.quantization import quantize_static, QuantizationMode from onnxruntime.quantization import CalibrationMethod # ---------- 动态量化 ---------- # 仅量化权重，不需要校准数据，适合 RNN/LSTM 等架构 quantize_dynamic( model_input="resnet50.onnx", model_output="resnet50_dynamic_int8.onnx", weight_type=QuantType.QInt8, ) # ---------- 静态量化 ---------- # 同时量化权重和激活值，需要校准数据集，效果更好 # 准备校准数据读取器 import numpy as np class CalibrationDataReader: def __init__(self, num_samples=100): self.data = [np.random.randn(1, 3, 224, 224).astype(np.float32) for _ in range(num_samples)] self.idx = 0 def get_next(self): if self.idx < len(self.data): datum = self.data[self.idx] self.idx += 1 return {"input": datum} return None def rewind(self): self.idx = 0 # 执行静态量化 quantize_static( model_input="resnet50.onnx", model_output="resnet50_static_int8.onnx", calibration_data_reader=CalibrationDataReader(num_samples=200), quant_format=QuantType.QInt8, per_channel=True, calibration_method=CalibrationMethod.MinMax, ) print("量化完成：resnet50_static_int8.onnx")

# 批量推理与内存管理 import onnxruntime as ort import numpy as np # IoTBinding — 支持用户管理 GPU 显存，减少内存拷贝 session = ort.InferenceSession( "resnet50.onnx", providers=["CUDAExecutionProvider", "CPUExecutionProvider"] ) # 准备批量数据 batch_size = 32 input_data = np.random.randn(batch_size, 3, 224, 224).astype(np.float32) # 方式1：直接推理（自动管理内存） outputs = session.run(["output"], {"input": input_data}) print(f"批量推理结果形状: {outputs[0].shape}") # 方式2：使用 IoTBinding（精细控制内存） io_binding = session.io_binding() # 将输入绑定到 GPU input_tensor = ort.OrtValue.ortvalue_from_numpy(input_data, "cuda", 0) io_binding.bind_ortvalue_input("input", input_tensor) # 将输出绑定到 GPU io_binding.bind_output("output", "cuda") # 执行推理 session.run_with_iobinding(io_binding) # 获取输出（从 GPU 拷贝回 CPU） output = io_binding.get_outputs()[0].numpy() print(f"IoBinding 推理结果形状: {output.shape}")

# TensorRT 从 ONNX 构建引擎 # pip install tensorrt import tensorrt as trt import numpy as np # 创建 logger 和 builder logger = trt.Logger(trt.Logger.WARNING) builder = trt.Builder(logger) # 创建网络定义（显式batch） network = builder.create_network( 1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH) ) # 创建 ONNX 解析器 parser = trt.OnnxParser(network, logger) # 解析 ONNX 模型 with open("resnet50.onnx", "rb") as f: if not parser.parse(f.read()): for err in range(parser.num_errors): print(f"解析错误 {err}: {parser.get_error(err)}") raise RuntimeError("ONNX 解析失败") print(f"网络层数: {network.num_layers}") print(f"网络输入数: {network.num_inputs}") print(f"网络输出数: {network.num_outputs}") # 配置 builder config = builder.create_builder_config() config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 30) # 1GB # 启用 FP16 if builder.platform_has_fast_fp16: config.set_flag(trt.BuilderFlag.FP16) print("已启用 FP16 精度") # 设置动态形状范围 (若模型有动态轴) profile = builder.create_optimization_profile() profile.set_shape( "input", min=(1, 3, 224, 224), opt=(8, 3, 224, 224), max=(32, 3, 224, 224) ) config.add_optimization_profile(profile) # 构建引擎 serialized_engine = builder.build_serialized_network(network, config) if serialized_engine is None: raise RuntimeError("引擎构建失败") # 序列化保存引擎 with open("resnet50.engine", "wb") as f: f.write(serialized_engine) print("TensorRT 引擎已保存: resnet50.engine")

# TensorRT 引擎加载与推理 import tensorrt as trt import numpy as np import pycuda.driver as cuda import pycuda.autoinit logger = trt.Logger(trt.Logger.WARNING) # 反序列化引擎 with open("resnet50.engine", "rb") as f: serialized_engine = f.read() runtime = trt.Runtime(logger) engine = runtime.deserialize_cuda_engine(serialized_engine) # 创建执行上下文 context = engine.create_execution_context() # 设置动态形状 (若需要) context.set_binding_shape(0, (1, 3, 224, 224)) # 分配 GPU 内存 input_idx = engine.get_binding_index("input") output_idx = engine.get_binding_index("output") input_shape = (1, 3, 224, 224) output_shape = (1, 1000) input_size = trt.volume(input_shape) * np.dtype(np.float32).itemsize output_size = trt.volume(output_shape) * np.dtype(np.float32).itemsize d_input = cuda.mem_alloc(input_size) d_output = cuda.mem_alloc(output_size) # 准备主机端数据 h_input = np.random.randn(*input_shape).astype(np.float32) h_output = np.empty(output_shape, dtype=np.float32) # 创建 CUDA 流 stream = cuda.Stream() # 执行推理 cuda.memcpy_htod_async(d_input, h_input, stream) context.execute_async_v2( bindings=[int(d_input), int(d_output)], stream_handle=stream.handle ) cuda.memcpy_dtoh_async(h_output, d_output, stream) stream.synchronize() print(f"推理完成，输出形状: {h_output.shape}") print(f"Top-5 索引: {np.argsort(h_output[0])[-5:][::-1]}")

# TensorRT INT8 量化校准 import tensorrt as trt # 自定义校准器 class MyCalibrator(trt.IInt8EntropyCalibrator2): def __init__(self, batch_size, num_batches): super().__init__() self.batch_size = batch_size self.num_batches = num_batches self.batch_idx = 0 self.device_input = cuda.mem_alloc( batch_size * 3 * 224 * 224 * np.dtype(np.float32).itemsize ) def get_batch_size(self): return self.batch_size def get_batch(self, names): if self.batch_idx >= self.num_batches: return None # 生成校准数据（实际使用时应加载真实数据） data = np.random.randn( self.batch_size, 3, 224, 224 ).astype(np.float32) cuda.memcpy_htod(self.device_input, data) self.batch_idx += 1 return [int(self.device_input)] def read_calibration_cache(self): # 读取缓存以加速多次构建 try: with open("calibration.cache", "rb") as f: return f.read() except FileNotFoundError: return None def write_calibration_cache(self, cache): with open("calibration.cache", "wb") as f: f.write(cache) # 构建 INT8 引擎 builder = trt.Builder(logger) network = builder.create_network( 1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH) ) # 解析 ONNX 模型... # (省略解析器部分，与之前相同) config = builder.create_builder_config() config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 30) # 启用 INT8 并设置校准器 config.set_flag(trt.BuilderFlag.INT8) config.int8_calibrator = MyCalibrator( batch_size=8, num_batches=50 ) # 也可同时启用 FP16 作为 INT8 的回退 if builder.platform_has_fast_fp16: config.set_flag(trt.BuilderFlag.FP16) engine_bytes = builder.build_serialized_network(network, config) with open("resnet50_int8.engine", "wb") as f: f.write(engine_bytes) print("INT8 TensorRT 引擎构建完成")

# Triton Inference Server 模型仓库配置 # 目录结构: # model_repository/ # resnet50/ # 1/ # model.plan # TensorRT 引擎文件 # config.pbtxt # 模型配置文件 # config.pbtxt 示例 # name: "resnet50" # platform: "tensorrt_plan" # max_batch_size: 0 # input [ # { # name: "input" # data_type: TYPE_FP32 # dims: [1, 3, 224, 224] # } # ] # output [ # { # name: "output" # data_type: TYPE_FP32 # dims: [1, 1000] # } # ] # instance_group [ # { count: 2 kind: KIND_GPU } # 2个GPU实例 # ] # dynamic_batching {} # ---------- Triton Python 客户端 ---------- import tritonclient.http as httpclient import numpy as np # 连接到 Triton 服务器 client = httpclient.InferenceServerClient(url="localhost:8000") # 检查模型状态 print(client.get_model_repository_index()) print(client.is_model_ready("resnet50")) # 构建推理请求 input_data = np.random.randn(1, 3, 224, 224).astype(np.float32) inputs = [httpclient.InferInput("input", input_data.shape, "FP32")] inputs[0].set_data_from_numpy(input_data) outputs = [httpclient.InferRequestedOutput("output")] # 执行推理 results = client.infer("resnet50", inputs, outputs=outputs) output = results.as_numpy("output") print(f"Triton 推理结果形状: {output.shape}")

# TensorFlow 模型转 TFLite import tensorflow as tf # 加载 Keras 模型 model = tf.keras.models.load_model("mobilenetv2.h5") # 转换为 TFLite converter = tf.lite.TFLiteConverter.from_keras_model(model) # 配置量化选项 converter.optimizations = [tf.lite.Optimize.DEFAULT] # 权重量化为 INT8 converter.target_spec.supported_types = [tf.float16] # 或者全 INT8 量化（需要代表性数据集） # converter.representative_dataset = representative_dataset # 转换并保存 tflite_model = converter.convert() with open("mobilenetv2.tflite", "wb") as f: f.write(tflite_model) print(f"TFLite 模型大小: {len(tflite_model) / 1024:.2f} KB")

# TFLite NNAPI 委托 — 利用 Android 硬件加速 import tensorflow as tf # 加载 TFLite 模型 interpreter = tf.lite.Interpreter( model_path="mobilenetv2.tflite", experimental_delegates=[ tf.lite.experimental.NnApiDelegate( accelerator_name="qti-dsp", # 指定 DSP 加速器 cache_dir="/data/local/tmp", cache_token="mobilenetv2_cache" ) ] ) interpreter.allocate_tensors() # 获取输入输出信息 input_details = interpreter.get_input_details() output_details = interpreter.get_output_details() # 执行推理 input_data = np.random.randn(1, 224, 224, 3).astype(np.float32) interpreter.set_tensor(input_details[0]["index"], input_data) interpreter.invoke() output = interpreter.get_tensor(output_details[0]["index"]) print(f"NNAPI 推理完成，输出形状: {output.shape}")

# PyTorch 模型转 CoreML # pip install coremltools import torch import coremltools as ct # 加载 PyTorch 模型 model = torch.load("mobilenetv2.pth", map_location="cpu") model.eval() # 追踪模型 example_input = torch.randn(1, 3, 224, 224) traced_model = torch.jit.trace(model, example_input) # 转换为 CoreML（MLModel 格式） mlmodel = ct.convert( traced_model, inputs=[ct.TensorType(name="input", shape=(1, 3, 224, 224))], outputs=[ct.TensorType(name="output")], convert_to="neuralnetwork", # 或 "mlprogram" (ANE加速) minimum_deployment_target=ct.target.iOS15, compute_units=ct.ComputeUnit.ALL, # CPU+GPU+ANE ) # 保存 .mlmodel 包 mlmodel.save("MobileNetV2.mlpackage") print("CoreML 模型已保存") # 使用 CoreML 进行推理 from PIL import Image import numpy as np # 加载 CoreML 模型 model = ct.models.MLModel("MobileNetV2.mlpackage") # 准备输入 image = Image.open("test.jpg").resize((224, 224)) input_dict = {"input": np.array(image).astype(np.float32).transpose(2, 0, 1)[None]} # 推理 result = model.predict(input_dict) print("CoreML 推理完成")

# Google Coral Edge TPU 编译 # 需要 edgetpu_compiler 工具 # 命令行: edgetpu_compiler mobilenetv2.tflite # Python 中执行编译 import subprocess import os result = subprocess.run( ["edgetpu_compiler", "--out_dir", "./edgetpu_models", "--num_segments", "1", "mobilenetv2.tflite"], capture_output=True, text=True ) print(result.stdout) if result.returncode == 0: print("Edge TPU 编译成功") else: print(f"编译失败: {result.stderr}") # Edge TPU 推理 from tflite_runtime.interpreter import Interpreter from tflite_runtime.interpreter import load_delegate interpreter = Interpreter( model_path="./edgetpu_models/mobilenetv2_edgetpu.tflite", experimental_delegates=[load_delegate("libedgetpu.so.1.0")] ) interpreter.allocate_tensors() print("Edge TPU 推理器已初始化")

// ONNX Runtime Web — 浏览器中运行 ONNX 模型 // npm install onnxruntime-web // Python 端：确保模型兼容 Web 后端 import onnx import onnxruntime as ort model = onnx.load("model.onnx") onnx.checker.check_model(model) // JavaScript 前端推理代码 /* import * as ort from 'onnxruntime-web'; async function runInference() { // 创建 ONNX Runtime 会话 const session = await ort.InferenceSession.create('./model.onnx', { executionProviders: ['wasm', 'webgl'], graphOptimizationLevel: 'all' }); // 准备输入张量 const input = new ort.Tensor( 'float32', new Float32Array(1 * 3 * 224 * 224), [1, 3, 224, 224] ); // 执行推理 const outputs = await session.run({ input }); console.log('推理结果:', outputs); } */ console.log("ONNX Runtime Web 部署示例 (详见注释)")

# TensorFlow 模型转 TFJS # pip install tensorflowjs import tensorflowjs as tfjs import tensorflow as tf # 加载 Keras 模型 model = tf.keras.models.load_model("mobilenetv2.h5") # 转换为 TFJS 格式 tfjs.converters.save_keras_model( model, "./tfjs_model" ) print("TFJS 模型已保存到 ./tfjs_model/") // JavaScript 前端推理 /* import * as tf from '@tensorflow/tfjs'; async function loadAndPredict() { // 加载模型 const model = await tf.loadLayersModel('./tfjs_model/model.json'); // 准备输入 const input = tf.randomNormal([1, 224, 224, 3]); // 推理 const output = model.predict(input); output.print(); // 释放张量内存 input.dispose(); output.dispose(); } */ console.log("TensorFlow.js 部署示例 (详见注释)")

# FastAPI 推理服务完整示例 # pip install fastapi uvicorn onnxruntime import io import numpy as np from PIL import Image from typing import Dict, List from pydantic import BaseModel import onnxruntime as ort from fastapi import FastAPI, File, UploadFile, HTTPException from fastapi.middleware.cors import CORSMiddleware import uvicorn # 创建应用 app = FastAPI(title="模型推理服务", version="1.0.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # 请求/响应模型 class PredictRequest(BaseModel): instances: List[List[List[List[float]]]] # batch x C x H x W class PredictResponse(BaseModel): predictions: List[List[float]] model_info: Dict[str, str] # 全局会话 session: ort.InferenceSession = None @app.on_event("startup") async def load_model(): global session try: session = ort.InferenceSession( "resnet50.onnx", providers=["CUDAExecutionProvider", "CPUExecutionProvider"] ) print(f"模型加载成功, 使用提供者: {session.get_providers()}") except Exception as e: print(f"模型加载失败: {e}") raise e @app.get("/health") async def health(): return {"status": "ok", "model_loaded": session is not None} @app.get("/metadata") async def metadata(): if session is None: raise HTTPException(503, "模型未加载") return { "inputs": [{"name": inp.name, "shape": inp.shape, "type": inp.type} for inp in session.get_inputs()], "outputs": [{"name": out.name, "shape": out.shape, "type": out.type} for out in session.get_outputs()], } @app.post("/predict", response_model=PredictResponse) async def predict(request: PredictRequest): if session is None: raise HTTPException(503, "模型未加载") input_data = np.array(request.instances, dtype=np.float32) # 前处理（例如 ImageNet 归一化） mean = np.array([0.485, 0.456, 0.406]).reshape(1, 3, 1, 1) std = np.array([0.229, 0.224, 0.225]).reshape(1, 3, 1, 1) input_data = (input_data / 255.0 - mean) / std # 推理 outputs = session.run(["output"], {"input": input_data}) return PredictResponse( predictions=outputs[0].tolist(), model_info={"name": "resnet50", "backend": session.get_providers()[0]} ) @app.post("/predict/image") async def predict_image(file: UploadFile = File(...)): # 图像上传推理 contents = await file.read() image = Image.open(io.BytesIO(contents)).resize((224, 224)) input_array = np.array(image).astype(np.float32).transpose(2, 0, 1)[None] # 归一化 mean = np.array([0.485, 0.456, 0.406]).reshape(1, 3, 1, 1) std = np.array([0.229, 0.224, 0.225]).reshape(1, 3, 1, 1) input_array = (input_array / 255.0 - mean) / std outputs = session.run(["output"], {"input": input_array}) top5 = np.argsort(outputs[0][0])[-5:][::-1] return { "filename": file.filename, "top5_predictions": top5.tolist(), "scores": outputs[0][0][top5].tolist() } if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

# Flask 推理服务（轻量级方案） from flask import Flask, request, jsonify import onnxruntime as ort import numpy as np app = Flask(__name__) # 加载模型 session = ort.InferenceSession("resnet50.onnx") @app.route("/predict", methods=["POST"]) def predict(): data = request.get_json() inputs = np.array(data["instances"], dtype=np.float32) # 推理 outputs = session.run(["output"], {"input": inputs}) return jsonify({ "predictions": outputs[0].tolist() }) @app.route("/batch_predict", methods=["POST"]) def batch_predict(): # 支持动态 batch data = request.get_json() inputs = np.array(data["instances"], dtype=np.float32) outputs = session.run(["output"], {"input": inputs}) return jsonify({ "batch_size": inputs.shape[0], "predictions": outputs[0].tolist() }) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, threads=4)

# Dockerfile — 模型推理服务容器化 # 多阶段构建：减小最终镜像体积 # 第一阶段：依赖安装 FROM python:3.10-slim as builder WORKDIR /app COPY requirements.txt . # 安装依赖 RUN pip install --no-cache-dir -r requirements.txt # 第二阶段：运行镜像 FROM nvidia/cuda:12.2-runtime-ubuntu22.04 WORKDIR /app # 复制 Python 和依赖 COPY --from=builder /usr/local/lib/python3.10 /usr/local/lib/python3.10 COPY --from=builder /usr/local/bin /usr/local/bin COPY --from=builder /app /app # 复制模型和代码 COPY app.py . COPY resnet50.onnx . # 暴露端口 EXPOSE 8000 # 启动服务 CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"] # ---------- requirements.txt ---------- # fastapi==0.104.0 # uvicorn==0.24.0 # onnxruntime-gpu==1.16.0 # numpy==1.24.0 # Pillow==10.1.0 # pydantic==2.5.0 # ---------- docker-compose.yml ---------- # version: "3.8" # services: # inference: # build: . # ports: # - "8000:8000" # deploy: # resources: # reservations: # devices: # - driver: nvidia # count: 1 # capabilities: [gpu] # 构建命令 # docker build -t inference-server:latest . # docker compose up -d print("Docker 部署配置示例 (详见注释)")

# Kubernetes 部署清单 # ---------- deployment.yaml ---------- # apiVersion: apps/v1 # kind: Deployment # metadata: # name: inference-server # spec: # replicas: 3 # selector: # matchLabels: # app: inference-server # template: # metadata: # labels: # app: inference-server # spec: # containers: # - name: inference # image: inference-server:latest # ports: # - containerPort: 8000 # resources: # requests: # memory: "2Gi" # cpu: "1000m" # limits: # memory: "4Gi" # cpu: "2000m" # nvidia.com/gpu: 1 # livenessProbe: # httpGet: # path: /health # port: 8000 # initialDelaySeconds: 30 # readinessProbe: # httpGet: # path: /health # port: 8000 # initialDelaySeconds: 5 # ---------- service.yaml ---------- # apiVersion: v1 # kind: Service # metadata: # name: inference-service # spec: # selector: # app: inference-server # ports: # - port: 80 # targetPort: 8000 # type: LoadBalancer # ---------- hpa.yaml (自动扩缩容) ---------- # apiVersion: autoscaling/v2 # kind: HorizontalPodAutoscaler # metadata: # name: inference-hpa # spec: # scaleTargetRef: # apiVersion: apps/v1 # kind: Deployment # name: inference-server # minReplicas: 2 # maxReplicas: 10 # metrics: # - type: Resource # resource: # name: cpu # target: # type: Utilization # averageUtilization: 70 # - type: Resource # resource: # name: memory # target: # type: Utilization # averageUtilization: 80 # 部署命令 # kubectl apply -f deployment.yaml # kubectl apply -f service.yaml # kubectl apply -f hpa.yaml print("Kubernetes 编排配置示例 (详见注释)")

# ONNX Runtime 性能分析 import onnxruntime as ort import time import numpy as np # 启用会话级别 profiling sess_options = ort.SessionOptions() sess_options.enable_profiling = True session = ort.InferenceSession( "resnet50.onnx", sess_options=sess_options, providers=["CUDAExecutionProvider"] ) # Benchmark input_data = np.random.randn(1, 3, 224, 224).astype(np.float32) num_warmup = 100 num_runs = 1000 # 预热 for _ in range(num_warmup): session.run(["output"], {"input": input_data}) # Benchmark start = time.perf_counter() for _ in range(num_runs): session.run(["output"], {"input": input_data}) end = time.perf_counter() avg_latency = (end - start) / num_runs * 1000 throughput = num_runs / (end - start) print(f"平均延迟: {avg_latency:.2f} ms") print(f"吞吐量: {throughput:.2f} req/s") # 获取 profiling 文件 prof_file = session.end_profiling() print(f"Profiling 文件: {prof_file}")

# TensorRT 引擎性能分析（trtexec 命令行） # 使用 TensorRT 自带的 trtexec 工具进行基准测试 # trtexec --onnx=resnet50.onnx --fp16 --workspace=1024 # --optBatch=1,8,32 --minBatch=1 --maxBatch=32 # --useSpinWait --warmUp=200 --duration=30 # --dumpProfile --separateProfileRun # 输出指标说明: # - Compute: GPU 计算时间 # - H2D: Host-to-Device 数据传输时间 # - D2H: Device-to-Host 数据传输时间 # - Total: 端到端推理延迟 # - Throughput: 吞吐量 (inferences/second) print("性能分析工具示例 (详见注释)")