Passerelle Ollama MQTT : exécuter un LLM local sur votre gateway et publier des analyses IA

Combiner Ollama et MQTT vous permet d'exécuter un modèle de langage à côté de vos machines et de l'alimenter avec des données de capteurs en direct, sans envoyer de mesures brutes à un service d'IA externe. Dans ce tutoriel, vous allez installer Ollama sur une passerelle Linux, créer un pont Python qui s'abonne aux topics de télémétrie, demande des analyses structurées à un modèle local et republie les résultats sur le broker pour les tableaux de bord et les alertes.

Pourquoi exécuter le LLM sur la passerelle ?

Un modèle local n'est pas toujours le bon choix, mais sur une passerelle edge il résout de vrais problèmes :

• Les données restent sur site. Les mesures sont traitées localement ; seule la courte analyse quitte la passerelle, si vous choisissez de la remonter.
• Aucune dépendance aux services d'IA sur Internet. L'analyse continue de fonctionner pendant les coupures et ne dépend pas de quotas d'API externes.
• Un coût prévisible. Une fois le matériel en place, il n'y a aucune facturation à la requête.

Les contreparties : des modèles plus petits, une génération plus lente sur du matériel modeste et le travail de mise à jour des modèles. Pour des explications, des classifications et de courts résumés de données numériques, les petits modèles suffisent généralement. Notre comparatif IA en périphérie ou IA cloud approfondit cette décision.

Architecture

sensors --MQTT--> broker --> ollama_bridge.py --HTTP localhost:11434--> Ollama
                    ^                |
                    |                v
                    +---- .../insights (JSON) <--+

Le pont est un client MQTT classique. Il fait tourner la boucle réseau dans un thread d'arrière-plan, place les messages intéressants dans une file d'attente et les traite un par un avec le modèle. Cette séparation est la décision de conception la plus importante de ce tutoriel : un appel au LLM peut durer plusieurs secondes, et bloquer aussi longtemps le callback MQTT entraîne des keepalives manqués et des déconnexions.

Étape 1 : installer Ollama et télécharger un modèle

Sur une passerelle Linux 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."

L'installateur enregistre un service systemd : l'API Ollama écoute donc sur http://localhost:11434 après un redémarrage. Laissez-la liée à localhost ; le pont tourne sur la même machine et rien ne justifie d'exposer l'API du modèle sur le réseau.

Installez les dépendances Python dans un environnement virtuel :

python3 -m venv ~/ollama-mqtt/.venv
source ~/ollama-mqtt/.venv/bin/activate
pip install ollama "paho-mqtt>=2.0"

Étape 2 : définir la forme d'une analyse

Le texte libre est difficile à exploiter en aval. Ollama prend en charge les sorties structurées : vous passez un schéma JSON dans le paramètre format et la réponse du modèle est contrainte à cette forme. Les tableaux de bord et les règles d'alerte peuvent alors lire directement des champs comme 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"]
}

Étape 3 : construire le pont Ollama MQTT

Le pont s'abonne à la télémétrie, conserve un court historique par appareil et déclenche une analyse lorsqu'une mesure franchit une règle simple ou lorsque l'intervalle d'analyse de l'appareil est écoulé. Remplacez 4c7a2e par votre propre chaîne aléatoire.

# 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                 # readings kept per device
MIN_INTERVAL = 120           # seconds between analyses per device
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()

Son fonctionnement :

1. Chaque message de télémétrie est stocké dans un historique borné par appareil.
2. Un dépassement de seuil ou un intervalle écoulé met une tâche d'analyse en file d'attente, mais jamais plus d'une par appareil et par intervalle.
3. Un seul thread de travail appelle Ollama, afin que le CPU de la passerelle ne soit pas surchargé par des générations parallèles.
4. L'analyse est publiée en retained message : les tableaux de bord affichent la dernière évaluation dès leur abonnement.

Choisir un modèle pour la passerelle

Sur une passerelle, le choix du modèle est un équilibre entre qualité, vitesse et mémoire. Quelques conseils pratiques :

1. Commencez petit. Un modèle de 3 milliards de paramètres est un choix par défaut raisonnable pour des analyses structurées à partir de courtes séries de mesures. Ne montez en gamme que si les réponses sont clairement trop superficielles pour votre cas d'usage.
2. Privilégiez les versions quantifiées. Les tags par défaut de la bibliothèque Ollama sont déjà quantifiés, ce qui rend viable le petit matériel. Évitez les variantes en pleine précision sur des passerelles sans GPU.
3. Laissez de la mémoire pour le reste. La passerelle fait aussi tourner le pont MQTT, éventuellement un broker local, des convertisseurs de protocole et le système d'exploitation. Surveillez la mémoire avec free -h pendant une génération.
4. Testez avec vos propres données. Collectez quelques dizaines de fenêtres de télémétrie réelles, incidents connus compris, et comparez la façon dont les modèles candidats les décrivent. Pour l'exploitation, la constance compte plus que l'éloquence.
5. Figez la version du modèle. Enregistrez le tag du modèle dans chaque analyse publiée, comme le fait le pont, pour savoir quel modèle a produit une recommandation donnée après une mise à jour.

Étape 4 : l'alimenter avec des données de test

Publiez quelques mesures normales puis une mesure chaude avec la CLI 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}"

Observez l'arrivée des analyses :

mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-4c7a2e -t "connectmq/demo/4c7a2e/site1/+/insights" -v

Vous devriez voir une analyse JSON avec une gravité comme warning, un court résumé mentionnant la hausse de la température et des vibrations, et une vérification recommandée. La formulation exacte varie selon le modèle.

Optimiser la configuration Ollama MQTT

Cette dernière ligne devient importante quand le volume de données augmente. Au lieu des mesures brutes, transmettez minimum, maximum, moyenne et tendance calculés en Python ; notre tutoriel sur les résumés de données capteurs par LLM détaille cette approche « faits d'abord ».

Passer en production

Sur la passerelle, exécutez le pont comme service systemd à côté d'Ollama, et connectez-le à un broker privé via 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)

Le broker peut être un broker local sur la passerelle, un broker MQTT managé comme ConnectMQ, ou les deux, la passerelle ne remontant alors que les analyses. Un point d'accès cloud est utile lorsque plusieurs sites doivent partager leurs analyses avec des tableaux de bord centraux : le portail ConnectMQ les affiche dans l'explorateur de topics en direct au fur et à mesure de leur arrivée. Le serveur MQTT public gratuit est parfait pour ce tutoriel, mais ses topics sont lisibles par tous : n'y faites pas transiter de vraies données d'usine.

Prochaines étapes : laissez les opérateurs interroger ces mêmes données avec un serveur MCP pour MQTT, ou permettez au modèle d'agir en toute sécurité sur les appareils grâce aux modèles présentés dans le contrôle d'appareils en langage naturel. Pour situer les modèles locaux parmi les autres techniques d'IA, consultez notre guide IA et IoT.

Démarrer avec ConnectMQ

Partagez les analyses IA de votre passerelle entre plusieurs sites avec ConnectMQ, un broker MQTT managé avec MQTT 5, TLS et WebSockets pour les tableaux de bord dans le navigateur. Créez un broker Serverless gratuit, sans carte bancaire, et publiez votre première analyse de LLM local en quelques minutes.

Questions fréquentes

Un Raspberry Pi peut-il faire tourner Ollama ?

Oui, un Raspberry Pi 5 avec 8 Go de RAM peut exécuter de petits modèles sur son CPU. Les réponses prennent quelques secondes : analysez donc des événements et des fenêtres périodiques plutôt que chaque message.

Ollama parle-t-il MQTT directement ?

Non. Ollama expose une API HTTP sur localhost. Un petit programme de pont se connecte au broker en tant que client MQTT et appelle l'API Ollama, comme le montre ce tutoriel.

Comment obtenir un JSON fiable de la part du modèle ?

Passez un schéma JSON dans le paramètre format d'Ollama pour utiliser les sorties structurées, réglez la température à 0 et validez tout de même le résultat avant de le publier.

Chaque message MQTT doit-il déclencher un appel au LLM ?

Non. Déclenchez l'analyse sur dépassement de seuil ou à intervalles réguliers, et limitez le nombre d'appels par appareil. Les modèles de langage sont bien plus lents que le débit de messages d'un topic chargé.