IoT MLOps: Machine-Learning-Modelle im Feld per MQTT überwachen

IoT MLOps umfasst die Bereitstellung, Überwachung und Aktualisierung von Machine-Learning-Modellen, die auf Geräten laufen, die Sie nicht sehen können. In diesem Leitfaden bauen Sie eine schlanke Überwachungsschleife auf: Edge-Geräte veröffentlichen Inferenzmetriken per MQTT, ein zentraler Python-Dienst misst die Drift, und verschlechterte Modelle lösen Alarme auf einem eigenen Topic aus.

Warum Modelle im Feld schlechter werden

Ein Modell, das im Labor gut abgeschnitten hat, verliert ab dem Tag der Auslieferung an Genauigkeit. Im IoT sind die Gründe physikalisch und oft schleichend:

• Sensoralterung und Neukalibrierung. Ein Schwingungssensor driftet, eine Kameralinse verstaubt, ein Mikrofon nimmt einen neuen Lüfter auf.
• Prozessänderungen. Ein neues Material, ein anderes Schichtmodell oder eine neue Produktvariante verschiebt die Eingangsverteilung.
• Saisonalität. Umgebungstemperatur und Luftfeuchte verändern die Grundwerte der Sensoren über Monate.
• Abweichungen bei der Bereitstellung. Quantisierung oder ein anderer Firmware-Build verändern die Ausgaben gegenüber dem Evaluierungsmodell leicht.

In einem Webdienst können Sie jede Anfrage protokollieren und Vorhersagen mit Nutzerfeedback vergleichen. Im Feld haben Sie meist überhaupt keine Labels, nur begrenzte Bandbreite und Geräte, die offline gehen. Deshalb stützt sich IoT MLOps auf Ersatzsignale: Eingangsstatistiken, Ausgabeverteilungen, Konfidenz und Latenz.

Was auf dem Gerät gemessen wird

Für jede Inferenz rohe Sensorfenster in die Cloud zu senden, würde den Sinn von Edge AI zunichtemachen. Stattdessen aggregiert jedes Gerät pro Intervall einige wenige Kennzahlen und veröffentlicht eine kompakte Metriknachricht.

Ein Metrik-Payload könnte so aussehen:

{
  "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
}

Topic-Design für IoT MLOps

Trennen Sie Modelltelemetrie von Prozesstelemetrie, damit Dashboards und Konsumenten gezielt abonnieren können:

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)

Wird das Status-Topic als Retained Message gehalten, erfährt jeder neue Subscriber sofort, welche Modellversion auf jedem Gerät läuft – die häufigste Frage bei einem Vorfall.

Betreibt Ihr Werk bereits einen Unified Namespace (UNS), dessen Topics der ISA-95-Hierarchie folgen (etwa acme/werk-stuttgart/montage/linie-1/zelle-3/...), hängen Sie den ml-Teilbaum am besten auf der passenden Ebene ein, zum Beispiel pro Standort oder pro Linie. So bleibt sichtbar, zu welcher Anlage ein Modell gehört, und die Instandhaltung findet Modellalarme neben den Maschinendaten.

Metriken von einem Edge-Gerät veröffentlichen

Das folgende Skript simuliert ein Edge-Gerät, das einen Klassifikator ausführt und pro Intervall eine Metriknachricht veröffentlicht. Auf echter Hardware würden Sie die Statistiken innerhalb Ihrer Inferenzschleife aufsummieren. Es nutzt den öffentlichen MQTT-Broker broker.connectmq.com, der für Tests und Prototypen gedacht ist; ersetzen Sie das Präfix durch Ihre eigene zufällige ID.

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  # Eingang langsam verschieben, um Sensoralterung zu simulieren
    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)

Erst das Histogramm mit festen Klassengrenzen macht die Drift zentral messbar, ohne Rohdaten zu übertragen.

Drift zentral erkennen

Der Monitor abonniert alle Geräte, vergleicht jedes eingehende Histogramm mit einem bei der Bereitstellung erfassten Referenzhistogramm und berechnet den Population Stability Index (PSI). Eine gängige Faustregel betrachtet einen PSI unter 0,1 als stabil, 0,1 bis 0,25 als beobachtungswürdig und über 0,25 als deutliche Verschiebung; stimmen Sie diese Bereiche jedoch auf Ihre eigenen Daten ab.

import json
import math
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/mlops-91bc/plant1/ml/vib-classifier"
reference = {}  # Gerät -> Referenzhistogramm
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()

Starten Sie beide Skripte, und Sie sehen, wie der Status mit zunehmender simulierter Drift von ok über watch zu degraded wechselt. Im Produktivbetrieb sollte das Referenzhistogramm aus den Validierungsdaten stammen, mit denen das Modell freigegeben wurde, und zusammen mit dem Modellartefakt gespeichert werden – nicht aus der ersten Nachricht nach dem Booten.

Auf Alarme reagieren

Ein Alarm ist nur nützlich, wenn jemand oder etwas darauf reagiert. Die Optionen, vom einfachsten bis zum am stärksten automatisierten Ansatz:

1. Eine Person benachrichtigen. Ein Node-RED-Flow oder ein kleiner Dienst abonniert das Alarm-Topic und versendet E-Mails oder Chatnachrichten.
2. Visuelle Alarmierung im Broker-Portal. Auf einem privaten ConnectMQ-Broker können Sie Schwellenwert-Alarmregeln für ein numerisches Feld wie psi festlegen und Treffer im Alarmprotokoll prüfen.
3. Labels sammeln. Driftet ein Gerät, lassen Sie es einige Rohproben auf einem separaten Topic hochladen, damit Ingenieure sie labeln und das Modell neu trainieren können.
4. Zurückrollen. Veröffentlichen Sie auf dem cmd-Topic des Geräts einen Befehl zum Laden der vorherigen Modellversion und bestätigen Sie dies über das Retained-Status-Topic.

Die Rollout-Seite, also wie Modelle überhaupt auf die Geräte gelangen, behandelt der Artikel ML-Modelle auf Edge-Geräten bereitstellen.

Canary-Rollouts und Schattenmodelle

Monitoring sagt Ihnen, wann sich ein Modell verschlechtert; Disziplin beim Rollout verhindert, dass ein neues Modell alles schlimmer macht. Zwei Muster funktionieren gut mit MQTT-Topics.

Canary-Rollout. Liefern Sie das neue Modell zuerst an eine kleine Gerätegruppe aus, zum Beispiel an fünf Prozent der Flotte oder eine Linie pro Werk. Da jede Metriknachricht modelVersion enthält, kann der Monitor Konfidenz, Klassenverteilung und Drift zwischen der Canary-Gruppe und dem Rest über denselben Zeitraum vergleichen. Übernehmen Sie das Modell erst, wenn die Canary-Gruppe mindestens so gesund aussieht wie das bisherige Modell, und rollen Sie andernfalls über dasselbe Befehls-Topic zurück.

Schattenmodus. Führen Sie das neue Modell auf demselben Gerät parallel zum aktuellen aus, handeln Sie aber nur auf Basis der Ausgabe des aktuellen Modells. Veröffentlichen Sie beide Vorhersagen in der Metriknachricht oder eine laufende Übereinstimmungsrate der beiden. Ein Schattenmodell, das bei einem großen Teil der Eingaben vom Produktivmodell abweicht, verdient eine genauere Prüfung, bevor es jemals übernimmt – selbst wenn beide Modelle zuversichtlich wirken.

Beide Muster benötigen vom Broker nichts weiter als zuverlässige Zustellung und klare Topics. Außerdem erleichtern sie die Analyse nach einem Vorfall: Wenn ein Alarm auslöst, lautet die erste Frage, ob er mit einem Versionswechsel zusammenhängt, und die Retained-Status-Topics beantworten das sofort.

Ein praktischer Hinweis zur Bandbreite: Eine Metriknachricht pro Minute und Gerät ist winzig, Uploads von Rohproben zum Labeln sind es nicht. Halten Sie Uploads auf einem separaten Topic, lösen Sie sie nur auf Anforderung aus und bleiben Sie unter der maximalen Nachrichtengröße Ihres Tarifs.

Checkliste für IoT MLOps

1. Jede Nachricht enthält den Modellnamen und die Version.
2. Geräte veröffentlichen in festen Intervallen aggregierte Eingangsstatistiken statt Rohdaten.
3. Zu jedem freigegebenen Modell wird eine Referenzverteilung gespeichert.
4. Drift- und Konfidenzschwellen sind dokumentiert und werden nach jedem Vorfall überprüft.
5. Alarme gehen an ein Topic, und mindestens ein Konsument macht daraus Benachrichtigungen für Menschen.
6. Das Zurückrollen ist ein getesteter Befehl, kein bloßer Plan.
7. Bei auftretender Drift werden gezielt Daten für das Nachtrainieren gesammelt.
8. Metriken und Alarme laufen im Produktivbetrieb über TLS mit gerätespezifischen Zugangsdaten.

Infrastruktur für flottenweites Monitoring

Metriken von Hunderten Geräten, ein Monitor, Dashboards und Rollout-Werkzeuge teilen sich einen gemeinsamen Nachrichtenbus. Ihn selbst zu betreiben bedeutet Patches, Zertifikate und Kapazitätsplanung. Ein verwalteter MQTT-Broker nimmt Ihnen diese Arbeit ab, und ein Cloud-MQTT-Broker ist von Geräten in jedem Netzwerk erreichbar. Mit Shared Subscriptions im kostenpflichtigen Serverless-Tarif können Sie mehrere Monitor-Instanzen betreiben, die sich die Metriklast teilen.

Derselbe Metrikstrom speist auch modellspezifische Arbeiten wie Predictive Maintenance mit Machine Learning, und Driftprüfungen sind eng verwandt mit der IoT-Anomalieerkennung: Die eine überwacht die Maschine, die andere das Modell. Das Gesamtbild von Edge- und Cloud-KI finden Sie im Leitfaden zu KI und IoT.

Überwachen Sie Ihre Modelle im Feld mit ConnectMQ

Erstellen Sie unter app.connectmq.com einen privaten Broker im kostenlosen Serverless-Tarif, ganz ohne Kreditkarte, verbinden Sie Ihre Geräte mit den generierten Zugangsdaten über Port 8883 und beobachten Sie im Live-Topic-Explorer, wie die Modellmetriken eintreffen. Der Drift-Monitor ist dann das obige Skript, ergänzt um Benutzername und Passwort.

Häufig gestellte Fragen

Worin unterscheidet sich IoT MLOps von klassischem MLOps?

Geräte im Feld haben selten Labels, verfügen über begrenzte Bandbreite und können offline sein. Das Monitoring stützt sich daher auf aggregierte Eingangs- und Ausgangsstatistiken statt auf Protokolle pro Anfrage. Auch Bereitstellung und Rollback müssen über unzuverlässige Verbindungen funktionieren.

Wie oft sollten Geräte Modellmetriken veröffentlichen?

Einmal pro Minute oder alle paar hundert Inferenzen ist ein guter Ausgangspunkt. Das ist häufig genug, um Drift innerhalb von Stunden zu erkennen, und sparsam genug, um deutlich unter den Ratenlimits für Nachrichten zu bleiben.

Welche Driftmetrik sollte ich verwenden?

Für Eingänge sind der Population Stability Index und Kolmogorow-Smirnow-Tests verbreitet, und die Überwachung der Konfidenz im unteren Perzentil deckt viele Probleme bei den Ausgaben auf. Nutzen Sie mehr als ein Signal, da jedes einzelne manche Fehlerbilder übersieht.

Kann der Broker Drift für mich erkennen?

Nein. Der Broker transportiert die Metriken; die Drifterkennung läuft in Ihrem eigenen Dienst, der sie abonniert. Sobald Ihr Dienst einen Driftwert veröffentlicht, können Schwellenwert-Alarmregeln im ConnectMQ-Portal einfache numerische Grenzen markieren.