Combining Ollama and MQTT lets you run a language model next to your machines and feed it live sensor data, without sending raw readings to an external AI service. In this tutorial you will install Ollama on a Linux gateway, build a Python bridge that subscribes to telemetry topics, asks a local model for structured insights and publishes the results back to the broker for dashboards and alerting.
Why Run the LLM on the Gateway?
A local model is not always the right choice, but on an edge gateway it solves real problems:
- Data stays on site. Readings are processed locally; only the short insight leaves the gateway, if you choose to publish it upstream.
- No dependency on internet AI services. The analysis keeps working through outages and does not depend on external API quotas.
- Predictable cost. Once the hardware is in place there is no per-request charge.
The trade-offs are smaller models, slower generation on modest hardware and the work of keeping models updated. For explanations, classifications and short summaries of numeric data, small models are usually adequate. Our edge AI vs cloud AI comparison covers this decision in more depth.
| Gateway class | What to expect |
|---|---|
| Raspberry Pi 5 with 8 GB RAM | Small models (1B to 3B parameters) run on CPU; responses take seconds, so analyze events, not every message |
| Mini PC with recent x86 CPU and 16 GB RAM | Comfortable with 3B to 8B models for periodic analysis |
| Jetson or a PC with a GPU | Larger models and faster responses, more frequent analysis |
Architecture
sensors --MQTT--> broker --> ollama_bridge.py --HTTP localhost:11434--> Ollama
^ |
| v
+---- .../insights (JSON) <--+
The bridge is a normal MQTT client. It keeps the network loop running in a background thread, puts interesting messages on a queue and processes them one at a time with the model. That separation is the most important design decision in this tutorial: an LLM call can take several seconds, and blocking the MQTT callback that long causes missed keepalives and disconnects.
Step 1: Install Ollama and Pull a Model
On a 64-bit Linux gateway:
curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.2:3b
ollama run llama3.2:3b "Reply with the word ready."
The installer registers a systemd service, so the Ollama API listens on http://localhost:11434 after reboot. Keep it bound to localhost; the bridge runs on the same machine and there is no reason to expose the model API to the network.
Install the Python dependencies in a virtual environment:
python3 -m venv ~/ollama-mqtt/.venv
source ~/ollama-mqtt/.venv/bin/activate
pip install ollama "paho-mqtt>=2.0"
Step 2: Define What an Insight Looks Like
Free text is hard to use downstream. Ollama supports structured outputs: you pass a JSON schema in the format parameter and the model's reply is constrained to that shape. Dashboards and alert rules can then read fields like severity directly.
{
"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"]
}
Step 3: Build the Ollama MQTT Bridge
The bridge subscribes to telemetry, keeps a short history per device and triggers an analysis when a reading crosses a simple rule or when the device's analysis interval has elapsed. Replace 4c7a2e with your own random string.
# 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()
How it behaves:
- Every telemetry message is stored in a bounded per-device history.
- A threshold breach or an elapsed interval queues an analysis job, but never more than one per device per interval.
- A single worker thread calls Ollama, so the gateway CPU is not overloaded by parallel generations.
- The insight is published retained, so dashboards show the latest assessment as soon as they subscribe.
Choosing a Model for the Gateway
Model choice on a gateway is a balance between quality, speed and memory. A few practical guidelines:
- Start small. A 3B model is a sensible default for structured insights from short readings. Move up only if the answers are clearly too shallow for your use case.
- Prefer quantized builds. The default tags in the Ollama library are already quantized, which is what makes small hardware viable. Avoid full-precision variants on CPU-only gateways.
- Leave memory for everything else. The gateway still runs the MQTT bridge, possibly a local broker, protocol converters and the operating system. Watch memory with
free -hwhile a generation is running. - Test with your own data. Collect a few dozen real telemetry windows, including known incidents, and compare how candidate models describe them. Consistency matters more than eloquence for operations.
- Pin the model version. Record the model tag in every published insight, as the bridge does, so you can tell which model produced a given recommendation after an upgrade.
Step 4: Feed It Test Data
Publish a few normal readings and then a hot one using the Mosquitto CLI:
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}"
Watch the insights arrive:
mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-4c7a2e -t "connectmq/demo/4c7a2e/site1/+/insights" -v
You should see a JSON insight with a severity such as warning, a short summary mentioning rising temperature and vibration, and a recommended check. The exact wording varies by model.
Tuning the Ollama MQTT Setup
| Problem | Cause | Fix |
|---|---|---|
| Bridge disconnects during analysis | LLM call running inside the MQTT callback | Keep the queue and worker thread pattern shown above |
| First response very slow | Model loading into memory | Set a longer keep-alive for the model, for example OLLAMA_KEEP_ALIVE=24h in the service environment |
| Queue fills up | Too many triggers for the hardware | Raise MIN_INTERVAL, tighten rules or use a smaller model |
| Output fields missing | Older Ollama without structured outputs | Update Ollama and the Python library |
| Insights contradict the data | Too much raw data or vague prompt | Send computed statistics instead of raw lists, lower temperature |
That last row matters as data volume grows. Rather than raw readings, pass min, max, mean and trend computed in Python; our tutorial on LLM sensor data summaries shows that facts-first approach in detail.
Going to Production
On the gateway, run the bridge as a systemd service next to Ollama, and connect it to a private broker over 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)
The broker can be a local one on the gateway, a managed MQTT broker such as ConnectMQ, or both, with the gateway publishing only insights upstream. A cloud endpoint is useful when several sites need to share insights with central dashboards: the ConnectMQ portal shows them in the live topic explorer as they arrive. The public MQTT test broker is perfect for this tutorial, but its topics are readable by everyone, so keep real plant data off it.
Next steps: let operators ask questions about the same data with an MCP server for MQTT, or allow the model to act on devices safely with the patterns in natural language device control. For an overview of where local models fit among other AI techniques, see our AI and IoT guide.
Get Started with ConnectMQ
Share your gateway's AI insights across sites with ConnectMQ, a hosted broker with MQTT 5, TLS and WebSockets for browser dashboards. Create a free Serverless broker, no credit card required, and publish your first local LLM insight in minutes.
Frequently Asked Questions
Can a Raspberry Pi run Ollama?
Yes, a Raspberry Pi 5 with 8 GB of RAM can run small models on its CPU. Responses take seconds, so analyze events and periodic windows rather than every message.
Does Ollama speak MQTT directly?
No. Ollama exposes an HTTP API on localhost. A small bridge program connects to the broker as an MQTT client and calls the Ollama API, as shown in this tutorial.
How do I get reliable JSON from the model?
Pass a JSON schema in Ollama's format parameter to use structured outputs, set temperature to 0 and still validate the result before publishing.
Should every MQTT message trigger an LLM call?
No. Trigger analysis on threshold breaches or at intervals, and limit calls per device. Language models are far slower than message rates on a busy topic.
