Tutorial de Function Calling de LLM para IoT: JSON Schemas Rígidos e Comandos MQTT Validados

O function calling de LLM para IoT transforma o entendimento que um modelo tem de um pedido em uma chamada estruturada como set_valve(valve_id="V-12", position_pct=40), que o seu código pode verificar e converter em um comando MQTT. Este tutorial se concentra na parte que a maioria das demonstrações pula: JSON schemas rígidos para ações de dispositivos, validação independente de cada chamada, um loop de correção para saídas inválidas e confirmações MQTT 5, para que nada do que o modelo produza chegue a um dispositivo sem verificação.

Function calling em um parágrafo

Com function calling (também chamado de tool calling), você envia ao modelo uma lista de funções com definições de parâmetros em JSON schema. Em vez de responder em texto, o modelo pode retornar o nome de uma função e um objeto JSON com os argumentos. Sua aplicação decide se vai executá-la. O modelo nunca executa nada por conta própria, e é exatamente por isso que o padrão é útil para dispositivos: a aplicação é um ponto de controle onde você pode impor regras.

O problema é que um schema no prompt é uma orientação, não uma imposição. Os modelos ocasionalmente retornam argumentos com o tipo errado, valores fora da faixa, campos extras ou IDs de dispositivo que não existem. Alguns provedores oferecem modos estritos que restringem a saída ao schema, mas regras semânticas (esta válvula está nesta linha? esta mudança é grande demais?) sempre exigem o seu próprio código.

Passo 1: projetar schemas rígidos para ações de dispositivos

Bons schemas para ações físicas compartilham algumas características:

Aqui está um módulo de schemas compartilhado pelo serviço de comandos e, idealmente, pelo firmware do dispositivo ou pelo 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()
    ]

Passo 2: validar duas vezes: schema, depois semântica

A primeira verificação é estrutural e usa a biblioteca jsonschema contra exatamente o mesmo schema que o modelo viu. A segunda é semântica: regras que dependem do estado ao vivo ou da política da planta, que nenhum schema consegue expressar.

# 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 da planta: no máximo 25 pontos 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

As mensagens de erro são escritas para o modelo ler. Isso é intencional: elas alimentam o loop de correção do próximo passo.

Passo 3: o loop de function calling do LLM com correção

Quando uma chamada falhar na validação, envie o erro de volta e deixe o modelo tentar de novo, uma ou no máximo duas vezes. Muitas falhas têm correção ("proponha um valor intermediário"), e um loop limitado evita tentativas sem fim. Este exemplo usa o Ollama com um modelo local capaz de usar ferramentas; a mesma estrutura funciona com a API de function calling de qualquer provedor.

# 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 seu próprio id aleatório
ACK_TOPIC = f"{PREFIX}/svc/acks"
MAX_REPAIRS = 2

live_state = {}   # preenchido a partir dos topics de estado retidos
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)   # aguarda a chegada do estado retido


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 executar um 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."))

Repare no que acontece com esse pedido de exemplo. Se a V-12 estiver atualmente em 40 por cento, a verificação semântica rejeita o salto de 50 pontos e diz ao modelo que proponha um valor intermediário. O modelo normalmente tenta de novo com 65, que passa e é publicado. Quaisquer chamadas válidas de um lote saem imediatamente, e apenas as rejeitadas são devolvidas para correção.

Modos estritos e decodificação restrita

Vários provedores de modelos e runtimes locais conseguem restringir a geração para que os argumentos das ferramentas sempre correspondam ao JSON schema declarado. Quando disponível, ative esse recurso: ele elimina a maioria dos erros de tipo e campos ausentes antes que cheguem ao seu validador. Mas ele não elimina a necessidade de validação, por três motivos. Primeiro, a decodificação restrita cobre apenas o subconjunto do JSON Schema que o runtime suporta, e palavras-chave como faixas numéricas ou padrões de string podem ser ignoradas. Segundo, uma chamada perfeitamente formada ainda pode estar errada para a situação atual, como um passo de válvula grande demais ou uma bomba em serviço crítico. Terceiro, você pode trocar de modelo ou de provedor no futuro, e a sua segurança não deve depender de um recurso de um único runtime. Trate a saída restrita como uma melhoria de qualidade e a validação no código como a garantia.

Passo 4: validar de novo no lado do dispositivo

O serviço de comandos não é a última linha de defesa. O gateway ou dispositivo que recebe .../cmd deve validar o mesmo schema, verificar se a mensagem não expirou e responder com uma confirmação no response topic:

# valve_gateway.py (lado do 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()

Instale as dependências com pip install ollama jsonschema "paho-mqtt>=2.0", inicie o gateway e depois o serviço de comandos.

Testando as definições de funções antes da implantação

As definições de funções merecem a mesma disciplina de testes de qualquer outra interface com equipamentos físicos. Monte uma suíte de testes em duas camadas:

1. Testes unitários do validador, sem modelo. Alimente check() com conjuntos de argumentos escritos à mão, válidos e inválidos, e verifique o resultado. Inclua valores de fronteira, tipos errados, IDs desconhecidos, campos extras e estado ao vivo obsoleto ou ausente.
2. Testes de comportamento do modelo. Mantenha uma lista de pedidos realistas com a função e os argumentos esperados, como "feche a V-11 completamente" ou "pare a bomba principal". Execute-os no modelo com temperatura 0 e registre com que frequência a primeira tentativa é válida, com que frequência o loop de correção a conserta e com que frequência termina em rejeição.

Repita os testes de comportamento sempre que mudar uma descrição, adicionar uma função ou trocar de modelo. Pequenas mudanças na redação de uma descrição podem alterar qual função o modelo escolhe, e você quer descobrir isso nos testes, e não no chão de fábrica. Registrar as chamadas rejeitadas em produção fornece novos casos de teste ao longo do tempo.

Checklist para function calling de LLM em IoT em produção

1. Os schemas são versionados e compartilhados entre o serviço de comandos e os dispositivos.
2. additionalProperties: false, enums para IDs e limites para todo número.
3. Validação estrutural com uma biblioteca de JSON Schema e, em seguida, verificações semânticas contra o estado ao vivo.
4. Um loop de correção limitado, com mensagens de erro claras e legíveis pelo modelo.
5. Os comandos carregam um ID, uma expiração e um response topic; o sucesso só é informado após a confirmação.
6. Funções de alto impacto exigem aprovação humana; funções de segurança nunca são expostas.
7. Toda chamada, rejeição e confirmação é registrada.

A demonstração usa o servidor MQTT público (gratuito) em broker.connectmq.com, adequado apenas para testes e protótipos. Para equipamentos reais, como as válvulas e bombas de uma estação de saneamento ou de uma usina, mude para a porta 8883 com TLS e o usuário, a senha e o client ID gerados para o seu broker; um broker MQTT gerenciado como o ConnectMQ cuida para você dos certificados, do suporte a MQTT 5 e das cotas de conexão. Para o design geral do agente, veja agentes LLM para dispositivos IoT; para uma interface mais amigável, veja controle de dispositivos por linguagem natural; e, para um contexto mais amplo, nosso guia de IA e IoT. A especificação do JSON Schema documenta todas as palavras-chave usadas acima.

Comece com o ConnectMQ

O ConnectMQ suporta os response topics, o correlation data e a expiração de mensagens do MQTT 5 dos quais esse padrão depende. Cadastre-se no plano Serverless gratuito, sem cartão de crédito, e execute hoje mesmo sua primeira chamada de função validada contra um broker real.

Perguntas frequentes

O JSON schema que envio ao modelo basta para manter os comandos seguros?

Não. Ele orienta o modelo, mas não garante uma saída em conformidade. Sempre valide os argumentos no seu próprio código e, novamente, no lado do dispositivo.

O que deve acontecer quando a chamada de função do modelo for inválida?

Devolva ao modelo uma mensagem de erro clara e permita uma ou duas tentativas de correção. Se ainda falhar, informe a rejeição ao usuário em vez de tentar adivinhar.

Por que usar MQTT 5 para comandos?

Os response topics e o correlation data oferecem uma forma padronizada de associar confirmações a comandos, e a expiração de mensagens impede que os dispositivos executem comandos obsoletos.

Um único pedido pode gerar várias chamadas de função?

Sim. Os modelos podem retornar várias chamadas de ferramenta em uma resposta. Valide cada uma de forma independente e publique apenas as que passarem.