Endüstriyel Veri için RAG: Ekipman Kılavuzlarını Canlı MQTT Telemetrisiyle Birleştirin

Endüstriyel veri için RAG, "hat 3'teki kompresör neden alarm veriyor ve neyi kontrol etmeliyim?" gibi soruları, bir teknisyenin normalde aynı anda idare etmek zorunda kaldığı iki kaynağı birleştirerek yanıtlar: ekipman dokümantasyonu ve makinenin güncel okumaları. Bu rehber; kılavuzların ve prosedürlerin nasıl indeksleneceğini, soru anında MQTT topic'lerinden canlı değerlerin nasıl çekileceğini ve kaynak gösteren sağlam yanıtlar üreten bir prompt'un nasıl oluşturulacağını, ayrıca fabrika sahasında önem taşıyan tuzakları gösteriyor.

Bir Tesiste Sade RAG Neden Yeterli Değil?

Erişimle zenginleştirilmiş üretim (RAG), belgelerinizden ilgili pasajları bulur ve bunları bir dil modeline verir; böylece model genel bilgiden değil, sizin içeriğinizden yanıt verebilir. Ofis belgeleri için hikâyenin tamamı budur. Endüstride ise yetersiz kalır, çünkü yararlı soruların çoğu bir makinenin mevcut durumuna bağlıdır:

• "Bu titreşim seviyesi bu pompa için normal mi?" sorusu hem kılavuzdaki limiti hem de güncel değeri gerektirir.
• "E-214 arıza kodu ne anlama geliyor ve hâlâ aktif mi?" sorusu arıza tablosunu ve canlı alarm durumunu gerektirir.
• "Soğutucuyu şimdi yeniden başlatabilir miyim?" sorusu yeniden başlatma prosedürünü ve güncel basınç ve sıcaklıkları gerektirir.

Yalnızca belgelere sahip bir model, değeri bilmeden limitleri aktarır. Yalnızca telemetriye sahip bir model ise sayıları ne anlama geldiklerini bilmeden görür. Bir asistanı bakım teknisyeni için yararlı kılan, ikisini birleştirmektir.

Endüstriyel Veri için RAG Mimarisi

Varlık kaydı, çoğu eğitimin atladığı ve sistemi çalışır kılan parçadır. Bir teknisyen "kompresör C-3" hakkında soru sorduğunda, kayıt size hangi kılavuzun geçerli olduğunu (başka açılardan benzer makineler arasında model numaraları farklıdır) ve okumalarının hangi topic'lerde tutulduğunu söyler.

{
  "C-3": {
    "description": "Screw air compressor, line 3",
    "model": "SC-75",
    "documents": ["sc75-operating-manual", "sc75-fault-codes", "wo-history-c3"],
    "topics": {
      "discharge_temp_c": "connectmq/demo/8b2e61/plant1/line3/c3/discharge_temp",
      "discharge_pressure_bar": "connectmq/demo/8b2e61/plant1/line3/c3/pressure",
      "active_fault": "connectmq/demo/8b2e61/plant1/line3/c3/fault"
    }
  }
}

ISA-95 seviyelerini (kuruluş, saha, alan, hat, hücre) izleyen bir topic yapısı bu eşlemeyi neredeyse mekanik hâle getirir. Makineleriniz Sparkplug B yayınlıyorsa, ConnectMQ bunu çözümleyip metrikleri okunabilir ISA-95 topic'lerine yeniden yayınlayabilir; bir RAG sistemi de bunları Protobuf çözücüsü olmadan tüketebilir. Sparkplug B ve Unified Namespace yazımız bu topic modelini açıklıyor.

Adım 1: Belgeleri Hazırlayın ve İndeksleyin

Yanıt kalitesini model seçiminden çok belge hazırlığı belirler.

1. Temiz metin çıkarın. PDF'leri metne dönüştürün ve tabloların sağlam kaldığını kontrol edin. Arıza kodu tabloları çoğu zaman en değerli içeriktir ve bozulma ihtimali en yüksek olanlardır.
2. Yapıya göre parçalayın. Sabit karakter sayılarına göre değil, başlıklara ve prosedür adımlarına göre bölün; böylece bir prosedür bütün kalır. Parçaları birkaç yüz kelimeyle sınırlayın.
3. Meta veri ekleyin. Her parçayla birlikte belge kimliğini, başlığı, bölümü, sayfayı ve ekipman modelini saklayın.
4. Embedding'e dönüştürün ve saklayın. Her parça için bir embedding hesaplayın ve meta veriyle birlikte tutun.

Aşağıdaki örnek, yerel bir embedding modeliyle Ollama'nın embedding API'sini kullanıyor ve indeksi NumPy'da tutuyor. pip install ollama numpy "paho-mqtt>=2.0" ve ollama pull nomic-embed-text ile kurun.

# index_docs.py
import json
import re
from pathlib import Path

import numpy as np
import ollama

EMBED_MODEL = "nomic-embed-text"
DOCS = Path("docs")          # one .md or .txt file per document, named <doc-id>.md


def chunks_for(path):
    text = path.read_text(encoding="utf-8")
    sections = re.split(r"\n(?=#{1,3} )", text)   # split before markdown headings
    for i, section in enumerate(sections):
        section = section.strip()
        if len(section) < 40:
            continue
        title = section.splitlines()[0].lstrip("# ").strip()
        yield {"doc": path.stem, "section": title, "chunk": i, "text": section[:3000]}


records = [c for p in sorted(DOCS.glob("*.md")) for c in chunks_for(p)]
vectors = []
for start in range(0, len(records), 32):
    batch = records[start:start + 32]
    resp = ollama.embed(model=EMBED_MODEL, input=["search_document: " + r["text"] for r in batch])
    vectors.extend(resp.embeddings)

matrix = np.array(vectors, dtype=np.float32)
matrix /= np.linalg.norm(matrix, axis=1, keepdims=True)
np.save("index.npy", matrix)
Path("index.json").write_text(json.dumps(records), encoding="utf-8")
print(f"indexed {len(records)} chunks")

search_document: ve search_query: önekleri, bu belirli embedding modelinin beklediği görev talimatlarıdır; diğer embedding modellerinin kendi kuralları vardır, bu yüzden kullandığınız modelin model kartını kontrol edin.

Adım 2: MQTT'den Bir Canlı Değer Önbelleği Tutun

Önbellek varlık topic'lerine abone olur ve en son değeri zaman damgasıyla birlikte saklar. Cihazlar retained değerler yayınladığından, önbellek başladıktan birkaç an sonra tamamlanır.

# live_cache.py
import json
import threading
import time
import paho.mqtt.client as mqtt

_cache, _lock = {}, threading.Lock()


def start(topics, client_id="rag-cache-8b2e61", host="broker.connectmq.com", port=1883):
    def on_connect(client, userdata, flags, reason_code, properties):
        for t in topics:
            client.subscribe(t, qos=1)

    def on_message(client, userdata, msg):
        raw = msg.payload.decode("utf-8", errors="replace")
        try:
            value = json.loads(raw)
        except json.JSONDecodeError:
            value = raw
        with _lock:
            _cache[msg.topic] = {"value": value, "receivedAt": time.time()}

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(host, port, keepalive=60)
    client.loop_start()
    return client


def snapshot(topic_map):
    now = time.time()
    out = {}
    with _lock:
        for name, topic in topic_map.items():
            entry = _cache.get(topic)
            out[name] = None if entry is None else {
                "value": entry["value"], "ageSeconds": round(now - entry["receivedAt"], 1)}
    return out

Her değerin yaşını bildirmek şarttır. İki saat önce gelmiş bir okuma "güncel" değildir ve model bunu belirtmelidir.

Adım 3: Getirin, Birleştirin ve Yanıtlayın

Soru anında varlığı belirleyin, pasajları yalnızca o varlığın belgelerinden getirin, canlı değerlerini okuyun ve her parçanın etiketlendiği bir prompt oluşturun.

# ask.py
import json
from pathlib import Path

import numpy as np
import ollama

import live_cache

EMBED_MODEL, CHAT_MODEL = "nomic-embed-text", "llama3.1:8b"
REGISTRY = json.loads(Path("assets.json").read_text())
RECORDS = json.loads(Path("index.json").read_text(encoding="utf-8"))
MATRIX = np.load("index.npy")

live_cache.start([t for a in REGISTRY.values() for t in a["topics"].values()])

SYSTEM = """You assist maintenance technicians.
Answer using ONLY the DOCUMENT passages and LIVE values provided.
Cite documents as [doc-id, section] and live values as [live: name].
If a live value is missing or older than 300 seconds, say it is unavailable or stale.
If the documents do not cover the question, say so. Never invent limits, codes or procedures.
Safety instructions in the documents (lockout, PPE) must be repeated, not shortened."""


def retrieve(question, docs, k=4):
    q = ollama.embed(model=EMBED_MODEL, input=["search_query: " + question]).embeddings[0]
    q = np.array(q, dtype=np.float32)
    q /= np.linalg.norm(q)
    scores = MATRIX @ q
    order = [i for i in np.argsort(-scores) if RECORDS[i]["doc"] in docs]
    return [RECORDS[i] | {"score": float(scores[i])} for i in order[:k]]


def ask(asset_id, question):
    asset = REGISTRY[asset_id]
    passages = retrieve(question, set(asset["documents"]))
    live = live_cache.snapshot(asset["topics"])
    context = "\n\n".join(f"DOCUMENT [{p['doc']}, {p['section']}]\n{p['text']}" for p in passages)
    live_text = "\n".join(f"LIVE [{k}]: {json.dumps(v)}" for k, v in live.items())
    prompt = (f"ASSET: {asset_id} - {asset['description']} (model {asset['model']})\n\n"
              f"{live_text}\n\n{context}\n\nQUESTION: {question}")
    resp = ollama.chat(model=CHAT_MODEL, options={"temperature": 0.1},
                       messages=[{"role": "system", "content": SYSTEM},
                                 {"role": "user", "content": prompt}])
    return resp.message.content


if __name__ == "__main__":
    import time
    time.sleep(2)   # allow retained values to arrive
    print(ask("C-3", "Discharge temperature looks high. Is it within limits and what should I check first?"))

Gerçek ekipman olmadan denemek için retained test değerleri yayınlayın:

mosquitto_pub -h broker.connectmq.com -p 1883 -i seed-8b2e61 -r -q 1 -t "connectmq/demo/8b2e61/plant1/line3/c3/discharge_temp" -m "104.5"
mosquitto_pub -h broker.connectmq.com -p 1883 -i seed-8b2e61 -r -q 1 -t "connectmq/demo/8b2e61/plant1/line3/c3/pressure" -m "7.9"
mosquitto_pub -h broker.connectmq.com -p 1883 -i seed-8b2e61 -r -q 1 -t "connectmq/demo/8b2e61/plant1/line3/c3/fault" -m "\"none\""

Herkese açık broker'daki retained verilerin kalıcı olacağı garanti edilmez; değerler kaybolursa yeniden yükleyin ve gerçek tesis verilerini asla herkese açık bir broker'a koymayın.

Endüstriyel RAG'e Özgü Tuzaklar

1. Doğru makine için yanlış kılavuz. Farklı firmware veya model numaralarına sahip benzer makinelerin limitleri farklıdır. Her şeyi aramak yerine, yukarıdaki gibi erişimi varlığın belge listesine göre filtreleyin.
2. Birimler ve ölçekleme. Bir topic ham sayımlar veya Fahrenheit yayınlarken kılavuz Celsius kullanıyorsa, model elmalarla armutları karşılaştırır. Birimleri önbellekte normalleştirin ve alan adında belirtin (discharge_temp_c).
3. Bayat değerler. Her zaman zaman damgalarını veya yaşları ekleyin ve modele eski değerleri kullanılamaz kabul etmesini söyleyin.
4. Çıkarma sırasında kaybolan tablolar. PDF dönüşümünden sonra arıza kodu tablolarını doğrulayın; bunları yapılandırılmış JSON olarak saklayıp kodları embedding'ler yerine doğrudan aramayı düşünün.
5. Güvenlik içeriği. Kilitleme ve etiketleme (lockout/tagout) adımları asla özetlenerek atlanmamalıdır. Bunu sistem prompt'unda belirtin ve test edin.
6. Değerlendirme seti yok. Beklenen yanıtlarıyla birlikte 30 ila 50 gerçek teknisyen sorusu toplayın ve parçalama, model veya prompt'ları her değiştirdiğinizde yeniden çalıştırın.

Prototipten Tesise

Canlı değerlerin okuma yolu hafiftir: her asistan örneği için birkaç wildcard aboneliğine sahip tek bir abone. Bunu Unified Namespace'inizi zaten taşıyan bir bulut MQTT broker üzerinde çalıştırmak, asistanın kendi kimlik bilgileriyle TLS üzerinden dashboard'lar ve historian'larla aynı veriyi görmesi anlamına gelir. ConnectMQ bunu, Sparkplug B çözümleme ve bir varlığın hangi topic'leri yayınladığını kontrol etmek için portalda bir ISA-95 cihaz görünümüyle birlikte barındırılan bir MQTT broker olarak sunar. Geliştirme sırasında, yüklenmiş test değerleri için herkese açık MQTT test broker'ı yeterlidir.

Asistanı daha ileri taşımak için değerleri bir MQTT MCP sunucusu aracılığıyla talep üzerine çekmesine izin verin, tekrarlayan soruları zamanlanmış LLM ile sensör verisi özetlerine dönüştürün veya tam bir endüstriyel yapay zeka copilot'unu keşfedin. Yapay zeka ve IoT rehberimiz bu desenleri bağlamına oturtuyor.

ConnectMQ ile Başlayın

RAG asistanınızı, retained message, ISA-95 topic tespiti ve yerleşik Sparkplug B çözümleme sunan ConnectMQ üzerinden canlı ve iyi yapılandırılmış telemetriyle besleyin. Kredi kartı gerektirmeden ücretsiz bir Serverless broker oluşturun ve ilk yanıtınızı gerçek makine verisine dayandırın.

Sık Sorulan Sorular

Endüstriyel veri için RAG ile fine-tuning arasındaki fark nedir?

RAG belgelerinizi ve canlı değerleri soru anında getirir; böylece yanıtlar güncel kalır ve kaynak gösterebilir. Fine-tuning modelin ağırlıklarını değiştirir ve bir makinenin mevcut durumunu bilemez.

Canlı değerleri neden bir veritabanı yerine MQTT'den çekmeli?

Retained MQTT mesajları her topic'in en son değerini anında verir; mevcut durumla ilgili bir sorunun ihtiyacı da tam olarak budur. Geçmişe yönelik sorular yine de bir zaman serisi veritabanına aittir.

Bir vektör veritabanına ihtiyacım var mı?

Birkaç bin parça için hayır; bellek içi bir NumPy indeksi yeterince hızlıdır. Koleksiyonlar büyüdüğünde veya çok sayıda kullanıcı eşzamanlı sorgu yaptığında bir vektör veritabanına geçin.

Asistanın limit uydurmasını nasıl engellerim?

Yalnızca sağlanan pasajlardan ve canlı değerlerden yanıt vermesini söyleyin, kaynak göstermeyi zorunlu kılın, erişimi doğru ekipman belgelerine göre filtreleyin ve sabit bir değerlendirme setiyle test edin.