Gerätesteuerung per natürlicher Sprache: Einfache Anfragen in validierte MQTT-Befehle umwandeln

Mit Gerätesteuerung per natürlicher Sprache können Menschen einfach sagen, was sie wollen – etwa „Mach das Büro wärmer“ oder „Schalte im Wohnzimmer alles aus“ –, statt sich durch Dashboards zu klicken. In diesem Tutorial bauen Sie einen Python-Dienst, in dem ein LLM Anfragen in strukturierte Funktionsaufrufe übersetzt, eine Validierungsschicht jeden Aufruf gegen ein Geräteregister prüft und nur gültige Befehle über MQTT an Geräte veröffentlicht werden, die sie bestätigen.

So funktioniert die Pipeline

Der Kerngedanke: Das Sprachmodell schreibt nie MQTT-Nachrichten. Es wählt lediglich aus einem Menü von Funktionen, die Sie definieren, und Ihr Code macht aus dieser Auswahl nach einer Prüfung Befehle.

1. Benutzeranfrage. Freitext aus einem Chatfenster, dem Transkript eines Sprachassistenten oder einem Messaging-Bot.
2. Kontext. Der Dienst gibt dem Modell eine kompakte Liste der Geräte mit ihren Anzeigenamen und Fähigkeiten.
3. Function Calling. Das Modell gibt null oder mehr Tool-Aufrufe zurück, zum Beispiel set_light(device="kitchen-ceiling", on=true, brightness=30).
4. Validierung. Der Code prüft, ob das Gerät existiert, die Fähigkeit unterstützt und die Werte im zulässigen Bereich liegen.
5. Veröffentlichen. Gültige Befehle gehen mit einer eindeutigen Befehls-ID über MQTT an .../<device>/set.
6. Bestätigen. Das Gerät übernimmt die Änderung und veröffentlicht seinen neuen Zustand, den der Dienst an den Benutzer zurückmeldet.

Voraussetzungen

• Python 3.10 oder neuer.
• Ollama läuft lokal mit einem Modell, das Tool Calling unterstützt, etwa qwen2.5:7b oder llama3.1:8b. Jeder Anbieter mit Function Calling funktioniert genauso; nur der API-Aufruf ändert sich.
• Ein MQTT-Broker. Für die Demo verwenden wir broker.connectmq.com.

ollama pull qwen2.5:7b
pip install ollama "paho-mqtt>=2.0"

Schritt 1: Das Geräteregister definieren

Das Register ist die einzige verbindliche Quelle dafür, was existiert und was es kann. Das Modell sieht eine Zusammenfassung davon; der Validator nutzt es vollständig.

# registry.py
DEVICES = {
    "kitchen-ceiling": {"name": "kitchen ceiling light", "room": "kitchen", "type": "light", "dimmable": True},
    "kitchen-under":   {"name": "kitchen under-cabinet light", "room": "kitchen", "type": "light", "dimmable": False},
    "living-lamp":     {"name": "living room floor lamp", "room": "living room", "type": "light", "dimmable": True},
    "living-blinds":   {"name": "living room blinds", "room": "living room", "type": "blinds"},
    "office-thermo":   {"name": "office thermostat", "room": "office", "type": "thermostat", "min_c": 16, "max_c": 26},
}


def describe_devices():
    lines = []
    for dev_id, d in DEVICES.items():
        extra = ""
        if d["type"] == "light":
            extra = " (dimmable)" if d["dimmable"] else " (on/off only)"
        if d["type"] == "thermostat":
            extra = f" (range {d['min_c']}-{d['max_c']} C)"
        lines.append(f"- {dev_id}: {d['name']}, room={d['room']}, type={d['type']}{extra}")
    return "\n".join(lines)

Schritt 2: Die Funktionen definieren

Jede Funktion entspricht einer Fähigkeit. Die Geräte-IDs im Schema aufzuzählen hilft dem Modell, gültige Ziele zu wählen.

# tools.py
from registry import DEVICES

LIGHTS = [k for k, d in DEVICES.items() if d["type"] == "light"]
BLINDS = [k for k, d in DEVICES.items() if d["type"] == "blinds"]
THERMOS = [k for k, d in DEVICES.items() if d["type"] == "thermostat"]

TOOLS = [
    {"type": "function", "function": {
        "name": "set_light",
        "description": "Turn a light on or off and optionally set brightness in percent (dimmable lights only).",
        "parameters": {"type": "object", "properties": {
            "device": {"type": "string", "enum": LIGHTS},
            "on": {"type": "boolean"},
            "brightness": {"type": "integer", "minimum": 1, "maximum": 100}},
            "required": ["device", "on"]}}},
    {"type": "function", "function": {
        "name": "set_blinds",
        "description": "Set blinds position in percent, 0 is closed and 100 is fully open.",
        "parameters": {"type": "object", "properties": {
            "device": {"type": "string", "enum": BLINDS},
            "position": {"type": "integer", "minimum": 0, "maximum": 100}},
            "required": ["device", "position"]}}},
    {"type": "function", "function": {
        "name": "set_thermostat",
        "description": "Set the target temperature in degrees Celsius.",
        "parameters": {"type": "object", "properties": {
            "device": {"type": "string", "enum": THERMOS},
            "target_c": {"type": "number"}},
            "required": ["device", "target_c"]}}},
]

Schritt 3: Jeden Aufruf validieren

Schemas lenken das Modell, binden es aber nicht. Gerade kleine lokale Modelle geben Zahlen manchmal als Strings zurück, erfinden Geräte-IDs oder ignorieren Wertebereiche. Der Validator normalisiert, was sich gefahrlos normalisieren lässt, und lehnt alles andere mit einer Begründung ab.

# validate.py
from registry import DEVICES


class Invalid(Exception):
    pass


def as_int(value, field):
    try:
        return int(round(float(value)))
    except (TypeError, ValueError):
        raise Invalid(f"{field} must be a number")


def validate(name, args):
    dev_id = args.get("device")
    dev = DEVICES.get(dev_id)
    if dev is None:
        raise Invalid(f"unknown device {dev_id!r}")
    if name == "set_light":
        if dev["type"] != "light":
            raise Invalid(f"{dev_id} is not a light")
        on = args.get("on")
        if isinstance(on, str):
            on = on.lower() in ("true", "on", "1")
        cmd = {"on": bool(on)}
        if "brightness" in args and args["brightness"] is not None:
            if not dev["dimmable"]:
                raise Invalid(f"{dev['name']} is not dimmable")
            b = as_int(args["brightness"], "brightness")
            if not 1 <= b <= 100:
                raise Invalid("brightness must be 1-100")
            cmd["brightness"] = b
        return dev_id, cmd
    if name == "set_blinds":
        if dev["type"] != "blinds":
            raise Invalid(f"{dev_id} is not blinds")
        p = as_int(args.get("position"), "position")
        if not 0 <= p <= 100:
            raise Invalid("position must be 0-100")
        return dev_id, {"position": p}
    if name == "set_thermostat":
        if dev["type"] != "thermostat":
            raise Invalid(f"{dev_id} is not a thermostat")
        try:
            t = round(float(args.get("target_c")), 1)
        except (TypeError, ValueError):
            raise Invalid("target_c must be a number")
        if not dev["min_c"] <= t <= dev["max_c"]:
            raise Invalid(f"target must be {dev['min_c']}-{dev['max_c']} C")
        return dev_id, {"target_c": t}
    raise Invalid(f"unknown function {name}")

Schritt 4: Der Dienst für die Gerätesteuerung per natürlicher Sprache

Jetzt verbinden Sie alles. Der Dienst liest eine Anfrage, fordert beim Modell Tool-Aufrufe an, validiert sie und veröffentlicht Befehle.

# nl_control.py
import json
import time
import uuid

import ollama
import paho.mqtt.client as mqtt

from registry import describe_devices
from tools import TOOLS
from validate import Invalid, validate

MODEL = "qwen2.5:7b"
PREFIX = "connectmq/demo/e3b90c/home"   # eigene Zufalls-ID verwenden

SYSTEM = (
    "You control smart home devices by calling functions. "
    "Only use device IDs from this list:\n" + describe_devices() + "\n"
    "If a request is ambiguous or no device matches, ask a short clarifying question instead of calling a function. "
    "Never guess temperatures outside the allowed range."
)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="nl-control-e3b90c")
client.on_message = lambda c, u, m: print(f"  state {m.topic.split('/')[-2]}: {m.payload.decode()}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.subscribe(f"{PREFIX}/+/state", qos=1)
client.loop_start()


def handle(request):
    resp = ollama.chat(
        model=MODEL,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": request}],
        tools=TOOLS,
        options={"temperature": 0},
    )
    calls = resp.message.tool_calls or []
    if not calls:
        print("assistant:", resp.message.content)
        return
    for call in calls:
        name, args = call.function.name, dict(call.function.arguments)
        try:
            dev_id, cmd = validate(name, args)
        except Invalid as err:
            print(f"rejected {name}({args}): {err}")
            continue
        cmd.update({"id": str(uuid.uuid4()), "ts": int(time.time() * 1000), "source": "nl"})
        client.publish(f"{PREFIX}/{dev_id}/set", json.dumps(cmd), qos=1)
        print(f"sent {dev_id}: {cmd}")


if __name__ == "__main__":
    while True:
        text = input("\nyou: ").strip()
        if text:
            handle(text)

Eine Temperatur von 0 macht die Tool-Auswahl reproduzierbarer, was Sie bei Steuerungsaufgaben wollen.

Schritt 5: Ein simuliertes Gerät hinzufügen und testen

Ein winziger Simulator abonniert alle set-Topics, übernimmt Befehle und veröffentlicht retained Zustände:

# sim_home.py
import json
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/e3b90c/home"
state = {}


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


def on_message(client, userdata, msg):
    dev = msg.topic.split("/")[-2]
    cmd = json.loads(msg.payload)
    state.setdefault(dev, {}).update({k: v for k, v in cmd.items() if k not in ("id", "ts", "source")})
    client.publish(f"{PREFIX}/{dev}/state", json.dumps({**state[dev], "lastCommand": cmd["id"]}), qos=1, retain=True)


c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sim-home-e3b90c")
c.on_connect = on_connect
c.on_message = on_message
c.connect("broker.connectmq.com", 1883, keepalive=60)
c.loop_forever()

Starten Sie den Simulator, dann den Dienst, und probieren Sie diese Anfragen aus:

Spannend sind die Fehlerfälle. Jeder wird entweder durch eine Rückfrage des Modells oder durch den Validator abgefangen, und nichts Ungültiges erreicht den Broker.

Warum das Modell nie MQTT-Payloads schreibt

Es ist verlockend, die Funktionsschicht wegzulassen und das Modell direkt ein Topic und einen JSON-Payload ausgeben zu lassen. Dieser Ansatz scheitert auf subtile Weise. Das Modell erfindet womöglich ein Topic, das richtig aussieht, aber nicht existiert, verwechselt Schlüsselnamen zwischen Gerätetypen oder erzeugt gültiges JSON mit falschen Einheiten. Jeder dieser Fehler erreicht den Broker, und manche Geräte ignorieren fehlerhafte Befehle stillschweigend – der Benutzer glaubt dann, die Änderung sei erfolgt.

Mit Function Calling wählt das Modell aus wenigen benannten Operationen, deren Parameter Sie definiert haben. Ihr Code verantwortet die Zuordnung von Operation zu Topic und Payload-Format, sodass eine geänderte Gerätefirmware oder Topic-Struktur nie eine Prompt-Änderung erfordert. Der Validator wird damit zur einzigen Stelle, an der Befehle entstehen, und lässt sich mit gewöhnlichen Unit-Tests ganz ohne Modell testen.

Folgeanfragen verarbeiten

Echte Gespräche bestehen selten aus einer einzigen Anfrage. Menschen sagen „etwas heller“ oder „das andere auch“. Um das zu unterstützen, behalten Sie die letzten Benutzer- und Assistentennachrichten in der messages-Liste und nehmen den aktuellen Zustand kürzlich angesprochener Geräte – gelesen aus den retained Zustands-Topics – in den System-Prompt auf. Das Modell kann „etwas heller“ dann der gerade geänderten Leuchte zuordnen und aus dem aktuellen Wert eine neue Helligkeit berechnen, während der Validator das Ergebnis weiterhin prüft. Begrenzen Sie den Verlauf auf wenige Runden, damit veralteter Kontext nicht in neue Anfragen einsickert.

Produktionsreif machen

Wenn Sie von der Demo zu einem echten Zuhause, Gebäude oder Werk übergehen, beachten Sie diese Punkte:

1. Private Zugangsdaten und TLS verwenden. Der öffentliche MQTT-Broker ist für jeden lesbar und nur für Tests und Prototypen gedacht. Für echte Geräte verbinden Sie sich über Port 8883 mit Zugangsdaten, die im ConnectMQ-Portal erzeugt wurden.
2. Destruktive oder ungewöhnliche Aktionen bestätigen lassen. Alle Lichter auszuschalten ist unproblematisch; ein Tor zu öffnen oder eine Alarmanlage zu deaktivieren sollte eine ausdrückliche Bestätigung erfordern.
3. Auf den Zustand warten, nicht nur veröffentlichen. Melden Sie Erfolg erst, wenn der Gerätezustand die Befehls-ID bestätigt.
4. Das Register generieren, nicht von Hand pflegen. Bauen Sie es aus retained Ankündigungs-Topics der Geräte auf, damit neue Geräte automatisch erscheinen. Im Werk ist ein Unified Namespace (UNS) nach ISA-95 die natürliche Quelle dafür, weil Topics wie acme/werk-muenchen/montage/linie-1/zelle-3/... bereits Standort, Linie und Zelle jedes Geräts beschreiben.
5. Alles protokollieren. Speichern Sie Anfrage, Tool-Aufrufe, Validierungsergebnisse und Bestätigungen für die Fehlersuche.

Dieselbe Validierungsdisziplin skaliert bis zu Industrieanlagen; unsere Leitfäden zu LLM Function Calling für IoT und LLM-Agenten für IoT-Geräte erweitern sie um Schema-Bibliotheken und menschliche Freigabe. Wenn Sie den gesamten Stack offline betreiben möchten, lesen Sie, wie Sie Ollama mit MQTT auf einem Gateway betreiben. Wie sich Sprachschnittstellen in andere KI-Techniken einordnen, erklärt unser Leitfaden zu KI und IoT.

Betreiben Sie den Steuerkanal auf einem verwalteten MQTT-Broker, erreichen Ihre Geräte, der Sprachdienst und die Dashboards einen einzigen sicheren Endpunkt über TCP, TLS oder WebSockets, ohne dass Sie zu Hause oder im Werk Ports öffnen müssen.

Starten Sie mit ConnectMQ

ConnectMQ bietet Ihnen einen verwalteten MQTT-Broker mit einem MQTT-Client im Browser, sodass Sie Befehle und Zustandsänderungen live beobachten können, während Sie Prompts testen. Erstellen Sie Ihren kostenlosen Serverless-Broker, ohne Kreditkarte, und lassen Sie Menschen Geräte in ihren eigenen Worten steuern.

Häufig gestellte Fragen

Brauche ich für die Gerätesteuerung per natürlicher Sprache ein großes Cloud-Modell?

Nein. Kleine lokale Modelle mit Tool-Calling-Unterstützung bewältigen einfache Gerätebefehle gut, wenn die Geräteliste kurz und die Funktionen klar sind. Die Validierung im Code deckt die Fälle ab, in denen sie danebenliegen.

Wie gehe ich mit mehrdeutigen Anfragen wie „Mach das Licht an“ um?

Weisen Sie das Modell im System-Prompt an, eine Rückfrage zu stellen, wenn mehrere Geräte passen. Ihr Dienst zeigt dem Benutzer die Frage und sendet die Antwort in der nächsten Runde zurück.

Kann ich Sprache statt Text verwenden?

Ja. Führen Sie zuerst eine Spracherkennung aus und speisen Sie das Transkript in dieselbe Pipeline ein. Behalten Sie Bestätigungen für riskante Aktionen bei, denn Transkriptionsfehler sind eine weitere Fehlerquelle.

Warum Befehle an ein set-Topic senden und ein separates state-Topic lesen?

Die Trennung von Befehlen und Zustand bedeutet, dass das state-Topic stets widerspiegelt, was das Gerät tatsächlich getan hat, und der retained Zustand gibt jedem Client sofort den aktuellen Wert.