Aprendizaje federado para IoT explicado: entrenar sin centralizar los datos

El aprendizaje federado para IoT es una forma de entrenar un único modelo compartido entre muchos dispositivos mientras cada uno se guarda para sí sus datos en bruto. En esta guía aprenderás cómo funcionan las rondas de entrenamiento, cuándo compensa la complejidad del aprendizaje federado, qué suele fallar y cómo un broker MQTT puede transportar el modelo global y las actualizaciones locales, con una pequeña simulación ejecutable que puedes probar contra broker.connectmq.com.

El problema que resuelve el aprendizaje federado

El machine learning estándar reúne los datos en un solo lugar y entrena allí. En IoT eso suele ser poco práctico. Las cámaras, los wearables y las máquinas generan datos sensibles o voluminosos; los clientes pueden prohibir que salgan de sus instalaciones, y los enlaces móviles encarecen subirlo todo. Sin embargo, un modelo entrenado solo con los datos de un dispositivo ve demasiada poca variedad para generalizar.

El aprendizaje federado invierte el flujo. El modelo viaja hacia los datos, cada dispositivo entrena brevemente con lo que tiene y solo las actualizaciones de parámetros resultantes hacen el camino de vuelta. Un coordinador combina esas actualizaciones en un modelo global mejor y lo vuelve a enviar. Las lecturas en bruto nunca salen del dispositivo.

Cómo funciona una ronda de entrenamiento federado

El algoritmo más habitual, Federated Averaging (FedAvg), repite estos pasos:

1. Difusión: el coordinador publica el modelo global actual con un número de ronda.
2. Selección: participa un subconjunto de dispositivos elegibles, a menudo los que están inactivos, cargando y con una conexión sin límite de datos.
3. Entrenamiento local: cada participante entrena unas pocas épocas con sus datos locales, partiendo de los pesos globales.
4. Informe: cada participante envía sus pesos actualizados (o la diferencia respecto a los pesos globales) y su número de muestras.
5. Agregación: el coordinador promedia las actualizaciones, ponderadas por número de muestras, para formar el siguiente modelo global.
6. Repetición hasta que la precisión deja de mejorar; después se despliega el modelo global para inferencia.

A continuación, los dispositivos ejecutan el modelo en local como edge AI convencional. Si la inferencia en el dispositivo es nueva para ti, lee primero qué es edge AI.

Cuándo encaja el aprendizaje federado en IoT

El aprendizaje federado es un buen candidato cuando:

• Los datos son sensibles: audio, vídeo, señales de salud o datos de producción propiedad de distintos clientes.
• Los datos son grandes o continuos, por lo que subirlos resulta caro.
• El comportamiento varía según la instalación (una mina en el norte de Chile no se comporta como una planta de autopartes en Querétaro) y quieres un único modelo que haya aprendido de todas.
• Los dispositivos tienen capacidad suficiente para entrenar, lo que suele significar gateways, PC industriales, móviles o equipos edge con Linux, más que pequeños microcontroladores.

A menudo es la elección equivocada cuando los datos no son sensibles y pueden subirse a bajo coste, porque el entrenamiento centralizado es más fácil de depurar y suele ser más preciso. Los dispositivos de clase microcontrolador ejecutan bien la inferencia, pero rara vez tienen memoria para entrenar; en esas flotas, el gateway es el participante federado natural. La guía qué es TinyML explica esos límites.

Retos con los que te encontrarás

El aprendizaje federado reduce la exposición de los datos en bruto, pero no es privado automáticamente. Trata las actualizaciones como datos sensibles y protégelas en tránsito y en reposo.

Comunicación del aprendizaje federado IoT sobre MQTT

Los frameworks de aprendizaje federado de código abierto suelen incluir su propio transporte, a menudo gRPC, que da por hecho que los dispositivos pueden mantener una conexión con el coordinador. Las flotas IoT ya cuentan con una capa de mensajería que tolera los enlaces intermitentes: MQTT. Trasladar el aprendizaje federado a MQTT resulta natural:

Dos restricciones merecen atención. Primero, el tamaño de los mensajes: un broker MQTT impone un tamaño máximo de mensaje, por ejemplo 64 KB en el área pública de pruebas, 256 KB en el plan Serverless (Free) de ConnectMQ y 1 MB en Serverless (Paid). Los modelos pequeños caben sin problema; los más grandes necesitan actualizaciones cuantizadas, compresión o división en varios mensajes. Segundo, el control de acceso: solo el coordinador debe poder publicar modelos globales y los dispositivos solo deben publicar sus propias actualizaciones, así que ejecuta los trabajos federados de producción en un broker privado con credenciales por dispositivo y TLS, nunca en un topic público. Si quieres repasar qué hace un broker MQTT, empieza por ahí.

Pruébalo: FedAvg sobre MQTT en un solo script

Esta simulación ejecuta un coordinador y tres dispositivos como clientes MQTT independientes en un único proceso de Python. Cada dispositivo tiene su propio conjunto de datos privado para un modelo lineal, extraído de una distribución distinta para imitar datos no IID. Solo los pesos y el número de muestras pasan por el 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))          # datos privados, nunca se publican
        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):                                # épocas locales de descenso de gradiente
            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                                        # actualización obsoleta de una ronda anterior
        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)                                             # dejar que se asienten las suscripciones
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()   # borrar el modelo retenido
for c in [coord.client] + [d.client for d in devices]:
    c.loop_stop()
    c.disconnect()

Cada ronda imprime los pesos agregados, que se acercan de forma constante a los valores reales aunque ningún dispositivo comparta nunca sus datos. Para observar el tráfico, suscríbete al prefijo impreso con mosquitto_sub -h broker.connectmq.com -p 1883 -i fl-viewer-42 -t '<prefix>/#' -v. Fíjate en lo pequeños que son los mensajes de actualización; una red neuronal real sería mucho más grande, y ahí es donde importan los límites de tamaño mencionados antes.

Lista de comprobación para producción

1. Autentica cada dispositivo con sus propias credenciales y TLS; vincula los client IDs a los dispositivos.
2. Firma o valida las actualizaciones en el coordinador: comprueba formas, rangos de valores y números de ronda.
3. Fija un plazo por ronda y agrega lo que haya llegado, para que un dispositivo lento no pueda bloquear el entrenamiento.
4. Mantén el modelo global retenido y versionado, y bórralo cuando termine una tarea.
5. Haz seguimiento de las métricas de cada ronda, como la pérdida de validación en un conjunto reservado propiedad del coordinador.
6. Considera la agregación segura o la privacidad diferencial si las propias actualizaciones son sensibles.

Para desplegar el modelo resultante en la flota, consulta desplegar modelos de ML en dispositivos edge. Para el equilibrio general entre edge y centralización, lee edge AI vs IA en la nube, y para el panorama completo, la guía de IA e IoT.

Ejecuta experimentos federados en ConnectMQ

El aprendizaje federado necesita un broker al que todos los participantes puedan llegar de forma segura desde cualquier red. ConnectMQ es un broker MQTT gestionado con TLS, caducidad de mensajes de MQTT 5 y suscripciones compartidas en los planes de pago, además de opciones Dedicated o BYOC cuando las actualizaciones del modelo deben permanecer en un entorno aislado. Crea un broker con el plan Serverless gratuito en app.connectmq.com, sin tarjeta de crédito. Para un experimento rápido como el anterior, basta con el servidor MQTT público gratuito, pensado para pruebas y prototipos, no para datos de producción.

Preguntas frecuentes

¿El aprendizaje federado es privado por defecto?

Mantiene los datos en bruto en los dispositivos, lo que reduce la exposición, pero las actualizaciones del modelo pueden seguir filtrando información. Combínalo con transporte cifrado, agregación segura o privacidad diferencial cuando los datos sean sensibles.

¿Pueden los microcontroladores participar en el aprendizaje federado?

Rara vez como participantes completos, porque entrenar requiere mucha más memoria que la inferencia. Un diseño habitual permite que un gateway entrene con los datos recogidos de los microcontroladores cercanos.

¿Por qué usar MQTT en lugar de gRPC para el aprendizaje federado?

MQTT se adapta a flotas con conectividad intermitente: los dispositivos se conectan hacia fuera, los mensajes retenidos entregan el modelo actual a quienes se incorporan tarde y QoS gestiona los reintentos. gRPC funciona bien cuando los participantes son servidores siempre accesibles.

¿Qué tamaño puede tener una actualización del modelo por MQTT?

Está limitado por el tamaño máximo de mensaje del broker. Los modelos más grandes necesitan actualizaciones cuantizadas o comprimidas, o dividir una actualización en varios mensajes con un número de secuencia.