Combinar Ollama y MQTT te permite ejecutar un modelo de lenguaje junto a tus máquinas y alimentarlo con datos de sensores en vivo, sin enviar las lecturas en bruto a un servicio de IA externo. En este tutorial instalarás Ollama en un gateway Linux, construirás un puente en Python que se suscribe a topics de telemetría, pide a un modelo local análisis estructurados y publica los resultados de vuelta en el broker para dashboards y alertas.
¿Por qué ejecutar el LLM en el gateway?
Un modelo local no siempre es la opción adecuada, pero en un gateway edge resuelve problemas reales:
- Los datos se quedan en planta. Las lecturas se procesan localmente; solo el breve análisis sale del gateway, si decides publicarlo hacia arriba.
- Sin dependencia de servicios de IA en internet. El análisis sigue funcionando durante los cortes y no depende de cuotas de API externas.
- Coste predecible. Una vez instalado el hardware, no hay cargo por petición.
A cambio, los modelos son más pequeños, la generación es más lenta en hardware modesto y hay que mantener los modelos actualizados. Para explicaciones, clasificaciones y resúmenes breves de datos numéricos, los modelos pequeños suelen ser suficientes. Nuestra comparativa IA en el edge frente a IA en la nube trata esta decisión con más detalle.
| Tipo de gateway | Qué esperar |
|---|---|
| Raspberry Pi 5 con 8 GB de RAM | Los modelos pequeños (de 1B a 3B parámetros) se ejecutan en CPU; las respuestas tardan segundos, así que analiza eventos, no cada mensaje |
| Mini PC con CPU x86 reciente y 16 GB de RAM | Funciona con soltura con modelos de 3B a 8B para análisis periódicos |
| Jetson o un PC con GPU | Modelos más grandes, respuestas más rápidas y análisis más frecuentes |
Arquitectura
sensors --MQTT--> broker --> ollama_bridge.py --HTTP localhost:11434--> Ollama
^ |
| v
+---- .../insights (JSON) <--+
El puente es un cliente MQTT normal. Mantiene el bucle de red en un hilo en segundo plano, pone los mensajes interesantes en una cola y los procesa de uno en uno con el modelo. Esa separación es la decisión de diseño más importante de este tutorial: una llamada a un LLM puede tardar varios segundos, y bloquear el callback MQTT durante tanto tiempo provoca keepalives perdidos y desconexiones.
Paso 1: instalar Ollama y descargar un modelo
En un gateway Linux de 64 bits:
curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.2:3b
ollama run llama3.2:3b "Reply with the word ready."
El instalador registra un servicio systemd, de modo que la API de Ollama escucha en http://localhost:11434 tras cada reinicio. Mantenla ligada a localhost; el puente se ejecuta en la misma máquina y no hay motivo para exponer la API del modelo a la red.
Instala las dependencias de Python en un entorno virtual:
python3 -m venv ~/ollama-mqtt/.venv
source ~/ollama-mqtt/.venv/bin/activate
pip install ollama "paho-mqtt>=2.0"
Paso 2: definir cómo es un análisis
El texto libre es difícil de aprovechar aguas abajo. Ollama admite salidas estructuradas: pasas un esquema JSON en el parámetro format y la respuesta del modelo queda restringida a esa forma. Así, los dashboards y las reglas de alerta pueden leer directamente campos como severity.
{
"type": "object",
"properties": {
"severity": { "type": "string", "enum": ["normal", "watch", "warning", "critical"] },
"summary": { "type": "string" },
"likely_causes": { "type": "array", "items": { "type": "string" }, "maxItems": 3 },
"recommended_check": { "type": "string" }
},
"required": ["severity", "summary", "recommended_check"]
}
Paso 3: construir el puente Ollama MQTT
El puente se suscribe a la telemetría, guarda un historial corto por dispositivo y lanza un análisis cuando una lectura supera una regla sencilla o cuando ha transcurrido el intervalo de análisis del dispositivo. El ejemplo se conecta al servidor MQTT público gratuito broker.connectmq.com, pensado solo para pruebas y no para datos de producción, así que sustituye 4c7a2e por tu propia cadena aleatoria.
# ollama_bridge.py
import json
import logging
import queue
import threading
import time
from collections import defaultdict, deque
import ollama
import paho.mqtt.client as mqtt
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")
BROKER, PORT = "broker.connectmq.com", 1883
PREFIX = "connectmq/demo/4c7a2e/site1"
MODEL = "llama3.2:3b"
HISTORY = 30 # lecturas guardadas por dispositivo
MIN_INTERVAL = 120 # segundos entre análisis por dispositivo
RULES = {"temp_c": 70.0, "vib_mm_s": 5.0}
SCHEMA = {
"type": "object",
"properties": {
"severity": {"type": "string", "enum": ["normal", "watch", "warning", "critical"]},
"summary": {"type": "string"},
"likely_causes": {"type": "array", "items": {"type": "string"}, "maxItems": 3},
"recommended_check": {"type": "string"},
},
"required": ["severity", "summary", "recommended_check"],
}
SYSTEM = (
"You analyze industrial sensor readings for maintenance technicians. "
"Base your answer only on the readings provided. "
"Phrase causes as possibilities, never as certainties. Keep the summary under 40 words."
)
history = defaultdict(lambda: deque(maxlen=HISTORY))
last_analysis = defaultdict(float)
work = queue.Queue(maxsize=50)
def on_connect(client, userdata, flags, reason_code, properties):
log.info("connected: %s", reason_code)
client.subscribe(f"{PREFIX}/+/telemetry", qos=1)
def on_message(client, userdata, msg):
device = msg.topic.split("/")[-2]
try:
reading = json.loads(msg.payload)
except json.JSONDecodeError:
return
history[device].append(reading)
breached = any(isinstance(reading.get(k), (int, float)) and reading[k] > v for k, v in RULES.items())
due = time.time() - last_analysis[device] > MIN_INTERVAL
if (breached or due) and len(history[device]) >= 5:
last_analysis[device] = time.time()
try:
work.put_nowait((device, list(history[device]), breached))
except queue.Full:
log.warning("queue full, skipping %s", device)
def analyze(device, readings, breached):
prompt = (
f"Device: {device}\nRule thresholds: {json.dumps(RULES)}\n"
f"Triggered by threshold breach: {breached}\n"
f"Last {len(readings)} readings, oldest first:\n{json.dumps(readings)}"
)
resp = ollama.chat(
model=MODEL,
messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": prompt}],
format=SCHEMA,
options={"temperature": 0},
)
return json.loads(resp.message.content)
def worker(client):
while True:
device, readings, breached = work.get()
started = time.time()
try:
insight = analyze(device, readings, breached)
except Exception as exc:
log.error("analysis failed for %s: %s", device, exc)
continue
insight.update({
"device": device,
"model": MODEL,
"trigger": "threshold" if breached else "interval",
"durationS": round(time.time() - started, 1),
"ts": int(time.time() * 1000),
})
client.publish(f"{PREFIX}/{device}/insights", json.dumps(insight), qos=1, retain=True)
log.info("%s -> %s: %s", device, insight["severity"], insight["summary"])
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="ollama-bridge-4c7a2e")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, keepalive=60)
threading.Thread(target=worker, args=(client,), daemon=True).start()
client.loop_forever()
Cómo se comporta:
- Cada mensaje de telemetría se guarda en un historial acotado por dispositivo.
- La superación de un umbral o el vencimiento de un intervalo encola un trabajo de análisis, pero nunca más de uno por dispositivo e intervalo.
- Un único hilo de trabajo llama a Ollama, para que la CPU del gateway no se sobrecargue con generaciones en paralelo.
- El análisis se publica como retenido, así que los dashboards muestran la última evaluación en cuanto se suscriben.
Elegir un modelo para el gateway
Elegir modelo en un gateway es un equilibrio entre calidad, velocidad y memoria. Algunas pautas prácticas:
- Empieza por lo pequeño. Un modelo de 3B es una opción razonable por defecto para análisis estructurados de lecturas cortas. Sube de tamaño solo si las respuestas son claramente demasiado superficiales para tu caso de uso.
- Prefiere versiones cuantizadas. Las etiquetas por defecto de la biblioteca de Ollama ya están cuantizadas, que es lo que hace viable el hardware pequeño. Evita las variantes de precisión completa en gateways solo con CPU.
- Deja memoria para todo lo demás. El gateway sigue ejecutando el puente MQTT, quizá un broker local, conversores de protocolo y el sistema operativo. Vigila la memoria con
free -hmientras se genera una respuesta. - Prueba con tus propios datos. Reúne unas cuantas decenas de ventanas reales de telemetría, incluidos incidentes conocidos, y compara cómo los describen los modelos candidatos. En operaciones, la consistencia importa más que la elocuencia.
- Fija la versión del modelo. Registra la etiqueta del modelo en cada análisis publicado, como hace el puente, para saber qué modelo generó una recomendación concreta tras una actualización.
Paso 4: alimentarlo con datos de prueba
Publica unas cuantas lecturas normales y luego una caliente con la CLI de Mosquitto:
for i in 1 2 3 4 5; do
mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-4c7a2e -q 1 \
-t "connectmq/demo/4c7a2e/site1/pump-2/telemetry" \
-m "{\"temp_c\": 6$i.0, \"vib_mm_s\": 2.$i, \"ts\": $(date +%s)000}"
sleep 1
done
mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-4c7a2e -q 1 \
-t "connectmq/demo/4c7a2e/site1/pump-2/telemetry" \
-m "{\"temp_c\": 74.5, \"vib_mm_s\": 5.8, \"ts\": $(date +%s)000}"
Observa cómo llegan los análisis:
mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-4c7a2e -t "connectmq/demo/4c7a2e/site1/+/insights" -v
Deberías ver un análisis JSON con una severidad como warning, un resumen breve que menciona el aumento de temperatura y vibración, y una comprobación recomendada. La redacción exacta varía según el modelo.
Ajustar la configuración de Ollama MQTT
| Problema | Causa | Solución |
|---|---|---|
| El puente se desconecta durante el análisis | La llamada al LLM se ejecuta dentro del callback MQTT | Mantén el patrón de cola e hilo de trabajo mostrado arriba |
| La primera respuesta es muy lenta | El modelo se está cargando en memoria | Configura un keep-alive más largo para el modelo, por ejemplo OLLAMA_KEEP_ALIVE=24h en el entorno del servicio |
| La cola se llena | Demasiados disparadores para el hardware | Aumenta MIN_INTERVAL, endurece las reglas o usa un modelo más pequeño |
| Faltan campos en la salida | Versión antigua de Ollama sin salidas estructuradas | Actualiza Ollama y la biblioteca de Python |
| Los análisis contradicen los datos | Demasiados datos en bruto o un prompt vago | Envía estadísticas calculadas en lugar de listas en bruto y baja la temperatura |
Esa última fila cobra importancia a medida que crece el volumen de datos. En lugar de lecturas en bruto, pasa el mínimo, el máximo, la media y la tendencia calculados en Python; nuestro tutorial sobre resúmenes de datos de sensores con LLM muestra en detalle ese enfoque de hechos primero.
Paso a producción
En el gateway, ejecuta el puente como servicio systemd junto a Ollama y conéctalo a un broker privado por TLS:
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)
El broker puede ser uno local en el gateway, un broker MQTT gestionado como ConnectMQ, o ambos, con el gateway publicando hacia arriba solo los análisis. Un endpoint en la nube resulta útil cuando varias plantas necesitan compartir análisis con dashboards centrales: el portal de ConnectMQ los muestra en el explorador de topics en vivo a medida que llegan. El servidor MQTT público gratuito es perfecto para este tutorial, pero sus topics son legibles por todo el mundo, así que no pongas en él datos reales de planta.
Siguientes pasos: deja que los operarios hagan preguntas sobre los mismos datos con un servidor MCP para MQTT, o permite que el modelo actúe sobre los dispositivos con seguridad aplicando los patrones de control de dispositivos con lenguaje natural. Para ver dónde encajan los modelos locales entre otras técnicas de IA, consulta nuestra guía de IA e IoT.
Empieza con ConnectMQ
Comparte los análisis de IA de tu gateway entre plantas con ConnectMQ, un broker MQTT gestionado con MQTT 5, TLS y WebSockets para dashboards en el navegador. Crea un broker Serverless gratuito, sin tarjeta de crédito, y publica tu primer análisis de un LLM local en minutos.
Preguntas frecuentes
¿Puede una Raspberry Pi ejecutar Ollama?
Sí, una Raspberry Pi 5 con 8 GB de RAM puede ejecutar modelos pequeños en su CPU. Las respuestas tardan segundos, así que analiza eventos y ventanas periódicas en lugar de cada mensaje.
¿Habla Ollama MQTT directamente?
No. Ollama expone una API HTTP en localhost. Un pequeño programa puente se conecta al broker como cliente MQTT y llama a la API de Ollama, como se muestra en este tutorial.
¿Cómo obtengo JSON fiable del modelo?
Pasa un esquema JSON en el parámetro format de Ollama para usar salidas estructuradas, fija la temperatura en 0 y, aun así, valida el resultado antes de publicarlo.
¿Debería cada mensaje MQTT disparar una llamada al LLM?
No. Lanza el análisis cuando se superen umbrales o a intervalos, y limita las llamadas por dispositivo. Los modelos de lenguaje son mucho más lentos que el ritmo de mensajes de un topic con mucho tráfico.
