Los resúmenes de datos de sensores con LLM convierten miles de lecturas en bruto en unos pocos párrafos que un operario puede leer en el relevo de turno: qué funcionó con normalidad, qué se desvió y qué requiere atención. En este tutorial recogerás telemetría de topics MQTT, calcularás las estadísticas en Python y pedirás a un modelo de lenguaje que redacte un informe de turno y explicaciones de anomalías en lenguaje claro basándose únicamente en esos hechos calculados.
¿Por qué no enviar sin más los datos en bruto al modelo?
Es tentador pegar un día de lecturas en un prompt y pedir un resumen. En la práctica, eso falla de tres maneras:
- La aritmética no es fiable. Los modelos de lenguaje son buenos con el lenguaje y flojos calculando medias sobre listas largas. Producirán números plausibles pero erróneos.
- El contexto es limitado y caro. La telemetría de alta frecuencia supera enseguida lo que cabe en un prompt, y cada token cuesta tiempo.
- Los resultados no son auditables. Si una cifra del informe es incorrecta, no puedes saber de dónde ha salido.
El patrón fiable reparte el trabajo. El código calcula todo lo que es un número: recuentos, medias, mínimos, máximos, tiempo por encima del umbral, huecos en los datos. El modelo recibe un documento JSON compacto de hechos y lo convierte en prosa. Cada cifra del informe puede rastrearse hasta el documento de hechos.
| Tarea | La hace | Por qué |
|---|---|---|
| Recoger lecturas | Suscriptor MQTT | Continuo y ligero |
| Estadísticas y comprobación de umbrales | Python | Exacto y comprobable |
| Detectar huecos y sensores sin datos recientes | Python | Aritmética temporal sencilla |
| Priorizar lo importante | Reglas en Python, el modelo para la redacción | Mantiene las prioridades deterministas |
| Narrativa, explicaciones, tono de relevo | LLM | Lo que los modelos hacen bien |
Arquitectura
sensors --MQTT--> broker --> collector.py (window buffer, stats) --> facts JSON
|
v
LLM (summary prompt)
|
broker <-- reports/shift (retained) <-+
El informe se publica de vuelta en el broker en un topic retenido, de modo que un dashboard, un gemelo digital 3D de la línea, un puente hacia un bot de chat o un servicio de impresión pueden recoger el último informe en cualquier momento.
Paso 1: publicar telemetría de prueba
Si todavía no tienes dispositivos, este simulador publica cada dos segundos temperatura, vibración y presión de dos máquinas, con algún pico de vibración ocasional. Sustituye b62d1f por tu propia cadena aleatoria.
# 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)
Paso 2: recoger y calcular los hechos
El colector se suscribe con un comodín, almacena las lecturas por máquina y métrica y, al final de cada ventana, construye el documento de hechos. Los umbrales proceden de la configuración, no del 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,
}
Conservar el número de superaciones, la hora de la primera y el valor máximo permite al modelo explicar una anomalía de forma concreta ("la vibración de press-2 superó 4,5 mm/s 6 veces, la primera a las 14:12, con un pico de 6,8") sin inventar nada.
Paso 3: pedir al LLM que redacte el resumen
El prompt importa tanto como los datos. Dile al modelo para qué sirve el informe, qué puede y qué no puede hacer y qué forma debe tener la salida. Este ejemplo usa un modelo local a través de Ollama, pero cualquier API de chat funciona con el mismo 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
Paso 4: ejecutar el bucle de ventanas y publicar el informe
# 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 pruebas; usa 8 horas para un 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)
La llamada al LLM es lenta comparada con el tráfico MQTT, pero como paho ejecuta su bucle de red en un hilo en segundo plano con loop_start(), los mensajes siguen llegando al búfer mientras se genera el informe. Publicar los hechos junto con el informe permite que cualquiera que lo lea compruebe cada cifra.
Lee el último informe desde cualquier lugar:
mosquitto_sub -h broker.connectmq.com -p 1883 -i reader-b62d1f -t "connectmq/demo/b62d1f/line1/reports/shift" -C 1
Elegir ventanas, umbrales y destinatarios
Los mismos hechos pueden producir informes muy distintos según cómo plantees la ventana y quién lea el resultado.
| Destinatario | Ventana | Enfoque |
|---|---|---|
| Operarios en el relevo | Un turno | Qué ha cambiado, qué sigue pendiente, qué vigilar a continuación |
| Equipo de mantenimiento | Un día o una semana | Superaciones repetidas, tendencias, máquinas que requieren inspección |
| Supervisores | Bajo demanda tras una alerta | Explicación breve de un evento con su contexto |
Los umbrales deben venir de la misma fuente que tus alarmas, como las especificaciones del equipo o los límites de proceso acordados con ingeniería, no del conocimiento general del modelo. Si un umbral cambia, el siguiente informe lo refleja automáticamente porque forma parte del documento de hechos.
Para informes basados en eventos, lanza el resumen cuando salte una alerta y pasa solo los hechos de la máquina afectada de los últimos 30 a 60 minutos, junto con la misma ventana del día anterior como referencia. El modelo redacta entonces una explicación centrada en lugar de un informe de turno completo. Mantener pequeño el documento de hechos también hace que la generación sea lo bastante rápida para que la explicación llegue mientras el evento sigue siendo relevante.
Cómo escribir mejores resúmenes de datos de sensores con LLM
Algunas lecciones prácticas que mejoran notablemente los informes:
- Aporta nombres y contexto. "press-2 (prensa hidráulica, línea 1)" produce mejores explicaciones que un ID sin más. Guarda las descripciones en la configuración y añádelas a los hechos.
- Ordena los problemas de antemano. Ordena las máquinas por gravedad en código antes de enviarlas. El modelo mantendrá casi siempre tu orden, así que lo más importante aparecerá primero.
- Incluye cómo es lo normal. Añadir las medias del turno anterior permite que el modelo diga "la temperatura está 3 grados por encima del turno anterior" usando números que has calculado tú.
- Separa los hechos de las hipótesis. Pide las posibles explicaciones como sugerencias que comprobar. Un modelo no puede saber que un rodamiento está fallando; sí puede sugerir revisarlo.
- Valida la salida. Una comprobación sencilla de que cada número del informe aparece también en el JSON de hechos detecta la mayoría de las cifras inventadas.
- Ten un plan B. Si el modelo no está disponible, publica un informe basado en una plantilla a partir de los mismos hechos para que el turno siga teniendo su resumen.
Para detectar las anomalías en primer lugar, combina esto con un detector adecuado de nuestra guía de detección de anomalías en IoT, y deja que el resumen explique lo que ha encontrado el detector en lugar de hacer la detección él mismo.
Dónde viven los datos
El colector funciona con cualquier fuente MQTT. En una planta de manufactura en Querétaro o en Bilbao podrías alimentarlo desde gateways de PLC que publican Sparkplug B; ConnectMQ puede decodificar Sparkplug B y republicarlo en topics ISA-95 legibles, mucho más fáciles de agregar y describir. Para pruebas basta con el servidor MQTT público gratuito, pero no para datos de producción: la telemetría de producción debe ir a un broker MQTT en la nube privado con TLS y credenciales por dispositivo. Si quieres mantener el modelo en planta, ejecútalo en un gateway como se muestra en Ollama con MQTT, y para respuestas que combinen valores en vivo con manuales de equipos consulta RAG para datos industriales. La visión de conjunto está en nuestra guía de IA e IoT.
Empieza con ConnectMQ
Recoge tu telemetría en ConnectMQ, un broker MQTT gestionado con explorador de topics en vivo, dashboard y reglas de alerta por umbral para las lecturas que resumen tus informes. Abre un broker Serverless gratuito, sin tarjeta de crédito, y publica hoy tu primer informe de turno escrito por IA.
Preguntas frecuentes
¿Puede el LLM calcular las estadísticas por sí mismo?
Puede intentarlo, pero los resultados no son fiables con muchos valores. Calcula las estadísticas en código y pásaselas al modelo, de modo que cada número del informe sea exacto y trazable.
¿Con qué frecuencia deben generarse los resúmenes?
Ajústala al ritmo del negocio: por turno para el relevo, cada hora para los supervisores o bajo demanda cuando salta una alerta. La ventana del colector simplemente sigue ese calendario.
¿Qué modelo debería usar?
Un modelo local pequeño basta para resúmenes breves y estructurados a partir de un documento de hechos. Los modelos más grandes redactan explicaciones más fluidas; el enfoque de hechos primero funciona con ambos.
¿Cómo evito que el modelo invente causas?
Indícale que formule las causas como cosas que comprobar, incluye solo hechos en el prompt y valida que los números de la salida coinciden con el JSON de hechos.
