推理代码评审:动态控制流和输入形状最容易漏
推理代码能输出结果,不代表它适合部署。本文检查输入形状、动态控制流、预热与回退路径,示例数字不作为性能结论。
1. 先写清部署形状与批处理规则
推理调优应先明确输入形状、并发模型、预热规则和验收口径。准确率、排队、计算与内存使用属于不同维度,需要分别记录和解释。
评审结论要绑定模型版本、运行时、硬件和输入分布。任何一项改变,都可能让原来的图优化和内存结论失效。
2. 按最小闭环验证
建议用最小调用链验证导出、预处理、引擎执行与后处理。只有在同一模型和同一输入条件下获得的记录才适合横向比较。
至少准备固定长度、最大长度和空输入三组断言,并记录每组输出形状。导出引擎或批处理规则变化后,用同一输入重新跑这些边界用例。
3. 参考实现与图示
下面用动态控制流展示部署时容易遗漏的路径。实际评审还要检查张量是否被隐式搬到 CPU,以及这些分支能否被目标运行时导出。
# 隐患示范:在模型推理内部交织 Python 动态控制流 def forward(self, input_ids): outputs = self.backbone(input_ids) logits = outputs.logits # 风险点 1: 依赖 Python 原生 list 操作与动态 CPU 转换 results = [] for logit in logits: if logit.sum().item() > 0: # 强制同步 GPU 显存到 CPU 内存! results.append(torch.argmax(logit)) return torch.stack(results)import ast import sys import onnx class CUDAStreamSyncVisitor(ast.NodeVisitor): """静态 AST 分析:检查 Python 推理代码中是否混入了显式/隐式 CUDA 同步调用""" def __init__(self): self.violations = [] def visit_Call(self, node): # 拦截 .item(), .tolist(), .cpu().numpy() 等同步方法 if isinstance(node.func, ast.Attribute): if node.func.attr in ["item", "tolist"]: self.violations.append((node.lineno, f"检测到违规同步调用: .{node.func.attr}()")) elif node.func.attr == "numpy": # 检查是否是在 cpu() 之后调用的 numpy self.violations.append((node.lineno, "检测到潜在的 GPU->CPU 拷贝调用: .numpy()")) self.generic_visit(node) def check_python_code_quality(filepath: str): print(f"正在对推理代码进行 AST 静态门禁检查: {filepath}") with open(filepath, "r", encoding="utf-8") as f: tree = ast.parse(f.read(), filename=filepath) visitor = CUDAStreamSyncVisitor() visitor.visit(tree) if visitor.violations: print("❌ 代码评审门禁未通过!发现以下隐性性能风险:") for line, msg in visitor.violations: print(f" - 第 {line} 行: {msg}") return False print("✅ Python 推理代码静态检查通过!") return True def check_onnx_model_gate(onnx_path: str): print(f"正在对 ONNX 模型结构进行门禁检查: {onnx_path}") try: model = onnx.load(onnx_path) onnx.checker.check_model(model) except Exception as e: print(f"❌ ONNX 模型结构合法性校验失败: {e}") return False # 检查 Dynamic Batch 维度是否进行了命名限制 for input_tensor in model.graph.input: shape = input_tensor.type.tensor_type.shape for dim in shape.dim: if dim.dim_param and dim.dim_param.startswith("unk"): print(f"⚠️ 警告: 输入 {input_tensor.name} 存在未命名的动态维度 {dim.dim_param}") print("✅ ONNX 模型门禁校验通过!") return True if __name__ == "__main__": # 模拟 CI 门禁入口 code_ok = check_python_code_quality(__file__) if not code_ok: sys.exit(1)4. 复核清单
- 动态形状是否有明确上限与回退实现。
- Python 分支是否阻断图编译或批处理。
- 预热、冷启动和稳定阶段是否分开统计。
- 超长输入与显存不足时是否返回可诊断错误。
把隐藏条件变成评审项
推理性能依赖的条件越隐蔽,发布后越难解释。把形状、运行时和失败路径写进评审,才有复现基础。