أنظمة الوكلاء المتعددة مع LangGraph
الفرق الهرمية والرسوم البيانية الفرعية المتداخلة
5 دقيقة للقراءة
عندما لا يكفي مشرف واحد
سيناريو إنتاجي حقيقي (يناير 2026):
احتاجت شركة SaaS مؤسسية إلى التعامل مع تذاكر دعم العملاء المعقدة التي تتطلب فرقاً متخصصة متعددة: فريق بحث للعثور على الوثائق ذات الصلة، وفريق تقني لتشخيص المشكلات، وفريق اتصالات لصياغة الردود. مشرف واحد مع 15+ عاملاً أصبح غير قابل للإدارة. بإعادة الهيكلة إلى فرق هرمية مع مشرفين فرعيين، حققوا:
- تقليل 40% في وقت الاستجابة
- معدل حل 85% من الاتصال الأول
- ملكية ومسؤولية واضحة لكل فريق
- تحديثات فريق مستقلة بدون إعادة نشر على مستوى النظام
هذا الدرس يعلمك: كيفية بناء أنظمة وكلاء متعددة هرمية حيث المشرفون يديرون فرقاً من العمال، وكيفية استخدام الرسوم البيانية الفرعية في LangGraph لتغليف منطق الفريق.
المعمارية الهرمية
في النظام الهرمي:
- المشرف الرئيسي ينسق الاستراتيجية عالية المستوى
- مشرفو الفرق يديرون عمالهم المتخصصين
- العمال ينفذون مهام محددة
- النتائج تتدفق لأعلى عبر التسلسل الهرمي
┌────────────────────┐
│ المشرف الرئيسي │
│ (الموجه الاستراتيجي) │
└─────────┬──────────┘
│
┌─────────────────┼─────────────────┐
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ فريق البحث │ │ الفريق التقني │ │ فريق الاتصالات│
│ مشرف │ │ مشرف │ │ مشرف │
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
│ │ │
┌───┼───┐ ┌───┼───┐ ┌───┼───┐
▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼
وثائق ويب API تشخيص اختبار إصلاح مسودة تحرير مراجعة
رؤية رئيسية: كل فريق هو رسم بياني فرعي كامل بنمط المشرف الخاص به. المشرف الرئيسي يتفاعل فقط مع مشرفي الفرق، وليس العمال الفرديين.
بناء الرسوم البيانية الفرعية للفرق
كل فريق يُنفذ كرسم بياني فرعي مستقل:
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
import operator
# =============================================================================
# الرسم البياني الفرعي لفريق البحث
# =============================================================================
class ResearchTeamState(TypedDict):
"""الحالة لعمليات فريق البحث."""
query: str
context: Optional[dict]
# مخرجات العمال
doc_results: Optional[str]
web_results: Optional[str]
api_results: Optional[str]
# مخرجات الفريق
research_summary: Optional[str]
confidence: float
def research_supervisor(state: ResearchTeamState) -> dict:
"""مشرف فريق البحث يقرر أي عامل يستدعي."""
query = state["query"]
if "documentation" in query.lower():
return {"next_worker": "doc_searcher"}
elif "api" in query.lower():
return {"next_worker": "api_researcher"}
else:
return {"next_worker": "web_searcher"}
def doc_searcher(state: ResearchTeamState) -> dict:
"""البحث في الوثائق الداخلية."""
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke(f"ابحث في الوثائق عن: {state['query']}")
return {"doc_results": response.content}
def web_searcher(state: ResearchTeamState) -> dict:
"""البحث في الويب للمعلومات."""
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke(f"ابحث في الويب عن: {state['query']}")
return {"web_results": response.content}
def research_synthesizer(state: ResearchTeamState) -> dict:
"""تجميع كل البحث في ملخص."""
llm = ChatOpenAI(model="gpt-4o")
all_results = []
if state.get("doc_results"):
all_results.append(f"الوثائق: {state['doc_results']}")
if state.get("web_results"):
all_results.append(f"الويب: {state['web_results']}")
combined = "\n\n".join(all_results)
response = llm.invoke(f"لخّص هذا البحث:\n{combined}")
return {
"research_summary": response.content,
"confidence": 0.85
}
def create_research_team() -> StateGraph:
"""إنشاء الرسم البياني الفرعي لفريق البحث."""
graph = StateGraph(ResearchTeamState)
graph.add_node("supervisor", research_supervisor)
graph.add_node("doc_searcher", doc_searcher)
graph.add_node("web_searcher", web_searcher)
graph.add_node("api_researcher", api_researcher)
graph.add_node("synthesizer", research_synthesizer)
# المشرف يوجه للعمال
graph.add_conditional_edges(
"supervisor",
route_research_worker,
{
"doc_searcher": "doc_searcher",
"web_searcher": "web_searcher",
"api_researcher": "api_researcher"
}
)
# جميع العمال يذهبون للمُجمّع
graph.add_edge("doc_searcher", "synthesizer")
graph.add_edge("web_searcher", "synthesizer")
graph.add_edge("synthesizer", END)
graph.set_entry_point("supervisor")
# حرج: التجميع بدون حافظ نقاط تفتيش
return graph.compile()
المشرف الرئيسي مع عقد الفرق
الرسم البياني الرئيسي يعامل كل فريق كعقدة واحدة:
# =============================================================================
# الرسم البياني الهرمي الرئيسي
# =============================================================================
class MainState(TypedDict):
"""الحالة للمشرف الرئيسي."""
# الإدخال
customer_request: str
ticket_id: str
# تعيينات الفرق
teams_needed: List[str]
current_team: Optional[str]
# نتائج الفرق (تتراكم من جميع الفرق)
team_results: Annotated[List[dict], operator.add]
# التحكم
iteration: int
max_iterations: int
# المخرجات
final_response: Optional[str]
resolution_status: str
# إنشاء الرسوم البيانية الفرعية للفرق
research_team = create_research_team()
technical_team = create_technical_team()
comms_team = create_comms_team()
def main_supervisor(state: MainState) -> dict:
"""
المشرف الرئيسي يقرر أي فريق يشترك.
ينسق الفرق فقط، وليس العمال الفرديين.
"""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
request = state["customer_request"]
completed_teams = [r["team"] for r in state.get("team_results", [])]
prompt = f"""حلل طلب العميل هذا وقرر أي فريق يشترك بعد ذلك.
الطلب: {request}
الفرق التي تمت استشارتها: {completed_teams}
الفرق المتاحة:
- research_team: إيجاد الوثائق والمراجع
- technical_team: تشخيص المشاكل واقتراح الإصلاحات
- comms_team: صياغة رد موجه للعميل
أجب باسم الفريق أو 'synthesize' إذا اكتمل العمل المطلوب."""
response = llm.invoke(prompt)
next_team = response.content.strip().lower()
return {
"current_team": next_team,
"iteration": state.get("iteration", 0) + 1
}
def research_team_node(state: MainState) -> dict:
"""تنفيذ الرسم البياني الفرعي لفريق البحث."""
team_input = {
"query": state["customer_request"],
"context": {"ticket_id": state["ticket_id"]}
}
result = research_team.invoke(team_input)
return {
"team_results": [{
"team": "research_team",
"summary": result.get("research_summary", ""),
"confidence": result.get("confidence", 0.5)
}]
}
# =============================================================================
# بناء الرسم البياني الرئيسي
# =============================================================================
def build_hierarchical_system():
"""بناء النظام الهرمي متعدد الوكلاء الكامل."""
graph = StateGraph(MainState)
# إضافة المشرف الرئيسي
graph.add_node("main_supervisor", main_supervisor)
# إضافة عقد الفرق (كل منها يغلف رسماً بيانياً فرعياً)
graph.add_node("research_team", research_team_node)
graph.add_node("technical_team", technical_team_node)
graph.add_node("comms_team", comms_team_node)
graph.add_node("synthesize", final_synthesizer)
# نقطة الدخول
graph.set_entry_point("main_supervisor")
# المشرف يوجه للفرق
graph.add_conditional_edges(
"main_supervisor",
route_to_team,
{
"research_team": "research_team",
"technical_team": "technical_team",
"comms_team": "comms_team",
"synthesize": "synthesize"
}
)
# الفرق تعود للمشرف للقرار التالي
graph.add_edge("research_team", "main_supervisor")
graph.add_edge("technical_team", "main_supervisor")
graph.add_edge("comms_team", "main_supervisor")
graph.add_edge("synthesize", END)
# حرج: فقط الرسم البياني الأب له حافظ نقاط تفتيش
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
return graph.compile(checkpointer=checkpointer)
حرج: التفتيش على الأب فقط (يناير 2026)
هذا أحد أهم الأنماط في أنظمة LangGraph الهرمية.
# ============================================================================
# صحيح: فقط الأب له حافظ نقاط تفتيش
# ============================================================================
# الرسوم البيانية الفرعية - بدون حافظ نقاط تفتيش
research_team = create_research_team() # .compile() بدون args
technical_team = create_technical_team() # .compile() بدون args
# الرسم البياني الأب - له حافظ نقاط تفتيش
main_app = main_graph.compile(checkpointer=PostgresSaver(...))
# ============================================================================
# خطأ: حافظات نقاط تفتيش متداخلة تسبب مشاكل
# ============================================================================
# لا تفعل هذا
research_team = create_research_team().compile(
checkpointer=PostgresSaver(...) # سيء - ينشئ تعارضات
)
لماذا التفتيش على الأب فقط:
- اتساق الحالة: حافظ الأب يلتقط الحالة الكاملة بما في ذلك جميع نتائج الرسوم البيانية الفرعية في كل خطوة
- لا حالات سباق: حافظات متعددة ستحاول حفظ الحالة بشكل مستقل
- مصدر واحد للحقيقة: الاستئناف والسفر عبر الزمن يعملان بشكل صحيح مع الحالة الموحدة
- تصحيح أبسط: مكان واحد لفحص تاريخ سير العمل
تنفيذ الفرق المتوازي
للفرق المستقلة، استخدم Send API:
from langgraph.constants import Send
def dispatch_to_all_teams(state: ParallelMainState) -> List[Send]:
"""إطلاق جميع الفرق بالتوازي."""
return [
Send("research_team", {"query": state["request"]}),
Send("technical_team", {"issue": state["request"]}),
Send("comms_team", {"content": state["request"], "tone": "professional"})
]
def collect_and_synthesize(state: ParallelMainState) -> dict:
"""جمع جميع النتائج المتوازية وإنشاء الاستجابة النهائية."""
results = state.get("team_results", [])
synthesis = f"جُمعت النتائج من {len(results)} فرق"
return {"final_response": synthesis}
# بناء الرسم البياني المتوازي
graph = StateGraph(ParallelMainState)
graph.add_conditional_edges("dispatcher", dispatch_to_all_teams)
أنماط تمرير الحالة
النمط 1: حقن السياق
def team_node_with_context(state: MainState) -> dict:
"""تمرير السياق من الحالة الرئيسية للفريق."""
team_input = {
"query": state["request"],
"context": {
"user_id": state.get("user_id"),
"previous_results": state.get("team_results", [])
}
}
result = team_subgraph.invoke(team_input)
return {"team_results": [{"team": "name", "result": result}]}
النمط 2: تحويل النتائج
def transform_team_result(state: MainState) -> dict:
"""تحويل مخرجات الفريق لتنسيق الحالة الرئيسية."""
raw_result = team_subgraph.invoke({"query": state["request"]})
return {
"team_results": [{
"team": "research",
"summary": raw_result.get("research_summary"),
"confidence": raw_result.get("confidence", 0.5),
"metadata": {"timestamp": datetime.now().isoformat()}
}]
}
معالجة الأخطاء عبر التسلسل الهرمي
def robust_team_node(state: MainState) -> dict:
"""عقدة فريق مع معالجة أخطاء شاملة."""
try:
result = team_subgraph.invoke({"query": state["request"]})
return {
"team_results": [{
"team": "research",
"status": "success",
"result": result
}]
}
except TimeoutError:
return {
"team_results": [{
"team": "research",
"status": "timeout",
"error": "انتهت مهلة تنفيذ الفريق"
}]
}
except Exception as e:
return {
"team_results": [{
"team": "research",
"status": "failed",
"error": str(e)
}]
}
def main_supervisor_handles_failures(state: MainState) -> dict:
"""المشرف الرئيسي يتكيف مع فشل الفرق."""
results = state.get("team_results", [])
failures = [r for r in results if r.get("status") == "failed"]
if failures:
failed_team = failures[-1]["team"]
if state["iteration"] < 3:
return {"current_team": failed_team} # إعادة المحاولة
else:
return {"current_team": "synthesize"} # التخطي والإنهاء
return normal_routing(state)
أسئلة المقابلة الشائعة
س1: "لماذا نستخدم الفرق الهرمية بدلاً من نمط المشرف المسطح؟"
إجابة قوية:
"الفرق الهرمية أفضل عندما يكون لديك: (1) مجالات خبرة مميزة تتطلب تنسيقاً متخصصاً - فريق البحث يعمل بشكل مختلف عن الفريق التقني. (2) أكثر من 5-7 عمال - الحمل المعرفي على مشرف واحد يصبح عالياً جداً. (3) الحاجة لتحديثات فريق مستقلة - يمكنك تعديل فريق البحث بدون لمس الفريق التقني. (4) حدود فريق واضحة - عندما يتجمع العمال طبيعياً حسب الوظيفة."
س2: "لماذا يجب أن يكون التفتيش على الأب فقط في الأنظمة الهرمية؟"
الإجابة:
"ثلاثة أسباب حرجة: (1) اتساق الحالة - حافظ الأب يلتقط الحالة الكاملة بما في ذلك جميع نتائج الرسوم البيانية الفرعية في كل خطوة رئيسية. (2) لا حالات سباق - حافظات متعددة ستحاول حفظ الحالة بشكل مستقل، مما ينشئ تعارضات. (3) مصدر واحد للحقيقة - تصحيح الاستئناف والسفر عبر الزمن يعملان بشكل صحيح لأن هناك تاريخ حالة موحد واحد."
س3: "كيف تتعامل مع فريق يفشل باستمرار؟"
الإجابة:
"نفّذ قاطع دائرة على مستوى المشرف الرئيسي. تتبع فشل الفريق في الحالة. بعد N فشل متتالي، علّم ذلك الفريق كـ 'متدهور' وإما: (1) وجّه لفريق بديل إذا متاح، (2) تخطى ذلك الفريق وجمّع بنتائج جزئية، (3) صعّد للمشغل البشري."
س4: "متى تستخدم التنفيذ المتوازي مقابل التسلسلي للفرق؟"
الإجابة:
"متوازي عندما تكون الفرق مستقلة - البحث والتحليل التقني والمسودة الأولية يمكن أن تعمل جميعها في وقت واحد. تسلسلي عندما تكون هناك تبعيات - فريق الاتصالات يحتاج نتائج البحث قبل الصياغة. النهج الهجين: شغّل الفرق المستقلة بالتوازي أولاً، ثم سلسل الفرق التابعة."
النقاط الرئيسية
- الفرق الهرمية = مشرف رئيسي + مشرفو فرق + عمال
- الرسوم البيانية الفرعية تغلف منطق كل فريق الكامل
- التفتيش على الأب فقط حرج لاتساق الحالة
- الحالة تُمرر صراحةً بين مستويات التسلسل الهرمي
- Send API يمكّن تنفيذ الفرق المتوازي
- معالجة الأخطاء في كل مستوى مع التدهور الأنيق
- عقد الفرق تغلف الرسوم البيانية الفرعية - الرسم البياني الرئيسي يرى الفرق كعقد مفردة
التالي: تعلم أنماط الاتصال بين الوكلاء لتمرير الرسائل المعقد في الدرس 3.
:::