L'apprentissage fédéré pour l'IoT expliqué : entraîner sans centraliser les données

L'apprentissage fédéré pour l'IoT permet d'entraîner un modèle commun sur de nombreux appareils tout en laissant chacun garder ses données brutes. Dans cet article, vous découvrirez comment se déroulent les cycles d'entraînement, quand l'apprentissage fédéré justifie sa complexité, ce qui pose généralement problème et comment un broker MQTT peut transporter le modèle global et les mises à jour locales, avec une petite simulation exécutable que vous pouvez tester sur broker.connectmq.com.

Le problème que résout l'apprentissage fédéré

Le machine learning classique rassemble les données en un seul endroit et y effectue l'entraînement. Pour l'IoT, c'est souvent irréalisable. Les caméras, les objets portables et les machines produisent des données sensibles ou volumineuses ; les clients peuvent interdire qu'elles quittent leurs sites ; et les liaisons cellulaires rendent coûteux l'envoi de tout. Pourtant, un modèle entraîné sur les seules données d'un appareil voit trop peu de diversité pour bien généraliser.

L'apprentissage fédéré inverse le flux. Le modèle voyage vers les données, chaque appareil s'entraîne brièvement sur ce dont il dispose, et seules les mises à jour de paramètres qui en résultent font le chemin retour. Un coordinateur combine ces mises à jour en un meilleur modèle global et le renvoie. Les mesures brutes ne quittent jamais l'appareil.

Comment se déroule un cycle d'entraînement fédéré

L'algorithme le plus courant, Federated Averaging (FedAvg), répète ces étapes :

1. Diffusion : le coordinateur publie le modèle global actuel avec un numéro de cycle.
2. Sélection : un sous-ensemble d'appareils éligibles participe, souvent ceux qui sont inactifs, en charge et sur une connexion non facturée au volume.
3. Entraînement local : chaque participant s'entraîne pendant quelques epochs sur ses données locales, en partant des poids globaux.
4. Remontée : chaque participant envoie ses poids mis à jour (ou la différence avec les poids globaux) et son nombre d'échantillons.
5. Agrégation : le coordinateur fait la moyenne des mises à jour, pondérée par le nombre d'échantillons, pour former le modèle global suivant.
6. Répétition jusqu'à ce que la précision cesse de s'améliorer, puis déploiement du modèle global pour l'inférence.

Les appareils exécutent ensuite le modèle localement, comme pour n'importe quelle application edge AI. Si l'inférence sur l'appareil est nouvelle pour vous, lisez d'abord ce qu'est l'edge AI.

Quand l'apprentissage fédéré convient à l'IoT

L'apprentissage fédéré est un bon candidat lorsque :

• Les données sont sensibles : audio, vidéo, signaux de santé ou données de production appartenant à différents clients.
• Les données sont volumineuses ou continues, ce qui rend leur envoi coûteux.
• Le comportement diffère selon les sites, et vous voulez un modèle unique qui a appris de tous.
• Les appareils ont assez de puissance pour s'entraîner, ce qui désigne généralement des passerelles, des PC industriels, des téléphones ou des boîtiers edge sous Linux plutôt que de petits microcontrôleurs.

C'est souvent le mauvais choix lorsque les données ne sont pas sensibles et peuvent être envoyées à moindre coût, car l'entraînement centralisé est plus simple à déboguer et généralement plus précis. Les appareils de classe microcontrôleur exécutent bien l'inférence mais ont rarement la mémoire nécessaire pour s'entraîner ; dans ces parcs, la passerelle est le participant fédéré naturel. Le guide ce qu'est le TinyML détaille ces limites.

Les défis que vous rencontrerez

L'apprentissage fédéré réduit l'exposition des données brutes, mais il n'est pas automatiquement confidentiel. Traitez les mises à jour comme des données sensibles et protégez-les en transit et au repos.

Communication de l'apprentissage fédéré IoT via MQTT

Les frameworks open source d'apprentissage fédéré fournissent généralement leur propre transport, souvent gRPC, qui suppose que les appareils peuvent maintenir une connexion avec le coordinateur. Les parcs IoT disposent déjà d'une couche de messagerie qui gère les liaisons intermittentes : MQTT. Transposer l'apprentissage fédéré sur MQTT est naturel :

Deux contraintes méritent l'attention. D'abord, la taille des messages : un broker MQTT impose une taille maximale de message, par exemple 64 Ko sur le serveur MQTT public gratuit, 256 Ko sur l'offre ConnectMQ Serverless (Free) et 1 Mo sur Serverless (Paid). Les petits modèles tiennent facilement ; les plus grands nécessitent des mises à jour quantifiées, une compression ou un découpage en plusieurs messages. Ensuite, le contrôle d'accès : seul le coordinateur doit pouvoir publier les modèles globaux, et les appareils ne doivent publier que leurs propres mises à jour ; exécutez donc les tâches fédérées de production sur un broker privé avec des identifiants par appareil et TLS, jamais sur un topic public. Si vous avez besoin d'un rappel sur ce que fait un broker MQTT, commencez par là.

Essayez : FedAvg via MQTT en un seul script

Cette simulation exécute un coordinateur et trois appareils sous forme de clients MQTT distincts dans un seul processus Python. Chaque appareil détient son propre jeu de données privé pour un modèle linéaire, tiré d'une distribution différente pour imiter des données non IID. Seuls les poids et les nombres d'échantillons transitent par le 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()

Chaque cycle affiche les poids agrégés, qui se rapprochent régulièrement des vraies valeurs alors qu'aucun appareil ne partage jamais ses points de données. Pour observer le trafic, abonnez-vous au préfixe affiché avec mosquitto_sub -h broker.connectmq.com -p 1883 -i fl-viewer-42 -t '<prefix>/#' -v. Remarquez la petite taille des messages de mise à jour ; un vrai réseau de neurones serait bien plus volumineux, et c'est là que les limites de taille évoquées plus haut entrent en jeu.

Checklist pour la production

1. Authentifiez chaque appareil avec ses propres identifiants et TLS ; liez les client IDs aux appareils.
2. Signez ou validez les mises à jour sur le coordinateur : vérifiez les dimensions, les plages de valeurs et les numéros de cycle.
3. Fixez une échéance par cycle et agrégez ce qui est arrivé, pour qu'un appareil lent ne puisse pas bloquer l'entraînement.
4. Gardez le modèle global en retained et versionné, et effacez-le à la fin d'une tâche.
5. Suivez des métriques par cycle, comme la perte de validation sur un jeu de test réservé détenu par le coordinateur.
6. Envisagez l'agrégation sécurisée ou la confidentialité différentielle si les mises à jour elles-mêmes sont sensibles.

Pour déployer le modèle obtenu sur le parc, consultez le déploiement de modèles ML sur des appareils edge. Pour les compromis entre périphérie et centralisation en général, lisez edge AI vs IA cloud, et pour une vue d'ensemble, le guide IA et IoT.

Menez vos expériences fédérées sur ConnectMQ

L'apprentissage fédéré nécessite un broker que chaque participant peut joindre de façon sécurisée depuis n'importe quel réseau. ConnectMQ est un broker MQTT managé avec TLS, l'expiration des messages MQTT 5 et les abonnements partagés sur les offres payantes, ainsi que des options Dedicated ou BYOC lorsque les mises à jour du modèle doivent rester dans un environnement isolé. Créez un broker avec l'offre gratuite Serverless sur app.connectmq.com, sans carte bancaire. Pour une expérience rapide comme celle ci-dessus, le serveur MQTT public gratuit suffit ; il reste réservé aux tests et ne doit jamais recevoir de données de production.

Questions fréquentes

L'apprentissage fédéré est-il confidentiel par défaut ?

Il conserve les données brutes sur les appareils, ce qui réduit l'exposition, mais les mises à jour du modèle peuvent tout de même laisser fuiter des informations. Associez-le à un transport chiffré, à l'agrégation sécurisée ou à la confidentialité différentielle lorsque les données sont sensibles.

Les microcontrôleurs peuvent-ils participer à l'apprentissage fédéré ?

Rarement en tant que participants à part entière, car l'entraînement exige bien plus de mémoire que l'inférence. Une conception courante confie l'entraînement à une passerelle, sur des données collectées auprès des microcontrôleurs voisins.

Pourquoi utiliser MQTT plutôt que gRPC pour l'apprentissage fédéré ?

MQTT convient aux parcs à connectivité intermittente : les appareils se connectent en sortie, les retained messages livrent le modèle courant aux retardataires et la QoS gère les nouvelles tentatives. gRPC fonctionne bien lorsque les participants sont des serveurs toujours joignables.

Quelle taille peut atteindre une mise à jour de modèle via MQTT ?

Elle est limitée par la taille maximale des messages du broker. Les modèles plus grands nécessitent des mises à jour quantifiées ou compressées, ou le découpage d'une mise à jour en plusieurs messages avec un numéro de séquence.