onnx-edge-deployment

name	onnx-edge-deployment
description	Déploiement Edge avec ONNX Runtime — conversion PyTorch/TF/Sklearn → ONNX, exécution cross-platform (CPU/GPU/TensorRT/OpenVINO/CoreML), optimisation de graphe, profilage et quantification avancée.
version	1.0.0
author	EVA
license	Privée EVA St-Étienne
platforms	["linux","macos","windows"]
metadata	{"EVA":{"tags":["onnx","onnx-runtime","tensorrt","openvino","coreml","quantization","model-optimization","edge-ai","cross-platform"],"related_skills":["tensorflow-lite-deep-dive","nvidia-jetson-deployment","model-optimization-edge","tinyml-fundamentals"]}}

ONNX Edge Deployment

Vue d'ensemble

ONNX (Open Neural Network Exchange) est le format d'échange universel entre frameworks ML. ONNX Runtime est le moteur d'inférence cross-platform qui exécute ces modèles sur CPU (x86/ARM), GPU (CUDA/TensorRT), NPU (OpenVINO) et Apple Silicon (CoreML).

Architecture ONNX Runtime Edge

┌─────────────────────────────────────────────────────┐
│         Modèle source (PyTorch / TF / Sklearn)       │
├─────────────────────────────────────────────────────┤
│              Conversion ONNX (torch.onnx / tf2onnx)  │
├─────────────────────────────────────────────────────┤
│              ONNX Runtime (cross-platform)            │
├──────┬──────┬──────┬──────┬──────┬──────┬──────────┤
│ CPU  │CUDA  │TRT   │OVINO │CoreML│XNNPACK│ DirectML│
│(MLAS)│(NVIDIA)│(Jetson)│(Intel)│(Apple)│ (ARM) │ (AMD)  │
└──────┴──────┴──────┴──────┴──────┴──────┴──────────┘

---

1. Conversion vers ONNX

1.1 Depuis PyTorch

import torch
import torch.onnx

class ModeleAudio(torch.nn.Module):
    def __init__(self, n_classes: int = 10):
        super().__init__()
        self.conv1 = torch.nn.Conv2d(1, 8, 3, padding=1)
        self.bn1 = torch.nn.BatchNorm2d(8)
        self.conv2 = torch.nn.Conv2d(8, 16, 3, padding=1, stride=2)
        self.bn2 = torch.nn.BatchNorm2d(16)
        self.pool = torch.nn.AdaptiveAvgPool2d(1)
        self.fc = torch.nn.Linear(16, n_classes)
    
    def forward(self, x):
        x = torch.relu(self.bn1(self.conv1(x)))
        x = torch.relu(self.bn2(self.conv2(x)))
        x = self.pool(x).flatten(1)
        return self.fc(x)

model = ModeleAudio()
model.eval()

# Entrée factice
dummy_input = torch.randn(1, 1, 32, 32)

# Export ONNX
torch.onnx.export(
    model,
    dummy_input,
    ,
    input_names=[],      
    output_names=[],
    dynamic_axes={                     
        : {: },
        : {: },
    },
    opset_version=,                 
    do_constant_folding=,          
    export_params=,                
)


 onnx
onnx_model = onnx.load()
onnx.checker.check_model(onnx_model)
()

1.2 Depuis TensorFlow / Keras

# Méthode 1 : tf2onnx (recommandé)
pip install tf2onnx

python3 -m tf2onnx.convert \
    --saved-model ./saved_model \
    --output model.onnx \
    --opset 18 \
    --inputs-as-nchw "input:0" \  # Format NCHW pour GPU
    --fold_const

# Méthode 2 : API Python tf2onnx
import tf2onnx
import tensorflow as tf

model = tf.keras.models.load_model("model.h5")
spec = (tf.TensorSpec((None, 32, 32, 1), tf.float32, name="input"))

output, _ = tf2onnx.convert.from_keras(
    model, 
    input_signature=[spec],
    opset=18,
)
with open("model.onnx", "wb") as f:
    f.write(output.SerializeToString())

1.3 Depuis Scikit-learn (modèles classiques)

# Conversion SVM, Random Forest, etc.
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType
from sklearn.ensemble import RandomForestClassifier

rf = RandomForestClassifier(n_estimators=100)
rf.fit(X_train, y_train)

initial_type = [("float_input", FloatTensorType([None, X_train.shape[1]]))]
onx = convert_sklearn(rf, initial_types=initial_type)

with open("rf_model.onnx", "wb") as f:
    f.write(onx.SerializeToString())

1.4 Bonnes pratiques de conversion

# 1. Forcer le format NCHW (meilleur perf GPU)
torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    opset_version=18,
)

# 2. Simplifier le graphe ONNX
# pip install onnx-simplifier
import onnxsim
model_simp, check = onnxsim.simplify("model.onnx")
assert check, "Échec de la simplification"
onnx.save(model_simp, "model_simplified.onnx")
print("Modèle simplifié avec succès")

# 3. Supprimer les noeuds Identity (inutiles)
import onnx
from onnx import optimizer
optimized = optimizer.optimize(
    onnx.load("model.onnx"),
    ["eliminate_deadend", "eliminate_identity", "fuse_consecutive_reshape"]
)

---

2. ONNX Runtime — Exécution

2.1 Sessions pour différentes plateformes

import onnxruntime as ort
import numpy as np

class SessionONNX:
    """Gestionnaire de sessions ONNX Runtime cross-platform."""
    
    def __init__(self, model_path: str, provider: str = "cpu"):
        self.model_path = model_path
        self.provider = provider
        self.session = self._creer_session()
        self._analyser_entrees_sorties()
    
    def _creer_session(self) -> ort.InferenceSession:
        """Crée une session avec le provider spécifié."""
        providers = self._mapper_providers()
        options = ort.SessionOptions()
        options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
        options.enable_pattern_optimization = True
        options.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL
        options.intra_op_num_threads = 4
        options.inter_op_num_threads = 2
        
        return ort.InferenceSession(self.model_path, options, providers=providers)
    
    def _mapper_providers(self) -> list:
        """Mapper du nom de provider vers la configuration ONNX."""
        mapping = {
            "cpu": ["CPUExecutionProvider"],
            "cuda": [
                ("CUDAExecutionProvider", {
                    "device_id": 0,
                    : ,
                    :  *  *  * ,  
                    : ,
                }),
                ,  
            ],
            : [
                (, {
                    : ,
                    :  *  *  * ,  
                    : ,
                    : ,
                    : ,
                    : ,
                }),
                ,
                ,
            ],
            : [
                (, {
                    : ,  
                }),
                ,
            ],
            : [
                (, {
                    : ,
                }),
            ],
            : [
                (, {}),
                ,
            ],
            : [
                (, {
                    : ,
                    : ,
                }),
            ],
        }
         mapping.get(.provider, mapping[])
    
     ():
        
        .inputs = {
            i.name: {
                : i.shape,
                : i.,
            }
             i  .session.get_inputs()
        }
        .outputs = {
            o.name: {
                : o.shape,
                : o.,
            }
             o  .session.get_outputs()
        }
    
     () -> :
        
        
         name, tensor  inputs.items():
             name  .inputs, 
             (tensor, np.ndarray):
                inputs[name] = tensor.astype(np.float32)
        
        
        outputs = .session.run(
            (.outputs.keys()),
            inputs,
        )
         ((.outputs.keys(), outputs))
    
     () -> :
        
         time
        
        
        dummy = {
            name: np.random.randn(*info[]).astype(np.float32)
             name, info  .inputs.items()
        }
        
        
         _  (warmup):
            .session.run((.outputs.keys()), dummy)
        
        
        latences = []
         _  (iterations):
            start = time.perf_counter()
            .session.run((.outputs.keys()), dummy)
            latences.append((time.perf_counter() - start) * )
        
         {
            : .provider,
            : (np.mean(latences)),
            : (np.percentile(latences, )),
            : (np.percentile(latences, )),
            :  / (np.mean(latences)),
        }



session = SessionONNX(, provider=)
resultats = session.inferer({: np.random.randn(, , , )})
bench = session.benchmark(iterations=)
()

2.2 Benchmark multi-provider

def benchmark_tous_providers(model_path: str) -> dict:
    """Compare les performances de tous les providers disponibles."""
    import onnxruntime as ort
    
    providers_disponibles = ort.get_available_providers()
    print(f"Providers disponibles : {providers_disponibles}")
    
    resultats = {}
    for provider in providers_disponibles:
        try:
            session = SessionONNX(model_path, provider=provider.replace("ExecutionProvider", "").lower())
            bench = session.benchmark(iterations=50)
            resultats[provider] = bench
        except Exception as e:
            resultats[provider] = {"erreur": str(e)}
    
    return resultats

2.3 Optimisation des sessions

# SessionOptions avancées
options = ort.SessionOptions()

# Niveau d'optimisation
options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_EXTENDED

# Optimisation spécifique
options.optimized_model_filepath = "model_optimized.onnx"

# Configuration mémoire
options.enable_cpu_mem_arena = True
options.enable_mem_reuse = True

# Parallélisme
options.intra_op_num_threads = 4    # threads par op
options.inter_op_num_threads = 2    # ops parallèles
options.execution_mode = ort.ExecutionMode.ORT_PARALLEL

# Logging
options.log_severity_level = 1  # 0=INFO, 1=WARN, 2=ERROR
options.log_verbosity_level = 0

---

3. Optimisation de Modèle ONNX

3.1 Simplification de graphe

# onnx-simplifier : élimine les noeuds superflus
pip install onnx-simplifier

python3 -m onnxsim input.onnx output_simplified.onnx \
    --skip-optimization \
    --check-n 3  # vérifier sur 3 échantillons

# Simplification avancée
import onnx
from onnx import optimizer, shape_inference

# Inférence de formes (resolve les shapes dynamiques)
model = onnx.load("model.onnx")
model_inferred = shape_inference.infer_shapes(model)

# Optimisation du graphe
passes = [
    "eliminate_identity",
    "eliminate_deadend",
    "eliminate_nop_dropout",
    "eliminate_nop_cast",
    "eliminate_nop_pad",
    "extract_constant_to_initializer",
    "fuse_consecutive_reshape",
    "fuse_consecutive_concats",
    "fuse_consecutive_squeezes",
    "fuse_matmul_add_bias_into_gemm",
    "fuse_pad_into_conv",
    "fuse_relu_into_conv",
]
optimized = optimizer.optimize(model_inferred, passes)
onnx.save(optimized, "model_optimized.onnx")

3.2 Quantification ONNX

# ONNX Runtime propose la quantification post-entraînement
from onnxruntime.quantization import quantize_dynamic, quantize_static
from onnxruntime.quantization import QuantType, CalibrationMethod

# Quantification dynamique (poids INT8, activations FP32)
quantize_dynamic(
    "model.onnx",
    "model_quant_dynamic.onnx",
    weight_type=QuantType.QInt8,  # ou QUInt8
    op_types_to_quantize=["MatMul", "Add", "Conv"],
    per_channel=True,
)

# Quantification statique (poids + activations INT8)
# Nécessite un dataset de calibration
from onnxruntime.quantization.calibrate import create_calibrator
from onnxruntime.quantization import CalibrationDataReader

class CalibrationReader(CalibrationDataReader):
    def __init__(self, data: np.ndarray, n_samples: int = 100):
        self.data = data
        self.n_samples = n_samples
        self.iterator = 0
    
    def get_next(self) -> dict:
        if self.iterator >= self.n_samples:
            return None
        batch = self.data[self.iterator:self.iterator + ]
        .iterator += 
         {: batch.astype(np.float32)}

calib_data = CalibrationReader(x_val, n_samples=)

quantize_static(
    ,
    ,
    calibration_data_reader=calib_data,
    quant_format=QuantType.QInt8,
    per_channel=,
    activation_type=QuantType.QUInt8,
    weight_type=QuantType.QInt8,
    calibrate_method=CalibrationMethod.MinMax,
    extra_options={
        : ,
        : ,
        : ,
    },
)

3.3 Operations fusions spécifiques

# Fusions manuelles possibles dans ONNX Runtime :
# Conv + BatchNorm → Conv (fusionnée)
# Conv + Add + Relu → Conv (fusionnée)
# MatMul + Add → Gemm
# Reshape + Softmax → Softmax (reshape intégré)
# LayerNorm → SimplifyLayerNorm

# Vérifier les fusions appliquées :
session = ort.InferenceSession("model.onnx")
print(f"Nombre de noeuds : {len(session._sess.graph.nodes())}")

---

4. Déploiement sur Cibles Edge

4.1 NVIDIA Jetson (TensorRT)

# TensorRT via ONNX Runtime
session = SessionONNX("model.onnx", provider="tensorrt")

# Ou conversion TensorRT native
import tensorrt as trt

TRT_LOGGER = trt.Logger(trt.Logger.WARNING)

def build_trt_engine(onnx_path: str, engine_path: str, 
                     fp16: bool = True) -> trt.ICudaEngine:
    """Construit un moteur TensorRT depuis ONNX."""
    with trt.Builder(TRT_LOGGER) as builder:
        network_flags = 1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
        
        with builder.create_network(network_flags) as network, \
             trt.OnnxParser(network, TRT_LOGGER) as parser:
            
            # Parser le modèle ONNX
            with open(onnx_path, "rb") as f:
                if not parser.parse(f.read()):
                    for err in range(parser.num_errors):
                        print(parser.get_error(err))
                    raise RuntimeError("Échec du parsing ONNX")
            
            config = builder.create_builder_config()
            config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 30)  # 1 GB
            
            if fp16 and builder.platform_has_fast_fp16:
                config.set_flag(trt.BuilderFlag.FP16)
                print("FP16 activé")
            
            
            profile = builder.create_optimization_profile()
            input_name = network.get_input().name
            input_shape = network.get_input().shape
            
            
            profile.set_shape(input_name, 
                            =(, *input_shape[:]),
                            opt=(, *input_shape[:]),
                            =(, *input_shape[:]))
            config.add_optimization_profile(profile)
            
            
            engine = builder.build_serialized_network(network, config)
             engine  :
                 RuntimeError()
            
             (engine_path, )  f:
                f.write(engine)
            
            ()
             engine

4.2 Intel OpenVINO

# OpenVINO : idéal pour CPU Intel et Myriad X (NCS2)

# Conversion vers OpenVINO IR
# pip install openvino-dev
!mo --input_model model.onnx --output_dir ./openvino_ir/ --data_type FP16

# Inférence avec OpenVINO
from openvino.runtime import Core

core = Core()
model = core.read_model("openvino_ir/model.xml")
compiled = core.compile_model(model, "CPU")  # ou GPU, MYRIAD

# Inférence
output = compiled([np.random.randn(1, 1, 32, 32)])

4.3 Apple CoreML

# CoreML pour Apple Neural Engine (iPhone, Mac M-series)

# Conversion via coremltools
import coremltools as ct

# Depuis ONNX
mlmodel = ct.converters.onnx.convert(
    model="model.onnx",
    minimum_ios_target="16",
    compute_precision=ct.transform.Float16ComputePrecision(),
)

# Depuis PyTorch directement
model = ModeleAudio()
model.eval()
traced = torch.jit.trace(model, torch.randn(1, 1, 32, 32))
mlmodel = ct.convert(
    traced,
    inputs=[ct.TensorType(shape=(1, 1, 32, 32))],
)

mlmodel.save("AudioModel.mlpackage")

4.4 Qualcomm SNPE / QNN

# Qualcomm Neural Processing SDK
snpe-onnx-to-dlc --input_model model.onnx --output_model model.dlc

# Conversion pour Adreno GPU
snpe-dlc-quantize --input_dlc model.dlc --input_list calibration_list.txt

# Inférence sur DSP/GPU
snpe-net-run --container model.dlc --input_list input_list.txt --use_dsp

---

5. Débogage et Validation

5.1 Vérification de la conformité ONNX

def valider_modele_onnx(model_path: str) -> dict:
    """Validation approfondie d'un modèle ONNX."""
    import onnx
    import onnxruntime as ort
    
    resultats = {"valide": True, "erreurs": [], "avertissements": []}
    
    # 1. Vérification structurelle
    try:
        model = onnx.load(model_path)
        onnx.checker.check_model(model)
        resultats["valide_structure"] = True
    except Exception as e:
        resultats["valide_structure"] = False
        resultats["erreurs"].append(f"Structure invalide : {e}")
        return resultats
    
    # 2. Version opset
    resultats["opset_version"] = model.opset_import[0].version
    
    # 3. Exécution test
    try:
        session = ort.InferenceSession(model_path)
        input_name = session.get_inputs()[0].name
        dummy = {input_name: np.random.randn(*session.get_inputs()[0].shape).astype(np.float32)}
        session.run(None, dummy)
        resultats["execution_test"] = True
    except Exception as e:
        resultats["execution_test"] = False
        resultats["erreurs"].append(f"Échec exécution : {e}")
    
    # 4. Graphe
    resultats[] = (model.graph.node)
    resultats[] = (model.graph.)
    resultats[] = (model.graph.output)
    resultats[] = (
        (t.dims) *   t  model.graph.initializer
    ) / ( * )
    
     resultats

5.2 Comparaison de précision

def comparer_precision(model_onnx_path: str, model_source, 
                       data_test: np.ndarray, n_echantillons: int = 100) -> dict:
    """Compare la précision du modèle ONNX vs modèle source."""
    
    from scipy.special import softmax
    
    session = SessionONNX(model_onnx_path, provider="cpu")
    erreurs = []
    
    for i in range(min(n_echantillons, len(data_test))):
        # Entrée source
        x = data_test[i:i+1]
        
        # Prédiction source
        if hasattr(model_source, 'predict'):
            pred_source = model_source.predict(x)
        else:
            model_source.eval()
            with torch.no_grad():
                pred_source = model_source(torch.from_numpy(x)).numpy()
        
        # Prédiction ONNX
        pred_onnx = session.inferer({"input": x.astype(np.float32)})
        pred_onnx = list(pred_onnx.values())[0]
        
        # Erreur relative
        diff = np.abs(pred_source - pred_onnx)
        erreurs.append(np.mean(diff))
    
    return {
        "erreur_moyenne": float(np.mean(erreurs)),
        "erreur_max": float(np.max(erreurs)),
        "erreur_std": float(np.std(erreurs)),
        "n_echantillons": min(n_echantillons, (data_test)),
    }

---

6. Surveillance et Métriques

6.1 Métriques de service

class MoniteurONNX:
    """Surveillance des performances d'inférence ONNX."""
    
    def __init__(self):
        self.metrics = {
            "latences": [],
            "debits": [],
            "erreurs": 0,
            "total": 0,
        }
    
    def mesurer(self, session: SessionONNX, entree: dict) -> dict:
        """Mesure avec monitoring."""
        import time
        
        self.metrics["total"] += 1
        start = time.perf_counter()
        
        try:
            sortie = session.inferer(entree)
            latence_ms = (time.perf_counter() - start) * 1000
            self.metrics["latences"].append(latence_ms)
            sortie["latence_ms"] = latence_ms
            return sortie
        except Exception as e:
            self.metrics["erreurs"] += 1
            raise
    
    def rapport(self) -> dict:
        """Rapport des métriques."""
        if not self.metrics["latences"]:
            return {: }
        
        l = .metrics[]
         {
            : .metrics[],
            : .metrics[] / (, .metrics[]) * ,
            : (np.mean(l)),
            : (np.percentile(l, )),
            : (np.percentile(l, )),
            : (np.percentile(l, )),
            :  / (np.mean(l)),
        }

---

Pièges Courants

1. Dynamic axes sans profile TensorRT : si le modèle a des axes dynamiques, TensorRT nécessite un optimization profile. Définir min/opt/max pour chaque dimension variable.

2. Opset version trop ancien : opset < 15 manque de support pour certaines ops modernes (LayerNorm, Softmax avec axes). Utiliser opset 18+.

3. Provider fallback silencieux : si le provider sélectionné n'est pas disponible, ONNX Runtime tombe sur CPU sans avertissement clair. Vérifier avec ort.get_available_providers().

4. Fusion BatchNorm manquée : BatchNorm ne se fusionne pas avec Conv si l'opérateur est défini indépendamment dans le graphe. Activer do_constant_folding=True à l'export.

5. Quantification statique sans calibration : la quantification statique sans calibration adaptée dégrade la précision de 5-15%. Utiliser un dataset de calibration représentatif.

6. TensorRT engine caching : la mise en cache des engines TensorRT est essentielle (le build prend 5-30 min). Activer trt_engine_cache_enable=True.

---

Références

• ONNX Runtime : https://onnxruntime.ai/
• ONNX Converter : https://github.com/onnx/onnx
• tf2onnx : https://github.com/onnx/tensorflow-onnx
• skl2onnx : https://github.com/onnx/sklearn-onnx
• onnx-simplifier : https://github.com/daquexian/onnx-simplifier
• TensorRT : https://developer.nvidia.com/tensorrt
• OpenVINO : https://docs.openvino.ai/
• CoreML : https://coremltools.readme.io/
• OnnxRuntime Quantization : https://onnxruntime.ai/docs/performance/quantization.html