Ollama MQTT Köprüsü: Gateway'inizde Yerel LLM Çalıştırın ve Yapay Zeka İçgörüleri Yayınlayın

Ollama ile MQTT'yi birleştirmek, ham okumaları harici bir yapay zeka servisine göndermeden bir dil modelini makinelerinizin hemen yanında çalıştırmanızı ve onu canlı sensör verisiyle beslemenizi sağlar. Bu eğitimde Ollama'yı bir Linux gateway'ine kuracak, telemetri topic'lerine abone olan, yerel bir modelden yapılandırılmış içgörüler isteyen ve sonuçları dashboard'lar ve alarmlar için broker'a geri yayınlayan bir Python köprüsü oluşturacaksınız.

LLM'i Neden Gateway Üzerinde Çalıştırmalı?

Yerel bir model her zaman doğru seçim değildir, ancak bir edge gateway'inde gerçek sorunları çözer:

• Veri sahada kalır. Okumalar yerel olarak işlenir; gateway'den yalnızca kısa içgörü çıkar, o da ancak siz üst sisteme yayınlamayı seçerseniz.
• İnternet tabanlı yapay zeka servislerine bağımlılık yoktur. Analiz kesintiler sırasında çalışmaya devam eder ve harici API kotalarına bağlı değildir.
• Öngörülebilir maliyet. Donanım bir kez kurulduktan sonra istek başına ücret yoktur.

Bunun karşılığında daha küçük modeller, mütevazı donanımda daha yavaş üretim ve modelleri güncel tutma zahmeti vardır. Sayısal verilere ilişkin açıklamalar, sınıflandırmalar ve kısa özetler için küçük modeller genellikle yeterlidir. Edge AI ile bulut AI karşılaştırmamız bu kararı daha ayrıntılı ele alıyor.

Mimari

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

Köprü normal bir MQTT client'ıdır. Ağ döngüsünü arka planda bir thread'de çalıştırır, ilgi çekici mesajları bir kuyruğa koyar ve bunları modelle birer birer işler. Bu ayrım bu eğitimdeki en önemli tasarım kararıdır: bir LLM çağrısı birkaç saniye sürebilir ve MQTT callback'ini bu kadar uzun süre bloke etmek kaçırılan keepalive'lara ve bağlantı kopmalarına yol açar.

Adım 1: Ollama'yı Kurun ve Bir Model İndirin

64 bit bir Linux gateway'inde:

curl -fsSL https://ollama.com/install.sh | sh
ollama pull llama3.2:3b
ollama run llama3.2:3b "Reply with the word ready."

Kurulum betiği bir systemd servisi kaydeder; böylece yeniden başlatmadan sonra Ollama API'si http://localhost:11434 adresinde dinler. Onu localhost'a bağlı tutun; köprü aynı makinede çalışır ve model API'sini ağa açmak için hiçbir neden yoktur.

Python bağımlılıklarını bir sanal ortama kurun:

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

Adım 2: Bir İçgörünün Neye Benzediğini Tanımlayın

Serbest metni sonraki adımlarda kullanmak zordur. Ollama yapılandırılmış çıktıları destekler: format parametresinde bir JSON şeması iletirsiniz ve modelin yanıtı bu biçimle sınırlandırılır. Dashboard'lar ve alarm kuralları da böylece severity gibi alanları doğrudan okuyabilir.

{
  "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"]
}

Adım 3: Ollama MQTT Köprüsünü Oluşturun

Köprü telemetriye abone olur, her cihaz için kısa bir geçmiş tutar ve bir okuma basit bir kuralı aştığında veya cihazın analiz aralığı dolduğunda bir analiz tetikler. 4c7a2e değerini kendi rastgele dizinizle değiştirin.

# 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()

Nasıl davranır:

1. Her telemetri mesajı, cihaz başına sınırlı bir geçmişte saklanır.
2. Bir eşik aşımı veya dolan bir aralık bir analiz işini kuyruğa ekler, ancak cihaz başına her aralıkta birden fazla değil.
3. Ollama'yı tek bir worker thread'i çağırır; böylece gateway CPU'su paralel üretimlerle aşırı yüklenmez.
4. İçgörü retained olarak yayınlanır; böylece dashboard'lar abone olur olmaz en son değerlendirmeyi gösterir.

Gateway İçin Model Seçmek

Bir gateway'de model seçimi kalite, hız ve bellek arasında bir dengedir. Birkaç pratik öneri:

1. Küçük başlayın. Kısa okumalardan yapılandırılmış içgörüler için 3B bir model mantıklı bir varsayılandır. Ancak yanıtlar kullanım senaryonuz için açıkça yüzeysel kalıyorsa büyütün.
2. Kuantize sürümleri tercih edin. Ollama kütüphanesindeki varsayılan etiketler zaten kuantizedir ve küçük donanımı uygulanabilir kılan da budur. Yalnızca CPU'lu gateway'lerde tam hassasiyetli varyantlardan kaçının.
3. Geri kalan her şey için bellek bırakın. Gateway hâlâ MQTT köprüsünü, muhtemelen yerel bir broker'ı, protokol dönüştürücüleri ve işletim sistemini çalıştırıyor. Bir üretim sürerken belleği free -h ile izleyin.
4. Kendi verilerinizle test edin. Bilinen olaylar dahil birkaç düzine gerçek telemetri penceresi toplayın ve aday modellerin bunları nasıl tanımladığını karşılaştırın. Operasyonda tutarlılık, belagatten daha önemlidir.
5. Model sürümünü sabitleyin. Köprünün yaptığı gibi model etiketini yayınlanan her içgörüye kaydedin; böylece bir yükseltmeden sonra belirli bir öneriyi hangi modelin ürettiğini anlayabilirsiniz.

Adım 4: Test Verisiyle Besleyin

Mosquitto CLI kullanarak birkaç normal okuma ve ardından bir yüksek sıcaklık okuması yayınlayın:

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}"

İçgörülerin gelişini izleyin:

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

warning gibi bir önem derecesi, yükselen sıcaklık ve titreşimden bahseden kısa bir özet ve önerilen bir kontrol içeren JSON bir içgörü görmelisiniz. Tam ifade modelden modele değişir.

Ollama MQTT Kurulumunu İnce Ayarlamak

Son satır, veri hacmi büyüdükçe önem kazanır. Ham okumalar yerine Python'da hesaplanmış minimum, maksimum, ortalama ve eğilim değerlerini iletin; LLM ile sensör verisi özetleme eğitimimiz bu olgu öncelikli yaklaşımı ayrıntılı olarak gösteriyor.

Üretime Geçiş

Gateway'de köprüyü Ollama'nın yanında bir systemd servisi olarak çalıştırın ve TLS üzerinden özel bir broker'a bağlayın:

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)

Broker, gateway üzerindeki yerel bir broker, ConnectMQ gibi yönetilen bir MQTT broker ya da gateway'in üst sisteme yalnızca içgörüleri yayınladığı bir kombinasyonla her ikisi olabilir. Birden fazla sahanın içgörüleri merkezi dashboard'larla paylaşması gerektiğinde bir bulut uç noktası faydalıdır: ConnectMQ portalı onları geldikleri anda canlı topic gezgininde gösterir. Herkese açık MQTT test broker'ı bu eğitim için idealdir, ancak topic'leri herkes tarafından okunabilir; bu yüzden gerçek tesis verilerini oradan uzak tutun.

Sonraki adımlar: operatörlerin aynı veri hakkında soru sorabilmesi için bir MQTT MCP sunucusu kurun veya doğal dil ile cihaz kontrolü yazısındaki desenlerle modelin cihazlar üzerinde güvenle işlem yapmasına izin verin. Yerel modellerin diğer yapay zeka teknikleri arasındaki yerine genel bir bakış için yapay zeka ve IoT rehberimize göz atın.

ConnectMQ ile Başlayın

Gateway'inizin yapay zeka içgörülerini, tarayıcı dashboard'ları için MQTT 5, TLS ve WebSocket sunan barındırılan bir broker olan ConnectMQ ile sahalar arasında paylaşın. Kredi kartı gerektirmeden ücretsiz bir Serverless broker oluşturun ve ilk yerel LLM içgörünüzü dakikalar içinde yayınlayın.

Sık Sorulan Sorular

Raspberry Pi Ollama çalıştırabilir mi?

Evet, 8 GB RAM'li bir Raspberry Pi 5 küçük modelleri CPU'su üzerinde çalıştırabilir. Yanıtlar saniyeler sürer; bu yüzden her mesajı değil, olayları ve periyodik pencereleri analiz edin.

Ollama doğrudan MQTT konuşur mu?

Hayır. Ollama, localhost üzerinde bir HTTP API sunar. Bu eğitimde gösterildiği gibi, küçük bir köprü programı broker'a bir MQTT client'ı olarak bağlanır ve Ollama API'sini çağırır.

Modelden nasıl güvenilir JSON alırım?

Yapılandırılmış çıktıları kullanmak için Ollama'nın format parametresinde bir JSON şeması iletin, temperature değerini 0'a ayarlayın ve yine de sonucu yayınlamadan önce doğrulayın.

Her MQTT mesajı bir LLM çağrısı tetiklemeli mi?

Hayır. Analizi eşik aşımlarında veya belirli aralıklarla tetikleyin ve cihaz başına çağrı sayısını sınırlayın. Dil modelleri, yoğun bir topic'teki mesaj hızlarından çok daha yavaştır.