El control de dispositivos con lenguaje natural permite que las personas digan lo que quieren, como "sube la temperatura de la oficina" o "apaga todo en el salón", en lugar de rebuscar entre dashboards. En este tutorial construirás un servicio en Python en el que un LLM traduce las peticiones a llamadas a funciones estructuradas, una capa de validación comprueba cada llamada frente a un registro de dispositivos y solo los comandos válidos se publican por MQTT hacia dispositivos que los confirman.
Cómo funciona la canalización
La idea clave es que el modelo de lenguaje nunca escribe mensajes MQTT. Solo elige de un menú de funciones que tú defines, y tu código convierte esas elecciones en comandos después de comprobarlas.
- Petición del usuario. Texto libre de un cuadro de chat, la transcripción de un asistente de voz o un bot de mensajería.
- Contexto. El servicio le da al modelo una lista compacta de dispositivos con sus nombres descriptivos y capacidades.
- Function calling. El modelo devuelve cero o más llamadas a herramientas, por ejemplo
set_light(device="kitchen-ceiling", on=true, brightness=30). - Validación. El código comprueba que el dispositivo existe, que admite la capacidad y que los valores están dentro de rango.
- Publicación. Los comandos válidos van a
.../<device>/setpor MQTT con un ID de comando único. - Confirmación. El dispositivo aplica el cambio y publica su nuevo estado, que el servicio comunica de vuelta al usuario.
| Capa | ¿De confianza? | Función |
|---|---|---|
| LLM | No | Interpretar la intención, elegir funciones y argumentos |
| Validador | Sí | Imponer registro, capacidades y rangos de valores |
| Broker MQTT | Sí | Entregar comandos y estado de forma fiable |
| Dispositivo | Sí | Aplicar el cambio dentro de sus propios límites e informar del estado |
Requisitos previos
- Python 3.10 o posterior.
- Ollama ejecutándose en local con un modelo que admita llamadas a herramientas, como
qwen2.5:7bollama3.1:8b. Cualquier proveedor con function calling funciona igual; solo cambia la llamada a la API. - Un broker MQTT. Para la demo usamos el servidor MQTT público gratuito en broker.connectmq.com, pensado para pruebas y no para datos de producción.
ollama pull qwen2.5:7b
pip install ollama "paho-mqtt>=2.0"
Paso 1: definir el registro de dispositivos
El registro es la única fuente de verdad sobre qué existe y qué puede hacer. El modelo ve un resumen; el validador lo usa completo.
# 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)
Paso 2: definir las funciones
Cada función corresponde a una capacidad. Enumerar los IDs de dispositivo en el esquema ayuda al modelo a elegir destinos válidos.
# 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"]}}},
]
Paso 3: validar cada llamada
Los esquemas orientan al modelo, pero no lo obligan. Los modelos locales pequeños, en particular, a veces devuelven números como cadenas, inventan IDs de dispositivo o ignoran los rangos. El validador normaliza lo que es seguro normalizar y rechaza todo lo demás indicando el motivo.
# 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}")
Paso 4: el servicio de control de dispositivos con lenguaje natural
Ahora conecta todas las piezas. El servicio lee una petición, pide al modelo las llamadas a herramientas, las valida y publica los comandos.
# 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" # usa tu propio id aleatorio
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)
Fijar la temperatura en 0 hace que la selección de herramientas sea más repetible, que es lo que quieres en tareas de control.
Paso 5: añadir un dispositivo simulado y probar
Un pequeño simulador se suscribe a todos los topics set, aplica los comandos y publica el estado retenido:
# 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()
Ejecuta el simulador, después el servicio, y prueba estas peticiones:
| Petición | Resultado esperado |
|---|---|
| Baja la luz del techo de la cocina al 30 por ciento | Una llamada a set_light con brillo 30 |
| Apaga todo en el salón | set_light para la lámpara y una pregunta o set_blinds, según la interpretación |
| Pon la oficina a 35 grados | Rechazada por el validador, o una respuesta aclaratoria del modelo |
| Atenúa la luz bajo los armarios | Rechazada: no es regulable |
| Enciende la luz del garaje | Pregunta aclaratoria, porque ese dispositivo no existe |
Los casos interesantes son los fallos. Cada uno se detecta o bien porque el modelo pide una aclaración o bien en el validador, y nada inválido llega al broker.
Por qué el modelo nunca escribe payloads MQTT
Es tentador saltarse la capa de funciones y pedir al modelo que genere directamente un topic y un payload JSON. Ese enfoque falla de maneras sutiles. El modelo puede inventar un topic que parece correcto pero no existe, confundir nombres de claves entre tipos de dispositivo o producir un JSON válido con unidades equivocadas. Cada uno de esos errores llega al broker, y algunos dispositivos ignorarán en silencio los comandos mal formados, dejando al usuario convencido de que el cambio se ha producido.
Con function calling, el modelo elige entre unas pocas operaciones con nombre cuyos parámetros has definido tú. Tu código se encarga de traducir cada operación a un topic y a un formato de payload, así que cambiar el firmware de un dispositivo o la estructura de topics nunca obliga a cambiar los prompts. El validador se convierte entonces en el único lugar donde se crean comandos, lo que facilita probarlo con tests unitarios corrientes y sin ningún modelo.
Gestionar peticiones de seguimiento
Las conversaciones reales rara vez se resuelven en un solo turno. La gente dice "un poco más de luz" o "la otra también". Para admitirlo, conserva los últimos mensajes del usuario y del asistente en la lista messages e incluye en el prompt del sistema el estado actual de los dispositivos tocados recientemente, leído de los topics de estado retenidos. Así el modelo puede asociar "un poco más de luz" con la lámpara que acaba de cambiar y calcular un nuevo brillo a partir del valor actual, mientras el validador sigue comprobando el resultado. Limita el historial a unos pocos turnos para que el contexto obsoleto no se cuele en peticiones nuevas.
Prepararlo para producción
Cuando pases de la demo a una vivienda, un edificio o una planta reales, ten en cuenta estos puntos:
- Usa credenciales privadas y TLS. El servidor MQTT público gratuito es legible por cualquiera. Para dispositivos reales, conéctate por el puerto 8883 con credenciales generadas en el portal de ConnectMQ.
- Confirma las acciones destructivas o inusuales. Apagar todas las luces no plantea problemas; abrir una cancela o desactivar una alarma debería requerir una confirmación explícita.
- Espera al estado, no te limites a publicar. Informa del éxito solo cuando el estado del dispositivo confirme el ID del comando.
- Genera el registro, no lo escribas a mano. Constrúyelo a partir de topics retenidos de anuncio de dispositivos para que los nuevos aparezcan automáticamente.
- Registra todo. Guarda la petición, las llamadas a herramientas, los resultados de validación y las confirmaciones para poder diagnosticar problemas.
La misma disciplina de validación escala a equipos industriales; nuestras guías sobre function calling de LLM para IoT y agentes LLM para dispositivos IoT la amplían con bibliotecas de esquemas y aprobación humana. Si quieres toda la pila sin conexión, consulta cómo ejecutar Ollama con MQTT en un gateway. Para entender dónde encajan las interfaces de lenguaje entre otras técnicas de IA, lee nuestra guía de IA e IoT.
Ejecutar el canal de control en un broker MQTT gestionado significa que tus dispositivos, el servicio de lenguaje y los dashboards llegan a un único endpoint seguro por TCP, TLS o WebSockets sin abrir puertos en casa ni en la planta.
Empieza con ConnectMQ
ConnectMQ te ofrece un broker MQTT gestionado con un cliente MQTT en el navegador para que veas en vivo los comandos y los cambios de estado mientras pruebas los prompts. Crea tu broker Serverless gratuito, sin tarjeta de crédito, y deja que las personas controlen los dispositivos con sus propias palabras.
Preguntas frecuentes
¿Necesito un gran modelo en la nube para el control de dispositivos con lenguaje natural?
No. Los modelos locales pequeños con soporte para llamadas a herramientas gestionan bien los comandos sencillos cuando la lista de dispositivos es corta y las funciones son claras. La validación en código cubre los casos en los que se equivocan.
¿Cómo gestiono peticiones ambiguas como "enciende la luz"?
Indica al modelo en el prompt del sistema que haga una pregunta aclaratoria cuando coincidan varios dispositivos. Tu servicio muestra la pregunta al usuario y envía la respuesta en el siguiente turno.
¿Puedo usar la voz en lugar de texto?
Sí. Ejecuta primero un reconocimiento de voz a texto y pasa la transcripción a la misma canalización. Mantén las confirmaciones para las acciones de riesgo, porque los errores de transcripción añaden otra fuente de fallos.
¿Por qué publicar los comandos en un topic set y leer un topic de estado aparte?
Separar los comandos del estado significa que el topic de estado siempre refleja lo que el dispositivo hizo realmente, y el estado retenido da a cada cliente el valor actual de inmediato.
