LLM-Zusammenfassungen von Sensordaten: Schichtberichte und Anomalieerklärungen aus MQTT erzeugen

LLM-Zusammenfassungen von Sensordaten verwandeln Tausende Rohmesswerte in wenige Absätze, die ein Bediener bei der Schichtübergabe lesen kann: was normal lief, was abgedriftet ist und was Aufmerksamkeit braucht. In diesem Tutorial sammeln Sie Telemetrie aus MQTT-Topics, berechnen die Statistiken in Python und lassen ein Sprachmodell einen Schichtbericht und verständliche Anomalieerklärungen schreiben – ausschließlich auf Basis dieser berechneten Fakten.

Warum nicht einfach die Rohdaten an das Modell senden?

Es ist verlockend, die Messwerte eines ganzen Tages in einen Prompt zu kopieren und um eine Zusammenfassung zu bitten. In der Praxis scheitert das auf drei Arten:

• Rechnen ist unzuverlässig. Sprachmodelle sind gut in Sprache und schwach darin, Mittelwerte über lange Listen zu berechnen. Sie liefern plausible, aber falsche Zahlen.
• Kontext ist begrenzt und teuer. Hochfrequente Telemetrie übersteigt schnell, was in einen Prompt passt, und jedes Token kostet Zeit.
• Ergebnisse sind nicht prüfbar. Ist eine Zahl im Bericht falsch, lässt sich nicht nachvollziehen, woher sie stammt.

Das zuverlässige Muster teilt die Arbeit auf. Der Code berechnet alles, was eine Zahl ist: Anzahlen, Mittelwerte, Minima, Maxima, Zeit über dem Grenzwert, Datenlücken. Das Modell erhält ein kompaktes JSON-Dokument mit Fakten und formt daraus Fließtext. Jede Zahl im Bericht lässt sich auf das Faktendokument zurückführen.

Architektur

sensors --MQTT--> broker --> collector.py (window buffer, stats) --> facts JSON
                                                              |
                                                              v
                                                   LLM (summary prompt)
                                                              |
                         broker <-- reports/shift (retained) <-+

Der Bericht wird auf einem retained Topic zurück an den Broker veröffentlicht, sodass ein Dashboard, eine Chatbot-Bridge oder ein Druckdienst jederzeit den neuesten Bericht abholen kann.

Schritt 1: Testtelemetrie veröffentlichen

Falls Sie noch keine Geräte haben, veröffentlicht dieser Simulator alle zwei Sekunden Temperatur, Vibration und Druck für zwei Maschinen, mit gelegentlichen Vibrationsspitzen. Ersetzen Sie b62d1f durch Ihre eigene zufällige Zeichenfolge.

# sim_line.py
import json
import random
import time
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/b62d1f/line1"
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sim-line-b62d1f")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    for machine in ("press-1", "press-2"):
        vib = random.gauss(2.0, 0.2)
        if machine == "press-2" and random.random() < 0.05:
            vib += random.uniform(3, 5)  # gelegentliche Spitze
        payload = {
            "temp_c": round(random.gauss(58 if machine == "press-1" else 63, 1.5), 2),
            "vib_mm_s": round(vib, 2),
            "pressure_bar": round(random.gauss(5.0, 0.1), 2),
            "ts": int(time.time() * 1000),
        }
        client.publish(f"{PREFIX}/{machine}/telemetry", json.dumps(payload), qos=0)
    time.sleep(2)

Schritt 2: Fakten sammeln und berechnen

Der Collector abonniert per Wildcard, puffert Messwerte pro Maschine und Metrik und erstellt am Ende jedes Zeitfensters das Faktendokument. Die Grenzwerte stammen aus der Konfiguration, nicht vom Modell.

# collector.py
import json
import statistics
import threading
import time
from collections import defaultdict

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/b62d1f/line1"
THRESHOLDS = {"temp_c": 65.0, "vib_mm_s": 4.5, "pressure_bar": 5.4}
UNITS = {"temp_c": "C", "vib_mm_s": "mm/s", "pressure_bar": "bar"}

buffer = defaultdict(list)       # (machine, metric) -> [(ts, value)]
last_seen = {}
lock = threading.Lock()


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


def on_message(client, userdata, msg):
    machine = msg.topic.split("/")[-2]
    try:
        data = json.loads(msg.payload)
    except json.JSONDecodeError:
        return
    ts = data.get("ts", int(time.time() * 1000))
    with lock:
        last_seen[machine] = ts
        for metric in THRESHOLDS:
            if isinstance(data.get(metric), (int, float)):
                buffer[(machine, metric)].append((ts, float(data[metric])))


def build_facts(window_start, window_end):
    with lock:
        items = {k: list(v) for k, v in buffer.items()}
        buffer.clear()
        seen = dict(last_seen)
    machines = defaultdict(dict)
    for (machine, metric), points in items.items():
        values = [v for _, v in points]
        limit = THRESHOLDS[metric]
        breaches = [(ts, v) for ts, v in points if v > limit]
        machines[machine][metric] = {
            "unit": UNITS[metric],
            "count": len(values),
            "mean": round(statistics.fmean(values), 2),
            "min": round(min(values), 2),
            "max": round(max(values), 2),
            "stdev": round(statistics.pstdev(values), 2),
            "threshold": limit,
            "breaches": len(breaches),
            "firstBreach": time.strftime("%H:%M:%S", time.localtime(breaches[0][0] / 1000)) if breaches else None,
            "peakDuringBreach": round(max(v for _, v in breaches), 2) if breaches else None,
        }
    stale = [m for m, ts in seen.items() if window_end * 1000 - ts > 60_000]
    return {
        "line": "line1",
        "window": {
            "start": time.strftime("%Y-%m-%d %H:%M", time.localtime(window_start)),
            "end": time.strftime("%Y-%m-%d %H:%M", time.localtime(window_end)),
        },
        "machines": machines,
        "staleMachines": stale,
    }

Da Anzahl der Überschreitungen, Zeitpunkt der ersten Überschreitung und Spitzenwert erhalten bleiben, kann das Modell eine Anomalie konkret erklären („Die Vibration an press-2 überschritt 6-mal 4,5 mm/s, erstmals um 14:12, mit einem Spitzenwert von 6,8“), ohne etwas zu erfinden.

Schritt 3: Das LLM die Zusammenfassung schreiben lassen

Der Prompt ist so wichtig wie die Daten. Sagen Sie dem Modell, wofür der Bericht gedacht ist, was es darf und was nicht, und welche Form die Ausgabe haben soll. Dieses Beispiel nutzt ein lokales Modell über Ollama, aber jede Chat-API funktioniert mit demselben Prompt.

# summarize.py
import json
import ollama

MODEL = "llama3.1:8b"

INSTRUCTIONS = """You write shift handover reports for production operators.
Use ONLY the numbers in the FACTS JSON. Do not calculate new numbers and do not guess causes as facts.
Structure:
1. One-sentence overall status.
2. 'Needs attention' bullet list: every metric with breaches > 0 and every stale machine. Quote count, first breach time and peak.
3. 'Normal operation' bullet list: one short line per machine.
4. 'Possible explanations' (max 3 bullets): plausible causes worded as suggestions to check, e.g. 'check ...'.
Keep it under 200 words. Plain text, no tables."""


def write_report(facts: dict) -> str:
    resp = ollama.chat(
        model=MODEL,
        messages=[
            {"role": "system", "content": INSTRUCTIONS},
            {"role": "user", "content": "FACTS:\n" + json.dumps(facts, indent=1)},
        ],
        options={"temperature": 0.2},
    )
    return resp.message.content

Schritt 4: Die Fensterschleife ausführen und den Bericht veröffentlichen

# main.py
import json
import time

import paho.mqtt.client as mqtt

import collector
from summarize import write_report

WINDOW_SECONDS = 600   # 10 Minuten zum Testen; für eine echte Schicht 8 Stunden
REPORT_TOPIC = f"{collector.PREFIX}/reports/shift"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="summarizer-b62d1f")
client.on_connect = collector.on_connect
client.on_message = collector.on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    start = time.time()
    time.sleep(WINDOW_SECONDS)
    facts = collector.build_facts(start, time.time())
    if not facts["machines"]:
        continue
    report = write_report(facts)
    message = {"facts": facts, "report": report, "generatedAt": int(time.time() * 1000)}
    client.publish(REPORT_TOPIC, json.dumps(message), qos=1, retain=True)
    print(report)

Der LLM-Aufruf ist im Vergleich zum MQTT-Verkehr langsam, doch da paho seine Netzwerkschleife mit loop_start() in einem Hintergrund-Thread ausführt, fließen weiterhin Nachrichten in den Puffer, während der Bericht erstellt wird. Weil die Fakten zusammen mit dem Bericht veröffentlicht werden, kann jeder Leser jede Zahl überprüfen.

Den neuesten Bericht können Sie von überall abrufen:

mosquitto_sub -h broker.connectmq.com -p 1883 -i reader-b62d1f -t "connectmq/demo/b62d1f/line1/reports/shift" -C 1

Zeitfenster, Grenzwerte und Zielgruppen wählen

Dieselben Fakten können sehr unterschiedliche Berichte ergeben – je nachdem, wie Sie das Zeitfenster festlegen und wer das Ergebnis liest.

Grenzwerte sollten aus derselben Quelle stammen wie Ihre Alarme, etwa aus Gerätespezifikationen oder mit der Technik abgestimmten Prozessgrenzen, nicht aus dem Allgemeinwissen des Modells. Ändert sich ein Grenzwert, spiegelt der nächste Bericht das automatisch wider, weil er Teil des Faktendokuments ist.

Für ereignisgesteuerte Berichte lösen Sie die Zusammenfassung aus, wenn ein Alarm auftritt, und übergeben nur die Fakten der betroffenen Maschine für die letzten 30 bis 60 Minuten sowie dasselbe Zeitfenster vom Vortag als Basislinie. Das Modell schreibt dann eine gezielte Erklärung statt eines vollständigen Schichtberichts. Ein kleines Faktendokument hält die Generierung zudem so schnell, dass die Erklärung eintrifft, solange das Ereignis noch relevant ist.

Bessere LLM-Zusammenfassungen von Sensordaten schreiben

Einige praktische Erfahrungen, die die Berichte spürbar verbessern:

1. Namen und Kontext mitgeben. „press-2 (Hydraulikpresse, Linie 1)“ führt zu besseren Erklärungen als eine nackte ID. Hinterlegen Sie Beschreibungen in der Konfiguration und fügen Sie sie den Fakten hinzu.
2. Probleme vorab priorisieren. Sortieren Sie die Maschinen vor dem Senden im Code nach Schweregrad. Das Modell behält Ihre Reihenfolge meist bei, sodass der wichtigste Punkt zuerst erscheint.
3. Mitliefern, wie „normal“ aussieht. Ergänzen Sie die Mittelwerte der vorherigen Schicht, dann kann das Modell „Die Temperatur liegt 3 Grad über der letzten Schicht“ mit von Ihnen berechneten Zahlen formulieren.
4. Fakten von Hypothesen trennen. Fordern Sie mögliche Erklärungen als Prüfvorschläge an. Ein Modell kann nicht wissen, dass ein Lager ausfällt; es kann aber vorschlagen, es zu prüfen.
5. Die Ausgabe validieren. Eine einfache Prüfung, ob jede Zahl im Bericht auch im Fakten-JSON vorkommt, fängt die meisten erfundenen Werte ab.
6. Einen Fallback bereithalten. Ist das Modell nicht verfügbar, veröffentlichen Sie einen vorlagenbasierten Bericht aus denselben Fakten, damit die Schicht trotzdem eine Zusammenfassung erhält.

Um die Anomalien überhaupt zu erkennen, kombinieren Sie dies mit einem richtigen Detektor aus unserem Leitfaden zur IoT-Anomalieerkennung und lassen die Zusammenfassung die Ergebnisse des Detektors erklären, statt die Erkennung selbst vorzunehmen.

Wo die Daten liegen

Der Collector funktioniert mit jeder MQTT-Quelle. In einer Fabrik speisen Sie ihn vielleicht aus SPS-Gateways, die Sparkplug B veröffentlichen; ConnectMQ kann Sparkplug B dekodieren und in lesbare ISA-95-Topics erneut veröffentlichen, die sich viel leichter aggregieren und beschreiben lassen. Betreibt das Werk einen Unified Namespace (UNS), abonniert der Collector einfach einen Zweig wie acme/werk-stuttgart/montage/linie-1/#; Werk, Bereich, Linie und Zelle stehen dann bereits im Topic und landen als Kontext im Faktendokument, etwa für einen Schichtbericht mit OEE-Kennzahlen pro Linie. Für Tests reicht der öffentliche MQTT-Broker, der nicht für Produktionsdaten gedacht ist; Produktionstelemetrie gehört jedoch auf einen privaten MQTT-Broker in der Cloud mit TLS und Zugangsdaten pro Gerät. Wenn Sie das Modell vor Ort behalten möchten, betreiben Sie es auf einem Gateway, wie in Ollama mit MQTT gezeigt, und für Antworten, die Live-Werte mit Gerätehandbüchern verbinden, lesen Sie RAG für Industriedaten. Das Gesamtbild behandelt unser Leitfaden zu KI und IoT.

Starten Sie mit ConnectMQ

Sammeln Sie Ihre Telemetrie auf ConnectMQ, einem verwalteten MQTT-Broker mit Live-Topic-Explorer, Dashboard und Grenzwert-Alarmregeln für genau die Messwerte, die Ihre Berichte zusammenfassen. Eröffnen Sie einen kostenlosen Serverless-Broker, ohne Kreditkarte, und veröffentlichen Sie noch heute Ihren ersten KI-geschriebenen Schichtbericht.

Häufig gestellte Fragen

Kann das LLM die Statistiken selbst berechnen?

Es kann es versuchen, aber bei vielen Werten sind die Ergebnisse unzuverlässig. Berechnen Sie die Statistiken im Code und übergeben Sie sie dem Modell, damit jede Zahl im Bericht exakt und nachvollziehbar ist.

Wie oft sollten Zusammenfassungen erzeugt werden?

Orientieren Sie sich am betrieblichen Rhythmus: pro Schicht für die Übergabe, stündlich für Schichtleiter oder auf Abruf, wenn ein Alarm auftritt. Das Fenster des Collectors folgt einfach diesem Zeitplan.

Welches Modell sollte ich verwenden?

Für kurze, strukturierte Zusammenfassungen aus einem Faktendokument genügt ein kleines lokales Modell. Größere Modelle schreiben flüssigere Erklärungen; der Fakten-zuerst-Ansatz funktioniert mit beiden.

Wie verhindere ich, dass das Modell Ursachen erfindet?

Weisen Sie es an, Ursachen als Prüfpunkte zu formulieren, nehmen Sie nur Fakten in den Prompt auf und prüfen Sie, ob die Zahlen in der Ausgabe mit dem Fakten-JSON übereinstimmen.