وكيل واحد أم وكلاء متعددون: أربعة اختبارات للحسم، ونمط الموجه (Router) الذي يفوز عادةً

لست بحاجة إلى نظام متعدد الوكلاء (multi-agent) لكتابة بريد إلكتروني، أو صياغة استعلام SQL، أو تحليل ملف PDF. استدعاء واحد للنموذج بهيكلة جيدة يكفي لإنجاز كل ذلك. ومع ذلك، تقوم العديد من قواعد الأكواد البرمجية للوكلاء بتوجيه طلب واحد عبر أربعة أو خمسة وكلاء فرعيين يتحدثون مع بعضهم البعض بنصوص نثرية. النتيجة: نظام بطيء، ومكلف، وصعب التصحيح (debugging).

الخطأ المعاكس يكلف نفس القدر. حشر مسار عمل متشعب، متعدد الأدوات، ومقيد بموافقات (approval-gated) في موجّه (prompt) واحد قد ينجح في العرض التجريبي (demo)، لكنه يفشل مع زيادة حجم العمل.

تحديد عدد الوكلاء الذي يحتاجه النظام هو أول قرار معماري في أي مشروع وكيل (agent). هذا القرار يحدد زمن الاستجابة (latency)، والتكلفة، ونطاق تأثير الخطأ (blast radius)، وما إذا كان بإمكانك كتابة اختبارات من الأساس. التوصية أدناه هي نقطة البداية الافتراضية لدينا: موجه واحد (router)، وعدد قليل من المتخصصين، وعمليات تسليم (handovers) تُمرر كبيانات بدلاً من محادثات. هناك أربعة اختبارات تحدد متى يجب الخروج عن هذا النمط. أما إطار العمل الذي يجب البناء عليه فهو سؤال منفصل، وتجيب عليه مقارنة أطر العمل المرتبطة أدناه.

طريقتان للوقوع في الخطأ

الموجّه المتجانس (The monolith prompt). استدعاء واحد للنموذج يحمل كل أداة، وكل شخصية (persona)، وكل مخطط (schema)، وكل حالة حافة (edge case) في موجّه نظام واحد. يكون هذا النهج رخيصاً وسريعاً طالما أنه يعمل. ولكن مع تراكم الأدوات والتعليمات، يبدأ النموذج في اختيار الأداة الخطأ. يتجاهل القيود السلبية مثل "لا تذكر الأسعار" وينحرف عن مخطط المخرجات (output schema).

هناك نتيجتان منشورتان تشرحان هذا التدهور. أظهر ليو وزملاؤه في جامعة ستانفورد ذلك في ورقة "Lost in the Middle". تستخدم النماذج المعلومات الموجودة في بداية ونهاية المدخلات الطويلة بشكل أفضل بكثير من المعلومات المدفونة في المنتصف. اختيار الأدوات يأخذ نفس الشكل. يوصي دليل استدعاء الدوال (function-calling) من OpenAI بتوفير أقل من 20 دالة في بداية الدور "للحصول على دقة أعلى". تنص وثائق البحث عن الأدوات من Anthropic على أن دقة الاختيار تتدهور بمجرد تجاوز 30 إلى 50 أداة متاحة. يعتمد الحد الدقيق على النموذج والموجّه الخاص بك، لكن الاتجاه العام ثابت: المزيد من الأدوات والتعليمات في موجّه واحد يعني اختياراً أقل موثوقية.

[User input] -> [One agent: 15 tools, 4 personas, 3 schemas] -> [Output]
                                  |
                                  v
                  attention drift, wrong tool, schema drift

السرب الحواري (The conversational swarm). تشجع بعض أطر العمل على تخصيص وكيل لكل اسم في عملك: باحث، كاتب، محرر، ناشر. يقوم كل وكيل بتسليم المهمة للتالي بلغة طبيعية. يبدو هذا سحرياً في العرض التجريبي، لكنه في بيئة الإنتاج (production) يُدخل عدم الحتمية (nondeterminism). يمكن للوكيل "أ" أن يطلب توضيحاً من الوكيل "ب" بعشرات الطرق. يصبح من الصعب كتابة اختبارات وحدة (unit tests) صارمة، وكل قفزة تضيف دورة كاملة لاستدعاء النموذج. تتخذ عائلات أطر العمل الثلاث مواقف مختلفة تجاه هذا الأمر، وكيف تختلف عائلات أطر العمل الثلاث هو المكان الذي يُتخذ فيه هذا القرار.

يشترك كلا الخطأين في سبب جذري واحد: تم اختيار البنية المعمارية قبل كتابة وتحديد مسار العمل (workflow).

الاختبارات الأربعة

السؤال ليس ما إذا كانت المهمة معقدة، بل ما إذا كان مسار العمل يحتاج إلى حالة معزولة (isolated state). إذا كانت المراحل المختلفة تحتاج إلى أدوات أو موجّهات أو صلاحيات (credentials) مختلفة، قم بالتقسيم. وإذا كانت تشترك في سياق واحد، احتفظ بوكيل واحد أو موجه (router) مبرمج مسبقاً. هناك أربعة اختبارات تحسم الأمر.

1. حجم كتالوج الأدوات. عندما يضطر وكيل واحد للاختيار من قائمة طويلة، تنخفض دقة الاختيار. قراءة قواعد البيانات، والكتابة في نظام إدارة علاقات العملاء (CRM)، واستدعاءات الدفع، وإجراءات البريد الإلكتروني والتقويم في كتالوج واحد هو الشكل المعتاد. الوكلاء المتخصصون الذين يرى كل منهم عدداً قليلاً من الأدوات يرتكبون أخطاء أقل. التوجيهات المذكورة أعلاه تعطي النطاق التقريبي. قم بقياس نقطة الانكسار (break point) في مسار عملك قبل اتخاذ القرار.

2. الشخصيات أو التعليمات المتضاربة. وضع تعليمات مثل "كن مدققاً قانونياً صارماً" و"كن كاتب محتوى دافئاً ومتعاطفاً" في نفس الموجّه يعطي مخرجات متواضعة في كليهما. الشخصيات المتضاربة مكانها وكلاء منفصلون.

3. عزل الصلاحيات والمخططات. الوكيل الذي يتحقق من سياسة الاسترجاع يجب ألا يمتلك رمز الكتابة (write token) الخاص ببوابة الدفع. والوكيل الذي يقرأ قاعدة معرفية عامة يجب ألا يصل إلى قاعدة بيانات الموارد البشرية. التقسيم هنا هو متطلب أمني، وليس تفضيلاً هندسياً. مبدأ تقليل البيانات في المادة 5(1)(ج) من القانون العام لحماية البيانات (GDPR) ينطبق على كل مكون من مكونات الذكاء الاصطناعي بقدر ما ينطبق على أي نظام آخر. قوانين حماية البيانات الخليجية تتبع نفس النهج، وقائمة التحقق من نظام حماية البيانات الشخصية (PDPL) تغطيها. إذا كان المكون لا يحتاج إلى وصول، فلا ينبغي أن يحصل عليه.

4. شكل مسار التنفيذ. المسار الخطي (أ، ثم ب، ثم ج) يتطلب وكيلاً واحداً بمخرجات مهيكلة. حتى السكربت البسيط بخطوة نموذج واحدة سيفي بالغرض. أما الحلقات (loops)، والفروع الشرطية، وإعادة التخطيط فتتطلب رسماً بيانياً يحتفظ بالحالة (stateful graph) مثل LangGraph.

تكلفة كل شكل

تعامل مع شكل التكاليف بدلاً من الاعتماد على معايير قياس (benchmarks) خاصة بشخص آخر. بالنسبة لمسار عمل يعالج N من الطلبات شهرياً:

• ▸وكيل واحد. دورة استدعاء واحدة للنموذج لكل طلب. زمن الاستجابة هو وقت وصول أول رمز (time to first token) زائد عملية توليد واحدة. التكلفة هي رموز الإدخال والإخراج لاستدعاء واحد. من الصعب تحديد مكان الفشل، لأن كل شيء يحدث في موجّه واحد.
• ▸موجه (Router) بالإضافة إلى متخصصين. استدعاء تصنيف واحد زائد استدعاء متخصص واحد. يتضاعف زمن الاستجابة والتكلفة تقريباً. من السهل تحديد مكان الفشل: إما أن الموجه اختار بشكل خاطئ أو أن المتخصص أخطأ. هذه هي نقطة البداية الافتراضية لدينا.
• ▸رسم بياني متعدد الوكلاء يحتفظ بالحالة (Stateful multi-agent graph). عدة استدعاءات لكل طلب، غالباً مع محاولات إعادة (retries) وحلقات. يزداد زمن الاستجابة والتكلفة مع عدد العقد (nodes) التي يتم المرور بها. في المقابل، تحصل على العزل، وقابلية الاختبار، وبوابات الموافقة عند عقد محددة.

دوّن زمن الاستجابة المستهدف من البداية للنهاية (end-to-end latency) ومدى تسامحك مع الأخطاء. ثم قم بقياس الأشكال الثلاثة على بياناتك الخاصة. إذا كان المستخدمون لا يستطيعون الانتظار عبر ثلاثة استدعاءات متتالية للنموذج، فإن المسار التفاعلي لا يمكنه تحمل رسم بياني عميق (deep graph). ادفع العمل الثقيل إلى مهمة غير متزامنة (asynchronous job) وأرجع إيصالاً (receipt) للمستخدم.

يغير الموجه (router) أيضاً من تكلفة الفشل. في النظام المتجانس (monolith)، الإجابة الخاطئة قد تأتي من أي من الأدوات الخمس عشرة أو الشخصيات الأربع. أما في الموجه، يُظهر التتبع (trace) استدعاء تصنيف واحداً واستدعاء متخصص واحداً. إما أن الموجه اختار بشكل خاطئ، وهو ما يُعد مثال تدريب مُصنف (labelled training example)، أو أن المتخصص أخطأ، وهو ما يعني موجّهاً ضيقاً يسهل إصلاحه. لهذا السبب يميل نمط الموجه إلى الفوز في الصيانة حتى عندما يخسر في زمن الاستجابة الخام.

عمليات التسليم الحتمية في بايثون

الفشل الأكثر شيوعاً في أكواد الوكلاء المتعددين هو التسليم الحواري (conversational handover). يكتب الوكيل "أ" رسالة حرة، ويقوم الوكيل "ب" بتحليلها، فينحرف شيء ما، وتفشل العملية. الحل هو جعل عمليات التسليم بيانات، وليس محادثة. كل انتقال هو مخطط (schema) مُتحقق منه، والمنسق (orchestrator) هو من يقرر من سيعمل تالياً. النموذج لا يفعل ذلك أبداً.

إليك النمط، باستخدام Pydantic للتحقق من الحالة (state validation) و LiteLLM بحيث يمكن لكل عقدة استخدام نموذج مختلف.

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

# Model choice lives in the environment, not the code. Use a small, fast model
# for routing (a small GPT or Claude model) and a larger model from the same
# families where the node has to reason. Pin exact IDs per environment.
ROUTER_MODEL = os.environ["ROUTER_MODEL"]
REASONING_MODEL = os.environ["REASONING_MODEL"]

# Global state schema. Every node reads and mutates this.
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)

# Structured outputs, so handovers are data rather than prose.
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

def log_transition(state: AgentState, message: str) -> None:
    state.execution_log.append(message)
    print(f"[{state.next_step.upper()}] {message}")

# Node 1: classify intent and pick the next specialist.
def run_classifier(state: AgentState) -> AgentState:
    log_transition(state, "Analyzing customer query for routing.")
    prompt = (
        f"Analyze this customer query: '{state.original_query}'. "
        "Decide if it is a billing issue or a general support issue."
    )
    response = completion(
        model=ROUTER_MODEL,
        messages=[{"role": "user", "content": prompt}],
        response_format=RoutingDecision,
    )
    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

# Node 2: billing specialist. Only this node holds billing credentials.
def run_billing_agent(state: AgentState) -> AgentState:
    log_transition(state, "Processing billing and refund eligibility.")
    prompt = (
        f"Customer query: {state.original_query}. "
        f"Calculate eligible refund. Customer ID: {state.customer_id}."
    )
    response = completion(
        model=REASONING_MODEL,
        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"
    log_transition(
        state,
        f"Refund {state.verification_status}: {state.refund_amount}. Handing to support."
    )
    return state

# Node 3: drafts the customer-facing message. No billing credentials in this context.
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=REASONING_MODEL,
        messages=[{"role": "user", "content": prompt}],
    )
    log_transition(state, f"Drafted response: {response.choices[0].message.content[:60]}...")
    state.next_step = "complete"
    return state

# Orchestrator. Code, not a model, decides who runs next.
def execute_workflow(initial_state: AgentState) -> AgentState:
    state = initial_state
    max_steps = 5  # hard iteration cap, a little above the longest expected path
    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 and state.next_step != "complete":
        log_transition(state, "Workflow halted: max iterations reached. Escalating to a person.")
    return state

if __name__ == "__main__":
    test_state = AgentState(
        customer_id="cust_9921",
        original_query=(
            "I was charged twice for my subscription yesterday. "
            "Please refund the duplicate charge."
        ),
    )
    final_state = execute_workflow(test_state)

هناك ثلاثة أشياء يجب ملاحظتها. أولاً، كل عملية تسليم هي نموذج Pydantic، لذلك فإن الاستجابة المشوهة تفشل بوضوح ومبكراً بدلاً من إفساد الحالة في المراحل التالية (downstream state). ثانياً، عقدة الفوترة هي الوحيدة التي تمتلك صلاحيات الفوترة. لا يمكن لوكيل الدعم لمسها، حتى لو حاول هجوم حقن الموجّه (prompt injection) ذلك. ثالثاً، المنسق (orchestrator) هو من يمتلك التحكم في مسار العمل (control flow). تتخذ النماذج قرارات بشأن البيانات؛ ولا تقرر من سيعمل تالياً.

قم بتبديل النموذج لكل عقدة. نموذج سريع ورخيص يقوم بالتصنيف؛ ونموذج أقوى يعمل فقط عندما تحتاج العقدة إلى الاستنتاج (reasoning). يمكن للتوجيه الدلالي (semantic routing) أن يأخذ هذا إلى أبعد من ذلك، مع التوجيه الدلالي لأرخص نموذج قادر يتم اختياره بناءً على محتوى الطلب، مرة واحدة لكل جلسة.

قابلية المراقبة وحراس الحلقات (Loop guards)

بمجرد التقسيم إلى عقد، يصبح تصحيح الأخطاء (debugging) مشكلة أنظمة موزعة. لا يمكنك البحث عن جمل الطباعة (print statements) عبر خمس عمليات مختلفة. يجب أن تُصدر كل عقدة تتبعات مهيكلة (structured traces) مرتبطة بـ trace_id واحد، بحيث يمكن إعادة تشغيل العملية الفاشلة من البداية إلى النهاية.

[Trace: request_4821]
├── Node: Classifier    (time to first token, generation, cost per call)
├── Node: Billing       (time to first token, generation, cost per call)
└── Node: Support       (time to first token, generation, cost per call)

أدوات التتبع مثل Langfuse مصممة خصيصاً لهذا الغرض. التتبعات هي أيضاً المادة الخام لـ التقييم على مستوى الخطوة، والذي يقيّم كل عملية تسليم بدلاً من الإجابة النهائية فقط.

الأمر الثاني غير القابل للتفاوض هو وضع حد أقصى صارم للتكرارات (iteration cap) على مستوى المنسق. إذا كان بإمكان عقدتين استدعاء بعضهما البعض، فهناك مسار يمكن أن يتبادلا فيه الأدوار (ping-pong) في حالة حافة (edge case) ويستنزفا رصيد الـ API دون رقابة. وجود حارس max_iterations، يتم تعيينه أعلى قليلاً من أطول مسار متوقع، بالإضافة إلى إيقاف سلس ينبه شخصاً مسؤولاً، هو أرخص تأمين متاح. الحد الأقصى الصارم للتكرارات والتسليم النظيف هما اثنان من الضوابط السبعة المذكورة في مقال مركز الوكلاء.

الأسئلة الشائعة

كيف نختار بين نمط الموجه (router) ورسم بياني كامل يحتفظ بالحالة (stateful graph)؟ ارسم مسار العمل. إذا كان المسار عبارة عن شجرة (تصنيف، ثم يعمل متخصص واحد، ثم ينتهي)، فالموجه يكفي، وسيكون أسرع وأرخص. أما إذا كان المسار يحتوي على حلقات (loops)، أو محاولات إعادة تعتمد على مخرجات الأدوات، أو عقد موافقة يمكنها إرسال العمل مرة أخرى إلى المراحل السابقة، فأنت بحاجة إلى رسم بياني يحتفظ بالحالة. لا تدفع ثمن تعقيد الرسم البياني الذي لن تستخدمه.

كيف نمنع الحلقات اللانهائية بين الوكلاء؟ عبر حارسين. عداد تكرار صارم في المنسق (orchestrator)، وليس في أي وكيل بمفرده، يتوقف بعد عدد ثابت من الانتقالات. وتحقق صارم من المخطط (schema validation) عند كل عملية تسليم. الوكيل الذي يتلقى حالة مشوهة يوجهها إلى شخص أو عقدة خطأ، ولا يعيدها أبداً إلى المرسل. حلقات التوضيح هي مصدر الفواتير الباهظة.

هل يمكننا المزج بين النماذج مفتوحة المصدر والمملوكة عبر العقد؟ نعم، وبالنسبة للعديد من أعباء عمل الإنتاج، يجب عليك ذلك. التصنيف والتوجيه يناسبان النماذج الأصغر مفتوحة الأوزان (open-weight) على وحدات معالجة الرسومات (GPUs) الخاصة بك أو عبر واجهة برمجة تطبيقات (API) رخيصة. احتفظ بالنماذج المملوكة الأقوى للعقدة أو العقدتين اللتين تحتاجان إلى استنتاج متعدد الخطوات. بوابة مثل LiteLLM تجعل التبديل مجرد تغيير في الإعدادات بدلاً من إعادة كتابة الكود. بالنسبة لجانب الاستنتاج المحلي، راجع دليل سرعة النماذج اللغوية الكبيرة المحلية (on-prem).

القاعدة العملية

ابدأ بوكيل واحد. هذا الأسبوع، قم بتشغيل مسار عملك الحقيقي من خلاله ودوّن أين ينكسر: أخطاء في اختيار الأدوات، انحراف المخطط، أو حدود صلاحيات لا يمكنك الدفاع عنها. قم بالتقسيم عند هذه الحدود بالضبط، وفقط عند هذه الحدود. احتفظ بالتحكم في مسار العمل (control flow) في المنسق الخاص بك، وليس في محادثة بين نموذج وآخر. ضع التتبع (tracing) وحدّ الحلقات (loop cap) في اليوم الأول، وليس بعد أول فاتورة باهظة.

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