استراتيجيات التنسيق

كيف تنسق وكلاء متعددين؟ استراتيجية التنسيق تحدد من يقرر ما يحدث بعد ذلك وكيف يتدفق العمل عبر النظام.

ثلاثة مناهج رئيسية

1. هرمي (مدير-عمال)

وكيل مدير مركزي يعيّن المهام وينسق العمال.

# التنسيق الهرمي مع CrewAI
from crewai import Crew, Process

crew = Crew(
    agents=[manager, researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.hierarchical,  # المدير يتحكم بالتدفق
    manager_agent=manager
)

# المدير يقرر:
# - أي وكيل يتعامل مع كل مهمة
# - متى ينتقل للمهمة التالية
# - كيف يتعامل مع الإخفاقات

المزايا: مسؤولية واضحة، تدفق متوقع، سهل التنقيح العيوب: عنق زجاجة المدير، نقطة فشل واحدة

2. ند-لند (تعاوني)

الوكلاء يتواصلون مباشرة بدون منسق مركزي.

# ند-لند مع تسليمات مباشرة
class PeerAgent:
    def __init__(self, name: str, peers: dict):
        self.name = name
        self.peers = peers  # الاسم -> نسخة الوكيل

    async def process(self, task: dict) -> dict:
        result = await self.do_work(task)

        # تحديد الند التالي بناءً على النتيجة
        if result["needs_review"]:
            return await self.peers["reviewer"].process(result)
        elif result["needs_more_data"]:
            return await self.peers["researcher"].process(result)
        else:
            return result

المزايا: لا عنق زجاجة، مرن، قابل للتكيف العيوب: تنسيق معقد، أصعب للتتبع، حلقات محتملة

3. سرب (سلوك ناشئ)

الوكلاء يعملون باستقلالية باتباع قواعد بسيطة؛ السلوك المعقد ينشأ.

# تنسيق شبيه بالسرب
class SwarmAgent:
    def __init__(self, role: str, shared_board: TaskBoard):
        self.role = role
        self.board = shared_board

    async def run(self):
        while True:
            # كل وكيل يختار مهام تطابق دوره
            task = await self.board.claim_task(
                filter=lambda t: t.type == self.role
            )
            if not task:
                await asyncio.sleep(1)
                continue

            result = await self.process(task)

            # نشر النتائج أو مهام جديدة للوحة
            if result.creates_new_tasks:
                for new_task in result.new_tasks:
                    await self.board.post_task(new_task)

            await self.board.complete_task(task.id, result)

المزايا: قابل للتوسع بشكل كبير، تنظيم ذاتي، متحمل للأخطاء العيوب: غير متوقع، صعب الفهم، كابوس للتنقيح

اختيار استراتيجيتك

السيناريو	أفضل استراتيجية
سير عمل محدد جيداً	هرمي
مهام ديناميكية متغيرة	ند-لند
توازي ضخم	سرب
الحاجة لسجل تدقيق	هرمي
تكيف في الوقت الفعلي	ند-لند
بحث/استكشاف	سرب

مثال هجين: سرب مُشرَف عليه

# مشرف + عمال سرب
class SupervisedSwarm:
    def __init__(self, supervisor: Agent, workers: list[SwarmAgent]):
        self.supervisor = supervisor
        self.workers = workers
        self.board = TaskBoard()

    async def execute(self, goal: str):
        # المشرف يقسم الهدف إلى مهام
        tasks = await self.supervisor.decompose(goal)
        for task in tasks:
            await self.board.post_task(task)

        # العمال يتجمعون على المهام
        worker_tasks = [w.run() for w in self.workers]

        # المشرف يراقب ويتدخل إذا لزم الأمر
        while not self.board.all_complete():
            status = self.board.get_status()
            if status.stuck_tasks:
                await self.supervisor.unstick(status.stuck_tasks)
            await asyncio.sleep(5)

ملاحظة نيردية: ابدأ بالهرمي، تطور إلى ند-لند عندما تفهم مجالك. السرب فقط للمشاكل القابلة للتوازي حقاً.