El RAG para datos industriales responde a preguntas como "¿por qué está en alarma el compresor de la línea 3 y qué debo revisar?" combinando dos fuentes que un técnico normalmente tiene que manejar a la vez: la documentación del equipo y las lecturas actuales de la máquina. Esta guía muestra cómo indexar manuales y procedimientos, obtener valores en vivo de topics MQTT en el momento de la pregunta y montar un prompt que produzca respuestas fundadas con citas, además de los escollos que importan en la planta.
Por qué el RAG básico no basta en una planta
La generación aumentada por recuperación (RAG) recupera pasajes relevantes de tus documentos y se los pasa a un modelo de lenguaje para que responda a partir de tu contenido en lugar de su conocimiento general. Para documentos de oficina, eso es todo. En la industria se queda corto, porque la mayoría de las preguntas útiles dependen del estado actual de una máquina:
- "¿Es normal este nivel de vibración para esta bomba?" necesita tanto el límite del manual como el valor actual.
- "¿Qué significa el código de fallo E-214 y sigue activo?" necesita la tabla de fallos y el estado de alarma en vivo.
- "¿Puedo reiniciar ya la enfriadora?" necesita el procedimiento de rearranque y las presiones y temperaturas actuales.
Un modelo solo con documentos citará límites sin conocer el valor. Un modelo solo con telemetría verá números sin saber qué significan. Combinar ambos es lo que hace que un asistente sea útil para un técnico de mantenimiento.
Arquitectura de RAG para datos industriales
| Componente | Función | Notas |
|---|---|---|
| Almacén de documentos | Manuales, procedimientos, tablas de fallos, órdenes de trabajo anteriores | Se fragmenta y vectoriza una vez y se actualiza cuando cambian los documentos |
| Índice vectorial | Encuentra pasajes similares a la pregunta | Una biblioteca como NumPy basta para colecciones pequeñas; a escala, usa una base de datos vectorial |
| Registro de activos | Relaciona los IDs de activo con documentos y topics MQTT | El pegamento entre documentos y datos en vivo |
| Caché de valores en vivo | Se suscribe a MQTT y guarda el último valor de cada topic | Los retained messages la llenan al instante al arrancar |
| Constructor de prompts | Combina pregunta, pasajes y valores en vivo | Etiqueta cada pieza con su fuente |
| LLM | Redacta la respuesta con citas | Local o alojado |
El registro de activos es la pieza que la mayoría de los tutoriales omite y la que hace que el sistema funcione. Cuando un técnico pregunta por el "compresor C-3", el registro te dice qué manual aplica (los números de modelo difieren entre máquinas por lo demás parecidas) y qué topics contienen sus lecturas.
{
"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"
}
}
}
Una estructura de topics que sigue los niveles de ISA-95 (empresa, planta, área, línea, celda) hace que esta correspondencia sea casi mecánica. Si tus máquinas publican Sparkplug B, ConnectMQ puede decodificarlo y republicar las métricas en topics ISA-95 legibles, que un sistema RAG puede consumir sin un decodificador de Protobuf. Nuestro artículo sobre Sparkplug B y el Unified Namespace explica ese modelo de topics.
Paso 1: preparar e indexar los documentos
La preparación de los documentos determina la calidad de las respuestas más que la elección del modelo.
- Extrae texto limpio. Convierte los PDF a texto y comprueba que las tablas han sobrevivido. Las tablas de códigos de fallo suelen ser el contenido más valioso y el que más probabilidades tiene de quedar destrozado.
- Fragmenta según la estructura. Divide por títulos y pasos de procedimiento, no por un número fijo de caracteres, para que cada procedimiento se mantenga unido. Limita los fragmentos a unos pocos cientos de palabras.
- Adjunta metadatos. Guarda con cada fragmento el ID del documento, el título, la sección, la página y el modelo del equipo.
- Vectoriza y almacena. Calcula un embedding por fragmento y guárdalo junto con los metadatos.
El ejemplo siguiente usa la API de embeddings de Ollama con un modelo de embeddings local y guarda el índice en NumPy. Instala con pip install ollama numpy "paho-mqtt>=2.0" y 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") # un archivo .md o .txt por documento, llamado <doc-id>.md
def chunks_for(path):
text = path.read_text(encoding="utf-8")
sections = re.split(r"\n(?=#{1,3} )", text) # divide antes de los 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")
Los prefijos search_document: y search_query: son las instrucciones de tarea que espera este modelo de embeddings en particular; otros modelos de embeddings tienen sus propias convenciones, así que consulta la ficha del modelo que uses.
Paso 2: mantener una caché de valores en vivo desde MQTT
La caché se suscribe a los topics de los activos y guarda el último valor con su marca de tiempo. Como los dispositivos publican valores retenidos, la caché está completa a los pocos instantes de arrancar.
# 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
Indicar la antigüedad de cada valor es esencial. Una lectura que llegó hace dos horas no es "actual", y el modelo debe decirlo.
Paso 3: recuperar, combinar y responder
En el momento de la pregunta, identifica el activo, recupera pasajes solo de los documentos de ese activo, lee sus valores en vivo y construye un prompt en el que cada pieza esté etiquetada.
# 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) # deja que lleguen los valores retenidos
print(ask("C-3", "Discharge temperature looks high. Is it within limits and what should I check first?"))
Para probarlo sin equipos reales, publica valores de prueba retenidos en el servidor MQTT público gratuito, que es legible por cualquiera y no debe recibir datos de producción:
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\""
No se garantiza que los datos retenidos en el broker público persistan, así que vuelve a sembrar los valores si desaparecen, y nunca pongas datos reales de planta en un broker público.
Escollos propios del RAG industrial
- Manual equivocado para la máquina correcta. Máquinas parecidas con firmware o números de modelo distintos tienen límites diferentes. Filtra la recuperación por la lista de documentos del activo, como arriba, en lugar de buscar en todo.
- Unidades y escalado. Si un topic publica cuentas en bruto o grados Fahrenheit mientras el manual usa Celsius, el modelo comparará peras con manzanas. Normaliza las unidades en la caché e indícalas en el nombre del campo (
discharge_temp_c). - Valores obsoletos. Incluye siempre marcas de tiempo o antigüedades, e indica al modelo que trate los valores antiguos como no disponibles.
- Tablas perdidas en la extracción. Verifica las tablas de códigos de fallo tras la conversión del PDF; plantéate guardarlas como JSON estructurado y buscar los códigos directamente en lugar de mediante embeddings.
- Contenido de seguridad. Los pasos de bloqueo y etiquetado nunca deben desaparecer en un resumen. Dilo en el prompt de sistema y pruébalo.
- Sin conjunto de evaluación. Reúne entre 30 y 50 preguntas reales de técnicos con sus respuestas esperadas y vuelve a ejecutarlas cada vez que cambies la fragmentación, los modelos o los prompts.
Del prototipo a la planta
La ruta de lectura de valores en vivo es ligera: un suscriptor por instancia del asistente con un puñado de suscripciones con comodines. Ejecutarla contra un broker MQTT en la nube que ya transporta tu Unified Namespace significa que el asistente ve los mismos datos que los dashboards, los historiadores o un gemelo digital 3D de la planta, por TLS y con sus propias credenciales. ConnectMQ lo ofrece como broker MQTT gestionado, con decodificación de Sparkplug B y una vista de dispositivos ISA-95 en el portal para comprobar qué topics publica un activo. Durante el desarrollo, el servidor público basta para valores de prueba sembrados.
Para llevar el asistente más lejos, deja que obtenga valores bajo demanda mediante un servidor MCP para MQTT, convierte las preguntas recurrentes en resúmenes de datos de sensores con LLM programados, o explora un copiloto de IA industrial completo. Nuestra guía de IA e IoT sitúa estos patrones en contexto.
Empieza con ConnectMQ
Alimenta tu asistente RAG con telemetría en vivo y bien estructurada desde ConnectMQ, con retained messages, detección de topics ISA-95 y decodificación de Sparkplug B integradas. Crea un broker Serverless gratuito, sin tarjeta de crédito, y fundamenta tu primera respuesta en datos reales de máquina.
Preguntas frecuentes
¿Qué diferencia hay entre RAG y fine-tuning para datos industriales?
El RAG recupera tus documentos y valores en vivo en el momento de la pregunta, así que las respuestas se mantienen actuales y pueden citar fuentes. El fine-tuning cambia los pesos del modelo y no puede conocer el estado actual de una máquina.
¿Por qué obtener los valores en vivo de MQTT en lugar de una base de datos?
Los mensajes MQTT retenidos dan al instante el último valor de cada topic, que es exactamente lo que necesita una pregunta sobre el estado actual. Las preguntas históricas siguen correspondiendo a una base de datos de series temporales.
¿Necesito una base de datos vectorial?
No para unos pocos miles de fragmentos; un índice NumPy en memoria es lo bastante rápido. Pasa a una base de datos vectorial cuando las colecciones crezcan mucho o muchos usuarios consulten a la vez.
¿Cómo evito que el asistente invente límites?
Indícale que responda solo a partir de los pasajes y valores en vivo proporcionados, exige citas, filtra la recuperación por los documentos del equipo correcto y prueba con un conjunto de evaluación fijo.
