أنماط التواصل

أنظمة الوكلاء المتعددين تحتاج طرقاً موثوقة لتبادل المعلومات. نمط التواصل الذي تختاره يؤثر على كل شيء من التأخير إلى تعقيد التنقيح.

ثلاثة أنماط أساسية

1. تمرير الرسائل

الوكلاء يتواصلون عبر رسائل منفصلة، مثل إرسال الإيميلات.

# تمرير الرسائل مع الطوابير
from dataclasses import dataclass
from typing import Any
import asyncio

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    content: Any
    message_type: str  # "request", "response", "broadcast"

class MessageBus:
    def __init__(self):
        self.queues: dict[str, asyncio.Queue] = {}

    async def send(self, message: AgentMessage):
        if message.receiver not in self.queues:
            self.queues[message.receiver] = asyncio.Queue()
        await self.queues[message.receiver].put(message)

    async def receive(self, agent_id: str) -> AgentMessage:
        if agent_id not in self.queues:
            self.queues[agent_id] = asyncio.Queue()
        return await self.queues[agent_id].get()

الأفضل لـ: الوكلاء ضعيفي الارتباط، العمليات غير المتزامنة، سجلات التدقيق

2. الحالة المشتركة

الوكلاء يقرؤون ويكتبون في مخزن ذاكرة مشترك.

# الحالة المشتركة مع وصول آمن للخيوط
import threading

class SharedState:
    def __init__(self):
        self._state = {}
        self._lock = threading.RLock()

    def read(self, key: str) -> Any:
        with self._lock:
            return self._state.get(key)

    def write(self, key: str, value: Any, agent_id: str):
        with self._lock:
            self._state[key] = {
                "value": value,
                "updated_by": agent_id,
                "timestamp": time.time()
            }

    def get_snapshot(self) -> dict:
        with self._lock:
            return dict(self._state)

الأفضل لـ: التعاون الفوري، التنسيق البسيط، فرق الوكلاء الصغيرة

3. مدفوع بالأحداث

الوكلاء ينشرون أحداثاً؛ الوكلاء المهتمون يشتركون ويتفاعلون.

# التواصل المدفوع بالأحداث
class EventBus:
    def __init__(self):
        self.subscribers: dict[str, list[callable]] = {}

    def subscribe(self, event_type: str, handler: callable):
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(handler)

    def publish(self, event_type: str, data: Any):
        for handler in self.subscribers.get(event_type, []):
            handler(data)

# الاستخدام
bus = EventBus()
bus.subscribe("research_complete", writer_agent.on_research_ready)
bus.subscribe("draft_ready", reviewer_agent.on_draft_ready)

الأفضل لـ: سير العمل التفاعلي، الأنظمة المنفصلة، المعماريات القابلة للتوسع

مقارنة الأنماط

النمط	التأخير	الارتباط	التنقيح	التوسع
تمرير الرسائل	متوسط	منخفض	سهل (السجلات)	عالي
الحالة المشتركة	منخفض	عالي	صعب	منخفض
مدفوع بالأحداث	منخفض	منخفض	متوسط	عالي

المناهج الهجينة

أنظمة الإنتاج غالباً تجمع الأنماط:

• الحالة المشتركة لسياق المهمة الحالية
• تمرير الرسائل لطلبات وكيل-إلى-وكيل
• الأحداث للإشعارات على مستوى النظام

ملاحظة نيردية: معظم أخطاء الأُطر تحدث عند حدود التواصل. سجّل كل رسالة في التطوير.

التالي: كيفية تنسيق هؤلاء الوكلاء المتواصلين. :::