مقارنة بين Multi-Agent و Single-Agent: متى تستحق تعقيدات البنية التحتية الاستثمار فعلياً؟

لا تحتاج إلى نظام multi-agent لكتابة بريد إلكتروني، أو صياغة استعلام SQL، أو تحليل ملف PDF. ومع ذلك، فإننا في Verel Systems ننقذ بانتظام الأكواد البرمجية للشركات حيث يتم توجيه مهمة بسيطة لاستخراج البيانات عبر خمسة وكلاء (agents) مختلفين في CrewAI، مما يكلف 0.15 دولار لكل عملية تشغيل ويستغرق 18 ثانية لإكمالها. هذا هو "سباغيتي الذكاء الاصطناعي" الكلاسيكي: هندسة مفرطة (over-engineered)، بطيئة، ومكلفة.

بالنسبة لمؤسسي الشركات الناشئة (SaaS) أو مسؤولي المشتريات في الشركات الكبرى في الولايات المتحدة أو منطقة الخليج العربي، فإن هذا الخيار المعماري ليس مجرد تفصيل هندسي بسيط، بل هو قرار مالي وتشغيلي حاسم. اختيار الإعداد الخاطئ يعني حرق آلاف الدولارات في فواتير واجهات برمجة التطبيقات (LLM API)، وتأخير وقت طرح منتجك في السوق (time-to-market)، والمخاطرة بخسارة العملاء بسبب أوقات الاستجابة البطيئة وغير المتوقعة.

تموت معظم مشاريع الذكاء الاصطناعي في مرحلة التجريب (pilot purgatory) لأن الفرق تبني شبكات وكلاء معقدة لحل مشكلات يمكن لموجّه نظام (system prompt) واحد مصمم بشكل جيد أن يحلها. ومع ذلك، فإن الخطأ المعاكس قاتل بنفس القدر. محاولة فرض عملية تجارية معقدة ومتعددة الخطوات، تحتوي على منطق متفرع وموافقات بشرية (human-in-the-loop)، داخل موجّه واحد ضخم (prompt monolith) تؤدي إلى تدهور جودة الموجّه، وتلوث نافذة السياق (context window)، وفقدان السيطرة الكامل على النظام.

إن الاختيار بين بنية single-agent وبنية multi-agent هو القرار الأكثر أهمية الذي ستتخذه عند الانتقال من مرحلة إثبات المفهوم (POC) إلى نظام جاهز للتشغيل الفعلي (production-grade). إليك كيفية اتخاذ هذا القرار بناءً على مقاييس هندسية صارمة ومقايضات تجارية واضحة، بعيداً عن البروباغندا المحيطة بإطارات العمل (frameworks).

فخ الموجّه الضخم (Monolith Prompt) مقابل الهندسة المفرطة للـ Multi-Agent

تعتمد بنية الـ single-agent على استدعاء واحد للنموذج (LLM call) - أو حلقة استدعاءات خطية - لمعالجة المدخلات، واختيار الأدوات، وإنتاج المخرجات. يحتوي الموجّه (prompt) على جميع التعليمات، وتعاريف الأدوات، وشخصية النظام (persona) بالكامل.

عند البدء، يكون هذا الأسلوب فعالاً للغاية وفعالاً من حيث التكلفة. يكون زمن الاستجابة (latency) منخفضاً لأنك تدفع فقط مقابل رحلة ذهاب وإياب واحدة إلى LLM، مما يحافظ على حد أدنى من التكاليف التشغيلية. ولكن مع نمو منطق العمل الخاص بك، تقع في فخ الموجّه الضخم (Monolith Prompt Trap). حيث تبدأ في إضافة المزيد من الأدوات، والمزيد من الحالات الاستثنائية (edge cases) للتعامل معها، والمزيد من قيود التنسيق.

[مدخلات المستخدم] ---> [موجّه الـ Single Agent الضخم] ---> [المخرجات]
                            | (15 أداة، 4 تعليمات للشخصية، 3 مخططات)
                            v
                 (تدهور الانتباه / هلوسة الأدوات)

عندما يتجاوز حجم الموجّه 2,000 token ويتخطى عدد الأدوات المتاحة خمس أدوات، يتدهور أداء النموذج (LLM) بشكل متسارع. يعاني النموذج من تدهور الانتباه المعروف بـ "الضياع في المنتصف" (lost in the middle). ويبدأ في هلوسة وسائط الأدوات (tool arguments)، وتجاهل القيود السلبية، والفشل في اتباع مخطط المخرجات (output schema). بالنسبة لتطبيق SaaS في بيئة التشغيل الفعلي، يترجم هذا مباشرة إلى تجارب مستخدم معطلة وأعباء دعم عملاء مرتفعة.

على الجانب الآخر تقع الهندسة المفرطة للـ multi-agent. وغالباً ما يكون هذا مدفوعاً بإطارات عمل الوكلاء عالية المستوى التي تشجعك على إنشاء "وكيل" (agent) لكل اسم أو دور في عملك. فينتهي بك الأمر مع "وكيل باحث" (Researcher Agent)، و"وكيل كاتب" (Writer Agent)، و"وكيل محرر" (Editor Agent)، و"وكيل ناشر" (Publisher Agent) يتحدثون جميعاً مع بعضهم البعض في حلقة محادثة مستمرة.

في بيئة التشغيل الفعلي (production)، يعد نمط المحادثة هذا كارثة مالية وتشغيلية. نظراً لأن عمليات تسليم المهام (handovers) غير منظمة وتعتمد على المحادثة، فإنك تفقد الحتمية (determinism). لا يمكنك كتابة اختبارات وحدة (unit tests) موثوقة لنظام قد يطلب فيه الوكيل (A) توضيحاً من الوكيل (B) بخمسين طريقة مختلفة. والأسوأ من ذلك، أن كل جولة تواصل بين وكيل وآخر تضيف ما بين 1.5 إلى 3 ثوانٍ من زمن الاستجابة (latency) وتستهلك آلاف الـ tokens غير الضرورية، مما يضخم فواتير البنية التحتية الشهرية بهدوء مع تقليل رضا المستخدمين.

مصفوفة القرار: متى يجب تقسيم الحالة (State)؟

المحرك المعماري الأساسي لاختيار نظام multi-agent ليس تعقيد المهمة؛ بل هو فصل الحالة (state separation).

إذا كانت الأجزاء المختلفة من سير العمل (workflow) تتطلب أدوات مختلفة، وموجّهات نظام مختلفة، وصلاحيات وصول مختلفة، فيجب عليك تقسيم الحالة. أما إذا كانت تشترك في نفس السياق، فاحتفظ بها في وكيل واحد (single agent) أو نمط توجيه بسيط (router pattern). الفشل في التقسيم عند الحاجة يعرضك لمخاطر جسيمة تتعلق بالامتثال وتسريب البيانات.

قيم سير العمل الخاص بك بناءً على هذه المعايير الهندسية الأربعة لتقليل أعباء التطوير وتكاليف التشغيل:

1. ▸كثافة الأدوات ودقة الاختيار (Tool Density and Selection Accuracy): عندما يُعرض على النموذج (LLM) أكثر من خمس أدوات، تنخفض دقة اختياره للأداة المناسبة. إذا كان سير عملك يتطلب 15 أداة مختلفة (مثل استعلامات قواعد البيانات، تحديثات Salesforce CRM، استدعاءات Stripe API، وإرسال البريد الإلكتروني)، فإن الوكيل الواحد سيفشل حتماً. يجب تقسيم هذه المهام بين وكلاء متخصصين، بحيث يتعامل كل منهم مع 3-4 أدوات فقط، لتجنب أخطاء التنفيذ المكلفة.
2. ▸تعارض موجّهات النظام (System Prompt Contradiction): إذا طلبت من النموذج في نفس الموجّه أن "يكون مدققاً قانونياً صارماً للغاية" و"كاتب نصوص إبداعياً ومتعاطفاً"، فسينتج نتائج متواضعة في كليهما. تتطلب هذه الشخصيات المتعارضة موجّهات نظام متميزة ويجب فصلها في وكلاء مستقلين لحماية جودة المخرجات.
3. ▸عزل الحالة والمخطط (State and Schema Isolation): في سير عمل دعم العملاء، لا ينبغي للوكيل الذي يتحقق من سياسة استرداد الأموال أن يمتلك رمز وصول الكتابة (write-access token) لـ Stripe API في نافذة السياق الخاصة به. من خلال عزل منطق تنفيذ الدفع في "وكيل معاملات" (Transaction Agent) مخصص يتلقى فقط مخططات الدفع التي تم التحقق منها، فإنك تقضي على المخاطر الأمنية ومخاطر الامتثال الحرجة.
4. ▸حتمية مسار التنفيذ (Execution Path Determinism): إذا كان سير عملك خطياً (الخطوة أ -> الخطوة ب -> الخطوة ج)، فاستخدم وكيلاً واحداً (single agent) مع مخرجات مهيكلة (structured outputs)، أو سكربت بسيطاً ومبرمجاً بشكل ثابت. يحافظ هذا على زمن استجابة منخفض ومتوقع. أما إذا كان سير عملك غير خطي، ويتطلب حلقات تكرارية، وتفرعاً شرطياً بناءً على مخرجات الأدوات، وإعادة تخطيط ديناميكية، فإن الرسم البياني متعدد الوكلاء ذو الحالة (stateful multi-agent graph) مثل LangGraph هو الخيار الصحيح.

مقاييس الأداء، زمن الاستجابة، والتكلفة

لجعل هذا الأمر ملموساً، قمنا بمقارنة ثلاثة أنماط معمارية في مهمة معقدة لدعم العملاء وتسوية الطلبات. تطلبت المهمة: تحليل بريد إلكتروني وارد، والاستعلام من قاعدة بيانات SQL داخلية عن حالة الطلب، والتحقق من سياسة الأهلية لاسترداد الأموال، وحساب استرداد جزئي، وتنفيذ عملية الاسترداد عبر بوابة دفع وهمية، وصياغة رد بريد إلكتروني مخصص.

قمنا بتشغيل 500 حالة اختبار باستخدام نموذج Claude 3.5 Sonnet كنموذج أساسي، وقارنا بين نمط Single Agent Monolith، ونمط Router Pattern (موجه واحد يوجه إلى استدعاءات LLM متخصصة وذات جولة واحدة)، ورسم بياني متعدد الوكلاء ذو حالة (Stateful Multi-Agent Graph).

قياس الأثر التجاري والمالي

لفهم التأثير الفعلي على شركة SaaS متنامية أو مؤسسة كبرى تعالج 10,000 معاملة شهرياً، دعنا نترجم هذه المقاييس المرجعية إلى تكاليف تشغيلية ووفورات مادية:

• ▸فخ الـ Single Agent Monolith: في حين أنه يكلف 120 دولاراً شهرياً فقط من رسوم واجهة برمجة التطبيقات (LLM API)، فإن معدل فشله البالغ 31.6% يعني 3,160 معاملة فاشلة كل شهر. يتطلب حل هذه الإخفاقات يدوياً موظفي دعم مخصصين. وبافتراض أن متوسط تكلفة الحل اليدوي يبلغ 3.00 دولارات لكل تذكرة من العمالة البشرية، فإن هذا يضيف تكلفة تشغيلية خفية قدرها 9,480 دولاراً شهرياً ويضر بشدة بمعدل الاحتفاظ بالعملاء.
• ▸ميزة الـ Stateful Multi-Agent Graph: تكلفة تشغيل هذا النمط تبلغ 580 دولاراً شهرياً في رسوم LLM API—بزيادة قدرها 460 دولاراً. ومع ذلك، نظراً لأن معدل الفشل ينخفض إلى 5.2% (520 حالة فاشلة فقط)، فإن تكاليف الحل اليدوي تتقلص إلى 1,560 دولاراً شهرياً.
• ▸صافي العائد على الاستثمار (Net ROI): من خلال الاستثمار في بنية stateful multi-agent، فإنك توفر 7,460 دولاراً شهرياً من التكاليف التشغيلية اليدوية، وتقلل من اختناقات استجابة دعم العملاء، وتحقق تجربة عملاء موثوقة للغاية.

إذا كان عملك لا يمكنه تحمل معدل خطأ بنسبة 5%، فإن الـ multi-agent graph هو الخيار الصحيح. أما إذا كنت تبني واجهة محادثة تفاعلية حيث يجب أن يكون زمن الاستجابة (latency) للمستخدم أقل من 3 ثوانٍ، فيجب عليك التحسين باتجاه نمط single-agent router.

تنفيذ عملية تسليم مهام حتمية (Deterministic Handover) في بيئة Multi-Agent باستخدام Python

من منظور تجاري، فإن برمجة منطق التوجيه (routing logic) بشكل ثابت بدلاً من ترك نماذج LLMs تتحدث بحرية هي استراتيجية مباشرة للحد من المخاطر. فهي تضمن التزام نظامك بإرشادات الامتثال والتشغيل الصارمة، مما يمنع سلوكيات LLM غير المتوقعة من إطلاق إجراءات قاعدة بيانات أو استدعاءات API غير مصرح بها. يوفر تنفيذ هذا النمط الحتمي على فرق الهندسة أسابيع من تصحيح الأخطاء (debugging) ويحمي مؤسستك من غرامات اتفاقية مستوى الخدمة (SLA) المكلفة.

إليك تنفيذاً جاهزاً للتشغيل الفعلي (production-grade) لموجه آلة الحالة (state-machine router) في Python باستخدام Pydantic للتحقق من صحة الحالة و LiteLLM لتنفيذ النموذج. يضمن هذا النمط التحقق الصارم من انتقالات الحالة (state transitions) قبل استدعاء الوكيل التالي.

import os
from typing import Literal, Dict, Any, Optional
from pydantic import BaseModel, Field
from litellm import completion

# Define the global state schema
class AgentState(BaseModel):
    customer_id: str
    original_query: str
    extracted_intent: Optional[str] = None
    refund_amount: float = 0.0
    verification_status: Literal["pending", "approved", "rejected"] = "pending"
    next_step: Literal["classifier", "billing_agent", "support_agent", "complete"] = "classifier"
    execution_log: list[str] = Field(default_factory=list)

# Define structured outputs for the routing decisions
class RoutingDecision(BaseModel):
    intent: str
    target_agent: Literal["billing_agent", "support_agent"]
    reason: str

class BillingAction(BaseModel):
    refund_approved: bool
    calculated_amount: float
    justification: str

# Helper to log state changes
def log_transition(state: AgentState, message: str):
    state.execution_log.append(message)
    print(f"[{state.next_step.upper()}] {message}")

# Agent 1: The Classifier (Routes based on intent)
def run_classifier(state: AgentState) -> AgentState:
    log_transition(state, "Analyzing customer query for routing.")
    
    prompt = f"Analyze this customer query: '{state.original_query}'. Determine if it is a billing issue or a general support issue."
    
    response = completion(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format=RoutingDecision
    )
    
    # Parse structured response
    decision = RoutingDecision.model_validate_json(response.choices[0].message.content)
    
    state.extracted_intent = decision.intent
    state.next_step = decision.target_agent
    log_transition(state, f"Routed to {decision.target_agent}. Reason: {decision.reason}")
    return state

# Agent 2: The Billing Specialist (Handles calculation)
def run_billing_agent(state: AgentState) -> AgentState:
    log_transition(state, "Processing billing and refund eligibility.")
    
    prompt = f"Customer Query: {state.original_query}. Calculate eligible refund. Customer ID: {state.customer_id}."
    
    response = completion(
        model="anthropic/claude-3-5-sonnet",
        messages=[{"role": "user", "content": prompt}],
        response_format=BillingAction
    )
    
    action = BillingAction.model_validate_json(response.choices[0].message.content)
    
    state.refund_amount = action.calculated_amount
    state.verification_status = "approved" if action.refund_approved else "rejected"
    state.next_step = "support_agent"  # Handover to support for drafting the final response
    
    log_transition(state, f"Refund {state.verification_status}: ${state.refund_amount}. Handing over to support.")
    return state

# Agent 3: The Support Agent (Drafts final communication)
def run_support_agent(state: AgentState) -> AgentState:
    log_transition(state, "Drafting final response to customer.")
    
    prompt = (
        f"Draft a response. Intent: {state.extracted_intent}. "
        f"Refund Status: {state.verification_status}. Amount: ${state.refund_amount}. "
        f"Original Query: {state.original_query}"
    )
    
    response = completion(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    
    log_transition(state, f"Drafted response: {response.choices[0].message.content[:60]}...")
    state.next_step = "complete"
    return state

# Orchestrator / State Machine Runner
def execute_workflow(initial_state: AgentState) -> AgentState:
    state = initial_state
    max_steps = 5
    step_count = 0
    
    while state.next_step != "complete" and step_count < max_steps:
        step_count += 1
        if state.next_step == "classifier":
            state = run_classifier(state)
        elif state.next_step == "billing_agent":
            state = run_billing_agent(state)
        elif state.next_step == "support_agent":
            state = run_support_agent(state)
        else:
            raise ValueError(f"Unknown state: {state.next_step}")
            
    if step_count >= max_steps:
        log_transition(state, "Workflow terminated: Max iterations reached to prevent infinite loop.")
        
    return state

# Example Execution
if __name__ == "__main__":
    # Ensure API keys are set in environment
    # os.environ["OPENAI_API_KEY"] = "your-key"
    # os.environ["ANTHROPIC_API_KEY"] = "your-key"
    
    test_state = AgentState(
        customer_id="cust_9921",
        original_query="I was charged twice for my subscription yesterday