IoT için LLM Function Calling Eğitimi: Katı JSON Şemaları ve Doğrulanmış MQTT Komutları

IoT için LLM function calling, bir modelin bir isteği anlamasını set_valve(valve_id="V-12", position_pct=40) gibi yapılandırılmış bir çağrıya dönüştürür; kodunuz bu çağrıyı kontrol edip bir MQTT komutuna çevirebilir. Bu eğitim, çoğu demonun atladığı kısma odaklanıyor: cihaz eylemleri için katı JSON şemaları, her çağrının bağımsız doğrulanması, geçersiz çıktılar için bir onarım döngüsü ve MQTT 5 onayları. Böylece modelin ürettiği hiçbir şey kontrol edilmeden bir cihaza ulaşmaz.

Tek Paragrafta Function Calling

Function calling (araç çağırma, tool calling olarak da bilinir) ile modele JSON şeması parametre tanımlarına sahip bir fonksiyon listesi gönderirsiniz. Model düzyazıyla yanıt vermek yerine bir fonksiyonun adını ve argümanlardan oluşan bir JSON nesnesi döndürebilir. Bunu çalıştırıp çalıştırmamaya uygulamanız karar verir. Model hiçbir şeyi kendisi çalıştırmaz ve desenin cihazlar için yararlı olmasının nedeni de tam olarak budur: uygulama, kuralları uygulayabileceğiniz bir darboğaz noktasıdır.

İşin püf noktası, prompt'taki bir şemanın uygulama değil yol gösterme olmasıdır. Modeller zaman zaman yanlış tipte argümanlar, aralık dışı değerler, fazladan alanlar veya var olmayan cihaz kimlikleri döndürür. Bazı sağlayıcılar çıktıyı şemayla sınırlayan katı modlar sunar, ancak anlamsal kurallar (bu vana bu hatta mı, bu değişiklik çok mu büyük) her zaman kendi kodunuzu gerektirir.

Adım 1: Cihaz Eylemleri için Katı Şemalar Tasarlayın

Fiziksel eylemler için iyi şemaların birkaç ortak özelliği vardır:

Komut servisi ve ideal olarak cihaz firmware'i ya da gateway tarafından paylaşılan bir şema modülü:

# schemas.py
DEVICE_ACTIONS = {
    "set_valve": {
        "description": "Set a control valve opening in percent.",
        "parameters": {
            "type": "object",
            "properties": {
                "valve_id": {"type": "string", "enum": ["V-11", "V-12", "V-21"]},
                "position_pct": {"type": "integer", "minimum": 0, "maximum": 100},
                "reason": {"type": "string", "minLength": 5, "maxLength": 200},
            },
            "required": ["valve_id", "position_pct", "reason"],
            "additionalProperties": False,
        },
    },
    "set_pump_state": {
        "description": "Start or stop a pump.",
        "parameters": {
            "type": "object",
            "properties": {
                "pump_id": {"type": "string", "enum": ["P-1", "P-2"]},
                "state": {"type": "string", "enum": ["running", "stopped"]},
                "reason": {"type": "string", "minLength": 5, "maxLength": 200},
            },
            "required": ["pump_id", "state", "reason"],
            "additionalProperties": False,
        },
    },
}


def as_tools():
    """Convert to the tool format used by OpenAI-compatible and Ollama chat APIs."""
    return [
        {"type": "function", "function": {"name": name, "description": spec["description"],
                                          "parameters": spec["parameters"]}}
        for name, spec in DEVICE_ACTIONS.items()
    ]

Adım 2: İki Kez Doğrulayın: Önce Şema, Sonra Anlam

İlk kontrol yapısaldır ve modelin gördüğü şemanın tıpatıp aynısına karşı jsonschema kütüphanesini kullanır. İkinci kontrol anlamsaldır: canlı duruma veya tesis politikasına bağlı, hiçbir şemanın ifade edemeyeceği kurallar.

# validator.py
from jsonschema import Draft202012Validator
from schemas import DEVICE_ACTIONS

VALIDATORS = {n: Draft202012Validator(s["parameters"]) for n, s in DEVICE_ACTIONS.items()}
MAX_VALVE_STEP = 25   # plant policy: no more than 25 points per command


class ValidationFailed(Exception):
    pass


def check(name, args, live_state):
    validator = VALIDATORS.get(name)
    if validator is None:
        raise ValidationFailed(f"Unknown function '{name}'. Allowed: {sorted(VALIDATORS)}")
    errors = sorted(validator.iter_errors(args), key=lambda e: list(e.path))
    if errors:
        details = "; ".join(f"{'/'.join(map(str, e.path)) or '(root)'}: {e.message}" for e in errors)
        raise ValidationFailed(f"Arguments for {name} are invalid: {details}")
    if name == "set_valve":
        current = live_state.get(args["valve_id"], {}).get("position_pct")
        if current is None:
            raise ValidationFailed(f"Current position of {args['valve_id']} is unknown; refusing to move it.")
        if abs(args["position_pct"] - current) > MAX_VALVE_STEP:
            raise ValidationFailed(
                f"Change from {current} to {args['position_pct']} exceeds the {MAX_VALVE_STEP}-point step limit. "
                f"Propose an intermediate value.")
    if name == "set_pump_state" and args["state"] == "stopped":
        if live_state.get(args["pump_id"], {}).get("duty") == "critical":
            raise ValidationFailed(f"{args['pump_id']} is on critical duty and cannot be stopped by the assistant.")
    return args

Hata mesajları modelin okuması için yazılmıştır. Bu bilinçli bir tercihtir: bir sonraki adımdaki onarım döngüsünü beslerler.

Adım 3: Onarımlı LLM Function Calling Döngüsü

Bir çağrı doğrulamadan geçemediğinde hatayı geri gönderin ve modelin en fazla bir veya iki kez yeniden denemesine izin verin. Pek çok hata düzeltilebilir ("ara bir değer öner") ve sınırlı bir döngü sonsuz denemeleri önler. Bu örnek, araç desteği olan yerel bir modelle Ollama kullanıyor; aynı yapı herhangi bir sağlayıcının function calling API'siyle çalışır.

# command_service.py
import json
import time
import uuid

import ollama
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

from schemas import as_tools
from validator import ValidationFailed, check

MODEL = "qwen2.5:7b"
PREFIX = "connectmq/demo/d07c5a/plant"     # use your own random id
ACK_TOPIC = f"{PREFIX}/svc/acks"
MAX_REPAIRS = 2

live_state = {}   # filled from retained state topics
acks = {}


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/+/state", qos=1)
    client.subscribe(ACK_TOPIC, qos=1)


def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    if msg.topic == ACK_TOPIC:
        acks[data["id"]] = data
    else:
        live_state[msg.topic.split("/")[-2]] = data


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="fc-svc-d07c5a", protocol=mqtt.MQTTv5)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
time.sleep(2)   # let retained state arrive


def publish_command(name, args):
    device = args.get("valve_id") or args.get("pump_id")
    cmd = {"id": str(uuid.uuid4()), "action": name, "args": args, "ts": int(time.time() * 1000)}
    props = Properties(PacketTypes.PUBLISH)
    props.ResponseTopic = ACK_TOPIC
    props.CorrelationData = cmd["id"].encode()
    props.MessageExpiryInterval = 15   # never execute a stale command
    client.publish(f"{PREFIX}/{device}/cmd", json.dumps(cmd), qos=1, properties=props)
    deadline = time.time() + 5
    while time.time() < deadline:
        if cmd["id"] in acks:
            return acks.pop(cmd["id"])
        time.sleep(0.1)
    return {"id": cmd["id"], "result": "no_ack"}


def run(request):
    messages = [
        {"role": "system", "content": "You operate plant equipment only through the provided functions. "
                                      "If a request cannot be done safely with them, explain why."},
        {"role": "user", "content": request},
    ]
    for attempt in range(MAX_REPAIRS + 1):
        resp = ollama.chat(model=MODEL, messages=messages, tools=as_tools(), options={"temperature": 0})
        calls = resp.message.tool_calls or []
        if not calls:
            return resp.message.content
        errors, results = [], []
        for call in calls:
            name, args = call.function.name, dict(call.function.arguments)
            try:
                check(name, args, live_state)
            except ValidationFailed as err:
                errors.append(str(err))
                continue
            results.append((name, args, publish_command(name, args)))
        if not errors:
            return results
        messages.append(resp.message)
        messages.append({"role": "user", "content": "Some function calls were rejected:\n- " + "\n- ".join(errors)
                         + "\nCorrect them or explain why the request cannot be done."})
    return {"rejected": errors}


if __name__ == "__main__":
    print(run("Open valve V-12 to 90 percent, the tank is filling too slowly."))

Bu örnek istekle ne olduğuna dikkat edin. V-12 şu anda yüzde 40'taysa, anlamsal kontrol 50 puanlık sıçramayı reddeder ve modele ara bir değer önermesini söyler. Model genellikle 65 ile yeniden dener; bu değer kontrolden geçer ve yayınlanır. Bir gruptaki geçerli çağrılar hemen gönderilir, yalnızca reddedilenler onarım için geri gönderilir.

Katı Modlar ve Kısıtlı Kod Çözme

Birçok model sağlayıcısı ve yerel çalışma ortamı, üretimi araç argümanları her zaman bildirilen JSON şemasıyla eşleşecek şekilde kısıtlayabilir. Mevcutsa açın: tip hatalarının ve eksik alanların çoğunu doğrulayıcınıza ulaşmadan ortadan kaldırır. Ancak doğrulama ihtiyacını üç nedenle ortadan kaldırmaz. Birincisi, kısıtlı kod çözme yalnızca çalışma ortamının desteklediği JSON Schema alt kümesini kapsar ve sayısal aralıklar veya metin desenleri gibi anahtar sözcükler göz ardı edilebilir. İkincisi, kusursuz biçimlendirilmiş bir çağrı bile mevcut durum için yanlış olabilir; örneğin çok büyük bir vana adımı veya kritik görevdeki bir pompa. Üçüncüsü, ileride model veya sağlayıcı değiştirebilirsiniz ve güvenliğiniz tek bir çalışma ortamının bir özelliğine bağlı olmamalıdır. Kısıtlı çıktıyı bir kalite iyileştirmesi, kod doğrulamasını ise güvence olarak görün.

Adım 4: Cihaz Tarafında Yeniden Doğrulayın

Komut servisi son savunma hattı değildir. .../cmd mesajını alan gateway veya cihaz aynı şemayı doğrulamalı, mesajın süresinin dolmadığını kontrol etmeli ve response topic üzerinden bir onayla yanıt vermelidir:

# valve_gateway.py (simulated device side)
import json
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties
from jsonschema import Draft202012Validator
from schemas import DEVICE_ACTIONS

PREFIX = "connectmq/demo/d07c5a/plant"
state = {"V-12": {"position_pct": 40}, "P-1": {"state": "running", "duty": "normal"}}


def on_connect(client, userdata, flags, reason_code, properties):
    for dev, s in state.items():
        client.publish(f"{PREFIX}/{dev}/state", json.dumps(s), qos=1, retain=True)
    client.subscribe(f"{PREFIX}/+/cmd", qos=1)


def on_message(client, userdata, msg):
    cmd = json.loads(msg.payload)
    spec = DEVICE_ACTIONS.get(cmd.get("action"))
    ok = spec is not None and not list(Draft202012Validator(spec["parameters"]).iter_errors(cmd.get("args", {})))
    device = msg.topic.split("/")[-2]
    if ok and cmd["action"] == "set_valve":
        state[device]["position_pct"] = cmd["args"]["position_pct"]
    elif ok and cmd["action"] == "set_pump_state":
        state[device]["state"] = cmd["args"]["state"]
    if ok:
        client.publish(f"{PREFIX}/{device}/state", json.dumps(state[device]), qos=1, retain=True)
    reply = Properties(PacketTypes.PUBLISH)
    if hasattr(msg.properties, "CorrelationData"):
        reply.CorrelationData = msg.properties.CorrelationData
    ack = {"id": cmd.get("id"), "result": "applied" if ok else "rejected_invalid"}
    client.publish(msg.properties.ResponseTopic, json.dumps(ack), qos=1, properties=reply)


c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="valve-gw-d07c5a", protocol=mqtt.MQTTv5)
c.on_connect = on_connect
c.on_message = on_message
c.connect("broker.connectmq.com", 1883, keepalive=60)
c.loop_forever()

Bağımlılıkları pip install ollama jsonschema "paho-mqtt>=2.0" ile kurun, önce gateway'i, ardından komut servisini başlatın.

Devreye Almadan Önce Fonksiyon Tanımlarını Test Etmek

Fonksiyon tanımları, fiziksel ekipmana açılan diğer her arayüzle aynı test disiplinini hak eder. İki katmanlı bir test paketi oluşturun:

1. Model olmadan doğrulayıcı birim testleri. Elle yazılmış geçerli ve geçersiz argüman setlerini check() fonksiyonuna verin ve sonucu doğrulayın. Sınır değerlerini, yanlış tipleri, bilinmeyen kimlikleri, fazladan alanları ve bayat veya eksik canlı durumu dahil edin.
2. Model davranışı testleri. "V-11'i tamamen kapat" veya "ana pompayı durdur" gibi, beklenen fonksiyon ve argümanlarla birlikte gerçekçi isteklerden oluşan bir liste tutun. Bunları temperature 0 ile modele karşı çalıştırın ve ilk denemenin ne sıklıkla geçerli olduğunu, onarım döngüsünün ne sıklıkla düzelttiğini ve ne sıklıkla bir retle sonuçlandığını kaydedin.

Bir açıklamayı değiştirdiğinizde, bir fonksiyon eklediğinizde veya model değiştirdiğinizde davranış testlerini yeniden çalıştırın. Bir açıklamadaki küçük ifade değişiklikleri modelin hangi fonksiyonu seçtiğini değiştirebilir ve bunu üretim sahasında değil testte keşfetmek istersiniz. Üretimdeki reddedilen çağrıları kaydetmek zamanla size yeni test senaryoları kazandırır.

IoT'de Üretim Ortamı için LLM Function Calling Kontrol Listesi

1. Şemalar sürümlenir ve komut servisi ile cihazlar arasında paylaşılır.
2. additionalProperties: false, kimlikler için enum'lar ve her sayı için sınırlar.
3. Bir JSON Schema kütüphanesiyle yapısal doğrulama, ardından canlı duruma karşı anlamsal kontroller.
4. Net, modelin okuyabileceği hata mesajlarıyla sınırlı bir onarım döngüsü.
5. Komutlar bir kimlik, süre dolumu ve response topic taşır; başarı yalnızca onaydan sonra bildirilir.
6. Yüksek etkili fonksiyonlar insan onayı gerektirir; güvenlik fonksiyonları asla açılmaz.
7. Her çağrı, ret ve onay kaydedilir.

Demo, herkese açık MQTT test broker'ını kullanıyor. Gerçek ekipman için TLS ile 8883 portuna ve broker'ınız için oluşturulan kullanıcı adı, parola ve client ID'ye geçin; ConnectMQ gibi yönetilen bir MQTT broker sertifikaları, MQTT 5 desteğini ve bağlantı kotalarını sizin için yönetir. Genel ajan tasarımı için IoT cihazları için LLM ajanları, daha kullanıcı dostu bir ön yüz için doğal dil ile cihaz kontrolü, daha geniş bağlam için de yapay zeka ve IoT rehberimize bakın. JSON Schema spesifikasyonu yukarıda kullanılan her anahtar sözcüğü belgeliyor.

ConnectMQ ile Başlayın

ConnectMQ, bu desenin dayandığı MQTT 5 response topic, correlation data ve mesaj süresi dolumu özelliklerini destekler. Kredi kartı gerektirmeden ücretsiz Serverless plana kaydolun ve ilk doğrulanmış fonksiyon çağrınızı bugün gerçek bir broker üzerinde çalıştırın.

Sık Sorulan Sorular

Modele gönderdiğim JSON şeması komutları güvende tutmaya yeter mi?

Hayır. Şema modele yol gösterir, ancak uyumlu çıktıyı garanti etmez. Argümanları her zaman kendi kodunuzda ve bir kez de cihaz tarafında doğrulayın.

Modelin fonksiyon çağrısı geçersiz olduğunda ne olmalı?

Modele net bir hata mesajı döndürün ve bir veya iki onarım denemesine izin verin. Hâlâ başarısız olursa tahmin yürütmek yerine reddi kullanıcıya bildirin.

Komutlar için neden MQTT 5 kullanmalı?

Response topic ve correlation data, onayları komutlarla eşleştirmenin standart bir yolunu sunar; mesaj süresi dolumu da cihazların bayat komutları yürütmesini engeller.

Tek bir istek birden fazla fonksiyon çağrısı üretebilir mi?

Evet. Modeller tek bir yanıtta birden fazla araç çağrısı döndürebilir. Her birini bağımsız olarak doğrulayın ve yalnızca geçenleri yayınlayın.