RAG für Industriedaten: Gerätehandbücher mit Live-MQTT-Telemetrie kombinieren

RAG für Industriedaten beantwortet Fragen wie „Warum meldet der Kompressor an Linie 3 einen Alarm, und was sollte ich prüfen?“, indem es zwei Quellen verbindet, zwischen denen ein Techniker sonst hin- und herwechseln muss: die Gerätedokumentation und die aktuellen Messwerte der Maschine. Dieser Leitfaden zeigt, wie Sie Handbücher und Verfahren indexieren, zum Zeitpunkt der Frage Live-Werte aus MQTT-Topics abrufen und einen Prompt zusammenstellen, der belegte Antworten mit Quellenangaben liefert – samt der Fallstricke, die in der Fertigungshalle zählen.

Warum einfaches RAG in einer Anlage nicht ausreicht

Retrieval-Augmented Generation (RAG) ruft relevante Passagen aus Ihren Dokumenten ab und übergibt sie einem Sprachmodell, damit es aus Ihren Inhalten statt aus Allgemeinwissen antwortet. Für Bürodokumente ist das schon alles. In der Industrie greift es zu kurz, weil die meisten nützlichen Fragen vom aktuellen Zustand einer Maschine abhängen:

• „Ist dieses Vibrationsniveau für diese Pumpe normal?“ braucht sowohl den Grenzwert aus dem Handbuch als auch den aktuellen Wert.
• „Was bedeutet Fehlercode E-214, und ist er noch aktiv?“ braucht die Fehlertabelle und den Live-Alarmzustand.
• „Kann ich den Kaltwassersatz jetzt neu starten?“ braucht das Neustartverfahren und die aktuellen Drücke und Temperaturen.

Ein Modell, das nur Dokumente kennt, zitiert Grenzwerte, ohne den Wert zu kennen. Ein Modell, das nur Telemetrie kennt, sieht Zahlen, ohne zu wissen, was sie bedeuten. Erst die Kombination macht einen Assistenten für Instandhaltungstechniker nützlich.

Architektur von RAG für Industriedaten

Das Asset-Register ist der Baustein, den die meisten Tutorials weglassen – und genau der, der das System funktionieren lässt. Fragt ein Techniker nach „Kompressor C-3“, sagt Ihnen das Register, welches Handbuch gilt (Modellnummern unterscheiden sich zwischen ansonsten ähnlichen Maschinen) und welche Topics seine Messwerte enthalten.

{
  "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"
    }
  }
}

Eine Topic-Struktur, die den ISA-95-Ebenen folgt (Unternehmen, Standort, Bereich, Linie, Zelle), macht diese Zuordnung nahezu mechanisch. Genau das ist das Prinzip eines Unified Namespace (UNS): Die ISA-95-Anlagenhierarchie wird zur Topic-Struktur auf dem Broker, etwa acme/werk-stuttgart/drucklufterzeugung/linie-3/c3/discharge_temp, und das Asset-Register muss nur noch auf diese Pfade verweisen. Veröffentlichen Ihre Maschinen Sparkplug B, kann ConnectMQ es dekodieren und die Metriken in lesbare ISA-95-Topics erneut veröffentlichen, die ein RAG-System ohne Protobuf-Decoder nutzen kann. Unser Artikel zu Sparkplug B und dem Unified Namespace erklärt dieses Topic-Modell.

Schritt 1: Die Dokumente aufbereiten und indexieren

Die Dokumentaufbereitung entscheidet mehr über die Antwortqualität als die Wahl des Modells.

1. Sauberen Text extrahieren. Konvertieren Sie PDFs in Text und prüfen Sie, ob Tabellen erhalten geblieben sind. Fehlercodetabellen sind oft der wertvollste Inhalt und werden am ehesten verstümmelt.
2. Nach Struktur in Chunks zerlegen. Teilen Sie an Überschriften und Verfahrensschritten, nicht nach fester Zeichenzahl, damit ein Verfahren zusammenbleibt. Halten Sie Chunks bei einigen hundert Wörtern.
3. Metadaten anhängen. Speichern Sie Dokument-ID, Titel, Abschnitt, Seite und Gerätemodell mit jedem Chunk.
4. Einbetten und speichern. Berechnen Sie pro Chunk ein Embedding und bewahren Sie es zusammen mit den Metadaten auf.

Das folgende Beispiel nutzt die Embedding-API von Ollama mit einem lokalen Embedding-Modell und hält den Index in NumPy. Installieren Sie mit pip install ollama numpy "paho-mqtt>=2.0" und ollama pull nomic-embed-text.

# 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")          # eine .md- oder .txt-Datei pro Dokument, benannt <doc-id>.md


def chunks_for(path):
    text = path.read_text(encoding="utf-8")
    sections = re.split(r"\n(?=#{1,3} )", text)   # vor Markdown-Überschriften teilen
    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")

Die Präfixe search_document: und search_query: sind die Aufgabenanweisungen, die genau dieses Embedding-Modell erwartet; andere Embedding-Modelle haben eigene Konventionen, prüfen Sie also die Model Card des verwendeten Modells.

Schritt 2: Einen Live-Wert-Cache aus MQTT pflegen

Der Cache abonniert die Asset-Topics und speichert den letzten Wert mit seinem Zeitstempel. Da Geräte retained Werte veröffentlichen, ist der Cache wenige Augenblicke nach dem Start vollständig.

# 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

Das Alter jedes Werts mitzuliefern ist unerlässlich. Ein Messwert, der vor zwei Stunden eintraf, ist nicht „aktuell“, und das Modell sollte das auch sagen.

Schritt 3: Abrufen, kombinieren und antworten

Zum Zeitpunkt der Frage identifizieren Sie das Asset, rufen Passagen nur aus den Dokumenten dieses Assets ab, lesen seine Live-Werte und bauen einen Prompt, in dem jedes Element gekennzeichnet ist.

# 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)   # retained Werte eintreffen lassen
    print(ask("C-3", "Discharge temperature looks high. Is it within limits and what should I check first?"))

Um es ohne echte Anlagen auszuprobieren, veröffentlichen Sie retained Testwerte:

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\""

Retained Daten auf dem öffentlichen Broker bleiben nicht garantiert erhalten; veröffentlichen Sie die Werte also erneut, wenn sie verschwinden, und legen Sie niemals echte Anlagendaten auf einem öffentlichen Broker ab.

Fallstricke speziell bei industriellem RAG

1. Falsches Handbuch für die richtige Maschine. Ähnliche Maschinen mit unterschiedlicher Firmware oder Modellnummer haben unterschiedliche Grenzwerte. Filtern Sie den Abruf wie oben nach der Dokumentliste des Assets, statt alles zu durchsuchen.
2. Einheiten und Skalierung. Veröffentlicht ein Topic Rohzählwerte oder Fahrenheit, während das Handbuch Celsius verwendet, vergleicht das Modell Äpfel mit Birnen. Normalisieren Sie Einheiten im Cache und nennen Sie sie im Feldnamen (discharge_temp_c).
3. Veraltete Werte. Geben Sie immer Zeitstempel oder Alter mit und weisen Sie das Modell an, alte Werte als nicht verfügbar zu behandeln.
4. Bei der Extraktion verlorene Tabellen. Prüfen Sie Fehlercodetabellen nach der PDF-Konvertierung; erwägen Sie, sie als strukturiertes JSON zu speichern und Codes direkt nachzuschlagen statt über Embeddings.
5. Sicherheitsinhalte. Schritte zu Lockout/Tagout dürfen niemals wegzusammengefasst werden. Schreiben Sie das in den System-Prompt und testen Sie es.
6. Kein Evaluierungsset. Sammeln Sie 30 bis 50 echte Technikerfragen mit erwarteten Antworten und lassen Sie sie erneut laufen, sobald Sie Chunking, Modelle oder Prompts ändern.

Vom Prototyp in die Anlage

Der Lesepfad für Live-Werte ist schlank: ein Subscriber pro Assistenteninstanz mit einer Handvoll Wildcard-Abonnements. Läuft er gegen einen Cloud-MQTT-Broker, der bereits Ihren UNS trägt, sieht der Assistent dieselben Daten wie Dashboards und Historians – über TLS mit eigenen Zugangsdaten. ConnectMQ bietet genau das als verwalteten MQTT-Broker, mit Sparkplug-B-Dekodierung und einer ISA-95-Geräteansicht im Portal, in der Sie prüfen können, welche Topics ein Asset veröffentlicht. Während der Entwicklung reicht der öffentliche MQTT-Broker für vorab veröffentlichte Testwerte; echte Anlagendaten gehören dort nicht hin.

Um den Assistenten weiterzuentwickeln, lassen Sie ihn Werte bei Bedarf über einen MCP-Server für MQTT abrufen, machen wiederkehrende Fragen zu geplanten LLM-Zusammenfassungen von Sensordaten oder erkunden einen vollständigen industriellen KI-Copiloten. Unser Leitfaden zu KI und IoT ordnet diese Muster ein.

Starten Sie mit ConnectMQ

Versorgen Sie Ihren RAG-Assistenten mit gut strukturierter Live-Telemetrie von ConnectMQ – mit Retained Messages, ISA-95-Topic-Erkennung und integrierter Sparkplug-B-Dekodierung. Erstellen Sie einen kostenlosen Serverless-Broker, ohne Kreditkarte, und stützen Sie Ihre erste Antwort auf echte Maschinendaten.

Häufig gestellte Fragen

Was ist der Unterschied zwischen RAG und Fine-Tuning für Industriedaten?

RAG ruft Ihre Dokumente und Live-Werte zum Zeitpunkt der Frage ab, sodass Antworten aktuell bleiben und Quellen nennen können. Fine-Tuning verändert die Gewichte des Modells und kann den aktuellen Zustand einer Maschine nicht kennen.

Warum Live-Werte aus MQTT statt aus einer Datenbank abrufen?

Retained MQTT-Messages liefern sofort den letzten Wert jedes Topics – genau das, was eine Frage zum aktuellen Zustand braucht. Historische Fragen gehören weiterhin in eine Zeitreihendatenbank.

Brauche ich eine Vektordatenbank?

Nicht für einige tausend Chunks; ein NumPy-Index im Arbeitsspeicher ist schnell genug. Wechseln Sie zu einer Vektordatenbank, wenn die Sammlungen groß werden oder viele Benutzer gleichzeitig abfragen.

Wie verhindere ich, dass der Assistent Grenzwerte erfindet?

Weisen Sie ihn an, nur aus den bereitgestellten Passagen und Live-Werten zu antworten, verlangen Sie Quellenangaben, filtern Sie den Abruf nach den richtigen Gerätedokumenten und testen Sie mit einem festen Evaluierungsset.