MLOps IoT : surveiller les modèles de machine learning sur le terrain via MQTT

Le MLOps IoT consiste à déployer, surveiller et mettre à jour des modèles de machine learning qui s'exécutent sur des équipements que vous ne voyez pas. Dans ce guide, vous allez construire une boucle de surveillance légère : les équipements edge publient des métriques d'inférence via MQTT, un service Python central mesure la dérive, et les modèles dégradés déclenchent des alertes sur un topic dédié.

Pourquoi les modèles se dégradent sur le terrain

Un modèle qui obtenait de bons scores en laboratoire commence à perdre en précision dès sa mise en service. En IoT, les causes sont physiques et souvent lentes :

• Vieillissement et recalibrage des capteurs. Un capteur de vibrations dérive, l'objectif d'une caméra se couvre de poussière, un microphone capte un nouveau ventilateur.
• Évolutions du procédé. Un nouveau matériau, une autre organisation des équipes ou une nouvelle variante de produit modifie la distribution des entrées.
• Saisonnalité. La température et l'humidité ambiantes font évoluer les valeurs de référence des capteurs au fil des mois.
• Écart de déploiement. La quantification ou une autre version de firmware modifie légèrement les sorties par rapport au modèle évalué.

Dans un service web, vous pouvez journaliser chaque requête et comparer les prédictions aux retours des utilisateurs. Sur le terrain, vous n'avez généralement aucun label, une bande passante limitée et des équipements qui se déconnectent. C'est pourquoi le MLOps IoT s'appuie sur des signaux indirects : statistiques des entrées, distributions des sorties, confiance et latence.

Ce qu'il faut mesurer sur l'équipement

Envoyer dans le cloud les fenêtres brutes des capteurs à chaque inférence va à l'encontre de l'intérêt de l'edge AI. Chaque équipement agrège plutôt quelques valeurs par intervalle et publie un message de métriques compact.

Un payload de métriques peut ressembler à ceci :

{
  "device": "press-07",
  "model": "vib-classifier",
  "modelVersion": "1.4.2",
  "window": "2026-10-16T10:00:00Z/PT1M",
  "inferences": 600,
  "inputMean": {"rms": 0.42, "peakHz": 118.0},
  "inputStd": {"rms": 0.05, "peakHz": 6.1},
  "classCounts": {"normal": 588, "imbalance": 9, "bearing": 3},
  "confMean": 0.91,
  "confP10": 0.74,
  "latencyMsP95": 14.2
}

Conception des topics pour le MLOps IoT

Séparez la télémétrie des modèles de la télémétrie du procédé afin que tableaux de bord et consommateurs puissent s'abonner précisément :

connectmq/plant1/ml/<model>/<device>/metrics
connectmq/plant1/ml/<model>/<device>/status        (retained: version, loaded at)
connectmq/plant1/ml/<model>/alerts
connectmq/plant1/ml/<model>/<device>/cmd           (rollout or rollback commands)

Retenir le topic de statut permet à tout nouvel abonné de savoir immédiatement quelle version de modèle tourne sur chaque équipement, ce qui est la question la plus fréquente lors d'un incident.

Publier des métriques depuis un équipement edge

Le script ci-dessous simule un équipement edge qui exécute un classifieur et publie un message de métriques par intervalle. Sur du matériel réel, vous accumuleriez les statistiques dans votre boucle d'inférence. Il utilise l'espace public du serveur MQTT public gratuit broker.connectmq.com, réservé aux tests et non aux données de production ; remplacez le préfixe par votre propre identifiant aléatoire.

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/mlops-91bc/plant1/ml/vib-classifier"
DEVICE = "press-07"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"{DEVICE}-{uuid.uuid4().hex[:6]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

status = {"modelVersion": "1.4.2", "loadedAt": time.time()}
client.publish(f"{PREFIX}/{DEVICE}/status", json.dumps(status), qos=1, retain=True)

drift = 0.0
while True:
    drift += 0.002  # décaler lentement l'entrée pour simuler le vieillissement du capteur
    rms = [random.gauss(0.42 + drift, 0.05) for _ in range(600)]
    conf = [min(1.0, max(0.0, random.gauss(0.91 - drift, 0.06))) for _ in range(600)]
    conf.sort()
    msg = {
        "device": DEVICE,
        "modelVersion": "1.4.2",
        "inferences": len(rms),
        "inputMean": {"rms": sum(rms) / len(rms)},
        "inputHist": {"rms": [sum(1 for v in rms if lo <= v < lo + 0.05)
                              for lo in [0.25 + 0.05 * i for i in range(10)]]},
        "confMean": sum(conf) / len(conf),
        "confP10": conf[len(conf) // 10],
        "ts": time.time(),
    }
    client.publish(f"{PREFIX}/{DEVICE}/metrics", json.dumps(msg), qos=1)
    time.sleep(10)

C'est l'histogramme à classes fixes qui rend la dérive mesurable de façon centralisée sans transférer de données brutes.

Détecter la dérive de façon centralisée

Le moniteur s'abonne à tous les équipements, compare chaque histogramme reçu à un histogramme de référence capturé lors du déploiement et calcule l'indice de stabilité de population (PSI). Une règle empirique courante considère un PSI inférieur à 0,1 comme stable, entre 0,1 et 0,25 comme à surveiller et supérieur à 0,25 comme un décalage significatif, mais ajustez ces seuils sur vos propres données.

import json
import math
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/mlops-91bc/plant1/ml/vib-classifier"
reference = {}  # équipement -> histogramme de référence
CONF_FLOOR = 0.70


def psi(expected, actual, eps=1e-4):
    e_total, a_total = sum(expected) or 1, sum(actual) or 1
    score = 0.0
    for e, a in zip(expected, actual):
        e_p = max(e / e_total, eps)
        a_p = max(a / a_total, eps)
        score += (a_p - e_p) * math.log(a_p / e_p)
    return score


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/+/metrics", qos=1)


def on_message(client, userdata, msg):
    m = json.loads(msg.payload)
    dev, hist = m["device"], m["inputHist"]["rms"]
    if dev not in reference:
        reference[dev] = hist
        print(f"reference captured for {dev}")
        return
    score = psi(reference[dev], hist)
    status = "ok"
    if score > 0.25 or m["confP10"] < CONF_FLOOR:
        status = "degraded"
    elif score > 0.1:
        status = "watch"
    print(f"{dev} psi={score:.3f} confP10={m['confP10']:.2f} -> {status}")
    if status != "ok":
        alert = {"device": dev, "modelVersion": m["modelVersion"],
                 "psi": round(score, 3), "confP10": round(m["confP10"], 3),
                 "status": status}
        client.publish(f"{PREFIX}/alerts", json.dumps(alert), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"mlops-monitor-{uuid.uuid4().hex[:6]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Exécutez les deux scripts et vous verrez le statut passer de ok à watch puis à degraded à mesure que la dérive simulée s'accumule. En production, l'histogramme de référence doit provenir des données de validation ayant servi à approuver le modèle, stockées avec l'artefact du modèle, et non du premier message reçu après le démarrage.

Agir sur les alertes

Une alerte n'est utile que si quelqu'un ou quelque chose y répond. Les options, de la plus simple à la plus automatisée :

1. Prévenir une personne. Un flux Node-RED ou un petit service s'abonne au topic d'alertes et envoie des e-mails ou des messages de chat.
2. Alertes visuelles dans le portail du broker. Sur un broker ConnectMQ privé, vous pouvez définir des règles d'alerte à seuil sur un champ numérique comme psi et consulter les déclenchements dans le journal des alertes.
3. Collecter des labels. Quand un équipement dérive, demandez-lui de téléverser quelques échantillons bruts sur un topic séparé pour que les ingénieurs puissent les étiqueter et réentraîner le modèle.
4. Revenir en arrière. Publiez une commande sur le topic cmd de l'équipement pour charger la version précédente du modèle, puis confirmez via le topic de statut retenu.

Le volet déploiement, c'est-à-dire la façon dont les modèles arrivent sur les équipements, est traité dans déployer des modèles de ML sur des équipements edge.

Déploiements canary et modèles fantômes

La surveillance vous indique quand un modèle se dégrade ; la rigueur des déploiements empêche un nouveau modèle d'aggraver la situation. Deux schémas fonctionnent bien avec les topics MQTT.

Déploiement canary. Envoyez d'abord le nouveau modèle à un petit groupe d'équipements, par exemple cinq pour cent du parc ou une ligne par usine. Comme chaque message de métriques contient modelVersion, le moniteur peut comparer la confiance, la distribution des classes et la dérive entre le groupe canary et le reste sur la même période. Ne généralisez que si le canary paraît au moins aussi sain que le modèle en place, et revenez en arrière avec le même topic de commande dans le cas contraire.

Mode fantôme (shadow). Exécutez le nouveau modèle en parallèle du modèle actuel sur le même équipement, mais n'agissez que sur la sortie du modèle actuel. Publiez les deux prédictions dans le message de métriques, ou publiez un taux d'accord glissant entre les deux. Un modèle fantôme qui est en désaccord avec le modèle de production sur une grande part des entrées mérite un examen approfondi avant de prendre le relais, même si les deux modèles semblent confiants.

Ces deux schémas n'exigent du broker rien d'autre qu'une livraison fiable et des topics clairs. Ils facilitent aussi l'analyse post-incident : lorsqu'une alerte se déclenche, la première question est de savoir si elle coïncide avec un changement de version, et les topics de statut retenus y répondent instantanément.

Une remarque pratique sur la bande passante : un message de métriques par minute et par équipement est minuscule, mais les téléversements d'échantillons bruts pour l'étiquetage ne le sont pas. Placez-les sur un topic séparé, déclenchez-les uniquement à la demande et restez sous la limite de taille de message de votre offre.

Liste de contrôle MLOps IoT

1. Chaque message contient le nom et la version du modèle.
2. Les équipements publient des statistiques d'entrée agrégées, et non des données brutes, à intervalle fixe.
3. Une distribution de référence est stockée avec chaque modèle publié.
4. Les seuils de dérive et de confiance sont documentés et revus après chaque incident.
5. Les alertes sont envoyées sur un topic, et au moins un consommateur les transforme en notifications humaines.
6. Le retour en arrière est une commande testée, pas un simple plan.
7. Les données de réentraînement sont collectées délibérément lorsqu'une dérive survient.
8. En production, métriques et alertes transitent en TLS avec des identifiants propres à chaque équipement.

L'infrastructure pour surveiller tout un parc

Les métriques de centaines d'équipements, un moniteur, des tableaux de bord et des outils de déploiement partagent tous un même bus de messages. Exploiter ce bus vous-même implique correctifs, certificats et planification de capacité. Un broker MQTT managé supprime ce travail, et un broker MQTT cloud est accessible depuis des équipements sur n'importe quel réseau. Les abonnements partagés de l'offre Serverless (Paid) vous permettent d'exécuter plusieurs instances du moniteur qui se répartissent la charge des métriques.

Le même flux de métriques alimente aussi des travaux propres à certains modèles, comme la maintenance prédictive par machine learning, et les contrôles de dérive sont proches cousins de la détection d'anomalies IoT : l'une surveille la machine, l'autre surveille le modèle. Pour une vue d'ensemble de l'IA en edge et dans le cloud, consultez le guide IA et IoT.

Surveillez vos modèles sur le terrain avec ConnectMQ

Créez un broker privé avec l'offre gratuite Serverless sur app.connectmq.com, sans carte bancaire, connectez-y vos équipements avec les identifiants générés sur le port 8883 et regardez les métriques des modèles arriver dans l'explorateur de topics en direct. Ajouter le moniteur de dérive revient à reprendre le script ci-dessus avec un nom d'utilisateur et un mot de passe.

Questions fréquentes

En quoi le MLOps IoT diffère-t-il du MLOps classique ?

Les équipements sur le terrain disposent rarement de labels, ont une bande passante limitée et peuvent être hors ligne ; la surveillance repose donc sur des statistiques agrégées des entrées et des sorties plutôt que sur des journaux par requête. Le déploiement et le retour en arrière doivent aussi fonctionner sur des liaisons peu fiables.

À quelle fréquence les équipements doivent-ils publier les métriques des modèles ?

Une fois par minute ou toutes les quelques centaines d'inférences est un bon point de départ. C'est assez fréquent pour repérer une dérive en quelques heures et assez léger pour rester largement sous les limites de débit de messages.

Quelle métrique de dérive utiliser ?

L'indice de stabilité de population et les tests de Kolmogorov-Smirnov sont courants pour les entrées, et le suivi de la confiance au percentile bas détecte de nombreux problèmes de sortie. Utilisez plusieurs signaux, car chacun passe à côté de certains modes de défaillance.

Le broker peut-il détecter la dérive à ma place ?

Non. Le broker transporte les métriques ; la détection de dérive s'exécute dans votre propre service abonné à celles-ci. Les règles d'alerte à seuil du portail ConnectMQ peuvent signaler des limites numériques simples dès que votre service publie un score de dérive.