LLM Function Calling im IoT: Strikte JSON-Schemas und validierte MQTT-Befehle

LLM Function Calling im IoT verwandelt das Verständnis eines Modells für eine Anfrage in einen strukturierten Aufruf wie set_valve(valve_id="V-12", position_pct=40), den Ihr Code prüfen und in einen MQTT-Befehl umwandeln kann. Dieses Tutorial konzentriert sich auf den Teil, den die meisten Demos auslassen: strikte JSON-Schemas für Geräteaktionen, unabhängige Validierung jedes Aufrufs, eine Reparaturschleife für ungültige Ausgaben und MQTT-5-Bestätigungen – damit nichts, was das Modell erzeugt, ungeprüft ein Gerät erreicht.

Function Calling in einem Absatz

Beim Function Calling (auch Tool Calling genannt) senden Sie dem Modell eine Liste von Funktionen mit Parameterdefinitionen als JSON-Schema. Statt in Fließtext zu antworten, kann das Modell den Namen einer Funktion und ein JSON-Objekt mit Argumenten zurückgeben. Ihre Anwendung entscheidet, ob sie ausgeführt wird. Das Modell führt selbst nie etwas aus – genau deshalb ist das Muster für Geräte so nützlich: Die Anwendung ist ein Nadelöhr, an dem Sie Regeln durchsetzen können.

Der Haken: Ein Schema im Prompt ist eine Anleitung, keine Durchsetzung. Modelle liefern gelegentlich Argumente mit falschem Typ, Werte außerhalb eines Bereichs, zusätzliche Felder oder Geräte-IDs, die es nicht gibt. Manche Anbieter bieten strikte Modi, die die Ausgabe auf das Schema beschränken, doch semantische Regeln (gehört dieses Ventil zu dieser Linie, ist diese Änderung zu groß) brauchen immer Ihren eigenen Code.

Schritt 1: Strikte Schemas für Geräteaktionen entwerfen

Gute Schemas für physische Aktionen haben einige Merkmale gemeinsam:

Hier ist ein Schema-Modul, das der Befehlsdienst und idealerweise auch die Gerätefirmware oder das Gateway gemeinsam nutzen:

# schemas.py
DEVICE_ACTIONS = {
    "set_valve": {
        "description": "Set a control valve opening in percent.",
        "parameters": {
            "type": "object",
            "properties": {
                "valve_id": {"type": "string", "enum": ["V-11", "V-12", "V-21"]},
                "position_pct": {"type": "integer", "minimum": 0, "maximum": 100},
                "reason": {"type": "string", "minLength": 5, "maxLength": 200},
            },
            "required": ["valve_id", "position_pct", "reason"],
            "additionalProperties": False,
        },
    },
    "set_pump_state": {
        "description": "Start or stop a pump.",
        "parameters": {
            "type": "object",
            "properties": {
                "pump_id": {"type": "string", "enum": ["P-1", "P-2"]},
                "state": {"type": "string", "enum": ["running", "stopped"]},
                "reason": {"type": "string", "minLength": 5, "maxLength": 200},
            },
            "required": ["pump_id", "state", "reason"],
            "additionalProperties": False,
        },
    },
}


def as_tools():
    """Convert to the tool format used by OpenAI-compatible and Ollama chat APIs."""
    return [
        {"type": "function", "function": {"name": name, "description": spec["description"],
                                          "parameters": spec["parameters"]}}
        for name, spec in DEVICE_ACTIONS.items()
    ]

Schritt 2: Doppelt validieren – erst Schema, dann Semantik

Die erste Prüfung ist strukturell und nutzt die Bibliothek jsonschema mit genau dem Schema, das auch das Modell gesehen hat. Die zweite Prüfung ist semantisch: Regeln, die vom Live-Zustand oder von Anlagenrichtlinien abhängen und die kein Schema ausdrücken kann.

# validator.py
from jsonschema import Draft202012Validator
from schemas import DEVICE_ACTIONS

VALIDATORS = {n: Draft202012Validator(s["parameters"]) for n, s in DEVICE_ACTIONS.items()}
MAX_VALVE_STEP = 25   # Anlagenrichtlinie: höchstens 25 Punkte pro Befehl


class ValidationFailed(Exception):
    pass


def check(name, args, live_state):
    validator = VALIDATORS.get(name)
    if validator is None:
        raise ValidationFailed(f"Unknown function '{name}'. Allowed: {sorted(VALIDATORS)}")
    errors = sorted(validator.iter_errors(args), key=lambda e: list(e.path))
    if errors:
        details = "; ".join(f"{'/'.join(map(str, e.path)) or '(root)'}: {e.message}" for e in errors)
        raise ValidationFailed(f"Arguments for {name} are invalid: {details}")
    if name == "set_valve":
        current = live_state.get(args["valve_id"], {}).get("position_pct")
        if current is None:
            raise ValidationFailed(f"Current position of {args['valve_id']} is unknown; refusing to move it.")
        if abs(args["position_pct"] - current) > MAX_VALVE_STEP:
            raise ValidationFailed(
                f"Change from {current} to {args['position_pct']} exceeds the {MAX_VALVE_STEP}-point step limit. "
                f"Propose an intermediate value.")
    if name == "set_pump_state" and args["state"] == "stopped":
        if live_state.get(args["pump_id"], {}).get("duty") == "critical":
            raise ValidationFailed(f"{args['pump_id']} is on critical duty and cannot be stopped by the assistant.")
    return args

Die Fehlermeldungen sind so formuliert, dass das Modell sie lesen kann. Das ist Absicht: Sie speisen die Reparaturschleife im nächsten Schritt.

Schritt 3: Die LLM-Function-Calling-Schleife mit Reparatur

Scheitert ein Aufruf an der Validierung, senden Sie den Fehler zurück und lassen das Modell es erneut versuchen – höchstens ein- oder zweimal. Viele Fehler lassen sich beheben („schlage einen Zwischenwert vor“), und eine begrenzte Schleife vermeidet endlose Wiederholungen. Dieses Beispiel nutzt Ollama mit einem lokalen Modell, das Tools unterstützt; dieselbe Struktur funktioniert mit der Function-Calling-API jedes Anbieters.

# command_service.py
import json
import time
import uuid

import ollama
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

from schemas import as_tools
from validator import ValidationFailed, check

MODEL = "qwen2.5:7b"
PREFIX = "connectmq/demo/d07c5a/plant"     # eigene Zufalls-ID verwenden
ACK_TOPIC = f"{PREFIX}/svc/acks"
MAX_REPAIRS = 2

live_state = {}   # aus retained Zustands-Topics befüllt
acks = {}


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


def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    if msg.topic == ACK_TOPIC:
        acks[data["id"]] = data
    else:
        live_state[msg.topic.split("/")[-2]] = data


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="fc-svc-d07c5a", protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
time.sleep(2)   # retained Zustand eintreffen lassen


def publish_command(name, args):
    device = args.get("valve_id") or args.get("pump_id")
    cmd = {"id": str(uuid.uuid4()), "action": name, "args": args, "ts": int(time.time() * 1000)}
    props = Properties(PacketTypes.PUBLISH)
    props.ResponseTopic = ACK_TOPIC
    props.CorrelationData = cmd["id"].encode()
    props.MessageExpiryInterval = 15   # niemals einen veralteten Befehl ausführen
    client.publish(f"{PREFIX}/{device}/cmd", json.dumps(cmd), qos=1, properties=props)
    deadline = time.time() + 5
    while time.time() < deadline:
        if cmd["id"] in acks:
            return acks.pop(cmd["id"])
        time.sleep(0.1)
    return {"id": cmd["id"], "result": "no_ack"}


def run(request):
    messages = [
        {"role": "system", "content": "You operate plant equipment only through the provided functions. "
                                      "If a request cannot be done safely with them, explain why."},
        {"role": "user", "content": request},
    ]
    for attempt in range(MAX_REPAIRS + 1):
        resp = ollama.chat(model=MODEL, messages=messages, tools=as_tools(), options={"temperature": 0})
        calls = resp.message.tool_calls or []
        if not calls:
            return resp.message.content
        errors, results = [], []
        for call in calls:
            name, args = call.function.name, dict(call.function.arguments)
            try:
                check(name, args, live_state)
            except ValidationFailed as err:
                errors.append(str(err))
                continue
            results.append((name, args, publish_command(name, args)))
        if not errors:
            return results
        messages.append(resp.message)
        messages.append({"role": "user", "content": "Some function calls were rejected:\n- " + "\n- ".join(errors)
                         + "\nCorrect them or explain why the request cannot be done."})
    return {"rejected": errors}


if __name__ == "__main__":
    print(run("Open valve V-12 to 90 percent, the tank is filling too slowly."))

Achten Sie darauf, was mit dieser Beispielanfrage passiert. Steht V-12 aktuell auf 40 Prozent, weist die semantische Prüfung den Sprung um 50 Punkte zurück und fordert das Modell auf, einen Zwischenwert vorzuschlagen. Das Modell versucht es typischerweise erneut mit 65, was die Prüfung besteht und veröffentlicht wird. Gültige Aufrufe innerhalb eines Batches gehen sofort hinaus, und nur die abgelehnten werden zur Reparatur zurückgeschickt.

Das Demo-Präfix .../plant/<gerät> ist bewusst flach gehalten. In einem Werk mit Unified Namespace (UNS) liegen Ventile und Pumpen unter ihrem ISA-95-Pfad, etwa acme/werk-muenchen/versorgung/kuehlwasser/v-12/cmd. Der Befehlsdienst kann dann semantische Regeln wie „gehört dieses Ventil zu dieser Linie?“ direkt aus dem Topic ableiten.

Strikte Modi und Constrained Decoding

Mehrere Modellanbieter und lokale Runtimes können die Generierung so einschränken, dass Tool-Argumente stets dem deklarierten JSON-Schema entsprechen. Wenn verfügbar, aktivieren Sie das: Es beseitigt die meisten Typfehler und fehlenden Felder, bevor sie Ihren Validator erreichen. Die Validierung ersetzt es aus drei Gründen nicht. Erstens deckt Constrained Decoding nur die Teilmenge von JSON Schema ab, die die Runtime unterstützt, und Schlüsselwörter wie numerische Bereiche oder String-Muster werden womöglich ignoriert. Zweitens kann ein perfekt geformter Aufruf für die aktuelle Situation dennoch falsch sein, etwa ein zu großer Ventilschritt oder eine Pumpe im kritischen Betrieb. Drittens wechseln Sie vielleicht später Modell oder Anbieter, und Ihre Sicherheit darf nicht von einer Funktion einer einzelnen Runtime abhängen. Betrachten Sie eingeschränkte Ausgaben als Qualitätsverbesserung und die Validierung im Code als Garantie.

Schritt 4: Auf der Geräteseite erneut validieren

Der Befehlsdienst ist nicht die letzte Verteidigungslinie. Das Gateway oder Gerät, das .../cmd empfängt, sollte dasselbe Schema validieren, prüfen, ob die Nachricht abgelaufen ist, und mit einer Bestätigung auf dem Response Topic antworten:

# valve_gateway.py (simulierte Geräteseite)
import json
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties
from jsonschema import Draft202012Validator
from schemas import DEVICE_ACTIONS

PREFIX = "connectmq/demo/d07c5a/plant"
state = {"V-12": {"position_pct": 40}, "P-1": {"state": "running", "duty": "normal"}}


def on_connect(client, userdata, flags, reason_code, properties):
    for dev, s in state.items():
        client.publish(f"{PREFIX}/{dev}/state", json.dumps(s), qos=1, retain=True)
    client.subscribe(f"{PREFIX}/+/cmd", qos=1)


def on_message(client, userdata, msg):
    cmd = json.loads(msg.payload)
    spec = DEVICE_ACTIONS.get(cmd.get("action"))
    ok = spec is not None and not list(Draft202012Validator(spec["parameters"]).iter_errors(cmd.get("args", {})))
    device = msg.topic.split("/")[-2]
    if ok and cmd["action"] == "set_valve":
        state[device]["position_pct"] = cmd["args"]["position_pct"]
    elif ok and cmd["action"] == "set_pump_state":
        state[device]["state"] = cmd["args"]["state"]
    if ok:
        client.publish(f"{PREFIX}/{device}/state", json.dumps(state[device]), qos=1, retain=True)
    reply = Properties(PacketTypes.PUBLISH)
    if hasattr(msg.properties, "CorrelationData"):
        reply.CorrelationData = msg.properties.CorrelationData
    ack = {"id": cmd.get("id"), "result": "applied" if ok else "rejected_invalid"}
    client.publish(msg.properties.ResponseTopic, json.dumps(ack), qos=1, properties=reply)


c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="valve-gw-d07c5a", protocol=mqtt.MQTTv5)
c.on_connect = on_connect
c.on_message = on_message
c.connect("broker.connectmq.com", 1883, keepalive=60)
c.loop_forever()

Installieren Sie die Abhängigkeiten mit pip install ollama jsonschema "paho-mqtt>=2.0", starten Sie das Gateway und dann den Befehlsdienst.

Funktionsdefinitionen vor dem Deployment testen

Funktionsdefinitionen verdienen dieselbe Testdisziplin wie jede andere Schnittstelle zu physischen Anlagen. Bauen Sie eine Testsuite in zwei Ebenen auf:

1. Unit-Tests für den Validator ohne Modell. Geben Sie handgeschriebene, gültige und ungültige Argumentsätze in check() ein und prüfen Sie das Ergebnis. Berücksichtigen Sie Grenzwerte, falsche Typen, unbekannte IDs, zusätzliche Felder sowie veralteten oder fehlenden Live-Zustand.
2. Verhaltenstests für das Modell. Führen Sie eine Liste realistischer Anfragen mit der erwarteten Funktion und den erwarteten Argumenten, etwa „Schließe V-11 vollständig“ oder „Stoppe die Hauptpumpe“. Lassen Sie sie mit Temperatur 0 gegen das Modell laufen und erfassen Sie, wie oft der erste Versuch gültig ist, wie oft die Reparaturschleife ihn korrigiert und wie oft er in einer Ablehnung endet.

Wiederholen Sie die Verhaltenstests, sobald Sie eine Beschreibung ändern, eine Funktion hinzufügen oder das Modell wechseln. Kleine Formulierungsänderungen in einer Beschreibung können verschieben, welche Funktion ein Modell wählt, und das wollen Sie im Test entdecken, nicht in der Produktionshalle. Protokollierte abgelehnte Aufrufe aus der Produktion liefern Ihnen mit der Zeit neue Testfälle.

Checkliste für LLM Function Calling im IoT-Produktivbetrieb

1. Schemas sind versioniert und werden von Befehlsdienst und Geräten gemeinsam genutzt.
2. additionalProperties: false, Enums für IDs und Grenzen für jede Zahl.
3. Strukturelle Validierung mit einer JSON-Schema-Bibliothek, danach semantische Prüfungen gegen den Live-Zustand.
4. Eine begrenzte Reparaturschleife mit klaren, für das Modell lesbaren Fehlermeldungen.
5. Befehle tragen eine ID, eine Ablaufzeit und ein Response Topic; Erfolg wird erst nach der Bestätigung gemeldet.
6. Funktionen mit großer Auswirkung erfordern eine menschliche Freigabe; Sicherheitsfunktionen werden nie bereitgestellt.
7. Jeder Aufruf, jede Ablehnung und jede Bestätigung wird protokolliert.

Die Demo nutzt den öffentlichen MQTT-Broker, der für Tests und Prototypen und nicht für Produktionsdaten gedacht ist. Für echte Anlagen wechseln Sie auf Port 8883 mit TLS sowie dem für Ihren Broker erzeugten Benutzernamen, Passwort und der Client-ID; ein verwalteter MQTT-Broker wie ConnectMQ übernimmt Zertifikate, MQTT-5-Unterstützung und Verbindungskontingente für Sie. Zum Gesamtdesign von Agenten lesen Sie LLM-Agenten für IoT-Geräte, für ein benutzerfreundlicheres Frontend die Gerätesteuerung per natürlicher Sprache und für den breiteren Kontext unseren Leitfaden zu KI und IoT. Die JSON-Schema-Spezifikation dokumentiert jedes oben verwendete Schlüsselwort.

Starten Sie mit ConnectMQ

ConnectMQ unterstützt die MQTT 5 Response Topics, Correlation Data und Message Expiry, auf denen dieses Muster beruht. Registrieren Sie sich für den kostenlosen Serverless-Tarif, ohne Kreditkarte, und führen Sie noch heute Ihren ersten validierten Funktionsaufruf gegen einen echten Broker aus.

Häufig gestellte Fragen

Reicht das JSON-Schema, das ich an das Modell sende, um Befehle sicher zu machen?

Nein. Es lenkt das Modell, garantiert aber keine konforme Ausgabe. Validieren Sie die Argumente immer in Ihrem eigenen Code und noch einmal auf der Geräteseite.

Was sollte passieren, wenn der Funktionsaufruf des Modells ungültig ist?

Geben Sie dem Modell eine klare Fehlermeldung zurück und erlauben Sie ein oder zwei Reparaturversuche. Scheitert es dann immer noch, melden Sie dem Benutzer die Ablehnung, statt zu raten.

Warum MQTT 5 für Befehle verwenden?

Response Topics und Correlation Data bieten einen standardisierten Weg, Bestätigungen den Befehlen zuzuordnen, und Message Expiry verhindert, dass Geräte veraltete Befehle ausführen.

Kann eine Anfrage mehrere Funktionsaufrufe erzeugen?

Ja. Modelle können mehrere Tool-Aufrufe in einer Antwort zurückgeben. Validieren Sie jeden unabhängig und veröffentlichen Sie nur diejenigen, die die Prüfung bestehen.