IA ONNX pour la compression
prédictive

Pourquoi une IA pour choisir l'algorithme ?

Universal Installer V2 supporte plusieurs algorithmes de compression : UXC (custom), Zstd, LZ4, Brotli et STORED (pas de compression). Chaque algorithme excelle sur un type de données différent : LZ4 est ultra-rapide sur les données répétitives, Zstd est optimal pour le texte structuré, UXC pour les binaires compilés, et STORED est le bon choix pour les données déjà compressées (JPEG, MP4, ZIP).

Sans IA, on choisit un algorithme global pour toute l'archive — compromis médiocre. Avec le modèle ONNX, chaque bloc de 64 KB est analysé indépendamment et reçoit l'algorithme optimal. Le tout en moins d'une milliseconde par bloc.

Le gain principal de l'IA n'est pas l'amélioration du ratio sur les données compressibles — c'est d'éviter de comprimer les données déjà compressées, économisant du temps CPU sans perdre de ratio.

Les 19 features d'entropie

Pour chaque bloc de 64 KB, 19 features sont calculées en Python pur avant l'inférence :

import numpy as np
from collections import Counter
import math

def extract_features(block: bytes) -> np.ndarray:
    # 1. Entropie de Shannon
    counts = Counter(block)
    total  = len(block)
    entropy = -sum((c/total) * math.log2(c/total) for c in counts.values())

    # 2-10. Distribution des bytes (9 buckets)
    hist = np.histogram(list(block), bins=9, range=(0, 256))[0] / total

    # 11. Taux de bytes nuls
    null_ratio = block.count(0) / total

    # 12. Taux de bytes répétitifs (run-length)
    runs = sum(1 for i in range(1, len(block)) if block[i] == block[i-1])
    run_ratio = runs / total

    # 13. Magic bytes (détection de format)
    is_jpeg = int(block[:3] == b'ÿØÿ')
    is_zip  = int(block[:4] == b'PK')
    is_elf  = int(block[:4] == b'ELF')

    # 14. Ratio printable ASCII
    printable = sum(1 for b in block if 32 <= b < 127) / total

    # 15-17. Variance, moyenne, écart-type
    arr = np.frombuffer(block, dtype=np.uint8).astype(float)
    mean, std, var = arr.mean(), arr.std(), arr.var()

    # 18. Taille du bloc
    size_ratio = len(block) / 65536

    # 19. Nombre de bytes uniques
    unique_ratio = len(counts) / 256

    return np.array([entropy, *hist, null_ratio, run_ratio,
                     is_jpeg, is_zip, is_elf, printable,
                     mean/255, std/128, var/16384,
                     size_ratio, unique_ratio])

Le modèle sklearn

Le modèle est un Random Forest de 100 arbres, entraîné sur 50 000 blocs échantillonnés depuis un corpus de 10 Go de fichiers variés (code source, binaires compilés, images, vidéos, archives, bases de données) :

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split

CLASSES = ['UXC', 'ZSTD', 'LZ4', 'STORED']

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

model = RandomForestClassifier(
    n_estimators=100,
    max_depth=12,
    n_jobs=-1
)
model.fit(X_train, y_train)
print(f"Accuracy: {model.score(X_test, y_test):.3f}")
# → Accuracy: 0.923

Export et runtime ONNX

from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType
import onnxruntime as ort

# Export ONNX
initial_type = [('float_input', FloatTensorType([None, 19]))]
onnx_model = convert_sklearn(model, initial_types=initial_type)
open('algo_selector.onnx', 'wb').write(onnx_model.SerializeToString())
# Taille : 2.3 MB

# Runtime ONNX — chargement une seule fois au démarrage
session = ort.InferenceSession('algo_selector.onnx',
    providers=['CPUExecutionProvider'])

def predict_algo(block: bytes) -> str:
    features = extract_features(block).astype(np.float32).reshape(1, -1)
    pred = session.run(None, {'float_input': features})[0]
    return CLASSES[pred[0]]
# Durée : < 0.5ms par bloc sur CPU standard

Inférence par bloc

Pendant la compression, chaque bloc de 64 KB est analysé et compressé avec l'algorithme prédit. Les blocs sont traités en parallèle (multi-threading) :

from concurrent.futures import ThreadPoolExecutor

def compress_block(block: bytes) -> tuple[str, bytes]:
    algo = predict_algo(block)
    if algo == 'STORED': return algo, block
    elif algo == 'LZ4':    return algo, lz4.frame.compress(block)
    elif algo == 'ZSTD':   return algo, zstd.compress(block, level=19)
    else:                    return algo, uxc_compress(block)

blocks = [data[i:i+65536] for i in range(0, len(data), 65536)]
with ThreadPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(compress_block, blocks))

Gains mesurés

Ré-entraîner sur vos données

Le modèle livré est entraîné sur un corpus générique. Sur un corpus spécifique (logs applicatifs, données médicales, images satellite), ré-entraîner améliore encore les performances :

from universal_installer.ai import AlgoSelectorTrainer

trainer = AlgoSelectorTrainer()

# Collecter des échantillons depuis votre corpus
trainer.collect_samples(
    corpus_path="path/to/your/data",
    n_samples=10000
)

# Entraîner (~2 min pour 10 000 blocs)
trainer.train()

# Exporter le modèle personnalisé
trainer.export_onnx("~/.universal_installer/models/custom.onnx")
# Le modèle custom est prioritaire sur le modèle par défaut

Conclusion

Le modèle ONNX d'Universal Installer V2 apporte un gain réel de 2 à 23% selon le type de données, avec un overhead d'inférence inférieur à 1ms par bloc. Le gain est maximal sur les archives mixtes où l'IA évite de comprimer les données déjà compressées — économisant du temps CPU et améliorant le ratio simultanément. Le modèle est ré-entraînable en 2 minutes sur votre propre corpus.