Contrôle d'appareils en langage naturel : transformer des demandes simples en commandes MQTT validées

Le contrôle d'appareils en langage naturel permet de dire simplement ce que l'on veut, par exemple « réchauffe un peu le bureau » ou « éteins tout dans le salon », au lieu de fouiller dans des tableaux de bord, que ce soit dans une maison connectée, un bâtiment ou un atelier. Dans ce tutoriel, vous allez créer un service Python dans lequel un LLM traduit les demandes en appels de fonctions structurés, une couche de validation vérifie chaque appel par rapport à un registre d'appareils, et seules les commandes valides sont publiées via MQTT vers des appareils qui en accusent réception.

Fonctionnement du pipeline

L'idée clé est que le modèle de langage n'écrit jamais de messages MQTT. Il se contente de choisir dans un menu de fonctions que vous définissez, et votre code transforme ces choix en commandes après les avoir vérifiés.

1. Demande de l'utilisateur. Du texte libre provenant d'une zone de chat, de la transcription d'un assistant vocal ou d'un bot de messagerie.
2. Contexte. Le service fournit au modèle une liste compacte des appareils, de leurs noms usuels et de leurs capacités.
3. Function calling. Le modèle renvoie zéro ou plusieurs appels d'outils, par exemple set_light(device="kitchen-ceiling", on=true, brightness=30).
4. Validation. Le code vérifie que l'appareil existe, qu'il prend en charge la capacité et que les valeurs sont dans la plage autorisée.
5. Publication. Les commandes valides partent vers .../<device>/set via MQTT avec un identifiant de commande unique.
6. Accusé de réception. L'appareil applique le changement et publie son nouvel état, que le service retransmet à l'utilisateur.

Prérequis

• Python 3.10 ou plus récent.
• Ollama en local avec un modèle prenant en charge l'appel d'outils, comme qwen2.5:7b ou llama3.1:8b. Tout fournisseur proposant le function calling fonctionne de la même manière ; seul l'appel d'API change.
• Un broker MQTT. Nous utilisons broker.connectmq.com, le serveur MQTT public gratuit, pour la démo.

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

Étape 1 : définir le registre d'appareils

Le registre est la source de vérité unique sur ce qui existe et ce que chaque appareil sait faire. Le modèle en voit un résumé ; le validateur l'utilise en entier.

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

Étape 2 : définir les fonctions

Chaque fonction correspond à une capacité. Énumérer les identifiants d'appareils dans le schéma aide le modèle à choisir des cibles valides.

# 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"]}}},
]

Étape 3 : valider chaque appel

Les schémas guident le modèle mais ne le contraignent pas. Les petits modèles locaux, en particulier, renvoient parfois des nombres sous forme de chaînes, inventent des identifiants d'appareils ou ignorent les plages. Le validateur normalise ce qui peut l'être sans risque et rejette tout le reste en indiquant la raison.

# 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}")

Étape 4 : le service de contrôle d'appareils en langage naturel

Il est temps de tout relier. Le service lit une demande, demande au modèle des appels d'outils, les valide et publie les commandes.

# 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"   # use your own random id

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)

Régler la température à 0 rend le choix des outils plus reproductible, ce qui est souhaitable pour des tâches de contrôle.

Étape 5 : ajouter un appareil simulé et tester

Un minuscule simulateur s'abonne à tous les topics set, applique les commandes et publie l'état en retained message :

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

Lancez le simulateur, puis le service, et essayez ces demandes :

Les cas intéressants sont les échecs. Chacun est intercepté, soit par le modèle qui demande une précision, soit par le validateur, et rien d'invalide n'atteint le broker.

Pourquoi le modèle n'écrit jamais de payloads MQTT

Il est tentant de sauter la couche de fonctions et de demander au modèle de produire directement un topic et un payload JSON. Cette approche échoue de manière subtile. Le modèle peut inventer un topic qui semble correct mais n'existe pas, confondre les noms de clés entre types d'appareils ou produire un JSON valide avec les mauvaises unités. Chacune de ces erreurs atteint le broker, et certains appareils ignorent silencieusement les commandes mal formées, laissant l'utilisateur croire que le changement a eu lieu.

Avec le function calling, le modèle choisit parmi quelques opérations nommées dont vous avez défini les paramètres. Votre code détient la correspondance entre opération, topic et format de payload : modifier le firmware d'un appareil ou l'organisation des topics n'exige donc jamais de modifier les prompts. Le validateur devient alors le seul endroit où les commandes sont créées, ce qui le rend facile à tester avec des tests unitaires classiques, sans aucun modèle.

Gérer les demandes de suivi

Les vraies conversations se limitent rarement à un seul échange. Les gens disent « un peu plus fort » ou « l'autre aussi ». Pour le prendre en charge, conservez les derniers messages de l'utilisateur et de l'assistant dans la liste messages et incluez dans le prompt système l'état actuel des appareils récemment modifiés, lu depuis les topics d'état en retained message. Le modèle peut alors associer « un peu plus fort » à la lumière qu'il vient de modifier et calculer une nouvelle luminosité à partir de la valeur actuelle, pendant que le validateur continue de vérifier le résultat. Limitez l'historique à quelques échanges pour éviter qu'un contexte périmé ne contamine les nouvelles demandes.

Passer en production

En passant de la démo à une vraie maison, un bâtiment ou une usine, gardez ces points à l'esprit :

1. Utilisez des identifiants privés et TLS. Le serveur MQTT public gratuit est lisible par tout le monde et réservé aux tests et aux prototypes. Pour de vrais appareils, connectez-vous sur le port 8883 avec des identifiants générés dans le portail ConnectMQ.
2. Faites confirmer les actions destructives ou inhabituelles. Éteindre toutes les lumières ne pose pas de problème ; ouvrir un portail ou désactiver une alarme doit exiger une confirmation explicite.
3. Attendez l'état, ne vous contentez pas de publier. N'annoncez un succès qu'une fois que l'état de l'appareil confirme l'identifiant de la commande.
4. Générez le registre au lieu de l'écrire à la main. Construisez-le à partir des topics d'annonce des appareils en retained message, afin que les nouveaux appareils apparaissent automatiquement.
5. Journalisez tout. Conservez la demande, les appels d'outils, les résultats de validation et les accusés de réception pour le dépannage.

La même discipline de validation s'applique aux équipements industriels ; nos guides sur le function calling LLM pour l'IoT et sur les agents LLM pour appareils IoT l'étendent avec des bibliothèques de schémas et une approbation humaine. Si vous voulez toute la pile hors ligne, découvrez comment exécuter Ollama avec MQTT sur une passerelle. Pour situer les interfaces en langage naturel parmi les autres techniques d'IA, lisez notre guide IA et IoT.

Faire passer le canal de contrôle par un broker MQTT managé signifie que vos appareils, le service de langage et les tableaux de bord rejoignent tous un point d'accès unique et sécurisé via TCP, TLS ou WebSockets, sans ouvrir de ports à la maison ou à l'usine. Les logiciels domotiques qui parlent MQTT, comme Jeedom ou Domoticz, peuvent d'ailleurs se connecter au même broker.

Démarrer avec ConnectMQ

ConnectMQ vous offre un broker MQTT managé avec un client MQTT intégré au navigateur, pour observer en direct commandes et changements d'état pendant que vous testez vos prompts. Créez votre broker Serverless (Free), sans carte bancaire, et laissez chacun piloter ses appareils avec ses propres mots.

Questions fréquentes

Faut-il un grand modèle cloud pour le contrôle d'appareils en langage naturel ?

Non. Les petits modèles locaux qui prennent en charge l'appel d'outils gèrent bien les commandes simples lorsque la liste d'appareils est courte et que les fonctions sont claires. La validation dans le code couvre les cas où ils se trompent.

Comment gérer des demandes ambiguës comme « allume la lumière » ?

Indiquez au modèle dans le prompt système de poser une question de clarification lorsque plusieurs appareils correspondent. Votre service affiche la question à l'utilisateur et renvoie sa réponse au tour suivant.

Puis-je utiliser la voix au lieu du texte ?

Oui. Passez d'abord par une reconnaissance vocale, puis injectez la transcription dans le même pipeline. Conservez les confirmations pour les actions risquées, car les erreurs de transcription ajoutent une source d'erreurs supplémentaire.

Pourquoi publier les commandes sur un topic set et lire un topic d'état séparé ?

Séparer commandes et état garantit que le topic d'état reflète toujours ce que l'appareil a réellement fait, et l'état en retained message fournit immédiatement la valeur actuelle à chaque client.