Controle de Dispositivos por Linguagem Natural: Pedidos Simples Viram Comandos MQTT Validados

O controle de dispositivos por linguagem natural permite que as pessoas digam o que querem, como "deixe o escritório mais quente" ou "desligue tudo na sala", em vez de procurar opções em dashboards. Neste tutorial você vai construir um serviço em Python no qual um LLM traduz pedidos em chamadas de função estruturadas, uma camada de validação confere cada chamada contra um registro de dispositivos e apenas comandos válidos são publicados via MQTT para dispositivos que os confirmam.

Como o pipeline funciona

A ideia central é que o modelo de linguagem nunca escreve mensagens MQTT. Ele apenas escolhe em um menu de funções que você define, e o seu código transforma essas escolhas em comandos depois de verificá-las.

1. Pedido do usuário. Texto livre vindo de uma caixa de chat, da transcrição de um assistente de voz ou de um bot de mensagens.
2. Contexto. O serviço entrega ao modelo uma lista compacta de dispositivos, com seus nomes amigáveis e recursos.
3. Function calling. O modelo retorna zero ou mais chamadas de ferramenta, por exemplo set_light(device="kitchen-ceiling", on=true, brightness=30).
4. Validação. O código verifica se o dispositivo existe, se ele suporta o recurso e se os valores estão dentro da faixa.
5. Publish. Comandos válidos vão para .../<device>/set via MQTT com um ID de comando exclusivo.
6. Confirmação. O dispositivo aplica a mudança e publica seu novo estado, que o serviço informa de volta ao usuário.

Pré-requisitos

• Python 3.10 ou mais recente.
• Ollama rodando localmente com um modelo que suporte tool calling, como qwen2.5:7b ou llama3.1:8b. Qualquer provedor com function calling funciona da mesma forma; só a chamada de API muda.
• Um broker MQTT. Usamos o servidor MQTT público (gratuito) em broker.connectmq.com na demonstração.

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

Passo 1: definir o registro de dispositivos

O registro é a única fonte de verdade sobre o que existe e o que cada coisa pode fazer. O modelo vê um resumo dele; o validador usa o registro inteiro.

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

Passo 2: definir as funções

Cada função corresponde a um recurso. Enumerar os IDs de dispositivo no schema ajuda o modelo a escolher alvos 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"]}}},
]

Passo 3: validar cada chamada

Os schemas orientam o modelo, mas não o obrigam. Modelos locais pequenos, em especial, às vezes retornam números como strings, inventam IDs de dispositivo ou ignoram faixas. O validador normaliza o que é seguro normalizar e rejeita todo o resto informando o 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}")

Passo 4: o serviço de controle de dispositivos por linguagem natural

Agora conecte tudo. O serviço lê um pedido, pede ao modelo as chamadas de ferramenta, valida-as e publica os 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"   # use seu próprio id aleatório

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)

Definir a temperatura como 0 torna a escolha das ferramentas mais reproduzível, que é o que você quer em tarefas de controle.

Passo 5: adicionar um dispositivo simulado e testar

Um pequeno simulador assina todos os topics set, aplica os comandos e publica o estado como 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()

Execute o simulador, depois o serviço, e experimente estes pedidos:

Os casos interessantes são as falhas. Cada uma é capturada ou pelo modelo, que pede esclarecimento, ou pelo validador, e nada inválido chega ao broker.

Por que o modelo nunca escreve payloads MQTT

É tentador pular a camada de funções e pedir ao modelo que gere diretamente um topic e um payload JSON. Essa abordagem falha de maneiras sutis. O modelo pode inventar um topic que parece correto, mas não existe, confundir nomes de chaves entre tipos de dispositivo ou produzir um JSON válido com as unidades erradas. Cada um desses erros chega ao broker, e alguns dispositivos simplesmente ignoram comandos malformados em silêncio, deixando o usuário acreditar que a mudança aconteceu.

Com function calling, o modelo escolhe entre algumas operações nomeadas cujos parâmetros você definiu. O seu código é dono do mapeamento entre operação, topic e formato do payload, então alterar o firmware de um dispositivo ou a organização dos topics nunca exige mudar prompts. O validador passa a ser o único lugar onde os comandos são criados, o que facilita testá-lo com testes unitários comuns, sem modelo nenhum.

Lidando com pedidos de continuação

Conversas reais raramente acontecem em uma só mensagem. As pessoas dizem "um pouco mais claro" ou "a outra também". Para suportar isso, mantenha as últimas mensagens do usuário e do assistente na lista messages e inclua no prompt de sistema o estado atual dos dispositivos alterados recentemente, lido dos topics de estado com retained message. Assim o modelo consegue entender que "um pouco mais claro" se refere à luz que acabou de mudar e calcular um novo brilho a partir do valor atual, enquanto o validador continua conferindo o resultado. Limite o histórico a poucas trocas para que contexto antigo não vaze para novos pedidos.

Preparando para produção

Ao passar da demonstração para uma casa, um prédio comercial ou uma planta reais, por exemplo uma fábrica em Manaus, tenha estes pontos em mente:

1. Use credenciais privadas e TLS. O servidor MQTT público pode ser lido por qualquer pessoa e serve apenas para testes e protótipos. Para dispositivos reais, conecte na porta 8883 com credenciais geradas no portal do ConnectMQ.
2. Confirme ações destrutivas ou incomuns. Desligar todas as luzes não tem problema; abrir um portão ou desativar um alarme deve exigir confirmação explícita.
3. Espere pelo estado, não apenas pelo publish. Informe sucesso somente depois que o estado do dispositivo confirmar o ID do comando.
4. Gere o registro em vez de escrevê-lo à mão. Monte-o a partir de topics de anúncio de dispositivos com retained message, para que novos dispositivos apareçam automaticamente.
5. Registre tudo. Armazene o pedido, as chamadas de ferramenta, os resultados da validação e as confirmações para facilitar a investigação de problemas.

A mesma disciplina de validação vale para equipamentos industriais; nossos guias sobre function calling de LLM para IoT e agentes LLM para dispositivos IoT a ampliam com bibliotecas de schema e aprovação humana. Se quiser toda a pilha offline, veja como executar o Ollama com MQTT em um gateway. Para entender onde as interfaces de linguagem se encaixam entre outras técnicas de IA, leia nosso guia de IA e IoT.

Executar o canal de controle em um broker MQTT gerenciado significa que seus dispositivos, o serviço de linguagem e os dashboards alcançam um único endpoint seguro via TCP, TLS ou WebSockets, sem abrir portas em casa ou na planta.

Comece com o ConnectMQ

O ConnectMQ oferece um broker MQTT gerenciado com um client MQTT no navegador, para você acompanhar comandos e mudanças de estado ao vivo enquanto testa os prompts. Crie seu broker Serverless gratuito, sem cartão de crédito, e deixe as pessoas controlarem dispositivos com as próprias palavras.

Perguntas frequentes

Preciso de um modelo grande na nuvem para controle de dispositivos por linguagem natural?

Não. Modelos locais pequenos com suporte a tool calling lidam bem com comandos simples quando a lista de dispositivos é curta e as funções são claras. A validação no código cobre os casos em que eles erram.

Como lidar com pedidos ambíguos como "acenda a luz"?

Instrua o modelo, no prompt de sistema, a fazer uma pergunta de esclarecimento quando vários dispositivos corresponderem. Seu serviço mostra a pergunta ao usuário e envia a resposta de volta na próxima interação.

Posso usar voz em vez de texto?

Sim. Execute primeiro o reconhecimento de fala e envie a transcrição para o mesmo pipeline. Mantenha as confirmações para ações arriscadas, porque erros de transcrição acrescentam mais uma fonte de enganos.

Por que publicar comandos em um topic set e ler um topic de estado separado?

Separar comandos de estado faz com que o topic de estado sempre reflita o que o dispositivo realmente fez, e o estado com retained message entrega a todo client o valor atual imediatamente.