Les résumés de données capteurs par LLM transforment des milliers de mesures brutes en quelques paragraphes qu'un opérateur peut lire lors de la relève : ce qui a fonctionné normalement, ce qui a dérivé et ce qui demande de l'attention. Dans ce tutoriel, vous allez collecter la télémétrie depuis des topics MQTT, calculer les statistiques en Python, puis demander à un modèle de langage de rédiger un rapport de poste et des explications d'anomalies en langage clair, fondés uniquement sur ces faits calculés.
Pourquoi ne pas envoyer directement les données brutes au modèle ?
Il est tentant de coller une journée de mesures dans un prompt et de demander un résumé. En pratique, cela échoue de trois façons :
- L'arithmétique n'est pas fiable. Les modèles de langage excellent avec la langue mais sont faibles pour calculer des moyennes sur de longues listes. Ils produiront des chiffres plausibles mais faux.
- Le contexte est limité et coûteux. Une télémétrie à haute fréquence dépasse vite ce qui tient dans un prompt, et chaque token coûte du temps.
- Les résultats ne sont pas vérifiables. Si un chiffre du rapport est faux, impossible de savoir d'où il vient.
Le schéma fiable répartit le travail. Le code calcule tout ce qui est numérique : comptages, moyennes, minimums, maximums, durée au-dessus d'un seuil, trous dans les données. Le modèle reçoit un document JSON compact de faits et le transforme en texte. Chaque chiffre du rapport peut être retracé jusqu'au document de faits.
| Tâche | Réalisée par | Pourquoi |
|---|---|---|
| Collecter les mesures | Abonné MQTT | Continu, léger |
| Statistiques et contrôle des seuils | Python | Exact et testable |
| Détecter les trous et les capteurs muets | Python | Simple calcul sur les horodatages |
| Hiérarchiser ce qui compte | Règles Python, modèle pour la formulation | Garde des priorités déterministes |
| Récit, explications, ton de la relève | LLM | Ce que les modèles font bien |
Architecture
sensors --MQTT--> broker --> collector.py (window buffer, stats) --> facts JSON
|
v
LLM (summary prompt)
|
broker <-- reports/shift (retained) <-+
Le rapport est republié sur le broker dans un topic en retained message : un tableau de bord, une passerelle vers un chatbot ou un service d'impression peut ainsi récupérer le dernier rapport à tout moment.
Étape 1 : publier de la télémétrie de test
Si vous n'avez pas encore d'appareils, ce simulateur publie toutes les deux secondes température, vibration et pression pour deux machines, avec un pic de vibration occasionnel. Remplacez b62d1f par votre propre chaîne aléatoire.
# 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) # occasional spike
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)
Étape 2 : collecter et calculer les faits
Le collecteur s'abonne avec un wildcard, met en mémoire tampon les mesures par machine et par métrique et, à la fin de chaque fenêtre, construit le document de faits. Les seuils proviennent de la configuration, pas du modèle.
# 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,
}
Conserver le nombre de dépassements, l'heure du premier dépassement et la valeur de pic permet au modèle d'expliquer concrètement une anomalie (« la vibration de press-2 a dépassé 4,5 mm/s à 6 reprises, la première fois à 14:12, avec un pic à 6,8 ») sans rien inventer.
Étape 3 : demander au LLM de rédiger le résumé
Le prompt compte autant que les données. Indiquez au modèle à quoi sert le rapport, ce qu'il peut faire ou non, et quelle forme doit prendre la sortie. Cet exemple utilise un modèle local via Ollama, mais n'importe quelle API de chat fonctionne avec le même 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
Étape 4 : lancer la boucle de fenêtres et publier le rapport
# main.py
import json
import time
import paho.mqtt.client as mqtt
import collector
from summarize import write_report
WINDOW_SECONDS = 600 # 10 minutes for testing; use 8 hours for a real shift
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)
L'appel au LLM est lent comparé au trafic MQTT, mais comme paho exécute sa boucle réseau dans un thread d'arrière-plan avec loop_start(), les messages continuent d'alimenter le tampon pendant la génération du rapport. Publier les faits avec le rapport permet à quiconque le lit de vérifier chaque chiffre.
Lisez le dernier rapport depuis n'importe où :
mosquitto_sub -h broker.connectmq.com -p 1883 -i reader-b62d1f -t "connectmq/demo/b62d1f/line1/reports/shift" -C 1
Choisir fenêtres, seuils et destinataires
Les mêmes faits peuvent produire des rapports très différents selon la fenêtre choisie et le lecteur visé.
| Destinataires | Fenêtre | Priorité |
|---|---|---|
| Opérateurs à la relève | Un poste | Ce qui a changé, ce qui reste ouvert, ce qu'il faut surveiller |
| Équipe de maintenance | Une journée ou une semaine | Dépassements répétés, tendances, machines à inspecter |
| Responsables | À la demande après une alerte | Courte explication d'un événement avec son contexte |
Les seuils doivent provenir de la même source que vos alarmes, comme les spécifications des équipements ou les limites de procédé convenues avec l'ingénierie, et non des connaissances générales du modèle. Si un seuil change, le rapport suivant en tient compte automatiquement puisqu'il fait partie du document de faits.
Pour des rapports déclenchés par événement, lancez le résumé lorsqu'une alerte se déclenche et ne transmettez que les faits de la machine concernée sur les 30 à 60 dernières minutes, ainsi que la même fenêtre de la veille comme référence. Le modèle rédige alors une explication ciblée au lieu d'un rapport de poste complet. Un document de faits réduit garde aussi une génération assez rapide pour que l'explication arrive pendant que l'événement est encore d'actualité.
Rédiger de meilleurs résumés de données capteurs par LLM
Quelques leçons pratiques qui améliorent nettement les rapports :
- Donnez des noms et du contexte. « press-2 (presse hydraulique, ligne 1) » produit de meilleures explications qu'un identifiant nu. Stockez les descriptions dans la configuration et ajoutez-les aux faits.
- Hiérarchisez les problèmes en amont. Triez les machines par gravité dans le code avant l'envoi. Le modèle conservera généralement votre ordre, et l'élément le plus important apparaîtra en premier.
- Indiquez à quoi ressemble la normale. Ajouter les moyennes du poste précédent permet au modèle de dire « la température est 3 degrés au-dessus du poste précédent » avec des chiffres que vous avez calculés.
- Séparez faits et hypothèses. Demandez des explications possibles sous forme de points à vérifier. Un modèle ne peut pas savoir qu'un roulement est en train de lâcher ; il peut suggérer de le contrôler.
- Validez la sortie. Un simple contrôle vérifiant que chaque chiffre du rapport figure aussi dans le JSON de faits détecte la plupart des valeurs inventées.
- Prévoyez une solution de repli. Si le modèle est indisponible, publiez un rapport basé sur un modèle de texte à partir des mêmes faits, afin que l'équipe reçoive tout de même un résumé.
Pour détecter les anomalies en amont, combinez cette approche avec un véritable détecteur issu de notre guide sur la détection d'anomalies IoT, et laissez le résumé expliquer les résultats du détecteur plutôt que de faire lui-même la détection.
Où résident les données
Le collecteur fonctionne avec n'importe quelle source MQTT. En usine, vous pouvez l'alimenter depuis des passerelles d'automates publiant en Sparkplug B ; ConnectMQ peut décoder le Sparkplug B et le republier dans des topics ISA-95 lisibles, bien plus faciles à agréger et à décrire. Pour des essais, le serveur MQTT public gratuit suffit, mais la télémétrie de production a sa place sur un broker MQTT cloud privé avec TLS et des identifiants par appareil. Si vous souhaitez garder le modèle sur site, exécutez-le sur une passerelle comme le montre Ollama avec MQTT ; pour des réponses combinant valeurs en direct et manuels d'équipement, consultez le RAG pour les données industrielles. La vue d'ensemble est présentée dans notre guide IA et IoT.
Démarrer avec ConnectMQ
Collectez votre télémétrie sur ConnectMQ, un broker MQTT managé doté d'un explorateur de topics en direct, d'un tableau de bord et de règles d'alerte par seuil pour les mesures que vos rapports résument. Ouvrez un broker Serverless gratuit, sans carte bancaire, et publiez dès aujourd'hui votre premier rapport de poste rédigé par l'IA.
Questions fréquentes
Le LLM peut-il calculer lui-même les statistiques ?
Il peut essayer, mais les résultats ne sont pas fiables sur un grand nombre de valeurs. Calculez les statistiques dans le code et transmettez-les au modèle, afin que chaque chiffre du rapport soit exact et traçable.
À quelle fréquence générer les résumés ?
Calquez-vous sur le rythme de l'activité : par poste pour la relève, toutes les heures pour les responsables, ou à la demande lorsqu'une alerte se déclenche. La fenêtre du collecteur suit simplement ce calendrier.
Quel modèle utiliser ?
Un petit modèle local suffit pour des résumés courts et structurés à partir d'un document de faits. Les modèles plus grands rédigent des explications plus fluides ; l'approche « faits d'abord » fonctionne avec les deux.
Comment empêcher le modèle d'inventer des causes ?
Demandez-lui de formuler les causes comme des points à vérifier, n'incluez que des faits dans le prompt et vérifiez que les chiffres de la sortie correspondent au JSON de faits.
