MQTT ile Yapay Zeka Ajan İletişimi: Çok Ajanlı Sistemler için Pub/Sub Desenleri

MQTT üzerinden yapay zeka ajan iletişimi, çok ajanlı sistemlere MQTT'yi IoT'nin varsayılanı yapan özelliklerin aynısını kazandırır: birbirinden ayrık göndericiler ve alıcılar, varlık (presence) tespiti, küçük mesajlar ve kusurlu ağlar üzerinde güvenilir teslimat. Bu açıklayıcı yazı, publish/subscribe'ın ajanlara neden iyi uyduğunu, keşif ve mesajlaşmayı hangi topic kurallarının çalışır kıldığını ve response topic, correlation data ve mesaj süresi dolumu gibi MQTT 5 özelliklerinin ajan etkileşimlerine nasıl karşılık geldiğini gösteriyor.

Ajanların Neden Bir Mesajlaşma Katmanına İhtiyacı Var?

Birden fazla ajanınız olduğunda, örneğin telemetriyi izleyen bir izleme ajanı, eylem öneren bir planlama ajanı ve özet yazan bir raporlama ajanı, bunların konuşmak için bir yola ihtiyacı vardır. Yaygın ilk deneme ajanlar arasında doğrudan HTTP çağrılarıdır. İki ajan için işe yarar, beşte ise sancılı hâle gelir:

• Her ajan diğer her ajanın adresini bilmek zorundadır.
• Bir ajan meşgulse veya yeniden başlıyorsa çağrılar başarısız olur ve her çağıranın yeniden deneme mantığı uygulaması gerekir.
• Bir olayı yaymak ("hat 2 durdu"), ilgilenen her ajanı tek tek çağırmak anlamına gelir.
• Ajanların birbirine ne söylediğini gözlemlemek için kimsenin tek bir yeri yoktur.

Bir broker, ajanların arasına topic'ler koyarak bunları çözer. Göndericiler bir topic'e yayın yapar, alıcılar abone olur ve MQTT broker yönlendirmeyi, dağıtımı (fan-out) ve kalıcı oturumlar için tamponlamayı üstlenir. Ajanlar, eşlerini yeniden yapılandırmadan eklenebilen, yeniden başlatılabilen veya ölçeklenebilen gevşek bağlı servislere dönüşür.

Neden Başka Bir Protokol Değil de MQTT?

Ajan trafiğini birkaç mesajlaşma sistemi taşıyabilir. Ajanlar cihazlarla birlikte yaşadığında MQTT'nin belirli avantajları vardır:

Son satır edge kurulumları için önemlidir: bir fabrika gateway'indeki ajanın yalnızca 8883 portunda giden bir bağlantıya ihtiyacı vardır, gelen trafik için güvenlik duvarı kuralı gerekmez.

Yapay Zeka Ajan İletişimi için Topic Kuralları

Bir broker'ı ajan ağına dönüştüren şey tutarlı bir namespace'tir. Basit ve uygulanabilir bir kural seti:

agents/<agent-id>/status        retained: online/offline (set as last will)
agents/<agent-id>/card          retained: capabilities, version, accepted message types
agents/<agent-id>/inbox         direct messages and requests to this agent
agents/<agent-id>/outbox/<n>    responses and results produced by this agent
agents/broadcast/<event>        events every agent may react to
tasks/<capability>              work queues, consumed by shared subscriptions

Retained Ajan Kartlarıyla Keşif

Her ajan başladığında kendini tanımlayan retained bir "kart" yayınlar:

{
  "id": "planner-1",
  "role": "maintenance planner",
  "version": "0.4.2",
  "accepts": ["plan.request"],
  "emits": ["plan.proposal"],
  "description": "Proposes maintenance actions from alarms and work order history."
}

Yeni bir ajan agents/+/card topic'ine abone olur ve broker retained mesajları abonelik anında ilettiği için tüm kartları hemen alır. Bu, ek altyapı gerektirmeyen eksiksiz bir servis kaydıdır. Bir ajanı kayıttan çıkarmak için kart topic'ine boş bir retained mesaj yayınlayın.

Last Will ile Varlık Takibi

Her ajan, durum topic'inde offline değerine sahip bir last will ile bağlanır ve bağlandıktan sonra retained bir online yayınlar. Ajan çökerse veya ağını kaybederse, broker onun yerine will mesajını yayınlar. agents/+/status topic'ine abone olan diğer ajanlar, keepalive süresi içinde bir eşin gittiğini öğrenir ve ona iş göndermeyi bırakabilir.

Doğrudan Mesajlar ve İstek/Yanıt

MQTT 5, istek/yanıtı temiz hâle getiren iki özellik ekler. İstekte bulunan taraf bir response topic (yanıtın nereye gönderileceği) ve correlation data (yanıtı soruyla eşleştirmek için opak bir kimlik) belirler. Bir mesaj süresi dolum aralığı eklemek, zamanında kimsenin işlemediği bir isteğin önemini yitirdikten çok sonra işlenmek yerine atılmasını sağlar.

Aşağıdaki örnek, MQTT 5 ile paho-mqtt 2.x kullanan bir yanıtlayıcı ajan ve bir istekte bulunan ajandır. 9e1f3b değerini kendi rastgele dizinizle değiştirin.

# responder.py - a "summarizer" agent that answers requests in its inbox
import json
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

BASE = "connectmq/demo/9e1f3b/agents"
ME = "summarizer-1"


def on_connect(client, userdata, flags, reason_code, properties):
    client.publish(f"{BASE}/{ME}/status", "online", qos=1, retain=True)
    client.publish(f"{BASE}/{ME}/card", json.dumps({"id": ME, "accepts": ["summarize.request"]}), qos=1, retain=True)
    client.subscribe(f"{BASE}/{ME}/inbox", qos=1)


def on_message(client, userdata, msg):
    props = msg.properties
    response_topic = getattr(props, "ResponseTopic", None)
    if not response_topic:
        return
    request = json.loads(msg.payload)
    text = request.get("text", "")
    # A real agent would call an LLM here
    result = {"summary": text[:80] + ("..." if len(text) > 80 else ""), "agent": ME}
    reply_props = Properties(PacketTypes.PUBLISH)
    if hasattr(props, "CorrelationData"):
        reply_props.CorrelationData = props.CorrelationData
    client.publish(response_topic, json.dumps(result), qos=1, properties=reply_props)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{ME}-9e1f3b", protocol=mqtt.MQTTv5)
client.will_set(f"{BASE}/{ME}/status", "offline", qos=1, retain=True)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_forever()

# requester.py - another agent asking the summarizer for help
import json
import threading
import uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

BASE = "connectmq/demo/9e1f3b/agents"
ME = "monitor-1"
REPLY_TOPIC = f"{BASE}/{ME}/outbox/replies"
pending = {}


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(REPLY_TOPIC, qos=1)
    userdata.set()


def on_message(client, userdata, msg):
    corr = getattr(msg.properties, "CorrelationData", b"").decode()
    if corr in pending:
        print("reply for", pending.pop(corr), "->", msg.payload.decode())


ready = threading.Event()
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"{ME}-9e1f3b",
                     protocol=mqtt.MQTTv5, userdata=ready)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=30)
client.loop_start()
ready.wait(10)

corr_id = str(uuid.uuid4())
pending[corr_id] = "alarm digest"
props = Properties(PacketTypes.PUBLISH)
props.ResponseTopic = REPLY_TOPIC
props.CorrelationData = corr_id.encode()
props.MessageExpiryInterval = 30   # seconds; drop the request if not handled in time
request = {"type": "summarize.request", "text": "Line 2 conveyor stopped three times this morning due to jam sensor trips at station 4."}
client.publish(f"{BASE}/summarizer-1/inbox", json.dumps(request), qos=1, properties=props)

input("waiting for replies, press Enter to quit\n")

Önce yanıtlayıcıyı, sonra istekte bulunanı çalıştırın ve tüm konuşmayı şununla izleyin:

mosquitto_sub -h broker.connectmq.com -p 1883 -V mqttv5 -i spy-9e1f3b -t "connectmq/demo/9e1f3b/#" -v

Shared Subscription ile Worker Havuzları

Bir görev türü yoğunlaştığında, aynı ajanın birkaç kopyasını çalıştırın ve $share/planners/tasks/plan gibi bir aboneliği paylaşmalarına izin verin. Broker her mesajı gruptaki yalnızca bir üyeye iletir; böylece ayrı bir kuyruk sistemi olmadan yük dengeleme elde edersiniz. ConnectMQ'da shared subscription, ücretli Serverless planı ve üzerinde kullanılabilir.

Ajanlar için Mesaj Tasarımı

Ajanlar doğal dil alışverişi yapar, ancak zarf yapılandırılmış olmalıdır. İyi bir mesaj şunları taşır:

1. Alıcıların düzyazıyı ayrıştırmadan yönlendirme yapabilmesi için tür (plan.request, plan.proposal).
2. Yanıtları ve takip mesajlarını birbirine bağlamak için kimlik ve korelasyon.
3. Denetim ve sıralama için gönderici ve zaman damgası.
4. Yapılandırılmış veriden ayrı, açıkça işaretlenmiş bir alanda doğal dil içeren payload.
5. Bayat talimatlara göre işlem yapılmaması için süre dolumu veya son tarih.

Mesajları küçük tutun. Büyük belgeleri gömmek yerine bir topic adı veya belge kimliği gibi referanslar iletin. Ücretsiz herkese açık MQTT broker mesajları 64 KB ile sınırlar; zaten küçük mesajlar her abone için daha hızlıdır.

Ajanlar Arası İletişimde Güvenlik

Çok ajanlı sistemler yeni bir risk doğurur: bir ajanın çıktısı başka bir ajanın girdisi olur. Bir ajan, enjekte edilmiş talimatlar içeren bir bakım notunu özetlerse, akıştaki sonraki bir ajan bunları emir olarak algılayabilir. Pratik savunmalar:

• Gelen her mesajı, bir ajanın kendi sistem prompt'unu geçersiz kılan talimatlar olarak değil, her zaman veri olarak ele alın.
• Her ajana kendi broker kimlik bilgilerini ve client ID'sini verin; cihaz komutu yayınlayabilen kimlik bilgilerini doğrulama yapan tek bir dağıtıcıda tutun. IoT cihazları için LLM ajanları rehberimiz bu politika kapısını ayrıntılı olarak gösteriyor.
• Ajan namespace'ine bir denetim client'ı abone ederek tüm ajan trafiğini kaydedin.
• İki ajan arasındaki bir döngünün broker'ı mesaja boğamaması için ajanlara hız sınırı koyun. Tam yönetilen bir MQTT servisi gibi flood korumalı ve plan bazlı kotaları olan broker'lar bir emniyet ağı sağlar, ancak asıl çözüm ajanlarda olmalıdır.

Kaçınılması Gereken Yaygın Hatalı Desenler

Bazı tasarımlar ilk başta kullanışlı görünür, sonra sorun çıkarır:

• Her şey için tek dev bir topic. Tüm ajan trafiğini tek bir topic'e yayınlamak, her ajanı her mesajı ayrıştırmaya zorlar. Bunun yerine ajan başına gelen kutuları ve türlendirilmiş olay topic'leri kullanın.
• Geveze ajan döngüleri. Birbirinin yayınlarına yanıt veren iki ajan sonsuz bir konuşma yaratabilir. Her mesaja bir kimlik verin, kendi mesajlarınızı yok sayın ve yanıt derinliğini sınırlayın.
• Büyük payload'lar. Ajan mesajlarına belgelerin veya görsellerin tamamını gömmek her aboneyi yavaşlatır. Referansları yayınlayın ve ajanların içeriği ihtiyaç duyduklarında almasına izin verin.
• Yanlış şeyleri retained yapmak. Talepleri veya sonuçları değil, durum ve yetenek kartlarını retained yapın; aksi hâlde yeni aboneler eski iş öğelerini alır.

Bunun Ajan Protokolleriyle İlişkisi

Ajanlara özgü protokoller bir asistanın araçları nasıl çağıracağını veya ajanların kendilerini nasıl tanımlayacağını belirler; MQTT ise mesajları çok sayıda taraf arasında taşıyan bir transport'tur. Birbirlerini tamamlarlar: bir MQTT MCP sunucusu, bir yapay zeka asistanının broker üzerindeki cihazlara ve ajanlara araçlar aracılığıyla ulaşmasını sağlar; yukarıdaki topic kuralları da otonom ajanların kendi aralarında koordinasyon kurmasını sağlar. Bağlantılı sistemlerde yapay zekanın büyük resmi için yapay zeka ve IoT rehberimize bakın; burada kullanılan protokol özellikleri hakkında daha fazla bilgi için de MQTT 5 özellikleri yazımızı okuyun.

ConnectMQ ile Başlayın

MQTT 5 response topic, mesaj süresi dolumu, retained message ve oturum süresi dolumu destekleyen bir bulut MQTT broker olan ConnectMQ ile ajanlarınıza ortak bir omurga verin. Kredi kartı gerektirmeden ücretsiz bir Serverless broker oluşturun ve ilk ajanlarınızı dakikalar içinde bağlayın.

Sık Sorulan Sorular

MQTT yapay zeka ajanları için yeterince hızlı mı?

Evet. Bir broker üzerinden mesaj teslimatı LLM çıkarımından çok daha hızlıdır, bu yüzden transport nadiren darboğaz olur. Asıl tasarım işi topic yapısında ve mesaj biçimlerindedir.

Ajanlar MQTT üzerinden birbirini nasıl bulur?

Her ajan bilinen bir topic desenine retained bir yetenek kartı yayınlar. Yeni ajanlar bir wildcard ile abone olur ve tüm kartları anında alır.

MQTT ile istek/yanıt yapabilir miyim?

Evet. MQTT 5 response topic ve correlation data standart bir istek/yanıt deseni sağlar; mesaj süresi dolumu da zamanında işlenmeyen istekleri atar.

Bir ajan çökerse ne olur?

Broker ajanın last will mesajını, genellikle offline durumunu, yayınlar; böylece diğer ajanlar o yeniden bağlanana kadar ona iş göndermeyi bırakır.