Tutoriel function calling LLM pour l'IoT : schémas JSON stricts et commandes MQTT validées

Le function calling LLM pour l'IoT transforme la compréhension qu'a un modèle d'une demande en un appel structuré comme set_valve(valve_id="V-12", position_pct=40), que votre code peut vérifier puis convertir en commande MQTT. Ce tutoriel se concentre sur la partie que la plupart des démos ignorent : des schémas JSON stricts pour les actions des appareils, une validation indépendante de chaque appel, une boucle de correction pour les sorties invalides et des accusés de réception MQTT 5, afin que rien de ce que produit le modèle n'atteigne un appareil sans contrôle.

Le function calling en un paragraphe

Avec le function calling (aussi appelé appel d'outils ou tool calling), vous envoyez au modèle une liste de fonctions dont les paramètres sont définis par un schéma JSON. Au lieu de répondre en prose, le modèle peut renvoyer le nom d'une fonction et un objet JSON d'arguments. C'est votre application qui décide de l'exécuter ou non. Le modèle n'exécute jamais rien lui-même, et c'est précisément ce qui rend ce pattern utile pour les appareils : l'application constitue un point de passage obligé où vous pouvez faire respecter des règles.

Le piège, c'est qu'un schéma dans le prompt est une indication, pas une contrainte. Les modèles renvoient parfois des arguments du mauvais type, des valeurs hors plage, des champs en trop ou des identifiants d'appareils inexistants. Certains fournisseurs proposent des modes stricts qui contraignent la sortie au schéma, mais les règles sémantiques (cette vanne est-elle sur cette ligne, ce changement est-il trop important) nécessitent toujours votre propre code.

Étape 1 : concevoir des schémas stricts pour les actions des appareils

Les bons schémas pour des actions physiques partagent quelques caractéristiques :

Voici un module de schémas partagé par le service de commande et, idéalement, par le firmware de l'appareil ou la passerelle :

# 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()
    ]

Étape 2 : valider deux fois, le schéma puis la sémantique

Le premier contrôle est structurel et utilise la bibliothèque jsonschema avec exactement le même schéma que celui vu par le modèle. Le second est sémantique : des règles qui dépendent de l'état en direct ou de la politique de l'usine, et qu'aucun schéma ne peut exprimer.

# 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   # plant policy: no more than 25 points per command


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

Les messages d'erreur sont rédigés pour être lus par le modèle. C'est voulu : ils alimentent la boucle de correction de l'étape suivante.

Étape 3 : la boucle de function calling LLM avec correction

Lorsqu'un appel échoue à la validation, renvoyez l'erreur et laissez le modèle réessayer, une ou deux fois au maximum. Beaucoup d'échecs sont corrigeables (« propose une valeur intermédiaire »), et une boucle bornée évite les tentatives sans fin. Cet exemple utilise Ollama avec un modèle local capable d'appeler des outils ; la même structure fonctionne avec l'API de function calling de n'importe quel fournisseur.

# 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"     # use your own random id
ACK_TOPIC = f"{PREFIX}/svc/acks"
MAX_REPAIRS = 2

live_state = {}   # filled from retained state topics
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)   # let retained state arrive


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   # never execute a stale command
    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."))

Observez ce qui se passe avec cette demande d'exemple. Si V-12 est actuellement à 40 %, le contrôle sémantique rejette le saut de 50 points et demande au modèle de proposer une valeur intermédiaire. Le modèle réessaie généralement avec 65, valeur qui passe le contrôle et est publiée. Dans un lot, les appels valides partent immédiatement, et seuls les appels rejetés sont renvoyés pour correction.

Modes stricts et décodage contraint

Plusieurs fournisseurs de modèles et runtimes locaux peuvent contraindre la génération pour que les arguments des outils correspondent toujours au schéma JSON déclaré. Lorsque c'est possible, activez cette option : elle élimine la plupart des erreurs de type et des champs manquants avant qu'ils n'atteignent votre validateur. Elle ne dispense pas pour autant de la validation, pour trois raisons. Premièrement, le décodage contraint ne couvre que le sous-ensemble de JSON Schema pris en charge par le runtime, et des mots-clés comme les plages numériques ou les motifs de chaînes peuvent être ignorés. Deuxièmement, un appel parfaitement formé peut tout de même être inadapté à la situation du moment, comme un pas de vanne trop important ou une pompe en service critique. Troisièmement, vous changerez peut-être de modèle ou de fournisseur, et votre sécurité ne doit pas dépendre d'une fonctionnalité propre à un runtime. Considérez la sortie contrainte comme une amélioration de la qualité et la validation dans le code comme la garantie.

Étape 4 : valider à nouveau côté appareil

Le service de commande n'est pas la dernière ligne de défense. La passerelle ou l'appareil qui reçoit .../cmd doit valider le même schéma, vérifier que le message n'a pas expiré et répondre par un accusé de réception sur le response topic :

# valve_gateway.py (simulated device side)
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()

Installez les dépendances avec pip install ollama jsonschema "paho-mqtt>=2.0", lancez la passerelle, puis le service de commande.

Tester les définitions de fonctions avant le déploiement

Les définitions de fonctions méritent la même rigueur de test que toute autre interface avec un équipement physique. Construisez une suite de tests en deux couches :

1. Des tests unitaires du validateur, sans modèle. Injectez dans check() des jeux d'arguments écrits à la main, valides et invalides, et vérifiez le résultat. Incluez des valeurs limites, des types erronés, des identifiants inconnus, des champs en trop et un état en direct périmé ou absent.
2. Des tests de comportement du modèle. Tenez une liste de demandes réalistes avec la fonction et les arguments attendus, comme « ferme complètement V-11 » ou « arrête la pompe principale ». Exécutez-les sur le modèle avec une température à 0 et mesurez la fréquence à laquelle la première tentative est valide, celle à laquelle la boucle de correction la répare et celle à laquelle elle aboutit à un rejet.

Relancez les tests de comportement chaque fois que vous modifiez une description, ajoutez une fonction ou changez de modèle. De petites modifications de formulation dans une description peuvent changer la fonction choisie par un modèle, et mieux vaut le découvrir en test que dans l'atelier. Journaliser les appels rejetés en production vous fournit au fil du temps de nouveaux cas de test.

Check-list pour le function calling LLM en production dans l'IoT

1. Les schémas sont versionnés et partagés entre le service de commande et les appareils.
2. additionalProperties: false, des énumérations pour les identifiants et des bornes pour chaque nombre.
3. Une validation structurelle avec une bibliothèque JSON Schema, puis des contrôles sémantiques par rapport à l'état en direct.
4. Une boucle de correction bornée avec des messages d'erreur clairs et lisibles par le modèle.
5. Les commandes portent un identifiant, une expiration et un response topic ; le succès n'est annoncé qu'après accusé de réception.
6. Les fonctions à fort impact exigent une approbation humaine ; les fonctions de sécurité ne sont jamais exposées.
7. Chaque appel, rejet et accusé de réception est journalisé.

La démo utilise le serveur MQTT public gratuit, réservé aux tests. Pour de vrais équipements, passez sur le port 8883 avec TLS et les nom d'utilisateur, mot de passe et client ID générés pour votre broker ; un broker MQTT managé comme ConnectMQ gère pour vous les certificats, la prise en charge de MQTT 5 et les quotas de connexion. Pour la conception globale de l'agent, consultez les agents LLM pour appareils IoT ; pour une interface plus conviviale, voyez le contrôle d'appareils en langage naturel ; et pour un contexte plus large, notre guide IA et IoT. La spécification JSON Schema documente chaque mot-clé utilisé ci-dessus.

Démarrer avec ConnectMQ

ConnectMQ prend en charge les response topics, les correlation data et l'expiration des messages de MQTT 5 sur lesquels repose ce pattern. Inscrivez-vous à l'offre Serverless gratuite, sans carte bancaire, et exécutez dès aujourd'hui votre premier appel de fonction validé sur un vrai broker.

Questions fréquentes

Le schéma JSON envoyé au modèle suffit-il à sécuriser les commandes ?

Non. Il guide le modèle mais ne garantit pas une sortie conforme. Validez toujours les arguments dans votre propre code, puis à nouveau côté appareil.

Que faire lorsque l'appel de fonction du modèle est invalide ?

Renvoyez un message d'erreur clair au modèle et autorisez une ou deux tentatives de correction. Si l'appel échoue encore, signalez le rejet à l'utilisateur au lieu de deviner.

Pourquoi utiliser MQTT 5 pour les commandes ?

Les response topics et les correlation data offrent un moyen standard d'associer les accusés de réception aux commandes, et l'expiration des messages empêche les appareils d'exécuter des commandes périmées.

Une demande peut-elle produire plusieurs appels de fonctions ?

Oui. Les modèles peuvent renvoyer plusieurs appels d'outils dans une même réponse. Validez chacun d'eux indépendamment et ne publiez que ceux qui passent les contrôles.