RAG para Dados Industriais: Combine Manuais de Equipamentos com Telemetria MQTT ao Vivo

O RAG para dados industriais responde a perguntas como "por que o compressor da linha 3 está em alarme e o que devo verificar?" combinando duas fontes que um técnico normalmente precisa conciliar sozinho: a documentação do equipamento e as leituras atuais da máquina. Este guia mostra como indexar manuais e procedimentos, buscar valores ao vivo em topics MQTT no momento da pergunta e montar um prompt que gere respostas fundamentadas e com citações, além das armadilhas que importam no chão de fábrica.

Por que o RAG simples não basta em uma planta

A geração aumentada por recuperação (RAG) busca trechos relevantes nos seus documentos e os entrega a um modelo de linguagem para que ele responda com base no seu conteúdo, e não no conhecimento geral. Para documentos de escritório, isso resolve tudo. Na indústria, não é suficiente, porque a maioria das perguntas úteis depende do estado atual de uma máquina:

• "Este nível de vibração é normal para esta bomba?" precisa tanto do limite do manual quanto do valor atual.
• "O que significa o código de falha E-214 e ele ainda está ativo?" precisa da tabela de falhas e do estado ao vivo do alarme.
• "Posso religar o chiller agora?" precisa do procedimento de religamento e das pressões e temperaturas atuais.

Um modelo só com documentos vai citar limites sem conhecer o valor. Um modelo só com telemetria vai ver números sem saber o que eles significam. Combinar os dois é o que torna um assistente útil para um técnico de manutenção.

Arquitetura de RAG para dados industriais

O registro de ativos é a peça que a maioria dos tutoriais deixa de fora e a que faz o sistema funcionar. Quando um técnico pergunta sobre o "compressor C-3", o registro informa qual manual se aplica (os números de modelo diferem entre máquinas que, de resto, são parecidas) e quais topics trazem suas leituras.

{
  "C-3": {
    "description": "Screw air compressor, line 3",
    "model": "SC-75",
    "documents": ["sc75-operating-manual", "sc75-fault-codes", "wo-history-c3"],
    "topics": {
      "discharge_temp_c": "connectmq/demo/8b2e61/plant1/line3/c3/discharge_temp",
      "discharge_pressure_bar": "connectmq/demo/8b2e61/plant1/line3/c3/pressure",
      "active_fault": "connectmq/demo/8b2e61/plant1/line3/c3/fault"
    }
  }
}

Uma estrutura de topics que segue os níveis da ISA-95 (empresa, site, área, linha, célula) torna esse mapeamento quase mecânico. Se as suas máquinas publicam Sparkplug B, o ConnectMQ consegue decodificá-lo e republicar as métricas em topics ISA-95 legíveis, que um sistema RAG pode consumir sem um decodificador Protobuf. Nosso artigo sobre Sparkplug B e o Unified Namespace explica esse modelo de topics.

Passo 1: preparar e indexar os documentos

A preparação dos documentos define a qualidade das respostas mais do que a escolha do modelo.

1. Extraia texto limpo. Converta PDFs em texto e confira se as tabelas sobreviveram. As tabelas de códigos de falha costumam ser o conteúdo mais valioso e o mais propenso a ser corrompido.
2. Divida pela estrutura. Separe nos títulos e nas etapas dos procedimentos, não em contagens fixas de caracteres, para que um procedimento fique inteiro. Mantenha os chunks com algumas centenas de palavras.
3. Anexe metadados. Armazene com cada chunk o ID do documento, o título, a seção, a página e o modelo do equipamento.
4. Gere embeddings e armazene. Calcule um embedding por chunk e guarde-o junto com os metadados.

O exemplo abaixo usa a API de embeddings do Ollama com um modelo de embeddings local e mantém o índice em NumPy. Instale com pip install ollama numpy "paho-mqtt>=2.0" e ollama pull nomic-embed-text.

# index_docs.py
import json
import re
from pathlib import Path

import numpy as np
import ollama

EMBED_MODEL = "nomic-embed-text"
DOCS = Path("docs")          # um arquivo .md ou .txt por documento, chamado <doc-id>.md


def chunks_for(path):
    text = path.read_text(encoding="utf-8")
    sections = re.split(r"\n(?=#{1,3} )", text)   # divide antes dos títulos markdown
    for i, section in enumerate(sections):
        section = section.strip()
        if len(section) < 40:
            continue
        title = section.splitlines()[0].lstrip("# ").strip()
        yield {"doc": path.stem, "section": title, "chunk": i, "text": section[:3000]}


records = [c for p in sorted(DOCS.glob("*.md")) for c in chunks_for(p)]
vectors = []
for start in range(0, len(records), 32):
    batch = records[start:start + 32]
    resp = ollama.embed(model=EMBED_MODEL, input=["search_document: " + r["text"] for r in batch])
    vectors.extend(resp.embeddings)

matrix = np.array(vectors, dtype=np.float32)
matrix /= np.linalg.norm(matrix, axis=1, keepdims=True)
np.save("index.npy", matrix)
Path("index.json").write_text(json.dumps(records), encoding="utf-8")
print(f"indexed {len(records)} chunks")

Os prefixos search_document: e search_query: são as instruções de tarefa que esse modelo de embeddings específico espera; outros modelos de embeddings têm suas próprias convenções, então confira o model card daquele que você usar.

Passo 2: manter um cache de valores ao vivo a partir do MQTT

O cache assina os topics dos ativos e armazena o valor mais recente com seu timestamp. Como os dispositivos publicam valores com retained message, o cache fica completo poucos instantes depois de iniciar.

# live_cache.py
import json
import threading
import time
import paho.mqtt.client as mqtt

_cache, _lock = {}, threading.Lock()


def start(topics, client_id="rag-cache-8b2e61", host="broker.connectmq.com", port=1883):
    def on_connect(client, userdata, flags, reason_code, properties):
        for t in topics:
            client.subscribe(t, qos=1)

    def on_message(client, userdata, msg):
        raw = msg.payload.decode("utf-8", errors="replace")
        try:
            value = json.loads(raw)
        except json.JSONDecodeError:
            value = raw
        with _lock:
            _cache[msg.topic] = {"value": value, "receivedAt": time.time()}

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=client_id)
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(host, port, keepalive=60)
    client.loop_start()
    return client


def snapshot(topic_map):
    now = time.time()
    out = {}
    with _lock:
        for name, topic in topic_map.items():
            entry = _cache.get(topic)
            out[name] = None if entry is None else {
                "value": entry["value"], "ageSeconds": round(now - entry["receivedAt"], 1)}
    return out

Informar a idade de cada valor é essencial. Uma leitura que chegou duas horas atrás não é "atual", e o modelo deve dizer isso.

Passo 3: recuperar, combinar e responder

No momento da pergunta, identifique o ativo, recupere trechos apenas dos documentos desse ativo, leia seus valores ao vivo e monte um prompt em que cada peça esteja identificada.

# ask.py
import json
from pathlib import Path

import numpy as np
import ollama

import live_cache

EMBED_MODEL, CHAT_MODEL = "nomic-embed-text", "llama3.1:8b"
REGISTRY = json.loads(Path("assets.json").read_text())
RECORDS = json.loads(Path("index.json").read_text(encoding="utf-8"))
MATRIX = np.load("index.npy")

live_cache.start([t for a in REGISTRY.values() for t in a["topics"].values()])

SYSTEM = """You assist maintenance technicians.
Answer using ONLY the DOCUMENT passages and LIVE values provided.
Cite documents as [doc-id, section] and live values as [live: name].
If a live value is missing or older than 300 seconds, say it is unavailable or stale.
If the documents do not cover the question, say so. Never invent limits, codes or procedures.
Safety instructions in the documents (lockout, PPE) must be repeated, not shortened."""


def retrieve(question, docs, k=4):
    q = ollama.embed(model=EMBED_MODEL, input=["search_query: " + question]).embeddings[0]
    q = np.array(q, dtype=np.float32)
    q /= np.linalg.norm(q)
    scores = MATRIX @ q
    order = [i for i in np.argsort(-scores) if RECORDS[i]["doc"] in docs]
    return [RECORDS[i] | {"score": float(scores[i])} for i in order[:k]]


def ask(asset_id, question):
    asset = REGISTRY[asset_id]
    passages = retrieve(question, set(asset["documents"]))
    live = live_cache.snapshot(asset["topics"])
    context = "\n\n".join(f"DOCUMENT [{p['doc']}, {p['section']}]\n{p['text']}" for p in passages)
    live_text = "\n".join(f"LIVE [{k}]: {json.dumps(v)}" for k, v in live.items())
    prompt = (f"ASSET: {asset_id} - {asset['description']} (model {asset['model']})\n\n"
              f"{live_text}\n\n{context}\n\nQUESTION: {question}")
    resp = ollama.chat(model=CHAT_MODEL, options={"temperature": 0.1},
                       messages=[{"role": "system", "content": SYSTEM},
                                 {"role": "user", "content": prompt}])
    return resp.message.content


if __name__ == "__main__":
    import time
    time.sleep(2)   # aguarda a chegada dos valores retidos
    print(ask("C-3", "Discharge temperature looks high. Is it within limits and what should I check first?"))

Para testar sem equipamentos reais, publique valores de teste com retained message:

mosquitto_pub -h broker.connectmq.com -p 1883 -i seed-8b2e61 -r -q 1 -t "connectmq/demo/8b2e61/plant1/line3/c3/discharge_temp" -m "104.5"
mosquitto_pub -h broker.connectmq.com -p 1883 -i seed-8b2e61 -r -q 1 -t "connectmq/demo/8b2e61/plant1/line3/c3/pressure" -m "7.9"
mosquitto_pub -h broker.connectmq.com -p 1883 -i seed-8b2e61 -r -q 1 -t "connectmq/demo/8b2e61/plant1/line3/c3/fault" -m "\"none\""

Não há garantia de que dados retidos no servidor MQTT público persistam, então publique os valores de novo se eles desaparecerem, e nunca coloque dados reais da planta no servidor MQTT público.

Armadilhas específicas do RAG industrial

1. Manual errado para a máquina certa. Máquinas semelhantes com firmware ou números de modelo diferentes têm limites diferentes. Filtre a recuperação pela lista de documentos do ativo, como acima, em vez de pesquisar em tudo.
2. Unidades e escalas. Se um topic publica contagens brutas ou Fahrenheit enquanto o manual usa Celsius, o modelo vai comparar alhos com bugalhos. Normalize as unidades no cache e indique-as no nome do campo (discharge_temp_c).
3. Valores obsoletos. Sempre inclua timestamps ou idades e instrua o modelo a tratar valores antigos como indisponíveis.
4. Tabelas perdidas na extração. Verifique as tabelas de códigos de falha após a conversão do PDF; considere armazená-las como JSON estruturado e consultar os códigos diretamente, em vez de via embeddings.
5. Conteúdo de segurança. As etapas de bloqueio e etiquetagem (lockout/tagout) nunca podem ser eliminadas em um resumo. Diga isso no prompt de sistema e teste.
6. Nenhum conjunto de avaliação. Reúna de 30 a 50 perguntas reais de técnicos com as respostas esperadas e execute-as novamente sempre que mudar a divisão em chunks, os modelos ou os prompts.

Do protótipo à planta

O caminho de leitura dos valores ao vivo é leve: um subscriber por instância do assistente, com algumas poucas assinaturas com curinga. Executá-lo em um broker MQTT na nuvem que já transporta o seu Unified Namespace significa que o assistente vê os mesmos dados que os dashboards, os historiadores e um eventual gêmeo digital da planta, via TLS e com credenciais próprias. O ConnectMQ oferece isso como um broker MQTT gerenciado, com decodificação de Sparkplug B e uma visão de dispositivos ISA-95 no portal para conferir quais topics um ativo publica. Durante o desenvolvimento, o servidor MQTT público basta para valores de teste publicados manualmente.

Para levar o assistente adiante, deixe-o buscar valores sob demanda por meio de um servidor MCP para MQTT, transforme perguntas recorrentes em resumos de dados de sensores com LLM agendados ou explore um copiloto de IA industrial completo. Nosso guia de IA e IoT coloca esses padrões em contexto.

Comece com o ConnectMQ

Alimente seu assistente RAG com telemetria ao vivo e bem estruturada do ConnectMQ, com retained messages, detecção de topics ISA-95 e decodificação de Sparkplug B integradas. Crie um broker Serverless gratuito, sem cartão de crédito, e fundamente sua primeira resposta em dados reais de máquinas.

Perguntas frequentes

Qual é a diferença entre RAG e fine-tuning para dados industriais?

O RAG recupera seus documentos e valores ao vivo no momento da pergunta, então as respostas ficam atualizadas e podem citar as fontes. O fine-tuning altera os pesos do modelo e não tem como conhecer o estado atual de uma máquina.

Por que buscar valores ao vivo no MQTT em vez de em um banco de dados?

As retained messages do MQTT entregam instantaneamente o valor mais recente de cada topic, que é exatamente o que uma pergunta sobre o estado atual precisa. Perguntas históricas continuam pertencendo a um banco de dados de séries temporais.

Preciso de um banco de dados vetorial?

Não para alguns milhares de chunks; um índice NumPy em memória é rápido o suficiente. Passe para um banco de dados vetorial quando as coleções crescerem muito ou quando muitos usuários fizerem consultas ao mesmo tempo.

Como impedir que o assistente invente limites?

Instrua-o a responder apenas com base nos trechos e valores ao vivo fornecidos, exija citações, filtre a recuperação pelos documentos corretos do equipamento e teste com um conjunto de avaliação fixo.