IoT'de Federe Öğrenme: Veriyi Merkezileştirmeden Model Eğitimi

IoT için federe öğrenme (federated learning), her cihaz kendi ham verisini kendine saklarken çok sayıda cihaz genelinde tek bir ortak model eğitmenin bir yoludur. Bu yazıda eğitim turlarının nasıl işlediğini, federe öğrenmenin karmaşıklığına ne zaman değdiğini, genellikle nelerin ters gittiğini ve bir MQTT broker'ın küresel modeli ve yerel güncellemeleri nasıl taşıyabileceğini öğrenecek; broker.connectmq.com üzerinde deneyebileceğiniz küçük, çalıştırılabilir bir simülasyon da bulacaksınız.

Federe Öğrenmenin Çözdüğü Problem

Standart makine öğrenimi veriyi tek bir yerde toplar ve orada eğitir. IoT için bu çoğu zaman pratik değildir. Kameralar, giyilebilir cihazlar ve makineler hassas veya hacimli veri üretir; müşteriler verinin tesislerinden çıkmasını yasaklayabilir; hücresel bağlantılar da her şeyi yüklemeyi pahalı hale getirir. Öte yandan yalnızca tek bir cihazın verisiyle eğitilen bir model, genelleme yapabilmek için çok az çeşitlilik görür.

Federe öğrenme akışı tersine çevirir. Model veriye gider, her cihaz elindekiyle kısa bir süre eğitim yapar ve geri yalnızca ortaya çıkan parametre güncellemeleri gelir. Bir koordinatör bu güncellemeleri daha iyi bir küresel modelde birleştirir ve yeniden dağıtır. Ham ölçümler cihazdan asla çıkmaz.

Bir Federe Eğitim Turu Nasıl İşler

En yaygın algoritma olan Federated Averaging (FedAvg) şu adımları tekrarlar:

1. Yayın: koordinatör mevcut küresel modeli bir tur numarasıyla yayınlar.
2. Seçim: uygun cihazların bir alt kümesi katılır; çoğu zaman boşta olan, şarj olan ve kotasız bir bağlantıdaki cihazlar.
3. Yerel eğitim: her katılımcı küresel ağırlıklardan başlayarak yerel verisi üzerinde birkaç epoch eğitim yapar.
4. Raporlama: her katılımcı güncellenmiş ağırlıklarını (veya küresel ağırlıklardan farkını) ve örnek sayısını gönderir.
5. Birleştirme: koordinatör, bir sonraki küresel modeli oluşturmak için güncellemelerin örnek sayısına göre ağırlıklı ortalamasını alır.
6. Tekrar: doğruluk artmayı bırakana kadar; ardından küresel model çıkarım için dağıtılır.

Cihazlar daha sonra modeli sıradan edge AI olarak yerelde çalıştırır. Cihaz üzerinde çıkarımda yeniyseniz önce edge AI nedir yazısını okuyun.

Federe Öğrenme IoT'ye Ne Zaman Uyar

Federe öğrenme şu durumlarda güçlü bir adaydır:

• Veri hassassa: ses, video, sağlık sinyalleri veya farklı müşterilere ait üretim verileri.
• Veri büyük veya sürekliyse ve yüklemesi maliyetliyse.
• Davranış tesisten tesise farklılık gösteriyorsa ve hepsinden öğrenmiş tek bir model istiyorsanız.
• Cihazların eğitim yapacak kadar hesaplama gücü varsa; bu genellikle küçük mikrodenetleyiciler yerine ağ geçitleri, endüstriyel PC'ler, telefonlar veya Linux edge kutuları anlamına gelir.

Veri hassas değilse ve ucuza yüklenebiliyorsa genellikle yanlış seçimdir; çünkü merkezi eğitimde hata ayıklamak daha basittir ve genellikle daha doğru sonuç verir. Mikrodenetleyici sınıfındaki cihazlar çıkarımı iyi çalıştırabilir, ancak eğitim için nadiren yeterli belleğe sahiptir; bu tür filolarda doğal federe katılımcı ağ geçididir. TinyML nedir rehberi bu sınırları ele alıyor.

Karşılaşacağınız Zorluklar

Federe öğrenme ham verinin açığa çıkmasını azaltır, ancak otomatik olarak gizlilik sağlamaz. Güncellemeleri hassas kabul edin ve aktarımda ve depolamada koruyun.

MQTT Üzerinden IoT Federe Öğrenme İletişimi

Açık kaynaklı federe öğrenme framework'leri genellikle kendi aktarım katmanlarıyla, çoğu zaman cihazların koordinatöre sürekli bağlı kalabileceğini varsayan gRPC ile gelir. IoT filolarının ise kesintili bağlantılarla başa çıkabilen bir mesajlaşma katmanı zaten vardır: MQTT. Federe öğrenmeyi MQTT'ye eşlemek doğaldır:

İki kısıt dikkat gerektirir. Birincisi mesaj boyutu: bir MQTT broker maksimum bir mesaj boyutu uygular; örneğin herkese açık test alanında 64 KB, ConnectMQ Serverless (Free) planında 256 KB ve Serverless (Paid) planında 1 MB. Küçük modeller kolayca sığar; daha büyükleri nicemlenmiş güncellemeler, sıkıştırma veya birden çok mesaja bölme gerektirir. İkincisi erişim kontrolü: küresel modelleri yalnızca koordinatör yayınlayabilmeli, cihazlar ise yalnızca kendi güncellemelerini yayınlayabilmelidir; bu yüzden üretimdeki federe işleri cihaz başına kimlik bilgileri ve TLS ile özel bir broker'da çalıştırın, asla herkese açık bir topic'te değil. Bir MQTT broker'ın ne yaptığı konusunda hafızanızı tazelemek isterseniz oradan başlayın.

Deneyin: Tek Betikte MQTT Üzerinden FedAvg

Bu simülasyon, bir koordinatörü ve üç cihazı tek bir Python sürecinde ayrı MQTT istemcileri olarak çalıştırır. Her cihaz, IID olmayan veriyi taklit etmek için farklı bir dağılımdan çekilmiş, doğrusal bir model için kendi özel veri kümesini tutar. Broker'dan yalnızca ağırlıklar ve örnek sayıları geçer.

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))          # özel veri, asla yayınlanmaz
        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):                                # gradyan inişinin yerel epoch'ları
            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                                        # eski bir turdan gelen bayat güncelleme
        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)                                             # aboneliklerin oturmasını bekle
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()   # retained modeli temizle
for c in [coord.client] + [d.client for d in devices]:
    c.loop_stop()
    c.disconnect()

Her tur, hiçbir cihaz veri noktalarını paylaşmamasına rağmen gerçek değerlere doğru istikrarla ilerleyen birleştirilmiş ağırlıkları yazdırır. Trafiği izlemek için yazdırılan öneke mosquitto_sub -h broker.connectmq.com -p 1883 -i fl-viewer-42 -t '<prefix>/#' -v ile abone olun. Güncelleme mesajlarının ne kadar küçük olduğuna dikkat edin; gerçek bir sinir ağı çok daha büyük olurdu ve yukarıdaki boyut sınırları tam da burada önem kazanır.

Üretim Kontrol Listesi

1. Her cihazın kimliğini kendi kimlik bilgileri ve TLS ile doğrulayın; client ID'leri cihazlara bağlayın.
2. Koordinatörde güncellemeleri imzalayın veya doğrulayın: şekilleri, değer aralıklarını ve tur numaralarını kontrol edin.
3. Bir tur süre sınırı belirleyin ve ne geldiyse onu birleştirin; böylece tek bir yavaş cihaz eğitimi durduramaz.
4. Küresel modeli retained ve sürümlenmiş tutun ve bir görev bittiğinde temizleyin.
5. Tur başına metrikleri izleyin; örneğin koordinatöre ait ayrılmış bir veri kümesindeki doğrulama kaybı.
6. Güncellemelerin kendisi hassassa güvenli birleştirmeyi veya diferansiyel gizliliği değerlendirin.

Ortaya çıkan modeli filoya dağıtmak için ML modellerini edge cihazlara dağıtma yazısına bakın. Genel olarak uç ile merkez arasındaki ödünleşimler için edge AI ve bulut yapay zeka yazısını, tüm tablo için de yapay zeka ve IoT rehberini okuyun.

Federe Deneylerinizi ConnectMQ Üzerinde Çalıştırın

Federe öğrenme, her katılımcının herhangi bir ağdan güvenli şekilde erişebileceği bir broker gerektirir. ConnectMQ; TLS, MQTT 5 mesaj sona erme süresi ve ücretli planlarda paylaşımlı abonelikler sunan, model güncellemelerinin izole bir ortamda kalması gerektiğinde Dedicated veya BYOC seçenekleri de olan barındırılan bir MQTT broker hizmetidir. app.connectmq.com adresinde ücretsiz Serverless planında kredi kartı gerektirmeden bir broker oluşturun. Yukarıdaki gibi hızlı bir deney için ücretsiz herkese açık MQTT broker yeterlidir.

Sık Sorulan Sorular

Federe öğrenme varsayılan olarak gizli midir?

Ham veriyi cihazlarda tutar ve bu da açığa çıkma riskini azaltır, ancak model güncellemeleri yine de bilgi sızdırabilir. Veri hassas olduğunda şifreli aktarım, güvenli birleştirme veya diferansiyel gizlilikle birleştirin.

Mikrodenetleyiciler federe öğrenmeye katılabilir mi?

Tam katılımcı olarak nadiren; çünkü eğitim, çıkarımdan çok daha fazla bellek gerektirir. Yaygın bir tasarım, yakındaki mikrodenetleyicilerden toplanan verilerle bir ağ geçidinin eğitim yapmasıdır.

Federe öğrenme için neden gRPC yerine MQTT kullanılmalı?

MQTT kesintili bağlantıya sahip filolara uygundur: cihazlar dışarıya doğru bağlanır, retained mesajlar mevcut modeli geç katılanlara iletir ve QoS yeniden denemeleri yönetir. gRPC ise katılımcıların her zaman erişilebilir sunucular olduğu durumlarda iyi çalışır.

MQTT üzerinden bir model güncellemesi ne kadar büyük olabilir?

Broker'ın maksimum mesaj boyutuyla sınırlıdır. Daha büyük modeller nicemlenmiş veya sıkıştırılmış güncellemeler ya da bir güncellemenin sıra numarasıyla birkaç mesaja bölünmesini gerektirir.