Function calling LLM IoT: tutorial con MQTT

El function calling de LLM para IoT convierte lo que el modelo entiende de una petición en una llamada estructurada como set_valve(valve_id="V-12", position_pct=40), que tu código puede comprobar y convertir en un comando MQTT. Este tutorial se centra en la parte que la mayoría de las demos se saltan: esquemas JSON estrictos para las acciones de los dispositivos, validación independiente de cada llamada, un bucle de reparación para la salida inválida y confirmaciones MQTT 5, de modo que nada de lo que produce el modelo llegue a un dispositivo sin comprobar.

Function calling en un párrafo

Con function calling (también llamado tool calling), envías al modelo una lista de funciones con definiciones de parámetros en JSON Schema. En lugar de responder en prosa, el modelo puede devolver el nombre de una función y un objeto JSON de argumentos. Tu aplicación decide si la ejecuta. El modelo nunca ejecuta nada por sí mismo, y precisamente por eso el patrón es útil para dispositivos: la aplicación es un punto de control en el que puedes imponer reglas.

El problema es que un esquema en el prompt es una orientación, no una imposición. Los modelos devuelven de vez en cuando argumentos con el tipo equivocado, valores fuera de rango, campos de más o IDs de dispositivo que no existen. Algunos proveedores ofrecen modos estrictos que restringen la salida al esquema, pero las reglas semánticas (si esta válvula está en esta línea, si este cambio es demasiado grande) siempre necesitan tu propio código.

Paso 1: diseñar esquemas estrictos para las acciones de los dispositivos

Los buenos esquemas para acciones físicas comparten algunos rasgos:

Regla	Ejemplo	Por qué
Una función por intención	`set_valve`, `start_pump`, no `control_device`	Espacio de argumentos más pequeño y verificable
Enumerar los identificadores	`"enum": ["V-11", "V-12"]`	El modelo no puede inventar dispositivos
Acotar cada número	`"minimum": 0, "maximum": 100`	Rechaza valores fuera de rango
Poner las unidades en los nombres	`position_pct`, `setpoint_c`	Elimina la ambigüedad de unidades
Prohibir campos adicionales	`"additionalProperties": false`	Detecta parámetros alucinados
Exigir un motivo	`"reason": {"maxLength": 200}`	Rastro de auditoría para cada acción

Este es un módulo de esquemas compartido por el servicio de comandos y, idealmente, por el firmware del dispositivo o el gateway:

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

Paso 2: validar dos veces, primero el esquema y luego la semántica

La primera comprobación es estructural y usa la biblioteca jsonschema contra exactamente el mismo esquema que vio el modelo. La segunda es semántica: reglas que dependen del estado en vivo o de la política de la planta, que ningún esquema puede expresar.

# 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   # política de planta: no más de 25 puntos por comando


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

Los mensajes de error están escritos para que los lea el modelo. Es deliberado: alimentan el bucle de reparación del siguiente paso.

Paso 3: el bucle de function calling del LLM con reparación

Cuando una llamada no supera la validación, devuelve el error y deja que el modelo lo intente de nuevo, una o dos veces como mucho. Muchos fallos tienen arreglo ("propón un valor intermedio"), y un bucle acotado evita los reintentos infinitos. Este ejemplo usa Ollama con un modelo local capaz de usar herramientas; la misma estructura funciona con la API de function calling de cualquier proveedor.

# 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"     # usa tu propio id aleatorio
ACK_TOPIC = f"{PREFIX}/svc/acks"
MAX_REPAIRS = 2

live_state = {}   # se rellena desde los topics de estado retenidos
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)   # deja que llegue el estado retenido


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   # nunca ejecutes un comando obsoleto
    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."))

Fíjate en lo que ocurre con esa petición de ejemplo. Si V-12 está ahora al 40 por ciento, la comprobación semántica rechaza el salto de 50 puntos y le pide al modelo que proponga un valor intermedio. El modelo normalmente lo reintenta con 65, que pasa la validación y se publica. Las llamadas válidas de un lote salen de inmediato, y solo las rechazadas se devuelven para su reparación.

Modos estrictos y decodificación restringida

Varios proveedores de modelos y runtimes locales pueden restringir la generación para que los argumentos de las herramientas cumplan siempre el JSON Schema declarado. Cuando esté disponible, actívalo: elimina la mayoría de los errores de tipo y de campos ausentes antes de que lleguen a tu validador. No elimina la necesidad de validar, por tres motivos. Primero, la decodificación restringida solo cubre el subconjunto de JSON Schema que admite el runtime, y palabras clave como los rangos numéricos o los patrones de cadena pueden ignorarse. Segundo, una llamada perfectamente formada puede seguir siendo errónea para la situación actual, como un salto de válvula demasiado grande o una bomba en servicio crítico. Tercero, puede que más adelante cambies de modelo o de proveedor, y tu seguridad no debe depender de una función de un runtime concreto. Trata la salida restringida como una mejora de calidad y la validación en código como la garantía.

Paso 4: validar de nuevo en el lado del dispositivo

El servicio de comandos no es la última línea de defensa. El gateway o dispositivo que recibe .../cmd debe validar el mismo esquema, comprobar que el mensaje no ha caducado y responder con una confirmación en el response topic:

# valve_gateway.py (lado del dispositivo simulado)
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()

Instala las dependencias con pip install ollama jsonschema "paho-mqtt>=2.0", arranca el gateway y después el servicio de comandos.

Probar las definiciones de funciones antes del despliegue

Las definiciones de funciones merecen la misma disciplina de pruebas que cualquier otra interfaz con equipos físicos. Construye un conjunto de pruebas en dos capas:

Tests unitarios del validador sin modelo. Pasa a check() conjuntos de argumentos escritos a mano, válidos e inválidos, y comprueba el resultado. Incluye valores límite, tipos erróneos, IDs desconocidos, campos de más y estado en vivo obsoleto o ausente.
Tests de comportamiento del modelo. Mantén una lista de peticiones realistas con la función y los argumentos esperados, como "cierra V-11 por completo" o "para la bomba principal". Ejecútalas contra el modelo con temperatura 0 y registra con qué frecuencia el primer intento es válido, con qué frecuencia lo arregla el bucle de reparación y con qué frecuencia termina en rechazo.

Vuelve a ejecutar los tests de comportamiento cada vez que cambies una descripción, añadas una función o cambies de modelo. Pequeños cambios de redacción en una descripción pueden alterar qué función elige un modelo, y quieres descubrirlo en las pruebas y no en la planta. Registrar las llamadas rechazadas en producción te proporciona nuevos casos de prueba con el tiempo.

Checklist de function calling LLM en IoT para producción

Los esquemas están versionados y se comparten entre el servicio de comandos y los dispositivos.
additionalProperties: false, enums para los IDs y límites para cada número.
Validación estructural con una biblioteca de JSON Schema y después comprobaciones semánticas frente al estado en vivo.
Un bucle de reparación acotado con mensajes de error claros y legibles para el modelo.
Los comandos llevan un ID, una caducidad y un response topic; el éxito solo se comunica tras la confirmación.
Las funciones de alto impacto requieren aprobación humana; las funciones de seguridad nunca se exponen.
Se registran todas las llamadas, rechazos y confirmaciones.

La demo usa el servidor MQTT público gratuito de ConnectMQ, válido solo para pruebas y prototipos, no para datos de producción. Para equipos reales, pasa al puerto 8883 con TLS y el usuario, la contraseña y el client ID generados para tu broker; un broker MQTT gestionado como ConnectMQ se encarga por ti de los certificados, el soporte de MQTT 5 y las cuotas de conexión. Para el diseño general del agente consulta agentes LLM para dispositivos IoT, para una interfaz más amigable consulta control de dispositivos con lenguaje natural, y para un contexto más amplio nuestra guía de IA e IoT. La especificación de JSON Schema documenta todas las palabras clave usadas arriba.

Empieza con ConnectMQ

ConnectMQ admite los response topics, los correlation data y la caducidad de mensajes de MQTT 5 en los que se basa este patrón. Regístrate en el plan Serverless gratuito, sin tarjeta de crédito, y ejecuta hoy tu primera llamada a función validada contra un broker real.

Preguntas frecuentes

¿Basta el esquema JSON que envío al modelo para que los comandos sean seguros?

No. Orienta al modelo, pero no garantiza una salida conforme. Valida siempre los argumentos en tu propio código y de nuevo en el lado del dispositivo.

¿Qué debe ocurrir cuando la llamada a función del modelo no es válida?

Devuelve al modelo un mensaje de error claro y permite uno o dos intentos de reparación. Si sigue fallando, comunica el rechazo al usuario en lugar de adivinar.

¿Por qué usar MQTT 5 para los comandos?

Los response topics y los correlation data ofrecen una forma estándar de emparejar las confirmaciones con los comandos, y la caducidad de mensajes impide que los dispositivos ejecuten comandos obsoletos.

¿Puede una petición generar varias llamadas a funciones?

Sí. Los modelos pueden devolver varias llamadas a herramientas en una misma respuesta. Valida cada una de forma independiente y publica solo las que superen la validación.

Tutorial de function calling LLM IoT: esquemas JSON estrictos y comandos MQTT validados