Resumos de Dados de Sensores com LLM: Relatórios de Turno e Explicações de Anomalias a partir do MQTT

Resumos de dados de sensores com LLM transformam milhares de leituras brutas em alguns parágrafos que um operador consegue ler na passagem de turno: o que funcionou normalmente, o que desviou e o que precisa de atenção. Neste tutorial você vai coletar telemetria de topics MQTT, calcular as estatísticas em Python e pedir a um modelo de linguagem que escreva um relatório de turno e explicações de anomalias em linguagem simples, baseados apenas nesses fatos calculados.

Por que não enviar os dados brutos direto para o modelo?

É tentador colar um dia inteiro de leituras em um prompt e pedir um resumo. Na prática, isso falha de três maneiras:

• A aritmética não é confiável. Modelos de linguagem são bons com linguagem e fracos em calcular médias sobre listas longas. Eles vão produzir números plausíveis, porém errados.
• O contexto é limitado e caro. Telemetria de alta frequência rapidamente ultrapassa o que cabe em um prompt, e cada token custa tempo.
• Os resultados não são auditáveis. Se um número do relatório estiver errado, não dá para saber de onde ele veio.

O padrão confiável divide o trabalho. O código calcula tudo o que é número: contagens, médias, mínimos, máximos, tempo acima do limite, lacunas nos dados. O modelo recebe um documento JSON compacto com os fatos e o transforma em texto corrido. Cada número do relatório pode ser rastreado até o documento de fatos.

Arquitetura

sensors --MQTT--> broker --> collector.py (window buffer, stats) --> facts JSON
                                                              |
                                                              v
                                                   LLM (summary prompt)
                                                              |
                         broker <-- reports/shift (retained) <-+

O relatório é publicado de volta no broker em um topic com retained message, para que um dashboard, uma ponte para chatbot ou um serviço de impressão possa pegar o relatório mais recente a qualquer momento.

Passo 1: publicar alguma telemetria de teste

Se você ainda não tem dispositivos, este simulador publica temperatura, vibração e pressão de duas máquinas a cada dois segundos, com um pico ocasional de vibração. Substitua b62d1f pela sua própria string aleatória.

# sim_line.py
import json
import random
import time
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/b62d1f/line1"
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="sim-line-b62d1f")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    for machine in ("press-1", "press-2"):
        vib = random.gauss(2.0, 0.2)
        if machine == "press-2" and random.random() < 0.05:
            vib += random.uniform(3, 5)  # pico ocasional
        payload = {
            "temp_c": round(random.gauss(58 if machine == "press-1" else 63, 1.5), 2),
            "vib_mm_s": round(vib, 2),
            "pressure_bar": round(random.gauss(5.0, 0.1), 2),
            "ts": int(time.time() * 1000),
        }
        client.publish(f"{PREFIX}/{machine}/telemetry", json.dumps(payload), qos=0)
    time.sleep(2)

Passo 2: coletar e calcular os fatos

O coletor assina com um curinga, armazena as leituras por máquina e métrica e, ao final de cada janela, monta o documento de fatos. Os limites vêm da configuração, não do modelo.

# collector.py
import json
import statistics
import threading
import time
from collections import defaultdict

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/b62d1f/line1"
THRESHOLDS = {"temp_c": 65.0, "vib_mm_s": 4.5, "pressure_bar": 5.4}
UNITS = {"temp_c": "C", "vib_mm_s": "mm/s", "pressure_bar": "bar"}

buffer = defaultdict(list)       # (machine, metric) -> [(ts, value)]
last_seen = {}
lock = threading.Lock()


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/+/telemetry", qos=0)


def on_message(client, userdata, msg):
    machine = msg.topic.split("/")[-2]
    try:
        data = json.loads(msg.payload)
    except json.JSONDecodeError:
        return
    ts = data.get("ts", int(time.time() * 1000))
    with lock:
        last_seen[machine] = ts
        for metric in THRESHOLDS:
            if isinstance(data.get(metric), (int, float)):
                buffer[(machine, metric)].append((ts, float(data[metric])))


def build_facts(window_start, window_end):
    with lock:
        items = {k: list(v) for k, v in buffer.items()}
        buffer.clear()
        seen = dict(last_seen)
    machines = defaultdict(dict)
    for (machine, metric), points in items.items():
        values = [v for _, v in points]
        limit = THRESHOLDS[metric]
        breaches = [(ts, v) for ts, v in points if v > limit]
        machines[machine][metric] = {
            "unit": UNITS[metric],
            "count": len(values),
            "mean": round(statistics.fmean(values), 2),
            "min": round(min(values), 2),
            "max": round(max(values), 2),
            "stdev": round(statistics.pstdev(values), 2),
            "threshold": limit,
            "breaches": len(breaches),
            "firstBreach": time.strftime("%H:%M:%S", time.localtime(breaches[0][0] / 1000)) if breaches else None,
            "peakDuringBreach": round(max(v for _, v in breaches), 2) if breaches else None,
        }
    stale = [m for m, ts in seen.items() if window_end * 1000 - ts > 60_000]
    return {
        "line": "line1",
        "window": {
            "start": time.strftime("%Y-%m-%d %H:%M", time.localtime(window_start)),
            "end": time.strftime("%Y-%m-%d %H:%M", time.localtime(window_end)),
        },
        "machines": machines,
        "staleMachines": stale,
    }

Guardar a contagem de violações, o horário da primeira violação e o valor de pico permite que o modelo explique uma anomalia de forma concreta ("a vibração na press-2 ultrapassou 4,5 mm/s 6 vezes, a primeira às 14:12, com pico de 6,8") sem inventar nada.

Passo 3: pedir ao LLM que escreva o resumo

O prompt importa tanto quanto os dados. Diga ao modelo para que serve o relatório, o que ele pode e não pode fazer e qual formato a saída deve ter. Este exemplo usa um modelo local via Ollama, mas qualquer API de chat funciona com o mesmo prompt.

# summarize.py
import json
import ollama

MODEL = "llama3.1:8b"

INSTRUCTIONS = """You write shift handover reports for production operators.
Use ONLY the numbers in the FACTS JSON. Do not calculate new numbers and do not guess causes as facts.
Structure:
1. One-sentence overall status.
2. 'Needs attention' bullet list: every metric with breaches > 0 and every stale machine. Quote count, first breach time and peak.
3. 'Normal operation' bullet list: one short line per machine.
4. 'Possible explanations' (max 3 bullets): plausible causes worded as suggestions to check, e.g. 'check ...'.
Keep it under 200 words. Plain text, no tables."""


def write_report(facts: dict) -> str:
    resp = ollama.chat(
        model=MODEL,
        messages=[
            {"role": "system", "content": INSTRUCTIONS},
            {"role": "user", "content": "FACTS:\n" + json.dumps(facts, indent=1)},
        ],
        options={"temperature": 0.2},
    )
    return resp.message.content

Passo 4: executar o loop de janelas e publicar o relatório

# main.py
import json
import time

import paho.mqtt.client as mqtt

import collector
from summarize import write_report

WINDOW_SECONDS = 600   # 10 minutos para testes; use 8 horas para um turno real
REPORT_TOPIC = f"{collector.PREFIX}/reports/shift"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="summarizer-b62d1f")
client.on_connect = collector.on_connect
client.on_message = collector.on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

while True:
    start = time.time()
    time.sleep(WINDOW_SECONDS)
    facts = collector.build_facts(start, time.time())
    if not facts["machines"]:
        continue
    report = write_report(facts)
    message = {"facts": facts, "report": report, "generatedAt": int(time.time() * 1000)}
    client.publish(REPORT_TOPIC, json.dumps(message), qos=1, retain=True)
    print(report)

A chamada ao LLM é lenta em comparação com o tráfego MQTT, mas, como o paho executa seu loop de rede em uma thread em segundo plano com loop_start(), as mensagens continuam chegando ao buffer enquanto o relatório é gerado. Publicar os fatos junto com o relatório significa que qualquer pessoa que o leia pode conferir cada número.

Leia o relatório mais recente de qualquer lugar:

mosquitto_sub -h broker.connectmq.com -p 1883 -i reader-b62d1f -t "connectmq/demo/b62d1f/line1/reports/shift" -C 1

Escolhendo janelas, limites e públicos

Os mesmos fatos podem gerar relatórios muito diferentes, dependendo de como você define a janela e de quem lê o resultado.

Os limites devem vir da mesma fonte dos seus alarmes, como especificações de equipamentos ou limites de processo acordados com a engenharia, e não do conhecimento geral do modelo. Se um limite mudar, o próximo relatório o reflete automaticamente, porque ele faz parte do documento de fatos.

Para relatórios orientados a eventos, dispare o resumo quando um alerta for acionado e envie apenas os fatos da máquina afetada nos últimos 30 a 60 minutos, mais a mesma janela do dia anterior como referência. O modelo então escreve uma explicação focada em vez de um relatório de turno completo. Manter o documento de fatos pequeno também deixa a geração rápida o suficiente para que a explicação chegue enquanto o evento ainda é relevante.

Escrevendo resumos de dados de sensores com LLM melhores

Algumas lições práticas que melhoram visivelmente os relatórios:

1. Dê nomes e contexto. "press-2 (prensa hidráulica, linha 1)" gera explicações melhores do que um ID sem contexto. Guarde as descrições na configuração e adicione-as aos fatos.
2. Classifique os problemas antes. Ordene as máquinas por severidade no código antes de enviar. O modelo geralmente mantém a sua ordem, então o item mais importante aparece primeiro.
3. Inclua como é o normal. Adicionar as médias do turno anterior permite que o modelo diga "a temperatura está 3 graus acima do último turno" usando números que você calculou.
4. Separe fatos de hipóteses. Peça as possíveis explicações como sugestões a verificar. Um modelo não tem como saber que um rolamento está falhando; ele pode sugerir que ele seja verificado.
5. Valide a saída. Uma verificação simples de que cada número do relatório também aparece no JSON de fatos pega a maioria dos valores inventados.
6. Tenha um plano B. Se o modelo estiver indisponível, publique um relatório baseado em template a partir dos mesmos fatos, para que o turno ainda receba um resumo.

Para detectar as anomalias em primeiro lugar, combine isso com um detector adequado do nosso guia de detecção de anomalias em IoT e deixe o resumo explicar o que o detector encontrou, em vez de fazer a detecção ele mesmo.

Onde os dados ficam

O coletor funciona com qualquer fonte MQTT. Em uma fábrica, você pode alimentá-lo a partir de gateways de CLP que publicam Sparkplug B; o ConnectMQ consegue decodificar Sparkplug B e republicá-lo em topics ISA-95 legíveis, que são muito mais fáceis de agregar e descrever. Para testes, o servidor MQTT público é suficiente, mas a telemetria de produção pertence a um broker MQTT na nuvem privado, com TLS e credenciais por dispositivo. Se quiser manter o modelo no local, execute-o em um gateway como mostrado em Ollama com MQTT e, para respostas que combinam valores ao vivo com manuais de equipamentos, veja RAG para dados industriais. O panorama mais amplo está no nosso guia de IA e IoT.

Comece com o ConnectMQ

Colete sua telemetria no ConnectMQ, um broker MQTT gerenciado com explorador de topics ao vivo, dashboard e regras de alerta por limite para as leituras que seus relatórios resumem. Abra um broker Serverless gratuito, sem cartão de crédito, e publique hoje mesmo seu primeiro relatório de turno escrito por IA.

Perguntas frequentes

O LLM pode calcular as estatísticas sozinho?

Ele pode tentar, mas os resultados não são confiáveis com muitos valores. Calcule as estatísticas no código e passe-as ao modelo, para que cada número do relatório seja exato e rastreável.

Com que frequência os resumos devem ser gerados?

Acompanhe o ritmo do negócio: por turno para a passagem de turno, de hora em hora para supervisores ou sob demanda quando um alerta for acionado. A janela do coletor simplesmente segue essa programação.

Qual modelo devo usar?

Um modelo local pequeno é suficiente para resumos curtos e estruturados a partir de um documento de fatos. Modelos maiores escrevem explicações mais fluentes; a abordagem de fatos primeiro funciona com qualquer um deles.

Como impedir que o modelo invente causas?

Instrua-o a formular as causas como itens a verificar, inclua apenas fatos no prompt e valide se os números da saída correspondem ao JSON de fatos.