1. 项目概述：为什么要把 scikit-learn 和 PyTorch 塞进同一个 ONNX 文件里？

你有没有遇到过这种场景：数据科学家用 scikit-learn 快速搭了个特征工程流水线——标准化、独热编码、多项式变换一气呵成，模型部分却用 PyTorch 写了个轻量级神经网络，因为业务指标要求对时序滑动窗口做动态预测；而 MLOps 工程师拿到这两个模块后，发现部署链路卡在了“接口不统一”上：scikit-learn 的 .predict() 返回 numpy 数组，PyTorch 的 model.forward() 输出 tensor，中间还要手动做 dtype 转换、device 搬运、batch 维度对齐……更糟的是，线上服务要同时支持 A/B 测试——一半请求走传统树模型路径，一半走深度学习路径，结果不得不维护两套 API、两套监控、两套版本灰度策略。这根本不是模型部署，这是在给系统打补丁。

这个标题里的“Unifying”不是修辞，是实打实的工程刚需。ONNX（Open Neural Network Exchange）在这里扮演的不是“格式转换器”，而是 跨框架的语义契约签署人 ——它强制所有参与方（scikit-learn、PyTorch、甚至后续可能接入的 XGBoost 或 TensorFlow）在计算图层面达成一致：输入张量形状怎么定义、数据类型如何约定、算子行为是否可复现、缺失值如何传播。我去年在一家做工业设备预测性维护的公司落地这个方案时，最深的体会是：ONNX 的价值不在“能转”，而在“转完之后敢上线”。它把原本散落在不同框架文档角落里的隐式约定，显式固化成一个 .onnx 文件里的 protobuf schema。比如 scikit-learn 的 StandardScaler 在 ONNX 中被编译为 ai.onnx.ml.Scale 算子，而 PyTorch 的 nn.Linear 则映射为标准 Gemm ，二者在 ONNX Runtime 执行时共享同一套内存布局和数值精度控制逻辑。这才是真正意义上的“统一”——不是把两个模型拼在一起，而是让它们在同一个计算原语层面上呼吸同一种空气。

适合谁来读？如果你正面临以下任一情况，这篇就是为你写的：

• 你手上有混合技术栈的模型（比如 scikit-learn 做预处理 + PyTorch 做核心推理），但每次上线都要写一堆胶水代码；
• 你的 CI/CD 流水线因框架差异被迫拆成多条分支，模型版本与预处理版本容易错配；
• 你在用 Triton Inference Server 或 ONNX Runtime 部署，却发现 scikit-learn 导出的 ONNX 缺少动态 batch 支持，而 PyTorch 导出的又无法自动处理 categorical 特征的 string-to-int 映射；
• 你想做模型可解释性分析（如 SHAP），但 SHAP 的 TreeExplainer 和 PyTorchDeepExplainer 根本不兼容同一套输入 pipeline。

这不是教你怎么点几下鼠标导出 ONNX，而是带你亲手把两个异构世界焊接到一起，并且焊点经得起生产环境的高温高压测试。

2. 整体设计思路：为什么必须绕开“先拼后转”，而要“先统一定义再分头实现”？

很多人第一反应是：先把 scikit-learn pipeline 用 skl2onnx 导出，再把 PyTorch 模型用 torch.onnx.export 导出，最后用 ONNX 的 compose 工具（比如 onnx.compose.merge_models ）强行拼接。我试过三次，全部在压测阶段崩溃。原因很直接： ONNX 的图拼接不是函数式编程里的 compose，而是图结构层面的硬连接，它不校验语义一致性 。举个真实案例——scikit-learn 的 OneHotEncoder 默认会把缺失值编码为全零向量，而 PyTorch 模型输入层如果没做特殊处理，就会把全零向量当作有效类别，导致预测偏差放大 300%。但 merge_models 根本不检查这个，它只看张量 shape 是否匹配。

所以我们的设计哲学是： 放弃“事后缝合”，转向“事前契约” 。整个流程分为三个不可跳过的阶段：

2.1 第一阶段：定义统一的 ONNX 接口契约（Interface Contract）

这不是写文档，而是用 ONNX 的 ModelProto 结构提前声明所有输入输出的“宪法”。我们用 Python 构造一个最小化的 onnx.ModelProto ，只包含 graph.input 和 graph.output 定义，不包含任何算子。关键参数必须显式指定：

• 输入名统一为 "features" ，类型为 FLOAT ，shape 为 [None, N] （N 是最终特征维度，需提前计算）；
• 输出名统一为 "logits" ，类型为 FLOAT ，shape 为 [None, C] （C 是分类数或回归目标数）；
• 添加 domain 属性标记为 "unified-deployment-v1" ，用于后续版本管理。

提示：这个契约文件 .onnx 不需要可执行，它只是个 schema 模板。我们用 onnx.helper.make_model 构造后立即保存，后续所有模块都必须严格遵循此 schema。我在实际项目中把这个文件纳入 Git LFS 管理，并在 CI 流程中加入 onnx.checker.check_model 验证，任何违反契约的 PR 都会被自动拒绝。

2.2 第二阶段：scikit-learn Pipeline 的 ONNX 化改造（非简单导出）

skl2onnx 默认导出的 pipeline 存在三个致命缺陷：

1. 静态 shape ： StandardScaler 的 transform 方法导出后，输入 shape 固定为 [1, N] ，无法支持变长 batch；
2. 缺失值黑洞 ： SimpleImputer 导出后，NaN 输入会触发 ONNX Runtime 的 InvalidArgument 异常，而非按 scikit-learn 逻辑填充；
3. 类别映射断裂 ： OrdinalEncoder 对字符串的编码映射在 ONNX 中丢失，导致线上 string 输入无法解析。

解决方案是 重写核心 transformer 的 ONNX 表达 。以 StandardScaler 为例，我们不调用 convert_sklearn ，而是手动构造 ONNX 算子序列：

# 计算 scaler 参数（均值、标准差）——必须在训练时固化
mean = pipeline.named_steps['scaler'].mean_
scale = pipeline.named_steps['scaler'].scale_

# 构造 ONNX 节点：Sub -> Div
sub_node = onnx.helper.make_node('Sub', inputs=['features', 'mean_const'], outputs=['sub_out'])
div_node = onnx.helper.make_node('Div', inputs=['sub_out', 'scale_const'], outputs=['scaled_features'])

# 将常量作为 initializer 注入 graph
mean_const = onnx.helper.make_tensor('mean_const', onnx.TensorProto.FLOAT, mean.shape, mean)
scale_const = onnx.helper.make_tensor('scale_const', onnx.TensorProto.FLOAT, scale.shape, scale)

这样导出的 ONNX 图里， Sub 和 Div 算子天然支持 dynamic batch（ [None, N] ），且 NaN 传播行为与 NumPy 一致。同理， OneHotEncoder 我们替换为 ai.onnx.ml.CategoryMapper 算子，显式传入 cats_strings 和 cats_int64s 参数，确保 string-to-int 映射永不丢失。

2.3 第三阶段：PyTorch 模型的契约对齐改造（非直接 export）

PyTorch 的 torch.onnx.export 默认会把 forward 方法的所有参数都列为输入，但我们契约里只允许一个 "features" 输入。因此必须重构模型入口：

class UnifiedModel(torch.nn.Module):
    def __init__(self, pytorch_core, feature_dim):
        super().__init__()
        self.pytorch_core = pytorch_core
        # 添加 dummy input 用于 ONNX trace —— 这是关键技巧
        self.dummy_input = torch.randn(1, feature_dim)  # shape 必须与契约一致
    
    def forward(self, features):
        # 强制类型和 device 一致（避免 tensor/tensor 混用）
        if features.dtype != torch.float32:
            features = features.to(torch.float32)
        return self.pytorch_core(features)

导出时使用 dynamic_axes={'features': {0: 'batch'}, 'logits': {0: 'batch'}} ，并设置 opset_version=15 （必须 ≥14 才支持 CategoryMapper ）。更重要的是，在导出前用 torch.jit.trace 对模型做一次轻量级 trace，捕获所有控制流分支——否则 ONNX 可能漏掉 dropout 或 layer norm 的条件逻辑。

整个设计的核心逻辑是： 契约先行，模块自治，验证闭环 。每个模块独立导出 ONNX，但导出前必须通过契约校验器（我们自研的 ContractValidator 类）检查输入输出 name、type、shape 是否 100% 匹配。这比任何后期拼接都可靠。

3. 核心细节解析：从 scikit-learn Pipeline 到 ONNX 的七处关键改造点

scikit-learn 的 ONNX 导出绝不是 convert_sklearn(pipeline) 一行代码的事。我在三个不同行业的项目中（金融风控、医疗影像预处理、IoT 传感器校准）反复验证，以下七处是必须手动干预的“雷区”，跳过任意一处都会导致线上 inference 结果漂移。

3.1 特征维度对齐：为什么 Pipeline.named_steps['step'].n_features_in_ 不可信？

n_features_in_ 是 scikit-learn 在 fit 时记录的输入维度，但它不反映 pipeline 中 ColumnTransformer 或 FeatureUnion 的实际输出维度。真实维度必须通过 模拟推理路径 计算：

1. 构造一个全零的 dummy input，shape 为 (1, raw_feature_count) ；
2. 逐层调用 pipeline.named_steps['step'].transform(dummy_input) ；
3. 记录每层输出的 .shape[1] ；
4. 最终维度 = 最后一层输出的列数。

我在某银行项目中发现， ColumnTransformer 对 numeric 和 categorical 列分别处理后， n_features_in_ 返回的是原始列数（12），但实际输出因 one-hot 展开变成 87 列。如果直接用 12 去定义 ONNX 输入 shape，ONNX Runtime 会静默截断后 75 列，导致模型完全失效。正确做法是把计算出的 87 写死进契约文件，并在 CI 中加入维度校验脚本：

# 检查 ONNX 文件输入维度是否匹配契约
onnxruntime-tools check-dim --input-model model.onnx --expected-input-shape "[None,87]"

3.2 缺失值（NaN）的 ONNX 语义： SimpleImputer 的三种策略如何精确映射？

SimpleImputer 的 strategy='mean' 、 'median' 、 'most_frequent' 在 ONNX 中没有直接对应算子。 skl2onnx 默认将其转为 ai.onnx.ml.Imputer ，但该算子在 ONNX Runtime 1.14+ 中已被标记为 deprecated，且不支持 string 类型的 most_frequent 。我们必须降级为原生 ONNX 算子组合：

• strategy='mean' ：用 ReduceMean + Where 实现。先用 IsNaN 生成 mask，再用 Where 将 NaN 替换为均值；
• strategy='most_frequent' （numeric）：用 Unique + TopK 找出现频次最高的值，再 Where 替换；
• strategy='most_frequent' （string）：这是最大难点。ONNX 不支持 string tensor，必须在预处理阶段将 string 映射为 int（用 OrdinalEncoder ），再对 int 应用 TopK 。

注意：所有 Where 算子的 condition 输入必须是 bool 类型，而 IsNaN 输出是 float ，需用 Cast 转换。我在某医疗项目中因漏掉 Cast ，导致 NaN 检测永远返回 false，模型在含缺失值的数据上准确率暴跌 40%。

3.3 类别型特征的双阶段编码：为什么不能只用 OneHotEncoder ？

OneHotEncoder 在 scikit-learn 中默认对 string 输入做 implicit ordinal encoding，但 ONNX 的 OneHotEncoder 算子要求输入必须是 int。如果直接导出，ONNX Runtime 会报 Invalid type for input 。正确路径是：

1. 先用 OrdinalEncoder 将 string → int（注意： handle_unknown='use_encoded_value' 必须设为 True，并指定 unknown_value=-1 ）；
2. 再用 OneHotEncoder 将 int → one-hot vector；
3. 在 ONNX 中， OrdinalEncoder 映射为 ai.onnx.ml.CategoryMapper ， OneHotEncoder 映射为 OneHotEncoder 算子。

关键细节： CategoryMapper 的 cats_strings 参数必须按 scikit-learn OrdinalEncoder.categories_[0] 的 原始顺序 传入，否则编码错位。我们用如下代码确保顺序一致：

# 获取 categories_ 并排序（按出现顺序，非字典序）
categories = encoder.categories_[0].tolist()
# 构造 cats_strings：必须与 sklearn fit 时的顺序完全一致
cats_strings = onnx.helper.make_tensor(
    'cats_strings', onnx.TensorProto.STRING, [len(categories)], 
    [c.encode('utf-8') for c in categories]
)

3.4 动态 Batch 支持： [None, N] 不是语法糖，是内存布局承诺

ONNX 的 None 表示 dynamic axis，但它在 ONNX Runtime 中触发的是 内存池重分配机制 。如果 pipeline 中有 StandardScaler 这类 stateful transformer，其内部常量（mean/scale）必须声明为 initializer 而非 input ，否则每次 batch size 变化时，ONNX Runtime 会尝试重新绑定常量，导致性能断崖式下跌。我们在某 IoT 项目中实测：batch size 从 1 变到 32 时，未声明 initializer 的模型延迟从 12ms 涨到 217ms；加上 initializer 后稳定在 14ms。

声明方式是在构造 ModelProto 时，将所有常量 tensor 加入 graph.initializer ：

graph = onnx.helper.make_graph(
    nodes=[sub_node, div_node],
    name="scaler_graph",
    inputs=[onnx.helper.make_tensor_value_info('features', onnx.TensorProto.FLOAT, [None, N])],
    outputs=[onnx.helper.make_tensor_value_info('scaled_features', onnx.TensorProto.FLOAT, [None, N])],
    initializer=[mean_const, scale_const]  # 关键！必须放这里
)

3.5 数值精度陷阱： float32 vs float64 的隐式降级

scikit-learn 默认使用 float64 进行中间计算（如 StandardScaler.mean_ ），但 ONNX Runtime 默认以 float32 执行。如果直接导出 float64 常量，ONNX Runtime 会静默 cast 为 float32 ，导致微小数值误差累积。在金融风控场景中，这种误差可能让某个样本的 score 从 0.499999 降到 0.499998，刚好跨过阈值 0.5，造成误拒。

解决方案是：在 pipeline fit 后， 强制将所有 float 参数转为 float32 ：

for step_name, step in pipeline.named_steps.items():
    if hasattr(step, 'mean_'):
        step.mean_ = step.mean_.astype(np.float32)
    if hasattr(step, 'scale_'):
        step.scale_ = step.scale_.astype(np.float32)
    if hasattr(step, 'categories_'):
        # categories_ 是 object 类型，需单独处理
        for i, cat in enumerate(step.categories_):
            step.categories_[i] = cat.astype(np.int32) if cat.dtype == 'int64' else cat

并在 ONNX 导出时指定 target_opset=15 ，确保所有算子支持 float32 常量。

3.6 自定义 Transformer 的 ONNX 注册：如何让 CustomFeatureSelector 被识别？

如果你写了继承 BaseEstimator 的自定义 transformer（比如基于方差阈值选择特征）， skl2onnx 默认不认识它。此时不能靠 register_converter （太重），而要用 onnx.numpy_helper.from_array 手动注入计算逻辑：

class CustomFeatureSelector(BaseEstimator, TransformerMixin):
    def __init__(self, threshold=0.1):
        self.threshold = threshold
    
    def fit(self, X, y=None):
        self.variances_ = np.var(X, axis=0)
        self.selected_indices_ = np.where(self.variances_ > self.threshold)[0]
        return self
    
    def transform(self, X):
        return X[:, self.selected_indices_]

# 手动构造 ONNX 节点：Gather + ConstantOfShape
indices_const = onnx.helper.make_tensor(
    'selected_indices', onnx.TensorProto.INT64, [len(selector.selected_indices_)], 
    selector.selected_indices_
)
gather_node = onnx.helper.make_node('Gather', inputs=['features', 'selected_indices'], outputs=['selected_features'])

Gather 算子天然支持 dynamic batch，且 indices 作为常量 initializer 注入，完全符合契约。

3.7 输出层标准化：为什么 predict_proba 必须转为 logits ？

scikit-learn 分类器的 predict_proba 输出概率分布（sum=1），但 ONNX 契约约定输出为 logits （未归一化）。如果直接导出 predict_proba ，ONNX Runtime 会执行 softmax，但线上服务通常需要 logits 做温度缩放（temperature scaling）或集成学习。我们必须拦截 predict_proba ，改用 decision_function 或 predict_log_proba ：

# 对于 LogisticRegression，用 decision_function
if hasattr(model, 'decision_function'):
    output_node = onnx.helper.make_node(
        'Identity', inputs=['decision_function_output'], outputs=['logits']
    )
# 对于 RandomForest，用 predict_log_proba（需 sklearn >=1.2）
elif hasattr(model, 'predict_log_proba'):
    output_node = onnx.helper.make_node(
        'Exp', inputs=['log_proba_output'], outputs=['logits']  # log_proba → proba
    )

这样导出的 ONNX 输出与 PyTorch 模型的 forward 输出语义完全一致，后续可直接做模型融合。

4. 实操过程：从零构建 unified ONNX 模型的完整流水线

现在我们把前面所有设计和细节，组装成一条可复现、可 CI/CD 的实操流水线。整个过程分为五个阶段，每个阶段都有明确的交付物和验证点。我在某车企的 ADAS 数据平台落地此流程，从第一次 commit 到生产上线共 11 天，其中 7 天花在调试这五个阶段的边界 case 上。

4.1 阶段一：契约定义与校验（耗时 0.5 天）

交付物 ： unified_contract.onnx （纯 schema 文件）
操作步骤 ：

1. 创建 contract_builder.py ：

import onnx
from onnx import helper, TensorProto

# 定义输入：features，shape [None, 128]（根据 3.1 计算的真实维度）
input_tensor = helper.make_tensor_value_info('features', TensorProto.FLOAT, [None, 128])
# 定义输出：logits，shape [None, 3]（3 分类任务）
output_tensor = helper.make_tensor_value_info('logits', TensorProto.FLOAT, [None, 3])

# 构建空 graph（无节点，只有 IO）
graph = helper.make_graph([], 'unified_contract', [input_tensor], [output_tensor])
model = helper.make_model(graph, opset_imports=[helper.make_opsetid("", 15)])
model.ir_version = 7
model.producer_name = "unified-contract-builder"
model.domain = "unified-deployment-v1"

# 保存
onnx.save(model, "unified_contract.onnx")

2. 运行 python contract_builder.py 生成文件；
3. 用 onnx.checker.check_model("unified_contract.onnx") 验证；
4. 将文件提交至 Git，并在 CI 中添加检查：

# .github/workflows/onnx-contract.yml
- name: Validate ONNX Contract
  run: |
    python -m onnx.checker unified_contract.onnx
    python -c "import onnx; m=onnx.load('unified_contract.onnx'); assert m.domain=='unified-deployment-v1'"

实操心得：契约文件必须包含 domain 字段，这是后续所有模块版本管理的基础。我们用 domain + model.version 实现灰度发布——新版本 pipeline 设置 domain="unified-deployment-v2" ，Triton Server 可根据请求 header 的 x-model-version 路由到不同 domain 的模型实例。

4.2 阶段二：scikit-learn Pipeline 的 ONNX 化（耗时 2 天）

交付物 ： sklearn_pipeline.onnx （仅含预处理逻辑）
操作步骤 ：

1. 准备训练好的 pipeline（假设已保存为 pipeline.pkl ）；
2. 加载并执行 3.1 的维度计算，确认输入维度为 128；
3. 编写 sklearn_to_onnx.py ，按 3.2–3.7 改造每个 transformer：

from skl2onnx.common.data_types import FloatTensorType
from skl2onnx import convert_sklearn
from skl2onnx.algebra.onnx_ops import OnnxSub, OnnxDiv, OnnxGather

# 手动构造 scaler 节点（替代 convert_sklearn）
scaler = pipeline.named_steps['scaler']
sub_node = OnnxSub('features', np.array(scaler.mean_, dtype=np.float32), output_names=['sub_out'])
div_node = OnnxDiv('sub_out', np.array(scaler.scale_, dtype=np.float32), output_names=['scaled_features'])

# 构造完整 graph
nodes = [sub_node, div_node]
initial_types = [('features', FloatTensorType([None, 128]))]
final_type = [('scaled_features', FloatTensorType([None, 128]))]

# 转换为 ONNX
onnx_model = convert_sklearn(
    pipeline, 
    initial_types=initial_types,
    final_types=final_type,
    options={id(pipeline): {'zipmap': False}}  # 关闭 zipmap，输出 logits
)

4. 用 onnxruntime.InferenceSession 加载并测试：

import numpy as np
sess = onnxruntime.InferenceSession("sklearn_pipeline.onnx")
# 输入 shape [1,128] 的 float32 array
input_data = np.random.randn(1, 128).astype(np.float32)
output = sess.run(None, {"features": input_data})[0]  # 应返回 [1,128]
assert output.shape == (1, 128)

5. 用 onnx.shape_inference.infer_shapes 补全 shape 信息，并保存。

注意： convert_sklearn 的 options 参数必须传入 {'zipmap': False} ，否则输出是 {"label": ..., "probability": ...} 字典结构，违反契约。我在某电商项目中因漏掉此参数，导致 Triton 解析失败，错误日志里只显示 invalid output format ，排查了 8 小时才发现是 zipmap 作祟。

4.3 阶段三：PyTorch 模型的 ONNX 化（耗时 1.5 天）

交付物 ： pytorch_model.onnx （仅含核心推理逻辑）
操作步骤 ：

1. 加载训练好的 PyTorch 模型（ .pth 文件）；
2. 实例化 UnifiedModel （见 2.3），传入真实特征维度 128；
3. 编写 pytorch_to_onnx.py ：

import torch
import onnx

model = UnifiedModel(torch.load("model.pth"), feature_dim=128)
model.eval()

# 创建 dummy input，shape 必须与契约一致
dummy_input = torch.randn(1, 128, dtype=torch.float32)

# 导出，关键参数：
torch.onnx.export(
    model,
    dummy_input,
    "pytorch_model.onnx",
    export_params=True,
    opset_version=15,
    do_constant_folding=True,
    input_names=["features"],
    output_names=["logits"],
    dynamic_axes={
        "features": {0: "batch"},
        "logits": {0: "batch"}
    }
)

4. 用 onnx.checker.check_model 验证；
5. 用 onnx.shape_inference.infer_shapes 补全 shape；
6. 用 onnxruntime 测试：

sess = onnxruntime.InferenceSession("pytorch_model.onnx")
input_data = np.random.randn(1, 128).astype(np.float32)
output = sess.run(None, {"features": input_data})[0]
assert output.shape == (1, 3)  # 3 分类 logits

实操心得： do_constant_folding=True 是必须的，它会把模型中的常量（如 nn.Linear.weight ）折叠进 ONNX 图，避免 runtime 时重复加载。但要注意，如果模型中有 torch.jit.script 的 control flow， do_constant_folding 可能失败，此时需设为 False 并手动优化。

4.4 阶段四：ONNX 图拼接与契约校验（耗时 1 天）

交付物 ： unified_model.onnx （完整 pipeline）
操作步骤 ：

1. 使用 onnx.compose.merge_models 拼接（注意顺序：先 sklearn，再 pytorch）：

import onnx
from onnx import compose

sklearn_onnx = onnx.load("sklearn_pipeline.onnx")
pytorch_onnx = onnx.load("pytorch_model.onnx")

# 拼接：sklearn 输出 'scaled_features' → pytorch 输入 'features'
unified_onnx = compose.merge_models(
    sklearn_onnx,
    pytorch_onnx,
    io_map=[("scaled_features", "features")]  # 关键：IO 映射必须精确
)

2. 用 onnx.shape_inference.infer_shapes 补全所有中间 shape；
3. 用 onnx.checker.check_model 验证；
4. 最关键的一步：契约校验 ：

def validate_contract(model_path, contract_path):
    model = onnx.load(model_path)
    contract = onnx.load(contract_path)
    
    # 检查输入名、类型、shape
    assert model.graph.input[0].name == contract.graph.input[0].name
    assert model.graph.input[0].type.tensor_type.elem_type == \
           contract.graph.input[0].type.tensor_type.elem_type
    assert model.graph.input[0].type.tensor_type.shape.dim[0].dim_param == "batch"  # dynamic axis
    
    # 检查输出名、类型、shape
    assert model.graph.output[0].name == contract.graph.output[0].name
    assert model.graph.output[0].type.tensor_type.elem_type == \
           contract.graph.output[0].type.tensor_type.elem_type
    assert len(model.graph.output[0].type.tensor_type.shape.dim) == 2
    assert model.graph.output[0].type.tensor_type.shape.dim[1].dim_value == 3

validate_contract("unified_model.onnx", "unified_contract.onnx")

5. 保存 unified_model.onnx 。

提示： io_map 参数必须写成 ("scaled_features", "features") ，不能反着写。我曾因写成 ("features", "scaled_features") ，导致拼接后输入名变成 features ，但实际数据流向是反的，模型输出全为 NaN。ONNX checker 不会报错，只能靠人工 review graph。

4.5 阶段五：端到端测试与性能压测（耗时 4 天）

交付物 ：测试报告 e2e_test_report.md ，包含 accuracy、latency、memory 指标
操作步骤 ：

1. 准备测试数据集（1000 个样本，覆盖 NaN、string、edge cases）；
2. 编写 e2e_test.py ，对比原始 pipeline + PyTorch 与 unified ONNX 的输出：

# 原始路径
original_output = pipeline.transform(X_test)  # sklearn preproc
original_output = torch_model(torch.tensor(original_output, dtype=torch.float32))  # pytorch
original_logits = original_output.detach().numpy()

# ONNX 路径
sess = onnxruntime.InferenceSession("unified_model.onnx")
onnx_output = sess.run(None, {"features": X_test.astype(np.float32)})[0]

# 计算最大绝对误差
max_error = np.max(np.abs(original_logits - onnx_output))
print(f"Max absolute error: {max_error:.6f}")  # 要求 < 1e-5

3. 用 locust 做压力测试：

# locustfile.py
from locust import HttpUser, task, between
import numpy as np

class ONNXUser(HttpUser):
    wait_time = between(0.1, 0.5)
    
    @task
    def predict(self):
        # 生成随机 batch
        data = np.random.randn(32, 128).astype(np.float32).tolist()
        self.client.post("/v1/predict", json={"features": data})

4. 监控指标：
   • Accuracy：与原始路径误差 < 1e-5；
   • P99 Latency：batch=32 时 < 50ms；
   • Memory：ONNX Runtime session 内存占用 < 1.2GB；
5. 生成报告并归档。

实操心得：压测时一定要用 onnxruntime.InferenceSession 的 providers=['CUDAExecutionProvider'] （如果 GPU 可用），否则 CPU 模式下的 latency 没参考价值。我在某云服务项目中，CPU 模式 P99 是 120ms，切到 CUDA 后降到 18ms，但忘了在 Triton 配置中启用 GPU，导致线上服务一直跑在 CPU 上，客户投诉延迟高。教训是：压测环境必须 100% 复刻生产环境的 provider 配置。

5. 常见问题与排查技巧实录：那些让你熬夜到凌晨三点的坑

我把过去两年踩过的所有坑，按发生频率和破坏力排序，整理成这张表。每一个问题背后，都有一段想删库跑路的回忆。

问题现象	根本原因	排查命令/技巧	解决方案	发生频率
ONNX Runtime 报 InvalidArgument: Input tensor has incorrect rank	scikit-learn pipeline 导出时， ColumnTransformer 的 remainder='passthrough' 未被正确处理，导致输出 shape 与契约不符	onnx.shape_inference.infer_shapes(model) 后，用 netron 查看 scaled_features 节点的 output shape	手动修改 ColumnTransformer 的 remainder 为 'drop' ，或用 FunctionTransformer 显式定义 passthrough 逻辑	⭐⭐⭐⭐⭐
模型输出全为 NaN	PyTorch 模型中存在 torch.nan_to_num ，但 ONNX 不支持该算子，导出时被忽略，导致 NaN 传播	onnxruntime.InferenceSession(model_path, providers=['CPUExecutionProvider']) 加 log_severity_level=0 参数，查看详细日志	替换 nan_to_num 为 torch.where(torch.isnan(x), torch.zeros_like(x), x) ，再导出	⭐⭐⭐⭐
Triton Server 启动失败，报 Failed to load model	ONNX 文件中 domain 字段为空，Triton 无法识别自定义 domain	python -c "import onnx; m=onnx.load('model.onnx'); print(m.domain)"	在 onnx.helper.make_model 时显式设置 domain="unified-deployment-v1"	⭐⭐⭐⭐
P99 Latency 波动剧烈（10ms ~ 500ms）	ONNX Runtime 的 memory pool 未预分配，每次 batch size 变化时触发 realloc	onnxruntime.SessionOptions().enable_mem_pattern = True （默认 True，但需确认）	在创建 Session 时显式设置 session_options = onnxruntime.SessionOptions(); session_options.enable_mem_pattern = True	⭐⭐⭐
String 类别输入报 Invalid type for input	CategoryMapper 的 cats_strings 是 bytes 类型，但 ONNX 要求 UTF-8 encoded string	onnx.numpy_helper.to_array(tensor).tolist() 查看实际值	用 c.encode('utf-8') 而非 str(c).encode() ，避免 b'b' 这种错误编码	⭐⭐⭐
onnx.checker.check_model 通过，但 onnxruntime 运行时报 Node input cannot be empty	某个 Constant 节点的 value 属性为空，常见于 OneHotEncoder 的 categories 未正确初始化	onnx.shape_inference.infer_shapes(model) 后，用 netron 检查所有 Constant 节点的 value 字段	在构造 CategoryMapper 时，确保 cats_strings 和 cats_int64s 长度相等，且不为空	⭐⭐
**CI