نمط المشرف — الغوص العميق في LangGraph: سير عمل الذكاء الاصطناعي الإنتاجية ذات الحالة — Nerd Level Tech

لماذا أنظمة الوكلاء المتعددة مهمة

سيناريو إنتاجي حقيقي (يناير 2026):

احتاجت شركة تقنية مالية إلى تحليل طلبات القروض مع فحوصات متخصصة متعددة: التحقق من درجة الائتمان، تحليل الدخل، كشف الاحتيال، والامتثال التنظيمي. أصبح وكيل متجانس واحد غير قابل للإدارة عند 5,000 سطر من الكود. بإعادة الهيكلة إلى نمط المشرف مع 4 وكلاء متخصصين فرعيين، قللوا تعقيد الكود بنسبة 60%، وحسّنوا الدقة بنسبة 25%، وأمكنهم تحديث كل متخصص بشكل مستقل دون إعادة نشر النظام بالكامل.

هذا الدرس يعلمك: كيفية تنفيذ نمط المشرف في LangGraph النقي لتنسيق وكلاء متخصصين متعددين في سير العمل الإنتاجي.

ما هو نمط المشرف؟

نمط المشرف هو بنية وكلاء متعددة حيث:

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

                    ┌─────────────────┐
                    │    المشرف       │
                    │   (المنسق)      │
                    └────────┬────────┘
                             │
          ┌──────────────────┼──────────────────┐
          ▼                  ▼                  ▼
    ┌──────────┐       ┌──────────┐       ┌──────────┐
    │   وكيل   │       │   وكيل   │       │   وكيل   │
    │  البحث   │       │ التحليل  │       │ الكتابة  │
    └──────────┘       └──────────┘       └──────────┘

رؤية رئيسية: المشرف لا يقوم أبداً بالعمل الفعلي. هو فقط ينسق، يوجه، ويتحقق من الجودة. هذا الفصل حاسم لقابلية الصيانة.

تنفيذ المشرف الإنتاجي (LangGraph 1.0.5)

from typing import TypedDict, Annotated, Literal, List, Optional
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
import operator

# -----------------------------------------------------------------------------
# تعريف الحالة
# -----------------------------------------------------------------------------
class SupervisorState(TypedDict):
    """الحالة المشتركة بين المشرف وجميع الوكلاء."""
    # الإدخال
    task: str

    # قرارات المشرف
    next_agent: Optional[str]
    agent_instructions: Optional[str]

    # مخرجات الوكلاء (تتراكم من جميع الوكلاء)
    agent_outputs: Annotated[List[dict], operator.add]

    # التحكم في سير العمل
    completed_agents: Annotated[List[str], operator.add]
    iteration: int
    max_iterations: int

    # المخرجات النهائية
    final_response: Optional[str]

# -----------------------------------------------------------------------------
# إعداد LLM
# -----------------------------------------------------------------------------
supervisor_llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

# -----------------------------------------------------------------------------
# عقدة المشرف
# -----------------------------------------------------------------------------
def supervisor_node(state: SupervisorState) -> dict:
    """
    يحلل المشرف المهمة ويقرر أي وكيل يستدعي بعد ذلك.
    هذا هو 'العقل' لنظام الوكلاء المتعددة.
    """
    task = state["task"]
    completed = state.get("completed_agents", [])
    outputs = state.get("agent_outputs", [])
    iteration = state.get("iteration", 0)
    max_iterations = state.get("max_iterations", 10)

    # التحقق من شروط الإنهاء
    if iteration >= max_iterations:
        return {
            "next_agent": "synthesizer",
            "agent_instructions": "تم الوصول للحد الأقصى للتكرارات. لخّص النتائج المتاحة.",
            "iteration": iteration + 1
        }

    # بناء السياق من المخرجات السابقة
    context = ""
    for output in outputs:
        context += f"\n[{output['agent']}]: {output['result'][:500]}..."

    # موجه المشرف - يستخدم LLM لاتخاذ القرار
    supervisor_prompt = f"""أنت مشرف ينسق فريق بحث متعدد الوكلاء.

المهمة: {task}
الوكلاء المكتملون: {completed}
المخرجات السابقة: {context if context else "لا شيء بعد"}

الوكلاء المتاحون:
- researcher: يجمع المعلومات والحقائق
- analyst: يحلل البيانات ويحدد الأنماط
- writer: يجمع النتائج في استجابة متماسكة
- synthesizer: ينشئ المخرجات النهائية

أجب بالتنسيق:
NEXT_AGENT: [اسم_الوكيل]
INSTRUCTIONS: [تعليمات محددة]"""

    response = supervisor_llm.invoke(supervisor_prompt)
    # تحليل الاستجابة...

    return {
        "next_agent": next_agent,
        "agent_instructions": instructions,
        "iteration": iteration + 1
    }

# -----------------------------------------------------------------------------
# عقد الوكلاء المتخصصين
# -----------------------------------------------------------------------------
def researcher_node(state: SupervisorState) -> dict:
    """وكيل البحث يجمع المعلومات."""
    task = state["task"]
    instructions = state.get("agent_instructions", "")

    prompt = f"""أنت متخصص في البحث.

المهمة الرئيسية: {task}
التعليمات المحددة: {instructions}

اجمع المعلومات والحقائق والبيانات ذات الصلة."""

    response = agent_llm.invoke(prompt)

    return {
        "agent_outputs": [{
            "agent": "researcher",
            "result": response.content,
            "instructions_received": instructions
        }],
        "completed_agents": ["researcher"]
    }

def analyst_node(state: SupervisorState) -> dict:
    """وكيل التحليل يحدد الأنماط والرؤى."""
    task = state["task"]
    instructions = state.get("agent_instructions", "")
    previous_outputs = state.get("agent_outputs", [])

    # الحصول على مخرجات البحث للتحليل
    research_context = ""
    for output in previous_outputs:
        if output["agent"] == "researcher":
            research_context += output["result"] + "\n"

    prompt = f"""أنت متخصص في التحليل.

المهمة الرئيسية: {task}
التعليمات المحددة: {instructions}

بيانات البحث للتحليل:
{research_context if research_context else "لا يوجد بحث سابق متاح."}

حلل البيانات، حدد الأنماط، واستخلص الرؤى."""

    response = agent_llm.invoke(prompt)

    return {
        "agent_outputs": [{
            "agent": "analyst",
            "result": response.content
        }],
        "completed_agents": ["analyst"]
    }

def writer_node(state: SupervisorState) -> dict:
    """وكيل الكتابة يجمع النتائج في نص متماسك."""
    task = state["task"]
    instructions = state.get("agent_instructions", "")
    previous_outputs = state.get("agent_outputs", [])

    # تجميع جميع المخرجات السابقة
    context = ""
    for output in previous_outputs:
        context += f"\n## مخرجات {output['agent'].upper()}:\n{output['result']}\n"

    prompt = f"""أنت متخصص في الكتابة.

المهمة الرئيسية: {task}
التعليمات المحددة: {instructions}

مخرجات الفريق:
{context if context else "لا توجد مخرجات سابقة متاحة."}

اكتب تجميعاً واضحاً ومتماسكاً للنتائج."""

    response = agent_llm.invoke(prompt)

    return {
        "agent_outputs": [{
            "agent": "writer",
            "result": response.content
        }],
        "completed_agents": ["writer"]
    }

def synthesizer_node(state: SupervisorState) -> dict:
    """عقدة التجميع النهائية - تنشئ الاستجابة النهائية."""
    task = state["task"]
    previous_outputs = state.get("agent_outputs", [])

    context = ""
    for output in previous_outputs:
        context += f"\n## {output['agent'].upper()}:\n{output['result']}\n"

    prompt = f"""أنشئ استجابة نهائية شاملة لهذه المهمة:

المهمة: {task}

مساهمات الفريق:
{context}

جمّع جميع المساهمات في استجابة نهائية واحدة منظمة جيداً."""

    response = supervisor_llm.invoke(prompt)

    return {
        "final_response": response.content,
        "completed_agents": ["synthesizer"]
    }

# -----------------------------------------------------------------------------
# بناء الرسم البياني
# -----------------------------------------------------------------------------
def build_supervisor_graph():
    """بناء رسم بياني المشرف متعدد الوكلاء الكامل."""
    graph = StateGraph(SupervisorState)

    # إضافة جميع العقد
    graph.add_node("supervisor", supervisor_node)
    graph.add_node("researcher", researcher_node)
    graph.add_node("analyst", analyst_node)
    graph.add_node("writer", writer_node)
    graph.add_node("synthesizer", synthesizer_node)

    # نقطة الدخول
    graph.set_entry_point("supervisor")

    # المشرف يوجه للوكلاء
    graph.add_conditional_edges(
        "supervisor",
        route_supervisor,
        {
            "researcher": "researcher",
            "analyst": "analyst",
            "writer": "writer",
            "synthesizer": "synthesizer"
        }
    )

    # جميع الوكلاء يعودون للمشرف (باستثناء المُجمّع)
    graph.add_edge("researcher", "supervisor")
    graph.add_edge("analyst", "supervisor")
    graph.add_edge("writer", "supervisor")

    # المُجمّع ينهي سير العمل
    graph.add_edge("synthesizer", END)

    return graph.compile()

مبادئ نمط المشرف الرئيسية

المبدأ	التنفيذ
منسق واحد	عقدة مشرف واحدة تتخذ جميع قرارات التوجيه
عزل المتخصصين	كل وكيل له مسؤولية واضحة ومحددة
حالة مشتركة	جميع الوكلاء يقرؤون/يكتبون لهيكل حالة مشترك
التحكم في التكرار	الحد الأقصى للتكرارات يمنع الحلقات اللانهائية
تراكم النتائج	مخفض `operator.add` يجمع جميع مخرجات الوكلاء

استراتيجيات قرار المشرف

الاستراتيجية 1: خط أنابيب تسلسلي

# المشرف يوجه دائماً: researcher -> analyst -> writer -> synthesizer
def sequential_supervisor(state):
    completed = state.get("completed_agents", [])
    if "researcher" not in completed:
        return {"next_agent": "researcher"}
    elif "analyst" not in completed:
        return {"next_agent": "analyst"}
    elif "writer" not in completed:
        return {"next_agent": "writer"}
    return {"next_agent": "synthesizer"}

الاستراتيجية 2: مدفوعة بـ LLM (ديناميكية)

# المشرف يستخدم LLM للقرار بناءً على المهمة والسياق
# (موضح في التنفيذ الرئيسي أعلاه)
# الأفضل للمهام المعقدة حيث التوجيه يعتمد على المحتوى

الاستراتيجية 3: التوجيه الشرطي

# التوجيه بناءً على كلمات نوع المهمة
def conditional_supervisor(state):
    task = state["task"].lower()
    if "analyze" in task or "data" in task:
        return {"next_agent": "analyst"}
    elif "write" in task or "report" in task:
        return {"next_agent": "writer"}
    return {"next_agent": "researcher"}

التحقق من الجودة في المشرف

المشرفون الإنتاجيون يجب أن يتحققوا من مخرجات الوكلاء قبل المتابعة:

def supervisor_with_quality_check(state: SupervisorState) -> dict:
    """مشرف يتحقق من مخرجات الوكلاء."""
    outputs = state.get("agent_outputs", [])

    # التحقق من جودة مخرجات آخر وكيل
    if outputs:
        last_output = outputs[-1]
        result = last_output.get("result", "")

        # فحوصات الجودة
        if len(result) < 100:
            # المخرجات قصيرة جداً - أعد المحاولة بتعليمات أكثر تحديداً
            return {
                "next_agent": last_output["agent"],
                "agent_instructions": f"المخرجات السابقة كانت مختصرة جداً. قدم تفاصيل أكثر."
            }

        if "error" in result.lower() or "unable" in result.lower():
            # الوكيل أبلغ عن صعوبة - جرب نهجاً بديلاً
            return {
                "next_agent": "researcher",
                "agent_instructions": "النهج السابق واجه مشاكل. جرب استراتيجية بحث مختلفة."
            }

    # تابع مع التوجيه الطبيعي
    return normal_routing_logic(state)

إضافة وكلاء متخصصين مخصصين

def create_specialist_agent(
    name: str,
    system_prompt: str,
    llm: ChatOpenAI = None
) -> callable:
    """دالة مصنع لإنشاء وكلاء متخصصين."""
    llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

    def agent_node(state: SupervisorState) -> dict:
        task = state["task"]
        instructions = state.get("agent_instructions", "")
        previous_outputs = state.get("agent_outputs", [])

        context = "\n".join([
            f"[{o['agent']}]: {o['result'][:500]}"
            for o in previous_outputs
        ])

        prompt = f"""{system_prompt}

المهمة: {task}
التعليمات: {instructions}
السياق: {context}"""

        response = llm.invoke(prompt)

        return {
            "agent_outputs": [{
                "agent": name,
                "result": response.content
            }],
            "completed_agents": [name]
        }

    return agent_node

# إنشاء متخصصين مخصصين
code_reviewer = create_specialist_agent(
    name="code_reviewer",
    system_prompt="أنت مراجع كود كبير. حلل الكود للأخطاء والمشاكل الأمنية وأفضل الممارسات."
)

security_analyst = create_specialist_agent(
    name="security_analyst",
    system_prompt="أنت متخصص أمني. حدد الثغرات واقترح التخفيفات."
)

# أضف للرسم البياني
graph.add_node("code_reviewer", code_reviewer)
graph.add_node("security_analyst", security_analyst)

معالجة الأخطاء في نمط المشرف

def robust_agent_node(state: SupervisorState) -> dict:
    """وكيل مع معالجة أخطاء شاملة."""
    try:
        # منطق الوكيل الطبيعي
        response = agent_llm.invoke(prompt)
        return {
            "agent_outputs": [{
                "agent": "researcher",
                "result": response.content,
                "status": "success"
            }],
            "completed_agents": ["researcher"]
        }
    except Exception as e:
        # سجّل الفشل لكن لا تعطل سير العمل
        return {
            "agent_outputs": [{
                "agent": "researcher",
                "result": f"خطأ: {str(e)}",
                "status": "failed",
                "error_type": type(e).__name__
            }],
            "completed_agents": ["researcher"]
        }

def supervisor_handles_failures(state: SupervisorState) -> dict:
    """مشرف يتعامل مع فشل الوكلاء بأناقة."""
    outputs = state.get("agent_outputs", [])

    # التحقق من الفشل
    failed_agents = [o for o in outputs if o.get("status") == "failed"]

    if failed_agents:
        last_failure = failed_agents[-1]
        if state["iteration"] < 3:
            return {
                "next_agent": last_failure["agent"],
                "agent_instructions": "المحاولة السابقة فشلت. حاول مرة أخرى بنهج أبسط."
            }
        else:
            return {
                "next_agent": "synthesizer",
                "agent_instructions": "بعض الوكلاء فشلوا. جمّع مع النتائج المتاحة."
            }

    return normal_routing(state)

أسئلة المقابلة الشائعة

س1: "لماذا نستخدم نمط المشرف بدلاً من وكيل واحد مع أدوات؟"

إجابة قوية:

"نمط المشرف أفضل عندما تكون لديك مهام معقدة تتطلب خبرة متخصصة. وكيل واحد مع أدوات يصبح صعب الإدارة مع 20+ أداة ويفقد التركيز. نمط المشرف يوفر: (1) فصل واضح للمسؤوليات - كل متخصص يتفوق في شيء واحد، (2) توسيع وتحديث مستقل للوكلاء، (3) إمكانية مراقبة أفضل - يمكنك تتبع أي وكيل فعل ماذا، (4) اختبار أسهل - اختبر كل متخصص بمعزل."

س2: "كيف تمنع الحلقات اللانهائية في نمط المشرف؟"

الإجابة:

"ثلاث آليات: (1) حد تكرار صارم عبر max_iterations في الحالة، يُفحص من المشرف قبل كل قرار. (2) تتبع completed_agents وتجنب التكرار غير الضروري. (3) recursion_limit في LangGraph وقت التجميع كشبكة أمان. المشرف نفسه يجب أن يكون لديه منطق للتوجيه إلى 'synthesizer' عند اكتمال العمل الكافي."

س3: "كيف تتعامل مع فشل الوكيل في الإنتاج؟"

الإجابة:

"لف كل عقدة وكيل في try-catch، وسجّل الفشل في الحالة مع حقل 'status'. يمكن للمشرف بعدها أن يقرر إعادة المحاولة بتعليمات مختلفة، أو التخطي لوكيل آخر، أو المتابعة للتجميع بنتائج جزئية. لا تدع فشل وكيل واحد يعطل سير العمل بالكامل."

س4: "متى تختار توجيه مدفوع بـ LLM مقابل قائم على القواعد؟"

الإجابة:

"قائم على القواعد لسير العمل المتوقع والمحدد جيداً حيث المسار معروف مسبقاً - إنه أسرع وأرخص. مدفوع بـ LLM للمهام المعقدة حيث التوجيه يعتمد على تحليل المحتوى أو تقييم الجودة أو الشروط الديناميكية. في الممارسة، استخدم هجيناً: قائم على القواعد للقرارات الواضحة، LLM للحالات الغامضة."

النقاط الرئيسية

نمط المشرف = منسق واحد + متخصصون متعددون
تراكم الحالة مع operator.add يحفظ جميع مخرجات الوكلاء
حدود التكرار تمنع حلقات المشرف-الوكيل اللانهائية
التوجيه المدفوع بـ LLM يمكّن تحليل المهام الديناميكي
دوال المصنع تجعل إضافة متخصصين جدد سهلة
كل وكيل يعود للمشرف للقرار التالي (باستثناء المُجمّع النهائي)
فحوصات الجودة في المشرف تضمن معايير المخرجات
معالجة الأخطاء تسمح بالتدهور الأنيق

التالي: تعلم فرق الوكلاء المتعددة الهرمية حيث المشرفون يديرون مشرفين فرعيين في الدرس 2.

:::