Aprendizado Federado para IoT Explicado: Treinamento sem Centralizar os Dados

O aprendizado federado para IoT é uma forma de treinar um único modelo compartilhado em muitos dispositivos enquanto cada um mantém seus dados brutos para si. Neste guia você vai aprender como funcionam as rodadas de treinamento, quando o aprendizado federado compensa a complexidade, o que costuma dar errado e como um broker MQTT pode transportar o modelo global e as atualizações locais, com uma pequena simulação executável que você pode testar no broker.connectmq.com, o servidor MQTT público (gratuito) do ConnectMQ.

O problema que o aprendizado federado resolve

O machine learning tradicional reúne os dados em um só lugar e treina ali. Em IoT, isso muitas vezes é inviável. Câmeras, wearables e máquinas produzem dados sensíveis ou volumosos; os clientes podem proibir que eles saiam de suas instalações; e links celulares tornam caro enviar tudo. Ainda assim, um modelo treinado apenas com os dados de um dispositivo vê pouca variedade para generalizar.

O aprendizado federado inverte o fluxo. O modelo viaja até os dados, cada dispositivo treina brevemente com o que tem e apenas as atualizações de parâmetros resultantes fazem o caminho de volta. Um coordenador combina essas atualizações em um modelo global melhor e o envia novamente. As leituras brutas nunca saem do dispositivo.

Como funciona uma rodada de treinamento federado

O algoritmo mais comum, o Federated Averaging (FedAvg), repete estes passos:

1. Difusão: o coordenador publica o modelo global atual com um número de rodada.
2. Seleção: um subconjunto de dispositivos elegíveis participa, muitas vezes aqueles que estão ociosos, carregando e em uma conexão sem tarifação por uso.
3. Treinamento local: cada participante treina por algumas épocas com seus dados locais, partindo dos pesos globais.
4. Relato: cada participante envia seus pesos atualizados (ou a diferença em relação aos pesos globais) e sua contagem de amostras.
5. Agregação: o coordenador calcula a média das atualizações, ponderada pela contagem de amostras, para formar o próximo modelo global.
6. Repetição até que a precisão pare de melhorar; então o modelo global é implantado para inferência.

Em seguida, os dispositivos executam o modelo localmente como edge AI comum. Se você é novo em inferência no dispositivo, leia antes o que é edge AI.

Quando o aprendizado federado se encaixa em IoT

O aprendizado federado é um bom candidato quando:

• Os dados são sensíveis: áudio, vídeo, sinais de saúde (dados pessoais sujeitos à LGPD) ou dados de produção pertencentes a clientes diferentes, como plantas de uma mesma indústria em Campinas, Joinville e Manaus.
• Os dados são grandes ou contínuos, de modo que enviá-los é caro.
• O comportamento difere de um local para outro, e você quer um único modelo que tenha aprendido com todos eles.
• Os dispositivos têm capacidade de processamento suficiente para treinar, o que normalmente significa gateways, PCs industriais, celulares ou equipamentos de borda Linux, e não pequenos microcontroladores.

Muitas vezes, é a escolha errada quando os dados não são sensíveis e podem ser enviados a baixo custo, porque o treinamento centralizado é mais simples de depurar e geralmente mais preciso. Dispositivos da classe dos microcontroladores executam bem a inferência, mas raramente têm memória para treinar; nessas frotas, o gateway é o participante federado natural. O guia o que é TinyML aborda esses limites.

Desafios que você vai encontrar

O aprendizado federado reduz a exposição dos dados brutos, mas não é automaticamente privado. Trate as atualizações como dados sensíveis e proteja-as em trânsito e em repouso.

Comunicação de aprendizado federado em IoT via MQTT

Os frameworks open source de aprendizado federado geralmente trazem seu próprio transporte, muitas vezes gRPC, que pressupõe que os dispositivos consigam manter uma conexão com o coordenador. As frotas IoT já têm uma camada de mensageria que lida com links intermitentes: o MQTT. Mapear o aprendizado federado sobre o MQTT é natural:

Duas restrições merecem atenção. Primeiro, o tamanho da mensagem: um broker MQTT impõe um tamanho máximo de mensagem, por exemplo 64 KB no servidor MQTT público, 256 KB no plano Serverless (Free) do ConnectMQ e 1 MB no Serverless (Paid). Modelos pequenos cabem com facilidade; os maiores precisam de atualizações quantizadas, compressão ou divisão em várias mensagens. Segundo, o controle de acesso: apenas o coordenador deve poder publicar modelos globais, e os dispositivos devem publicar apenas as suas próprias atualizações; portanto, rode jobs federados de produção em um broker privado com credenciais por dispositivo e TLS, nunca em um topic público. Se quiser relembrar o que faz um broker MQTT, comece por aí.

Experimente: FedAvg via MQTT em um único script

Esta simulação executa um coordenador e três dispositivos como clientes MQTT separados em um único processo Python. Cada dispositivo tem seu próprio conjunto de dados privado para um modelo linear, extraído de uma distribuição diferente para imitar dados não IID. Apenas pesos e contagens de amostras passam pelo 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))          # dados privados, nunca publicados
        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 locais de gradiente descendente
            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                                        # atualização obsoleta de uma rodada antiga
        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)                                             # aguarda as assinaturas se estabilizarem
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()   # limpa o modelo retido
for c in [coord.client] + [d.client for d in devices]:
    c.loop_stop()
    c.disconnect()

Cada rodada imprime os pesos agregados, que se aproximam de forma constante dos valores verdadeiros, mesmo que nenhum dispositivo jamais compartilhe seus pontos de dados. Para acompanhar o tráfego, assine o prefixo impresso com mosquitto_sub -h broker.connectmq.com -p 1883 -i fl-viewer-42 -t '<prefix>/#' -v. Repare como as mensagens de atualização são pequenas; uma rede neural real seria muito maior, e é aí que os limites de tamanho citados acima importam.

Checklist de produção

1. Autentique cada dispositivo com suas próprias credenciais e TLS; vincule os client IDs aos dispositivos.
2. Assine ou valide as atualizações no coordenador: verifique formatos, faixas de valores e números de rodada.
3. Defina um prazo por rodada e agregue o que tiver chegado, para que um único dispositivo lento não trave o treinamento.
4. Mantenha o modelo global retido e versionado e limpe-o quando uma tarefa terminar.
5. Acompanhe métricas por rodada, como a perda de validação em um conjunto reservado que pertence ao coordenador.
6. Considere agregação segura ou privacidade diferencial se as próprias atualizações forem sensíveis.

Para implantar o modelo resultante na frota, veja como implantar modelos de ML em dispositivos de borda. Para os trade-offs entre borda e processamento central em geral, leia edge AI vs IA na nuvem e, para o panorama completo, o guia de IA e IoT.

Rode experimentos federados no ConnectMQ

O aprendizado federado precisa de um broker que todos os participantes consigam alcançar com segurança a partir de qualquer rede. O ConnectMQ é um broker MQTT gerenciado com TLS, expiração de mensagens do MQTT 5 e assinaturas compartilhadas nos planos pagos, além de opções Dedicated ou BYOC quando as atualizações de modelo precisam ficar em um ambiente isolado. Crie um broker no plano gratuito Serverless em app.connectmq.com, sem cartão de crédito. Para um experimento rápido como o acima, o servidor MQTT público (gratuito) é suficiente; ele serve para testes e protótipos, não para dados de produção.

Perguntas frequentes

O aprendizado federado é privado por padrão?

Ele mantém os dados brutos nos dispositivos, o que reduz a exposição, mas as atualizações de modelo ainda podem vazar informações. Combine-o com transporte criptografado, agregação segura ou privacidade diferencial quando os dados forem sensíveis.

Microcontroladores podem participar do aprendizado federado?

Raramente como participantes completos, porque o treinamento exige muito mais memória do que a inferência. Um projeto comum deixa um gateway treinar com os dados coletados de microcontroladores próximos.

Por que usar MQTT em vez de gRPC para aprendizado federado?

O MQTT é adequado a frotas com conectividade intermitente: os dispositivos se conectam de saída, as mensagens retidas entregam o modelo atual a quem entra depois e o QoS cuida das retentativas. O gRPC funciona bem quando os participantes são servidores sempre acessíveis.

Qual o tamanho máximo de uma atualização de modelo via MQTT?

Ele é limitado pelo tamanho máximo de mensagem do broker. Modelos maiores precisam de atualizações quantizadas ou comprimidas, ou da divisão de uma atualização em várias mensagens com um número de sequência.