RAG for Industrial Data: Combine Equipment Manuals with Live MQTT Telemetry

RAG for industrial data answers questions like "why is the compressor on line 3 alarming and what should I check?" by combining two sources a technician normally has to juggle: the equipment documentation and the machine's current readings. This guide shows how to index manuals and procedures, pull live values from MQTT topics at question time, and assemble a prompt that produces grounded answers with citations, plus the pitfalls that matter on a factory floor.

Why Plain RAG Is Not Enough in a Plant

Retrieval-augmented generation (RAG) retrieves relevant passages from your documents and gives them to a language model so it can answer from your content instead of from general knowledge. For office documents, that is the whole story. In industry it falls short, because most useful questions depend on the current state of a machine:

• "Is this vibration level normal for this pump?" needs both the manual's limit and the current value.
• "What does fault code E-214 mean and is it still active?" needs the fault table and the live alarm state.
• "Can I restart the chiller now?" needs the restart procedure and the current pressures and temperatures.

A model with only documents will quote limits without knowing the value. A model with only telemetry will see numbers without knowing what they mean. Combining both is what makes an assistant useful to a maintenance technician.

Architecture of RAG for Industrial Data

The asset registry is the piece most tutorials leave out and the one that makes the system work. When a technician asks about "compressor C-3", the registry tells you which manual applies (model numbers differ between otherwise similar machines) and which topics hold its readings.

{
  "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"
    }
  }
}

A topic structure that follows ISA-95 levels (enterprise, site, area, line, cell) makes this mapping almost mechanical. If your machines publish Sparkplug B, ConnectMQ can decode it and republish the metrics into readable ISA-95 topics, which a RAG system can consume without a Protobuf decoder. Our article on Sparkplug B and the Unified Namespace explains that topic model.

Step 1: Prepare and Index the Documents

Document preparation decides answer quality more than the choice of model.

1. Extract clean text. Convert PDFs to text and check tables survived. Fault code tables are often the most valuable content and the most likely to be mangled.
2. Chunk by structure. Split on headings and procedure steps, not fixed character counts, so a procedure stays together. Keep chunks to a few hundred words.
3. Attach metadata. Store document ID, title, section, page and equipment model with every chunk.
4. Embed and store. Compute an embedding per chunk and keep it with the metadata.

The example below uses Ollama's embedding API with a local embedding model and keeps the index in NumPy. Install with pip install ollama numpy "paho-mqtt>=2.0" and 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")          # one .md or .txt file per document, named <doc-id>.md


def chunks_for(path):
    text = path.read_text(encoding="utf-8")
    sections = re.split(r"\n(?=#{1,3} )", text)   # split before markdown headings
    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")

The search_document: and search_query: prefixes are the task instructions this particular embedding model expects; other embedding models have their own conventions, so check the model card of the one you use.

Step 2: Keep a Live Value Cache From MQTT

The cache subscribes to the asset topics and stores the latest value with its timestamp. Because devices publish retained values, the cache is complete within moments of starting.

# 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

Reporting the age of each value is essential. A reading that arrived two hours ago is not "current", and the model should say so.

Step 3: Retrieve, Combine and Answer

At question time, identify the asset, retrieve passages only from that asset's documents, read its live values and build a prompt where every piece is labeled.

# 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)   # allow retained values to arrive
    print(ask("C-3", "Discharge temperature looks high. Is it within limits and what should I check first?"))

To try it without real equipment, publish retained test values:

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\""

Retained data on the public broker is not guaranteed to persist, so reseed values if they disappear, and never place real plant data on a public broker.

Pitfalls Specific to Industrial RAG

1. Wrong manual for the right machine. Similar machines with different firmware or model numbers have different limits. Filter retrieval by the asset's document list, as above, rather than searching everything.
2. Units and scaling. If a topic publishes raw counts or Fahrenheit while the manual uses Celsius, the model will compare apples to oranges. Normalize units in the cache and name them in the field (discharge_temp_c).
3. Stale values. Always include timestamps or ages, and instruct the model to treat old values as unavailable.
4. Tables lost in extraction. Verify fault code tables after PDF conversion; consider storing them as structured JSON and looking codes up directly instead of via embeddings.
5. Safety content. Lockout and tagout steps must never be summarized away. Say so in the system prompt and test it.
6. No evaluation set. Collect 30 to 50 real technician questions with expected answers and rerun them whenever you change chunking, models or prompts.

From Prototype to Plant

The read path for live values is light: one subscriber per assistant instance with a handful of wildcard subscriptions. Running it against a cloud MQTT broker that already carries your Unified Namespace means the assistant sees the same data as dashboards and historians, over TLS with its own credentials. ConnectMQ provides that as a hosted MQTT broker, with Sparkplug B decoding and an ISA-95 devices view in the portal to check which topics an asset publishes. During development, the public MQTT test broker is enough for seeded test values.

To take the assistant further, let it fetch values on demand through an MCP server for MQTT, turn recurring questions into scheduled LLM sensor data summaries, or explore a full industrial AI copilot. Our AI and IoT guide puts these patterns in context.

Get Started with ConnectMQ

Feed your RAG assistant live, well-structured telemetry from ConnectMQ, with retained messages, ISA-95 topic detection and Sparkplug B decoding built in. Create a free Serverless broker, no credit card required, and ground your first answer in real machine data.

Frequently Asked Questions

What is the difference between RAG and fine-tuning for industrial data?

RAG retrieves your documents and live values at question time, so answers stay current and can cite sources. Fine-tuning changes the model's weights and cannot know a machine's current state.

Why pull live values from MQTT instead of a database?

Retained MQTT messages give the latest value of every topic instantly, which is exactly what a question about the current state needs. Historical questions still belong in a time-series database.

Do I need a vector database?

Not for a few thousand chunks; an in-memory NumPy index is fast enough. Move to a vector database when collections grow large or many users query concurrently.

How do I keep the assistant from inventing limits?

Instruct it to answer only from the provided passages and live values, require citations, filter retrieval by the correct equipment documents and test with a fixed evaluation set.