Ponte Ollama MQTT: Execute um LLM Local no Seu Gateway e Publique Insights de IA

Combinar Ollama e MQTT permite executar um modelo de linguagem ao lado das suas máquinas, como as bombas de uma estação de saneamento ou os compressores de um frigorífico, e alimentá-lo com dados de sensores ao vivo, sem enviar leituras brutas a um serviço de IA externo. Neste tutorial você vai instalar o Ollama em um gateway Linux, construir uma ponte em Python que assina topics de telemetria, pede a um modelo local insights estruturados e publica os resultados de volta no broker para dashboards e alertas.

Por que executar o LLM no gateway?

Um modelo local nem sempre é a escolha certa, mas em um gateway de borda ele resolve problemas reais:

• Os dados ficam no local. As leituras são processadas localmente; apenas o insight curto sai do gateway, se você optar por publicá-lo para fora.
• Sem dependência de serviços de IA na internet. A análise continua funcionando durante quedas de conexão e não depende de cotas de APIs externas.
• Custo previsível. Com o hardware instalado, não há cobrança por requisição.

As contrapartidas são modelos menores, geração mais lenta em hardware modesto e o trabalho de manter os modelos atualizados. Para explicações, classificações e resumos curtos de dados numéricos, modelos pequenos geralmente bastam. Nossa comparação entre IA na borda e IA na nuvem aprofunda essa decisão.

Arquitetura

sensors --MQTT--> broker --> ollama_bridge.py --HTTP localhost:11434--> Ollama
                    ^                |
                    |                v
                    +---- .../insights (JSON) <--+

A ponte é um client MQTT comum. Ela mantém o loop de rede rodando em uma thread em segundo plano, coloca as mensagens interessantes em uma fila e as processa uma de cada vez com o modelo. Essa separação é a decisão de projeto mais importante deste tutorial: uma chamada ao LLM pode levar vários segundos, e bloquear o callback MQTT por tanto tempo causa perda de keepalives e desconexões.

Passo 1: instalar o Ollama e baixar um modelo

Em um gateway Linux de 64 bits:

curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.2:3b
ollama run llama3.2:3b "Reply with the word ready."

O instalador registra um serviço systemd, então a API do Ollama escuta em http://localhost:11434 após a reinicialização. Mantenha-a vinculada ao localhost; a ponte roda na mesma máquina e não há motivo para expor a API do modelo à rede.

Instale as dependências Python em um ambiente virtual:

python3 -m venv ~/ollama-mqtt/.venv
source ~/ollama-mqtt/.venv/bin/activate
pip install ollama "paho-mqtt>=2.0"

Passo 2: definir o formato de um insight

Texto livre é difícil de usar nas etapas seguintes. O Ollama suporta saídas estruturadas: você passa um JSON schema no parâmetro format e a resposta do modelo fica restrita a esse formato. Dashboards e regras de alerta podem então ler campos como severity diretamente.

{
  "type": "object",
  "properties": {
    "severity": { "type": "string", "enum": ["normal", "watch", "warning", "critical"] },
    "summary": { "type": "string" },
    "likely_causes": { "type": "array", "items": { "type": "string" }, "maxItems": 3 },
    "recommended_check": { "type": "string" }
  },
  "required": ["severity", "summary", "recommended_check"]
}

Passo 3: construir a ponte Ollama MQTT

A ponte assina a telemetria, mantém um histórico curto por dispositivo e dispara uma análise quando uma leitura viola uma regra simples ou quando o intervalo de análise do dispositivo termina. Substitua 4c7a2e pela sua própria string aleatória.

# ollama_bridge.py
import json
import logging
import queue
import threading
import time
from collections import defaultdict, deque

import ollama
import paho.mqtt.client as mqtt

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

BROKER, PORT = "broker.connectmq.com", 1883
PREFIX = "connectmq/demo/4c7a2e/site1"
MODEL = "llama3.2:3b"
HISTORY = 30                 # leituras mantidas por dispositivo
MIN_INTERVAL = 120           # segundos entre análises por dispositivo
RULES = {"temp_c": 70.0, "vib_mm_s": 5.0}

SCHEMA = {
    "type": "object",
    "properties": {
        "severity": {"type": "string", "enum": ["normal", "watch", "warning", "critical"]},
        "summary": {"type": "string"},
        "likely_causes": {"type": "array", "items": {"type": "string"}, "maxItems": 3},
        "recommended_check": {"type": "string"},
    },
    "required": ["severity", "summary", "recommended_check"],
}

SYSTEM = (
    "You analyze industrial sensor readings for maintenance technicians. "
    "Base your answer only on the readings provided. "
    "Phrase causes as possibilities, never as certainties. Keep the summary under 40 words."
)

history = defaultdict(lambda: deque(maxlen=HISTORY))
last_analysis = defaultdict(float)
work = queue.Queue(maxsize=50)


def on_connect(client, userdata, flags, reason_code, properties):
    log.info("connected: %s", reason_code)
    client.subscribe(f"{PREFIX}/+/telemetry", qos=1)


def on_message(client, userdata, msg):
    device = msg.topic.split("/")[-2]
    try:
        reading = json.loads(msg.payload)
    except json.JSONDecodeError:
        return
    history[device].append(reading)
    breached = any(isinstance(reading.get(k), (int, float)) and reading[k] > v for k, v in RULES.items())
    due = time.time() - last_analysis[device] > MIN_INTERVAL
    if (breached or due) and len(history[device]) >= 5:
        last_analysis[device] = time.time()
        try:
            work.put_nowait((device, list(history[device]), breached))
        except queue.Full:
            log.warning("queue full, skipping %s", device)


def analyze(device, readings, breached):
    prompt = (
        f"Device: {device}\nRule thresholds: {json.dumps(RULES)}\n"
        f"Triggered by threshold breach: {breached}\n"
        f"Last {len(readings)} readings, oldest first:\n{json.dumps(readings)}"
    )
    resp = ollama.chat(
        model=MODEL,
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": prompt}],
        format=SCHEMA,
        options={"temperature": 0},
    )
    return json.loads(resp.message.content)


def worker(client):
    while True:
        device, readings, breached = work.get()
        started = time.time()
        try:
            insight = analyze(device, readings, breached)
        except Exception as exc:
            log.error("analysis failed for %s: %s", device, exc)
            continue
        insight.update({
            "device": device,
            "model": MODEL,
            "trigger": "threshold" if breached else "interval",
            "durationS": round(time.time() - started, 1),
            "ts": int(time.time() * 1000),
        })
        client.publish(f"{PREFIX}/{device}/insights", json.dumps(insight), qos=1, retain=True)
        log.info("%s -> %s: %s", device, insight["severity"], insight["summary"])


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="ollama-bridge-4c7a2e")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, keepalive=60)
threading.Thread(target=worker, args=(client,), daemon=True).start()
client.loop_forever()

Como ela se comporta:

1. Cada mensagem de telemetria é armazenada em um histórico limitado por dispositivo.
2. Uma violação de limite ou um intervalo vencido coloca um trabalho de análise na fila, mas nunca mais de um por dispositivo por intervalo.
3. Uma única thread de trabalho chama o Ollama, para que a CPU do gateway não fique sobrecarregada com gerações paralelas.
4. O insight é publicado como retained message, então os dashboards mostram a avaliação mais recente assim que fazem a assinatura.

Escolhendo um modelo para o gateway

A escolha do modelo em um gateway é um equilíbrio entre qualidade, velocidade e memória. Algumas orientações práticas:

1. Comece pequeno. Um modelo de 3B é um padrão sensato para insights estruturados a partir de leituras curtas. Suba de tamanho apenas se as respostas forem claramente superficiais demais para o seu caso de uso.
2. Prefira versões quantizadas. As tags padrão da biblioteca do Ollama já são quantizadas, e é isso que torna viável o hardware pequeno. Evite variantes de precisão total em gateways só com CPU.
3. Deixe memória para todo o resto. O gateway ainda executa a ponte MQTT, talvez um broker local, conversores de protocolo e o sistema operacional. Acompanhe a memória com free -h enquanto uma geração estiver em andamento.
4. Teste com seus próprios dados. Colete algumas dezenas de janelas reais de telemetria, incluindo incidentes conhecidos, e compare como os modelos candidatos as descrevem. Para a operação, consistência importa mais do que eloquência.
5. Fixe a versão do modelo. Registre a tag do modelo em cada insight publicado, como a ponte faz, para saber qual modelo produziu determinada recomendação depois de uma atualização.

Passo 4: alimentar com dados de teste

Publique algumas leituras normais e depois uma quente usando o CLI do Mosquitto:

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-4c7a2e -q 1 \
    -t "connectmq/demo/4c7a2e/site1/pump-2/telemetry" \
    -m "{\"temp_c\": 6$i.0, \"vib_mm_s\": 2.$i, \"ts\": $(date +%s)000}"
  sleep 1
done
mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-4c7a2e -q 1 \
  -t "connectmq/demo/4c7a2e/site1/pump-2/telemetry" \
  -m "{\"temp_c\": 74.5, \"vib_mm_s\": 5.8, \"ts\": $(date +%s)000}"

Acompanhe a chegada dos insights:

mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-4c7a2e -t "connectmq/demo/4c7a2e/site1/+/insights" -v

Você deve ver um insight em JSON com uma severidade como warning, um resumo curto mencionando o aumento de temperatura e vibração e uma verificação recomendada. A redação exata varia de acordo com o modelo.

Ajustando a configuração Ollama MQTT

Essa última linha ganha importância à medida que o volume de dados cresce. Em vez de leituras brutas, passe mínimo, máximo, média e tendência calculados em Python; nosso tutorial sobre resumos de dados de sensores com LLM mostra em detalhes essa abordagem de fatos primeiro.

Indo para produção

No gateway, execute a ponte como um serviço systemd ao lado do Ollama e conecte-a a um broker privado via TLS:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)

O broker pode ser um broker local no gateway, um broker MQTT gerenciado como o ConnectMQ, ou ambos, com o gateway publicando para fora apenas os insights. Um endpoint na nuvem é útil quando várias unidades, como plantas em Campinas e em Manaus, precisam compartilhar insights com dashboards centrais: o portal do ConnectMQ os mostra no explorador de topics ao vivo assim que chegam. O servidor MQTT público (gratuito) é perfeito para este tutorial e para protótipos, mas não para produção: seus topics podem ser lidos por qualquer pessoa, então mantenha dados reais da planta longe dele.

Próximos passos: permita que os operadores façam perguntas sobre os mesmos dados com um servidor MCP para MQTT ou deixe o modelo agir nos dispositivos com segurança usando os padrões de controle de dispositivos por linguagem natural. Para uma visão geral de onde os modelos locais se encaixam entre outras técnicas de IA, veja nosso guia de IA e IoT.

Comece com o ConnectMQ

Compartilhe os insights de IA do seu gateway entre sites com o ConnectMQ, um broker MQTT gerenciado com MQTT 5, TLS e WebSockets para dashboards no navegador. Crie um broker no plano Serverless (Free), sem cartão de crédito, e publique seu primeiro insight de LLM local em minutos.

Perguntas frequentes

Um Raspberry Pi consegue executar o Ollama?

Sim, um Raspberry Pi 5 com 8 GB de RAM consegue executar modelos pequenos na CPU. As respostas levam segundos, então analise eventos e janelas periódicas em vez de cada mensagem.

O Ollama fala MQTT diretamente?

Não. O Ollama expõe uma API HTTP no localhost. Um pequeno programa de ponte se conecta ao broker como client MQTT e chama a API do Ollama, como mostrado neste tutorial.

Como obter JSON confiável do modelo?

Passe um JSON schema no parâmetro format do Ollama para usar saídas estruturadas, defina a temperatura como 0 e, mesmo assim, valide o resultado antes de publicar.

Toda mensagem MQTT deve disparar uma chamada ao LLM?

Não. Dispare a análise em violações de limite ou em intervalos e limite as chamadas por dispositivo. Modelos de linguagem são muito mais lentos do que a taxa de mensagens de um topic movimentado.