Federated Learning für IoT erklärt: Training ohne zentrale Datensammlung

Federated Learning für IoT ist ein Verfahren, ein gemeinsames Modell über viele Geräte hinweg zu trainieren, während jedes Gerät seine Rohdaten für sich behält. In diesem Beitrag erfahren Sie, wie die Trainingsrunden ablaufen, wann sich die Komplexität von Federated Learning lohnt, was typischerweise schiefgeht und wie ein MQTT-Broker das globale Modell und die lokalen Updates transportieren kann, inklusive einer kleinen lauffähigen Simulation, die Sie gegen broker.connectmq.com ausprobieren können.

Das Problem, das Federated Learning löst

Klassisches Machine Learning sammelt Daten an einem Ort und trainiert dort. Für IoT ist das oft unpraktikabel. Kameras, Wearables und Maschinen erzeugen sensible oder umfangreiche Daten; Kunden untersagen womöglich, dass sie ihre Standorte verlassen; und Mobilfunkverbindungen machen das Hochladen aller Daten teuer. Ein Modell, das nur mit den Daten eines Geräts trainiert wurde, sieht jedoch zu wenig Vielfalt, um zu generalisieren.

Federated Learning kehrt den Fluss um. Das Modell wandert zu den Daten, jedes Gerät trainiert kurz mit dem, was es hat, und nur die resultierenden Parameter-Updates wandern zurück. Ein Koordinator kombiniert diese Updates zu einem besseren globalen Modell und verteilt es erneut. Rohe Messwerte verlassen das Gerät nie.

So läuft eine föderierte Trainingsrunde ab

Der gängigste Algorithmus, Federated Averaging (FedAvg), wiederholt diese Schritte:

1. Verteilen: Der Koordinator veröffentlicht das aktuelle globale Modell mit einer Rundennummer.
2. Auswählen: Eine Teilmenge der geeigneten Geräte nimmt teil, oft solche, die gerade ungenutzt sind, laden und an einer Verbindung ohne Volumenbegrenzung hängen.
3. Lokales Training: Jeder Teilnehmer trainiert einige Epochen mit seinen lokalen Daten, ausgehend von den globalen Gewichten.
4. Melden: Jeder Teilnehmer sendet seine aktualisierten Gewichte (oder die Differenz zu den globalen Gewichten) und die Anzahl seiner Samples.
5. Aggregieren: Der Koordinator mittelt die Updates, gewichtet nach Anzahl der Samples, und bildet daraus das nächste globale Modell.
6. Wiederholen, bis sich die Genauigkeit nicht mehr verbessert, und dann das globale Modell für die Inferenz bereitstellen.

Die Geräte führen das Modell anschließend lokal als gewöhnliche Edge AI aus. Wenn Inferenz auf dem Gerät für Sie neu ist, lesen Sie zuerst was Edge AI ist.

Wann Federated Learning zu IoT passt

Federated Learning ist ein starker Kandidat, wenn:

• Daten sensibel sind: Audio, Video, Gesundheitssignale oder Produktionsdaten verschiedener Kunden.
• Daten umfangreich oder kontinuierlich anfallen, sodass das Hochladen teuer ist.
• sich das Verhalten je Standort unterscheidet und Sie ein Modell möchten, das von allen gelernt hat.

Ein typisches Szenario im DACH-Maschinenbau: Ein Hersteller möchte ein Modell für Predictive Maintenance verbessern, das auf den Maschinen in den Werken vieler Kunden läuft. Die Betreiber geben ihre Prozess- und Produktionsdaten aus Gründen des Know-how-Schutzes oder der DSGVO nicht heraus, akzeptieren aber, dass ein Edge-Gateway an der Maschine lokal trainiert und nur Modell-Updates teilt.

• Geräte genug Rechenleistung zum Trainieren haben, was meist Gateways, Industrie-PCs, Smartphones oder Linux-Edge-Boxen bedeutet und keine kleinen Mikrocontroller.

Oft ist es die falsche Wahl, wenn die Daten nicht sensibel sind und sich günstig hochladen lassen, denn zentrales Training ist einfacher zu debuggen und meist genauer. Geräte der Mikrocontroller-Klasse können gut Inferenz ausführen, haben aber selten den Speicher zum Trainieren; in solchen Flotten ist das Gateway der natürliche föderierte Teilnehmer. Der Leitfaden was TinyML ist behandelt diese Grenzen.

Herausforderungen, denen Sie begegnen werden

Federated Learning verringert die Preisgabe von Rohdaten, ist aber nicht automatisch privat. Behandeln Sie Updates als sensibel und schützen Sie sie bei Übertragung und Speicherung.

Federated-Learning-IoT-Kommunikation über MQTT

Open-Source-Frameworks für Federated Learning bringen in der Regel ihren eigenen Transport mit, oft gRPC, der voraussetzt, dass Geräte eine Verbindung zum Koordinator halten können. IoT-Flotten verfügen bereits über eine Messaging-Schicht, die mit unterbrochenen Verbindungen umgehen kann: MQTT. Federated Learning auf MQTT abzubilden liegt nahe:

Wenn Ihre Maschinendaten bereits in einem Unified Namespace (UNS) liegen, dessen Topic-Struktur der ISA-95-Hierarchie von Werk, Bereich und Linie folgt, halten Sie die fl/-Topics als eigenen Zweig daneben, damit Trainingsverkehr und Betriebsdaten getrennt berechtigt und überwacht werden können.

Zwei Einschränkungen verdienen Aufmerksamkeit. Erstens die Nachrichtengröße: Ein MQTT-Broker setzt eine maximale Nachrichtengröße durch, zum Beispiel 64 KB auf dem öffentlichen MQTT-Broker, 256 KB im ConnectMQ-Plan Serverless (Free) und 1 MB bei Serverless (Paid). Kleine Modelle passen problemlos; größere brauchen quantisierte Updates, Kompression oder eine Aufteilung auf mehrere Nachrichten. Zweitens die Zugriffskontrolle: Nur der Koordinator sollte globale Modelle veröffentlichen dürfen, und Geräte sollten nur ihre eigenen Updates veröffentlichen. Betreiben Sie produktive föderierte Jobs daher auf einem privaten Broker mit Zugangsdaten pro Gerät und TLS, niemals auf einem öffentlichen Topic. Wenn Sie auffrischen möchten, was ein MQTT-Broker macht, beginnen Sie dort.

Ausprobieren: FedAvg über MQTT in einem Skript

Diese Simulation betreibt einen Koordinator und drei Geräte als separate MQTT-Clients in einem einzigen Python-Prozess. Jedes Gerät hält seinen eigenen privaten Datensatz für ein lineares Modell, gezogen aus einer jeweils anderen Verteilung, um Nicht-IID-Daten nachzubilden. Nur Gewichte und Sample-Anzahlen gehen über den Broker.

pip install "paho-mqtt>=2.0" numpy

import json
import threading
import time
import uuid

import numpy as np
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
P = f"connectmq/demo/{uuid.uuid4().hex[:8]}/fl/demo-task"
TRUE_W, TRUE_B = np.array([2.0, -1.0, 0.5]), 0.3
ROUNDS, N_DEVICES = 10, 3
done = threading.Event()


def make_client(name, on_connect, on_message):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{name}-{uuid.uuid4().hex[:6]}")
    c.on_connect, c.on_message = on_connect, on_message
    c.connect(HOST, PORT, keepalive=60)
    c.loop_start()
    return c


class Device:
    def __init__(self, name, n, shift, seed):
        rng = np.random.default_rng(seed)
        self.X = rng.normal(shift, 1.0, (n, 3))          # private data, never published
        self.y = self.X @ TRUE_W + TRUE_B + rng.normal(0, 0.1, n)
        self.name, self.last_round = name, -1
        self.client = make_client(name, self.on_connect, self.on_global)

    def on_connect(self, client, userdata, flags, reason_code, properties):
        client.subscribe(f"{P}/global", qos=1)

    def on_global(self, client, userdata, msg):
        if not msg.payload:
            return
        g = json.loads(msg.payload)
        if g["round"] <= self.last_round or g["round"] >= ROUNDS:
            return
        self.last_round = g["round"]
        w, b = np.array(g["w"]), g["b"]
        for _ in range(5):                                # local epochs of gradient descent
            err = self.X @ w + b - self.y
            w -= 0.05 * self.X.T @ err / len(self.y)
            b -= 0.05 * err.mean()
        update = {"round": g["round"], "n": len(self.y), "w": w.tolist(), "b": float(b)}
        client.publish(f"{P}/update/{self.name}", json.dumps(update), qos=1)


class Coordinator:
    def __init__(self):
        self.round, self.updates = 0, {}
        self.client = make_client("coordinator", self.on_connect, self.on_update)

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

    def publish_global(self, w, b):
        msg = {"round": self.round, "w": list(map(float, w)), "b": float(b)}
        self.client.publish(f"{P}/global", json.dumps(msg), qos=1, retain=True)

    def on_update(self, client, userdata, msg):
        u = json.loads(msg.payload)
        if u["round"] != self.round:
            return                                        # stale update from an old round
        self.updates[msg.topic.rsplit("/", 1)[-1]] = u
        if len(self.updates) < N_DEVICES:
            return
        total = sum(x["n"] for x in self.updates.values())
        w = sum(np.array(x["w"]) * x["n"] for x in self.updates.values()) / total
        b = sum(x["b"] * x["n"] for x in self.updates.values()) / total
        print(f"round {self.round}: w={np.round(w, 3)} b={b:.3f}")
        self.round, self.updates = self.round + 1, {}
        self.publish_global(w, b)
        if self.round >= ROUNDS:
            done.set()


coord = Coordinator()
devices = [Device("gw-a", 200, -1.0, 1), Device("gw-b", 120, 0.0, 2), Device("gw-c", 80, 1.5, 3)]
time.sleep(2)                                             # let subscriptions settle
print("topics under", P)
coord.publish_global(np.zeros(3), 0.0)
done.wait(timeout=120)
print("true  : w=", TRUE_W, "b=", TRUE_B)

coord.client.publish(f"{P}/global", b"", qos=1, retain=True).wait_for_publish()   # clear retained model
for c in [coord.client] + [d.client for d in devices]:
    c.loop_stop()
    c.disconnect()

Jede Runde gibt die aggregierten Gewichte aus, die sich stetig den wahren Werten nähern, obwohl kein Gerät jemals seine Datenpunkte teilt. Um den Datenverkehr zu beobachten, abonnieren Sie das ausgegebene Präfix mit mosquitto_sub -h broker.connectmq.com -p 1883 -i fl-viewer-42 -t '<prefix>/#' -v. Beachten Sie, wie klein die Update-Nachrichten sind; ein echtes neuronales Netz wäre weitaus größer, und genau dort spielen die oben genannten Größenlimits eine Rolle.

Checkliste für die Produktion

1. Authentifizieren Sie jedes Gerät mit eigenen Zugangsdaten und TLS; binden Sie Client IDs an Geräte.
2. Signieren oder validieren Sie Updates auf dem Koordinator: Prüfen Sie Formen, Wertebereiche und Rundennummern.
3. Legen Sie eine Rundenfrist fest und aggregieren Sie, was angekommen ist, damit ein langsames Gerät das Training nicht aufhalten kann.
4. Halten Sie das globale Modell als Retained Message und versioniert vor und löschen Sie es, wenn eine Aufgabe endet.
5. Verfolgen Sie Metriken pro Runde, etwa den Validierungsverlust auf einem zurückgehaltenen Datensatz, der dem Koordinator gehört.
6. Erwägen Sie Secure Aggregation oder Differential Privacy, wenn die Updates selbst sensibel sind.

Wie Sie das resultierende Modell auf die Flotte bringen, lesen Sie unter ML-Modelle auf Edge-Geräten bereitstellen. Zu den allgemeinen Zielkonflikten zwischen Edge und Zentrale lesen Sie Edge AI vs. Cloud AI und für das Gesamtbild den Leitfaden zu KI und IoT.

Föderierte Experimente auf ConnectMQ durchführen

Federated Learning braucht einen Broker, den jeder Teilnehmer aus jedem Netzwerk sicher erreichen kann. ConnectMQ ist ein verwalteter MQTT-Broker mit TLS, MQTT-5-Message-Expiry und Shared Subscriptions in kostenpflichtigen Plänen sowie Dedicated- oder BYOC-Optionen, wenn Modell-Updates in einer isolierten Umgebung bleiben müssen. Erstellen Sie einen Broker im kostenlosen Serverless-Plan unter app.connectmq.com, ohne Kreditkarte. Für ein schnelles Experiment wie das obige reicht der kostenlose öffentliche MQTT-Broker, der für Tests und Prototypen gedacht ist, nicht für echte Modell-Updates.

Häufig gestellte Fragen

Ist Federated Learning standardmäßig privat?

Es hält Rohdaten auf den Geräten, was die Preisgabe verringert, doch Modell-Updates können trotzdem Informationen verraten. Kombinieren Sie es bei sensiblen Daten mit verschlüsselter Übertragung, Secure Aggregation oder Differential Privacy.

Können Mikrocontroller an Federated Learning teilnehmen?

Als vollwertige Teilnehmer nur selten, weil Training deutlich mehr Speicher braucht als Inferenz. Ein gängiges Design lässt ein Gateway mit Daten trainieren, die von Mikrocontrollern in der Nähe gesammelt wurden.

Warum MQTT statt gRPC für Federated Learning?

MQTT eignet sich für Flotten mit unterbrochener Konnektivität: Geräte verbinden sich ausgehend, Retained Messages liefern Nachzüglern das aktuelle Modell, und QoS kümmert sich um Wiederholungen. gRPC funktioniert gut, wenn die Teilnehmer stets erreichbare Server sind.

Wie groß darf ein Modell-Update über MQTT sein?

Es ist durch die maximale Nachrichtengröße des Brokers begrenzt. Größere Modelle brauchen quantisierte oder komprimierte Updates oder die Aufteilung eines Updates auf mehrere Nachrichten mit einer Sequenznummer.