RAG pour les données industrielles : combiner manuels d'équipement et télémétrie MQTT en direct

Le RAG pour les données industrielles répond à des questions comme « pourquoi le compresseur de la ligne 3 est-il en alarme et que dois-je vérifier ? » en combinant deux sources qu'un technicien doit habituellement jongler : la documentation de l'équipement et les mesures actuelles de la machine. Ce guide montre comment indexer manuels et procédures, récupérer les valeurs en direct depuis des topics MQTT au moment de la question et assembler un prompt qui produit des réponses fondées et sourcées, ainsi que les pièges qui comptent en atelier.

Pourquoi un RAG classique ne suffit pas en usine

La génération augmentée par récupération (RAG) extrait des passages pertinents de vos documents et les fournit à un modèle de langage pour qu'il réponde à partir de votre contenu plutôt que de ses connaissances générales. Pour des documents bureautiques, c'est tout. Dans l'industrie, ce n'est pas suffisant, car la plupart des questions utiles dépendent de l'état actuel d'une machine :

• « Ce niveau de vibration est-il normal pour cette pompe ? » nécessite à la fois la limite du manuel et la valeur actuelle.
• « Que signifie le code défaut E-214 et est-il toujours actif ? » nécessite la table des défauts et l'état d'alarme en direct.
• « Puis-je redémarrer le groupe froid maintenant ? » nécessite la procédure de redémarrage ainsi que les pressions et températures actuelles.

Un modèle qui ne dispose que des documents citera des limites sans connaître la valeur. Un modèle qui ne dispose que de la télémétrie verra des chiffres sans savoir ce qu'ils signifient. C'est la combinaison des deux qui rend un assistant utile à un technicien de maintenance.

Architecture d'un RAG pour données industrielles

Le registre des actifs est l'élément que la plupart des tutoriels omettent, et c'est pourtant lui qui fait fonctionner le système. Quand un technicien pose une question sur le « compresseur C-3 », le registre indique quel manuel s'applique (les références diffèrent entre des machines par ailleurs similaires) et quels topics contiennent ses mesures.

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

Une structure de topics qui suit les niveaux ISA-95 (entreprise, site, zone, ligne, cellule) rend cette correspondance presque mécanique. Si vos machines publient en Sparkplug B, ConnectMQ peut le décoder et republier les métriques dans des topics ISA-95 lisibles, qu'un système RAG peut consommer sans décodeur Protobuf. Notre article sur Sparkplug B et l'Unified Namespace explique ce modèle de topics.

Étape 1 : préparer et indexer les documents

La préparation des documents détermine la qualité des réponses davantage que le choix du modèle.

1. Extrayez un texte propre. Convertissez les PDF en texte et vérifiez que les tableaux ont survécu. Les tables de codes défaut sont souvent le contenu le plus précieux et le plus susceptible d'être abîmé.
2. Découpez selon la structure. Coupez sur les titres et les étapes de procédure, pas selon un nombre fixe de caractères, afin qu'une procédure reste d'un seul tenant. Limitez les fragments à quelques centaines de mots.
3. Associez des métadonnées. Stockez avec chaque fragment l'identifiant du document, son titre, la section, la page et le modèle d'équipement.
4. Vectorisez et stockez. Calculez un embedding par fragment et conservez-le avec les métadonnées.

L'exemple ci-dessous utilise l'API d'embeddings d'Ollama avec un modèle d'embedding local et conserve l'index dans NumPy. Installez avec pip install ollama numpy "paho-mqtt>=2.0" et 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")

Les préfixes search_document: et search_query: sont les instructions de tâche attendues par ce modèle d'embedding en particulier ; d'autres modèles d'embedding ont leurs propres conventions, vérifiez donc la fiche du modèle que vous utilisez.

Étape 2 : maintenir un cache de valeurs en direct depuis MQTT

Le cache s'abonne aux topics des actifs et stocke la dernière valeur avec son horodatage. Comme les appareils publient des valeurs en retained message, le cache est complet quelques instants après le démarrage.

# 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

Indiquer l'âge de chaque valeur est indispensable. Une mesure arrivée il y a deux heures n'est pas « actuelle », et le modèle doit le signaler.

Étape 3 : récupérer, combiner et répondre

Au moment de la question, identifiez l'actif, récupérez des passages uniquement dans les documents de cet actif, lisez ses valeurs en direct et construisez un prompt où chaque élément est étiqueté.

# 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?"))

Pour essayer sans équipement réel, publiez des valeurs de test en 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\""

La persistance des données retenues sur le serveur MQTT public gratuit n'est pas garantie : republiez les valeurs si elles disparaissent, et ne placez jamais de vraies données d'usine sur un serveur public.

Pièges propres au RAG industriel

1. Le mauvais manuel pour la bonne machine. Des machines similaires avec des firmwares ou des références différents ont des limites différentes. Filtrez la recherche sur la liste de documents de l'actif, comme ci-dessus, plutôt que de chercher partout.
2. Unités et mises à l'échelle. Si un topic publie des valeurs brutes ou des degrés Fahrenheit alors que le manuel utilise des degrés Celsius, le modèle comparera des choux et des carottes. Normalisez les unités dans le cache et indiquez-les dans le nom du champ (discharge_temp_c).
3. Valeurs périmées. Incluez toujours des horodatages ou des âges, et demandez au modèle de traiter les valeurs anciennes comme indisponibles.
4. Tableaux perdus à l'extraction. Vérifiez les tables de codes défaut après la conversion des PDF ; envisagez de les stocker en JSON structuré et de rechercher les codes directement plutôt que via les embeddings.
5. Contenu lié à la sécurité. Les étapes de consignation ne doivent jamais disparaître dans un résumé. Précisez-le dans le prompt système et testez-le.
6. Aucun jeu d'évaluation. Rassemblez 30 à 50 vraies questions de techniciens avec les réponses attendues et rejouez-les chaque fois que vous modifiez le découpage, les modèles ou les prompts.

Du prototype à l'usine

Le chemin de lecture des valeurs en direct est léger : un abonné par instance d'assistant avec une poignée d'abonnements par wildcard. En l'exécutant sur un broker MQTT cloud qui porte déjà votre Unified Namespace, l'assistant voit les mêmes données que les tableaux de bord et les historiens, via TLS et avec ses propres identifiants. ConnectMQ le propose sous forme de broker MQTT managé, avec décodage Sparkplug B et une vue des appareils ISA-95 dans le portail pour vérifier quels topics publie un actif. Pendant le développement, le serveur MQTT public gratuit suffit pour des valeurs de test.

Pour aller plus loin, laissez l'assistant récupérer des valeurs à la demande via un serveur MCP pour MQTT, transformez les questions récurrentes en résumés de données capteurs par LLM planifiés, ou explorez un copilote IA industriel complet. Notre guide IA et IoT remet ces patterns en contexte.

Démarrer avec ConnectMQ

Alimentez votre assistant RAG avec une télémétrie en direct et bien structurée grâce à ConnectMQ, qui intègre retained messages, détection des topics ISA-95 et décodage Sparkplug B. Créez un broker Serverless gratuit, sans carte bancaire, et fondez votre première réponse sur de vraies données machine.

Questions fréquentes

Quelle différence entre RAG et fine-tuning pour les données industrielles ?

Le RAG récupère vos documents et vos valeurs en direct au moment de la question : les réponses restent à jour et peuvent citer leurs sources. Le fine-tuning modifie les poids du modèle et ne peut pas connaître l'état actuel d'une machine.

Pourquoi récupérer les valeurs en direct depuis MQTT plutôt que depuis une base de données ?

Les retained messages MQTT fournissent instantanément la dernière valeur de chaque topic, exactement ce dont a besoin une question sur l'état actuel. Les questions historiques relèvent toujours d'une base de données de séries temporelles.

Ai-je besoin d'une base de données vectorielle ?

Pas pour quelques milliers de fragments ; un index NumPy en mémoire est assez rapide. Passez à une base vectorielle lorsque les collections deviennent volumineuses ou que de nombreux utilisateurs interrogent le système simultanément.

Comment empêcher l'assistant d'inventer des limites ?

Demandez-lui de répondre uniquement à partir des passages et valeurs en direct fournis, exigez des citations, filtrez la recherche sur les documents du bon équipement et testez avec un jeu d'évaluation fixe.