IoT için MLOps: Sahadaki Makine Öğrenmesi Modellerini MQTT ile İzlemek

IoT için MLOps, göremediğiniz cihazlarda çalışan makine öğrenmesi modellerini dağıtma, izleme ve güncelleme pratiğidir. Bu rehberde hafif bir izleme döngüsü kuracaksınız: uç cihazlar çıkarım metriklerini MQTT ile yayınlar, merkezi bir Python servisi kaymayı (drift) ölçer ve bozulan modeller ayrı bir topic'te uyarıları tetikler.

Sahadaki Modeller Neden Bozulur

Laboratuvarda iyi skor alan bir model, sahaya çıktığı gün doğruluk kaybetmeye başlar. IoT'de nedenler fizikseldir ve çoğu zaman yavaş ilerler:

• Sensör yaşlanması ve yeniden kalibrasyon. Bir titreşim sensörü kayar, bir kamera lensi tozlanır, bir mikrofon yeni bir fanın sesini almaya başlar.
• Proses değişiklikleri. Yeni bir malzeme, farklı bir vardiya düzeni veya yeni bir ürün varyantı girdi dağılımını kaydırır.
• Mevsimsellik. Ortam sıcaklığı ve nem, aylar içinde sensör referans değerlerini değiştirir.
• Dağıtım uyumsuzluğu. Kuantizasyon veya farklı bir firmware derlemesi, çıktıları değerlendirme modeline kıyasla biraz değiştirir.

Bir web servisinde her isteği kaydedip tahminleri kullanıcı geri bildirimiyle karşılaştırabilirsiniz. Sahada ise genellikle hiç etiket yoktur, bant genişliği sınırlıdır ve cihazlar çevrimdışı olur. Bu yüzden IoT için MLOps vekil sinyallere dayanır: girdi istatistikleri, çıktı dağılımları, güven ve gecikme.

Cihazda Neler Ölçülmeli

Her çıkarım için ham sensör pencerelerini buluta göndermek, uç yapay zekanın amacını ortadan kaldırır. Bunun yerine her cihaz aralık başına birkaç sayıyı toplar ve kompakt bir metrik mesajı yayınlar.

Bir metrik payload'u şöyle görünebilir:

{
  "device": "press-07",
  "model": "vib-classifier",
  "modelVersion": "1.4.2",
  "window": "2026-10-16T10:00:00Z/PT1M",
  "inferences": 600,
  "inputMean": {"rms": 0.42, "peakHz": 118.0},
  "inputStd": {"rms": 0.05, "peakHz": 6.1},
  "classCounts": {"normal": 588, "imbalance": 9, "bearing": 3},
  "confMean": 0.91,
  "confP10": 0.74,
  "latencyMsP95": 14.2
}

IoT için MLOps'ta Topic Tasarımı

Dashboard'ların ve tüketicilerin hassas şekilde abone olabilmesi için model telemetrisini proses telemetrisinden ayrı tutun:

connectmq/plant1/ml/<model>/<device>/metrics
connectmq/plant1/ml/<model>/<device>/status        (retained: version, loaded at)
connectmq/plant1/ml/<model>/alerts
connectmq/plant1/ml/<model>/<device>/cmd           (rollout or rollback commands)

Status topic'ini retained tutmak, her yeni abonenin hangi cihazın hangi model sürümünü çalıştırdığını anında öğrenmesi demektir; bu da bir olay sırasında en sık sorulan sorudur.

Bir Uç Cihazdan Metrik Yayınlamak

Aşağıdaki betik, bir sınıflandırıcı çalıştıran ve her aralıkta bir metrik mesajı yayınlayan bir uç cihazı simüle eder. Gerçek donanımda istatistikleri çıkarım döngünüzün içinde biriktirirsiniz. broker.connectmq.com'un genel test alanını kullanır; önekteki değeri kendi rastgele kimliğinizle değiştirin.

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/mlops-91bc/plant1/ml/vib-classifier"
DEVICE = "press-07"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"{DEVICE}-{uuid.uuid4().hex[:6]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

status = {"modelVersion": "1.4.2", "loadedAt": time.time()}
client.publish(f"{PREFIX}/{DEVICE}/status", json.dumps(status), qos=1, retain=True)

drift = 0.0
while True:
    drift += 0.002  # sensör yaşlanmasını simüle etmek için girdiyi yavaşça kaydır
    rms = [random.gauss(0.42 + drift, 0.05) for _ in range(600)]
    conf = [min(1.0, max(0.0, random.gauss(0.91 - drift, 0.06))) for _ in range(600)]
    conf.sort()
    msg = {
        "device": DEVICE,
        "modelVersion": "1.4.2",
        "inferences": len(rms),
        "inputMean": {"rms": sum(rms) / len(rms)},
        "inputHist": {"rms": [sum(1 for v in rms if lo <= v < lo + 0.05)
                              for lo in [0.25 + 0.05 * i for i in range(10)]]},
        "confMean": sum(conf) / len(conf),
        "confP10": conf[len(conf) // 10],
        "ts": time.time(),
    }
    client.publish(f"{PREFIX}/{DEVICE}/metrics", json.dumps(msg), qos=1)
    time.sleep(10)

Ham veri göndermeden kaymayı merkezi olarak ölçülebilir kılan şey, sabit aralıklı (bin) histogramdır.

Kaymayı Merkezi Olarak Tespit Etmek

İzleyici tüm cihazlara abone olur, gelen her histogramı dağıtım anında yakalanmış bir referans histogramla karşılaştırır ve Popülasyon Kararlılık İndeksi'ni (PSI) hesaplar. Yaygın bir pratik kural, 0,1'in altındaki PSI'yı kararlı, 0,1 ile 0,25 arasını izlenmeye değer, 0,25'in üzerini ise önemli bir kayma olarak kabul eder; ancak bu bantları kendi verinize göre ayarlayın.

import json
import math
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/mlops-91bc/plant1/ml/vib-classifier"
reference = {}  # cihaz -> referans histogram
CONF_FLOOR = 0.70


def psi(expected, actual, eps=1e-4):
    e_total, a_total = sum(expected) or 1, sum(actual) or 1
    score = 0.0
    for e, a in zip(expected, actual):
        e_p = max(e / e_total, eps)
        a_p = max(a / a_total, eps)
        score += (a_p - e_p) * math.log(a_p / e_p)
    return score


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


def on_message(client, userdata, msg):
    m = json.loads(msg.payload)
    dev, hist = m["device"], m["inputHist"]["rms"]
    if dev not in reference:
        reference[dev] = hist
        print(f"reference captured for {dev}")
        return
    score = psi(reference[dev], hist)
    status = "ok"
    if score > 0.25 or m["confP10"] < CONF_FLOOR:
        status = "degraded"
    elif score > 0.1:
        status = "watch"
    print(f"{dev} psi={score:.3f} confP10={m['confP10']:.2f} -> {status}")
    if status != "ok":
        alert = {"device": dev, "modelVersion": m["modelVersion"],
                 "psi": round(score, 3), "confP10": round(m["confP10"], 3),
                 "status": status}
        client.publish(f"{PREFIX}/alerts", json.dumps(alert), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"mlops-monitor-{uuid.uuid4().hex[:6]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

İki betiği de çalıştırdığınızda, simüle edilen kayma biriktikçe durumun ok'tan watch'a, oradan da degraded'a geçtiğini göreceksiniz. Üretimde referans histogram, açılıştan sonraki ilk mesajdan değil, modeli onaylamak için kullanılan doğrulama verisinden gelmeli ve model artefaktıyla birlikte saklanmalıdır.

Uyarılara Göre Harekete Geçmek

Bir uyarı ancak biri ya da bir şey ona yanıt verirse faydalıdır. En basitinden en otomatiğine seçenekler:

1. Bir kişiyi bilgilendirin. Bir Node-RED akışı veya küçük bir servis uyarı topic'ine abone olur ve e-posta ya da sohbet mesajı gönderir.
2. Broker portalında görsel uyarı. Özel bir ConnectMQ broker'ında psi gibi sayısal bir alan üzerinde eşik tabanlı uyarı kuralları tanımlayabilir ve tetiklenmeleri uyarı günlüğünde inceleyebilirsiniz.
3. Etiket toplayın. Bir cihaz kaydığında, mühendislerin etiketleyip yeniden eğitebilmesi için ondan ayrı bir topic'e birkaç ham örnek yüklemesini isteyin.
4. Geri alın. Cihazın cmd topic'inde önceki model sürümünü yükleten bir komut yayınlayın, ardından retained status topic'i üzerinden doğrulayın.

Dağıtım tarafı, yani modellerin cihazlara ilk etapta nasıl ulaştığı, ML modellerini uç cihazlara dağıtmak yazısında ele alınıyor.

Kanarya Dağıtımları ve Gölge Modeller

İzleme, bir modelin ne zaman bozulduğunu söyler; dağıtım disiplini ise yeni bir modelin işleri daha da kötüleştirmesini önler. MQTT topic'leriyle iki kalıp iyi çalışır.

Kanarya dağıtımı. Yeni modeli önce küçük bir cihaz grubuna gönderin; örneğin filonun yüzde beşine veya her fabrikada bir hatta. Her metrik mesajı modelVersion taşıdığından izleyici, aynı dönem için kanarya grubu ile geri kalanlar arasında güveni, sınıf dağılımını ve kaymayı karşılaştırabilir. Yalnızca kanarya en az mevcut model kadar sağlıklı göründüğünde yaygınlaştırın; değilse aynı komut topic'iyle geri alın.

Gölge mod. Yeni modeli aynı cihazda mevcut modelin yanında çalıştırın, ancak yalnızca mevcut modelin çıktısına göre hareket edin. Her iki tahmini de metrik mesajında yayınlayın ya da ikisi arasındaki anlık uyum oranını yayınlayın. Girdilerin büyük bir kısmında üretim modeliyle uyuşmayan bir gölge model, her iki model de kendinden emin görünse bile, devralmadan önce daha yakından incelenmeyi hak eder.

Her iki kalıp da broker'dan güvenilir teslimat ve net topic'ler dışında hiçbir şey gerektirmez. Ayrıca olay sonrası analizi kolaylaştırırlar: bir uyarı tetiklendiğinde ilk soru bunun bir sürüm değişikliğiyle ilişkili olup olmadığıdır ve retained status topic'leri bunu anında yanıtlar.

Bant genişliği hakkında pratik bir not: her cihazdan dakikada bir gelen metrik mesajı çok küçüktür, ancak etiketleme için yapılan ham örnek yüklemeleri öyle değildir. Yüklemeleri ayrı bir topic'te tutun, yalnızca talep üzerine tetikleyin ve planınızın mesaj boyutu sınırının altında kalın.

IoT için MLOps Kontrol Listesi

1. Her mesaj model adını ve sürümünü taşır.
2. Cihazlar sabit bir aralıkta ham veri değil, toplanmış girdi istatistikleri yayınlar.
3. Yayınlanan her modelle birlikte bir referans dağılım saklanır.
4. Kayma ve güven eşikleri belgelenir ve her olaydan sonra gözden geçirilir.
5. Uyarılar bir topic'e gider ve en az bir tüketici bunları insanlara yönelik bildirimlere dönüştürür.
6. Geri alma bir plan değil, test edilmiş bir komuttur.
7. Kayma oluştuğunda yeniden eğitim verisi bilinçli olarak toplanır.
8. Üretimde metrikler ve uyarılar, cihaz başına kimlik bilgileriyle TLS üzerinden taşınır.

Filo Genelinde İzleme için Altyapı

Yüzlerce cihazdan gelen metrikler, bir izleyici, dashboard'lar ve dağıtım araçları tek bir mesaj yolunu paylaşır. Bu yolu kendiniz işletmek; yama, sertifika ve kapasite planlaması demektir. Tam yönetilen bir MQTT servisi bu işi ortadan kaldırır; bir bulut MQTT broker ise her ağdaki cihazlardan erişilebilir. Ücretli Serverless planındaki shared subscription'lar, metrik yükünü paylaşan birkaç izleyici örneği çalıştırmanıza olanak tanır.

Aynı metrik akışı, makine öğrenmesi ile kestirimci bakım gibi modele özgü çalışmaları da besler; kayma kontrolleri ise IoT anomali tespitinin yakın akrabasıdır: biri makineyi, diğeri modeli izler. Uç ve bulut yapay zekasının büyük resmi için yapay zeka ve IoT rehberine bakın.

Sahadaki Modellerinizi ConnectMQ ile İzleyin

app.connectmq.com adresinde ücretsiz Serverless planında kredi kartı gerektirmeden özel bir broker oluşturun, cihazlarınızı oluşturulan kimlik bilgileriyle 8883 portu üzerinden ona yönlendirin ve model metriklerinin canlı topic gezginine gelişini izleyin. Kayma izleyicisini eklemek, yukarıdaki betiğe bir kullanıcı adı ve parola eklemekten ibarettir.

Sık Sorulan Sorular

IoT için MLOps normal MLOps'tan nasıl farklıdır?

Sahadaki cihazlarda nadiren etiket bulunur, bant genişlikleri sınırlıdır ve çevrimdışı olabilirler; bu yüzden izleme, istek başına kayıtlar yerine toplanmış girdi ve çıktı istatistiklerine dayanır. Dağıtım ve geri alma da güvenilmez bağlantılar üzerinden çalışabilmelidir.

Cihazlar model metriklerini ne sıklıkla yayınlamalı?

Dakikada bir veya birkaç yüz çıkarımda bir iyi bir başlangıç noktasıdır. Bu, kaymayı saatler içinde yakalayacak kadar sık ve mesaj hızı sınırlarının rahatça altında kalacak kadar küçüktür.

Hangi kayma metriğini kullanmalıyım?

Girdiler için Popülasyon Kararlılık İndeksi ve Kolmogorov-Smirnov testleri yaygındır; düşük yüzdelik güveni izlemek ise birçok çıktı sorununu yakalar. Her biri bazı arıza türlerini kaçırdığından birden fazla sinyal kullanın.

Broker kaymayı benim için tespit edebilir mi?

Hayır. Broker metrikleri taşır; kayma tespiti, bunlara abone olan kendi servisinizde çalışır. Servisiniz bir kayma skoru yayınladığında, ConnectMQ portalındaki eşik tabanlı uyarı kuralları basit sayısal sınırları işaretleyebilir.