Ollama-MQTT-Bridge: Ein lokales LLM auf dem Gateway betreiben und KI-Erkenntnisse veröffentlichen

Wenn Sie Ollama und MQTT kombinieren, können Sie ein Sprachmodell direkt neben Ihren Maschinen betreiben und es mit Live-Sensordaten versorgen, ohne Rohmesswerte an einen externen KI-Dienst zu senden. In diesem Tutorial installieren Sie Ollama auf einem Linux-Gateway und bauen eine Python-Bridge, die Telemetrie-Topics abonniert, ein lokales Modell nach strukturierten Erkenntnissen fragt und die Ergebnisse für Dashboards und Alarmierung an den Broker zurückveröffentlicht.

Warum das LLM auf dem Gateway betreiben?

Ein lokales Modell ist nicht immer die richtige Wahl, doch auf einem Edge-Gateway löst es echte Probleme:

• Die Daten bleiben vor Ort. Messwerte werden lokal verarbeitet; nur die kurze Erkenntnis verlässt das Gateway – sofern Sie sie überhaupt nach oben veröffentlichen. Für viele Mittelständler im DACH-Raum, die Maschinendaten als Betriebsgeheimnis behandeln, ist das ein zentrales Argument.
• Keine Abhängigkeit von KI-Diensten im Internet. Die Analyse funktioniert auch bei Ausfällen weiter und hängt nicht von externen API-Kontingenten ab.
• Planbare Kosten. Ist die Hardware erst einmal vorhanden, fallen keine Kosten pro Anfrage an.

Die Kehrseite sind kleinere Modelle, langsamere Generierung auf bescheidener Hardware und der Aufwand, Modelle aktuell zu halten. Für Erklärungen, Klassifizierungen und kurze Zusammenfassungen numerischer Daten reichen kleine Modelle meist aus. Unser Vergleich Edge AI vs. Cloud AI behandelt diese Entscheidung ausführlicher.

Architektur

sensors --MQTT--> broker --> ollama_bridge.py --HTTP localhost:11434--> Ollama
                    ^                |
                    |                v
                    +---- .../insights (JSON) <--+

Die Bridge ist ein normaler MQTT-Client. Sie lässt die Netzwerkschleife in einem Hintergrund-Thread laufen, legt interessante Nachrichten in eine Warteschlange und verarbeitet sie nacheinander mit dem Modell. Diese Trennung ist die wichtigste Designentscheidung in diesem Tutorial: Ein LLM-Aufruf kann mehrere Sekunden dauern, und wird der MQTT-Callback so lange blockiert, kommt es zu verpassten Keepalives und Verbindungsabbrüchen.

Schritt 1: Ollama installieren und ein Modell laden

Auf einem 64-Bit-Linux-Gateway:

curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.2:3b
ollama run llama3.2:3b "Reply with the word ready."

Das Installationsprogramm registriert einen systemd-Dienst, sodass die Ollama-API nach einem Neustart auf http://localhost:11434 lauscht. Lassen Sie sie an localhost gebunden; die Bridge läuft auf demselben Rechner, und es gibt keinen Grund, die Modell-API ins Netzwerk zu öffnen.

Installieren Sie die Python-Abhängigkeiten in einer virtuellen Umgebung:

python3 -m venv ~/ollama-mqtt/.venv
source ~/ollama-mqtt/.venv/bin/activate
pip install ollama "paho-mqtt>=2.0"

Schritt 2: Festlegen, wie eine Erkenntnis aussieht

Freitext ist in nachgelagerten Systemen schwer zu verwenden. Ollama unterstützt strukturierte Ausgaben: Sie übergeben im Parameter format ein JSON-Schema, und die Antwort des Modells wird auf diese Form beschränkt. Dashboards und Alarmregeln können dann Felder wie severity direkt auslesen.

{
  "type": "object",
  "properties": {
    "severity": { "type": "string", "enum": ["normal", "watch", "warning", "critical"] },
    "summary": { "type": "string" },
    "likely_causes": { "type": "array", "items": { "type": "string" }, "maxItems": 3 },
    "recommended_check": { "type": "string" }
  },
  "required": ["severity", "summary", "recommended_check"]
}

Schritt 3: Die Ollama-MQTT-Bridge bauen

Die Bridge abonniert Telemetrie, hält einen kurzen Verlauf pro Gerät und stößt eine Analyse an, wenn ein Messwert eine einfache Regel verletzt oder das Analyseintervall des Geräts abgelaufen ist. Ersetzen Sie 4c7a2e durch Ihre eigene zufällige Zeichenfolge.

# ollama_bridge.py
import json
import logging
import queue
import threading
import time
from collections import defaultdict, deque

import ollama
import paho.mqtt.client as mqtt

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

BROKER, PORT = "broker.connectmq.com", 1883
PREFIX = "connectmq/demo/4c7a2e/site1"
MODEL = "llama3.2:3b"
HISTORY = 30                 # gespeicherte Messwerte pro Gerät
MIN_INTERVAL = 120           # Sekunden zwischen Analysen pro Gerät
RULES = {"temp_c": 70.0, "vib_mm_s": 5.0}

SCHEMA = {
    "type": "object",
    "properties": {
        "severity": {"type": "string", "enum": ["normal", "watch", "warning", "critical"]},
        "summary": {"type": "string"},
        "likely_causes": {"type": "array", "items": {"type": "string"}, "maxItems": 3},
        "recommended_check": {"type": "string"},
    },
    "required": ["severity", "summary", "recommended_check"],
}

SYSTEM = (
    "You analyze industrial sensor readings for maintenance technicians. "
    "Base your answer only on the readings provided. "
    "Phrase causes as possibilities, never as certainties. Keep the summary under 40 words."
)

history = defaultdict(lambda: deque(maxlen=HISTORY))
last_analysis = defaultdict(float)
work = queue.Queue(maxsize=50)


def on_connect(client, userdata, flags, reason_code, properties):
    log.info("connected: %s", reason_code)
    client.subscribe(f"{PREFIX}/+/telemetry", qos=1)


def on_message(client, userdata, msg):
    device = msg.topic.split("/")[-2]
    try:
        reading = json.loads(msg.payload)
    except json.JSONDecodeError:
        return
    history[device].append(reading)
    breached = any(isinstance(reading.get(k), (int, float)) and reading[k] > v for k, v in RULES.items())
    due = time.time() - last_analysis[device] > MIN_INTERVAL
    if (breached or due) and len(history[device]) >= 5:
        last_analysis[device] = time.time()
        try:
            work.put_nowait((device, list(history[device]), breached))
        except queue.Full:
            log.warning("queue full, skipping %s", device)


def analyze(device, readings, breached):
    prompt = (
        f"Device: {device}\nRule thresholds: {json.dumps(RULES)}\n"
        f"Triggered by threshold breach: {breached}\n"
        f"Last {len(readings)} readings, oldest first:\n{json.dumps(readings)}"
    )
    resp = ollama.chat(
        model=MODEL,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": prompt}],
        format=SCHEMA,
        options={"temperature": 0},
    )
    return json.loads(resp.message.content)


def worker(client):
    while True:
        device, readings, breached = work.get()
        started = time.time()
        try:
            insight = analyze(device, readings, breached)
        except Exception as exc:
            log.error("analysis failed for %s: %s", device, exc)
            continue
        insight.update({
            "device": device,
            "model": MODEL,
            "trigger": "threshold" if breached else "interval",
            "durationS": round(time.time() - started, 1),
            "ts": int(time.time() * 1000),
        })
        client.publish(f"{PREFIX}/{device}/insights", json.dumps(insight), qos=1, retain=True)
        log.info("%s -> %s: %s", device, insight["severity"], insight["summary"])


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="ollama-bridge-4c7a2e")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, keepalive=60)
threading.Thread(target=worker, args=(client,), daemon=True).start()
client.loop_forever()

So verhält sie sich:

1. Jede Telemetrienachricht wird in einem begrenzten Verlauf pro Gerät gespeichert.
2. Eine Grenzwertüberschreitung oder ein abgelaufenes Intervall reiht einen Analyseauftrag ein, aber nie mehr als einen pro Gerät und Intervall.
3. Ein einzelner Worker-Thread ruft Ollama auf, damit die Gateway-CPU nicht durch parallele Generierungen überlastet wird.
4. Die Erkenntnis wird retained veröffentlicht, sodass Dashboards die neueste Bewertung zeigen, sobald sie abonnieren.

In einem Werk mit Unified Namespace (UNS) liegen die Erkenntnisse idealerweise direkt neben den Rohdaten der Anlage in der ISA-95-Hierarchie, etwa unter acme/werk-muenchen/pumpenhaus/linie-1/pumpe-2/insights. So finden MES, Instandhaltung und Dashboards Messwerte und KI-Bewertung am selben Ort.

Ein Modell für das Gateway wählen

Die Modellwahl auf einem Gateway ist eine Abwägung zwischen Qualität, Geschwindigkeit und Speicher. Einige praktische Richtlinien:

1. Klein anfangen. Ein 3B-Modell ist ein sinnvoller Standard für strukturierte Erkenntnisse aus kurzen Messreihen. Steigen Sie nur auf, wenn die Antworten für Ihren Anwendungsfall eindeutig zu oberflächlich sind.
2. Quantisierte Varianten bevorzugen. Die Standard-Tags in der Ollama-Bibliothek sind bereits quantisiert – genau das macht kleine Hardware erst praxistauglich. Vermeiden Sie Varianten mit voller Genauigkeit auf reinen CPU-Gateways.
3. Speicher für alles andere freihalten. Auf dem Gateway laufen weiterhin die MQTT-Bridge, eventuell ein lokaler Broker, Protokollkonverter und das Betriebssystem. Beobachten Sie den Speicher mit free -h, während eine Generierung läuft.
4. Mit eigenen Daten testen. Sammeln Sie einige Dutzend echte Telemetriefenster, einschließlich bekannter Vorfälle, und vergleichen Sie, wie die Kandidatenmodelle sie beschreiben. Für den Betrieb zählt Konsistenz mehr als Eloquenz.
5. Die Modellversion festschreiben. Erfassen Sie den Modell-Tag in jeder veröffentlichten Erkenntnis, wie es die Bridge tut, damit Sie nach einem Upgrade erkennen, welches Modell eine bestimmte Empfehlung erzeugt hat.

Schritt 4: Testdaten einspeisen

Veröffentlichen Sie mit dem Mosquitto-CLI einige normale Messwerte und anschließend einen heißen:

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-4c7a2e -q 1 \
    -t "connectmq/demo/4c7a2e/site1/pump-2/telemetry" \
    -m "{\"temp_c\": 6$i.0, \"vib_mm_s\": 2.$i, \"ts\": $(date +%s)000}"
  sleep 1
done
mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-4c7a2e -q 1 \
  -t "connectmq/demo/4c7a2e/site1/pump-2/telemetry" \
  -m "{\"temp_c\": 74.5, \"vib_mm_s\": 5.8, \"ts\": $(date +%s)000}"

Beobachten Sie, wie die Erkenntnisse eintreffen:

mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-4c7a2e -t "connectmq/demo/4c7a2e/site1/+/insights" -v

Sie sollten eine JSON-Erkenntnis mit einem Schweregrad wie warning, einer kurzen Zusammenfassung, die steigende Temperatur und Vibration erwähnt, und einer empfohlenen Prüfung sehen. Der genaue Wortlaut variiert je nach Modell.

Das Ollama-MQTT-Setup optimieren

Die letzte Zeile wird mit wachsendem Datenvolumen wichtig. Übergeben Sie statt Rohmesswerten in Python berechnetes Minimum, Maximum, Mittelwert und Trend; unser Tutorial zu LLM-Zusammenfassungen von Sensordaten zeigt diesen Fakten-zuerst-Ansatz im Detail.

Der Weg in die Produktion

Betreiben Sie die Bridge auf dem Gateway als systemd-Dienst neben Ollama und verbinden Sie sie über TLS mit einem privaten Broker:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)

Der Broker kann ein lokaler auf dem Gateway sein, ein verwalteter MQTT-Broker wie ConnectMQ oder beides, wobei das Gateway nur die Erkenntnisse nach oben veröffentlicht. Ein Cloud-Endpunkt ist nützlich, wenn mehrere Standorte Erkenntnisse mit zentralen Dashboards teilen müssen: Das ConnectMQ-Portal zeigt sie im Live-Topic-Explorer, sobald sie eintreffen. Der öffentliche MQTT-Broker ist ideal für dieses Tutorial, aber seine Topics sind für jeden lesbar – halten Sie echte Anlagendaten davon fern.

Nächste Schritte: Lassen Sie Bediener mit einem MCP-Server für MQTT Fragen zu denselben Daten stellen, oder erlauben Sie dem Modell mit den Mustern aus der Gerätesteuerung per natürlicher Sprache, sicher auf Geräte einzuwirken. Einen Überblick, wo lokale Modelle unter anderen KI-Techniken ihren Platz haben, gibt unser Leitfaden zu KI und IoT.

Starten Sie mit ConnectMQ

Teilen Sie die KI-Erkenntnisse Ihres Gateways standortübergreifend mit ConnectMQ, einem verwalteten MQTT-Broker mit MQTT 5, TLS und WebSockets für Browser-Dashboards. Erstellen Sie einen kostenlosen Serverless-Broker, ohne Kreditkarte, und veröffentlichen Sie in wenigen Minuten Ihre erste lokale LLM-Erkenntnis.

Häufig gestellte Fragen

Kann ein Raspberry Pi Ollama ausführen?

Ja, ein Raspberry Pi 5 mit 8 GB RAM kann kleine Modelle auf seiner CPU ausführen. Antworten dauern Sekunden, analysieren Sie daher Ereignisse und periodische Zeitfenster statt jeder einzelnen Nachricht.

Spricht Ollama direkt MQTT?

Nein. Ollama stellt eine HTTP-API auf localhost bereit. Ein kleines Bridge-Programm verbindet sich als MQTT-Client mit dem Broker und ruft die Ollama-API auf, wie in diesem Tutorial gezeigt.

Wie bekomme ich zuverlässiges JSON vom Modell?

Übergeben Sie im Ollama-Parameter format ein JSON-Schema, um strukturierte Ausgaben zu nutzen, setzen Sie die Temperatur auf 0 und validieren Sie das Ergebnis trotzdem vor dem Veröffentlichen.

Sollte jede MQTT-Nachricht einen LLM-Aufruf auslösen?

Nein. Lösen Sie Analysen bei Grenzwertüberschreitungen oder in Intervallen aus und begrenzen Sie die Aufrufe pro Gerät. Sprachmodelle sind weit langsamer als die Nachrichtenrate auf einem stark frequentierten Topic.