الاختبار ومشروع التخرج
مشروع التخرج: نظام بحث إنتاجي
8 دقيقة للقراءة
يجمع مشروع التخرج هذا جميع المفاهيم من الدورة لبناء نظام بحث متعدد الوكلاء جاهز للإنتاج. ستنفذ إدارة الحالة مع المخفضات، تنسيق نمط المشرف، التفتيش للاستمرارية، موافقة الإنسان في الحلقة، استعادة الأخطاء، ومراقبة LangSmith.
نظرة عامة على المشروع
نظام البحث الإنتاجي هو سير عمل متعدد الوكلاء يقوم بـ:
- البحث في المواضيع باستخدام مصادر متعددة
- تحليل المعلومات المجمعة للرؤى الرئيسية
- كتابة تقارير مهنية
- مراجعة الجودة من خلال مشرف
- الموافقة عبر الإنسان في الحلقة
- الاستمرار في الحالة لسير العمل طويل المدى
- تتبع التنفيذ للتصحيح والمراقبة
يوضح هذا النظام الأنماط الواقعية المستخدمة في تطبيقات الذكاء الاصطناعي الإنتاجية.
بنية النظام
+------------------+
| المشرف |
| (المنسق) |
+--------+---------+
|
+--------------------+--------------------+
| | |
v v v
+-------+--------+ +-------+--------+ +-------+--------+
| الباحث | | المحلل | | الكاتب |
| (جمع البيانات) | | (استخراج الرؤى) | | (صياغة التقرير) |
+----------------+ +----------------+ +----------------+
| | |
+--------------------+--------------------+
|
v
+--------+---------+
| المراجعة البشرية |
| (الموافقة/التنقيح)|
+--------+---------+
|
v
+--------+---------+
| الإنهاء |
| (إضافة البيانات) |
+------------------+
التنفيذ الكامل
تعريف الحالة
"""
نظام البحث الإنتاجي - تعريف الحالة
يوضح مخطط الحالة هذا:
1. حقول الإدخال للتكوين
2. المخرجات المتراكمة باستخدام المخفضات
3. حقول مخرجات العمال
4. حقول التحكم للتوجيه
5. البيانات الوصفية للتتبع
"""
from typing import TypedDict, Annotated, Literal, Optional
from langgraph.graph import StateGraph, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langsmith import traceable
import operator
import os
from datetime import datetime
# تمكين تتبع LangSmith للمراقبة الإنتاجية
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production-research-system"
class ResearchDocument(TypedDict):
"""هيكل لمستندات البحث."""
source: str
content: str
query: str
timestamp: str
relevance_score: Optional[float]
class Message(TypedDict):
"""هيكل للرسائل بين الوكلاء."""
from_agent: str
to_agent: str
message_type: str
content: str
timestamp: str
class ResearchState(TypedDict):
"""
الحالة الكاملة لنظام البحث الإنتاجي.
الفئات:
1. تكوين الإدخال
2. البيانات المتراكمة (مع المخفضات)
3. مخرجات العمال
4. تدفق التحكم
5. التتبع/البيانات الوصفية
"""
# === تكوين الإدخال ===
query: str # استعلام البحث من المستخدم
max_iterations: int # حد الأمان للتكرارات
quality_threshold: int # الحد الأدنى لنقاط الجودة (0-100)
require_human_approval: bool # هل المراجعة البشرية مطلوبة
# === البيانات المتراكمة (المخفضات) ===
documents: Annotated[list[ResearchDocument], operator.add]
messages: Annotated[list[Message], operator.add]
errors: Annotated[list[dict], operator.add]
# === مخرجات العمال ===
analysis: Optional[str] # التحليل من المحلل
report: Optional[str] # التقرير من الكاتب
quality_score: Optional[int] # تقييم الجودة (0-100)
# === تدفق التحكم ===
next_worker: Literal[
"researcher",
"analyzer",
"writer",
"quality_check",
"human_review",
"finalize",
"error_handler",
"done"
]
current_phase: Literal["research", "analysis", "writing", "review", "complete"]
# === التتبع/البيانات الوصفية ===
iteration: int
started_at: str
completed_at: Optional[str]
approved: bool
approval_notes: Optional[str]
feedback: Optional[str]
عقد العمال
"""
عقد العمال - الوكلاء الذين يقومون بالعمل الفعلي.
كل عامل:
1. مزين بـ @traceable لرؤية LangSmith
2. يستقبل الحالة الكاملة، يُرجع تحديثات جزئية
3. يضيف رسائل للتواصل بين الوكلاء
4. يتعامل مع الأخطاء بأناقة
"""
@traceable(name="researcher", run_type="chain", tags=["worker", "llm-call"])
def researcher(state: ResearchState) -> dict:
"""
جمع مستندات البحث بناءً على الاستعلام.
في الإنتاج، هذا سيقوم بـ:
- استدعاء APIs البحث (Google، Bing، قواعد بيانات أكاديمية)
- كشط المواقع ذات الصلة
- الاستعلام من قواعد المعرفة الداخلية
- الاسترجاع من متاجر المتجهات
"""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# استخدام التغذية الراجعة إذا طُلبت المراجعة
query = state["query"]
if state.get("feedback"):
query = f"{query}\n\nتركيز إضافي بناءً على التغذية الراجعة: {state['feedback']}"
# محاكاة البحث مع LLM (في الإنتاج، استخدم APIs بحث حقيقية)
prompt = f"""أنت مساعد بحث. ابحث عن الحقائق والمعلومات الرئيسية حول:
الاستعلام: {query}
قدم نتائجك ككائن JSON بهذا الهيكل:
{{
"facts": [
{{"fact": "...", "source": "...", "confidence": "high/medium/low"}},
...
],
"summary": "ملخص موجز للنتائج"
}}
تضمين 3-5 حقائق رئيسية مع المصادر."""
try:
response = llm.invoke(prompt)
# إنشاء مستند البحث
doc: ResearchDocument = {
"source": "llm_research",
"content": response.content,
"query": query,
"timestamp": datetime.now().isoformat(),
"relevance_score": 0.85
}
# إنشاء رسالة الحالة
message: Message = {
"from_agent": "researcher",
"to_agent": "supervisor",
"message_type": "status",
"content": f"تم العثور على نتائج بحث لـ: {query[:50]}...",
"timestamp": datetime.now().isoformat()
}
return {
"documents": [doc],
"messages": [message],
"iteration": state["iteration"] + 1,
"current_phase": "research"
}
except Exception as e:
error = {
"node": "researcher",
"error": str(e),
"timestamp": datetime.now().isoformat()
}
return {
"errors": [error],
"next_worker": "error_handler"
}
@traceable(name="analyzer", run_type="chain", tags=["worker", "llm-call"])
def analyzer(state: ResearchState) -> dict:
"""
تحليل المستندات المجمعة للرؤى.
يتضمن التحليل:
- استخراج المواضيع الرئيسية
- تقييم الأدلة
- تقييم الثقة
- تحديد الفجوات
"""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# دمج جميع محتويات المستندات
doc_contents = "\n\n---\n\n".join([
f"المصدر: {doc['source']}\nالمحتوى: {doc['content']}"
for doc in state["documents"]
])
prompt = f"""حلل نتائج البحث هذه وقدم تحليلاً منظماً:
مستندات البحث:
{doc_contents}
الاستعلام الأصلي: {state['query']}
قدم التحليل بهذا التنسيق:
## المواضيع الرئيسية
- الموضوع 1: الوصف
- الموضوع 2: الوصف
...
## الأدلة الداعمة
لكل موضوع، اذكر الأدلة من المستندات.
## تقييم الثقة
- الثقة العامة: عالية/متوسطة/منخفضة
- السبب: لماذا هذا المستوى من الثقة
## فجوات المعلومات
ما البحث الإضافي الذي قد يكون مطلوباً؟
## التوصيات
بناءً على التحليل، ما هي النقاط الرئيسية؟"""
try:
response = llm.invoke(prompt)
message: Message = {
"from_agent": "analyzer",
"to_agent": "supervisor",
"message_type": "analysis_complete",
"content": f"اكتمل التحليل ({len(response.content)} حرف)",
"timestamp": datetime.now().isoformat()
}
return {
"analysis": response.content,
"messages": [message],
"iteration": state["iteration"] + 1,
"current_phase": "analysis"
}
except Exception as e:
error = {
"node": "analyzer",
"error": str(e),
"timestamp": datetime.now().isoformat()
}
return {
"errors": [error],
"next_worker": "error_handler"
}
@traceable(name="writer", run_type="chain", tags=["worker", "llm-call"])
def writer(state: ResearchState) -> dict:
"""
كتابة تقرير بحث مهني.
هيكل التقرير:
- ملخص تنفيذي
- النتائج الرئيسية
- المنهجية
- التحليل التفصيلي
- الاستنتاجات
- المراجع
"""
llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
prompt = f"""اكتب تقرير بحث مهني بناءً على التالي:
الاستعلام: {state['query']}
التحليل:
{state['analysis']}
مستندات المصدر: {len(state['documents'])} مستند تم تحليله
اكتب تقريراً كاملاً بهذا الهيكل:
# تقرير البحث: [أنشئ عنواناً مناسباً]
## الملخص التنفيذي
ملخص من 2-3 فقرات للنتائج الرئيسية.
## المنهجية
كيف تم إجراء البحث.
## النتائج الرئيسية
النتائج التفصيلية مع الأدلة الداعمة.
## التحليل
تحليل متعمق للنتائج.
## الاستنتاجات
الاستنتاجات النهائية والآثار.
## التوصيات
توصيات قابلة للتنفيذ بناءً على النتائج.
## المراجع
قائمة المصادر المستخدمة في هذا البحث.
اجعل التقرير مهنياً، منظماً جيداً، وشاملاً."""
try:
response = llm.invoke(prompt)
message: Message = {
"from_agent": "writer",
"to_agent": "supervisor",
"message_type": "report_complete",
"content": f"تمت صياغة التقرير ({len(response.content)} حرف)",
"timestamp": datetime.now().isoformat()
}
return {
"report": response.content,
"messages": [message],
"iteration": state["iteration"] + 1,
"current_phase": "writing"
}
except Exception as e:
error = {
"node": "writer",
"error": str(e),
"timestamp": datetime.now().isoformat()
}
return {
"errors": [error],
"next_worker": "error_handler"
}
عقدة المشرف
"""
عقدة المشرف - ينسق العمال ويتحقق من الجودة.
المشرف:
1. يحدد أي عامل يجب أن ينفذ التالي
2. يتحقق من جودة المخرجات
3. يقرر متى يطلب المراجعة البشرية
4. يتعامل مع حدود التكرار
5. يوجه إلى معالج الأخطاء عند الحاجة
"""
@traceable(name="supervisor", run_type="chain", tags=["supervisor"])
def supervisor(state: ResearchState) -> dict:
"""
المنسق المركزي لسير عمل البحث.
منطق القرار:
1. التحقق من حد التكرار (الأمان)
2. التحقق من الأخطاء (الاستعادة)
3. التوجيه بناءً على التقدم الحالي
4. تقييم الجودة قبل المراجعة البشرية
"""
# الأمان: التحقق من حد التكرار
if state["iteration"] >= state["max_iterations"]:
message: Message = {
"from_agent": "supervisor",
"to_agent": "system",
"message_type": "iteration_limit",
"content": f"تم الوصول للحد الأقصى للتكرارات ({state['max_iterations']})",
"timestamp": datetime.now().isoformat()
}
return {
"next_worker": "done",
"messages": [message],
"completed_at": datetime.now().isoformat()
}
# التحقق من الأخطاء الأخيرة
if state.get("errors") and len(state["errors"]) > 0:
recent_error = state["errors"][-1]
if recent_error.get("handled") != True:
return {"next_worker": "error_handler"}
# منطق القرار بناءً على الحالة الحالية
# لا توجد مستندات بعد -> يحتاج بحث
if not state.get("documents") or len(state["documents"]) == 0:
return {"next_worker": "researcher"}
# لديه مستندات لكن لا تحليل -> يحتاج تحليل
if not state.get("analysis"):
return {"next_worker": "analyzer"}
# التحقق من جودة التحليل
if state.get("analysis"):
analysis_length = len(state["analysis"])
if analysis_length < 200: # موجز جداً
message: Message = {
"from_agent": "supervisor",
"to_agent": "analyzer",
"message_type": "revision_request",
"content": "التحليل يحتاج المزيد من العمق والتفصيل",
"timestamp": datetime.now().isoformat()
}
return {
"next_worker": "analyzer",
"feedback": "الرجاء تقديم تحليل أكثر تفصيلاً مع أدلة محددة",
"messages": [message],
"analysis": None
}
# لديه تحليل لكن لا تقرير -> يحتاج كتابة
if not state.get("report"):
return {"next_worker": "writer"}
# التحقق من جودة التقرير
if state.get("report"):
report_length = len(state["report"])
if report_length < 500: # موجز جداً
return {
"next_worker": "writer",
"feedback": "الرجاء توسيع التقرير بمزيد من التفاصيل",
"report": None
}
# لديه تقرير، التحقق مما إذا كانت الموافقة البشرية مطلوبة
if state.get("require_human_approval", True) and not state.get("approved"):
return {"next_worker": "human_review"}
# تمت الموافقة بالفعل أو لا حاجة للموافقة -> إنهاء
if state.get("approved") or not state.get("require_human_approval", True):
return {"next_worker": "finalize"}
return {
"next_worker": "done",
"completed_at": datetime.now().isoformat()
}
عقدة المراجعة البشرية
"""
عقدة المراجعة البشرية - تنفذ نمط الإنسان في الحلقة.
تستخدم interrupt() من LangGraph للإيقاف/الاستئناف الأصلي.
سير العمل يتوقف هنا وينتظر المدخلات البشرية.
"""
@traceable(name="human_review", run_type="chain", tags=["human-in-loop"])
def human_review(state: ResearchState) -> Command:
"""
الإيقاف للموافقة البشرية على تقرير البحث.
دالة interrupt():
1. تحفظ الحالة الحالية إلى نقطة تفتيش
2. تُرجع بيانات المقاطعة للمستدعي
3. توقف التنفيذ حتى الاستئناف
خيارات الاستئناف:
- approve: قبول التقرير
- revise: طلب تغييرات مع تغذية راجعة
- reject: رفض وإنهاء سير العمل
"""
review_data = {
"type": "report_approval",
"query": state["query"],
"report_preview": state["report"][:1000] if state["report"] else "",
"report_length": len(state["report"]) if state["report"] else 0,
"documents_count": len(state.get("documents", [])),
"iterations": state["iteration"],
"options": ["approve", "revise", "reject"],
"instructions": "الرجاء مراجعة التقرير واختيار إجراء"
}
# هذا يوقف التنفيذ ويُرجع review_data للمستدعي
decision = interrupt(review_data)
# التنفيذ يستأنف هنا عندما يقدم الإنسان المدخلات
action = decision.get("action", "reject")
feedback = decision.get("feedback", "")
notes = decision.get("notes", "")
if action == "approve":
message: Message = {
"from_agent": "human",
"to_agent": "supervisor",
"message_type": "approval",
"content": f"تمت الموافقة على التقرير. ملاحظات: {notes}",
"timestamp": datetime.now().isoformat()
}
return Command(
goto="finalize",
update={
"approved": True,
"approval_notes": notes,
"messages": [message],
"current_phase": "review"
}
)
elif action == "revise":
message: Message = {
"from_agent": "human",
"to_agent": "supervisor",
"message_type": "revision_request",
"content": f"طُلبت المراجعة: {feedback}",
"timestamp": datetime.now().isoformat()
}
return Command(
goto="supervisor",
update={
"feedback": feedback,
"report": None,
"analysis": None,
"messages": [message],
"current_phase": "review"
}
)
else: # reject
message: Message = {
"from_agent": "human",
"to_agent": "system",
"message_type": "rejection",
"content": f"تم رفض التقرير. السبب: {feedback}",
"timestamp": datetime.now().isoformat()
}
return Command(
goto=END,
update={
"approved": False,
"approval_notes": f"مرفوض: {feedback}",
"messages": [message],
"completed_at": datetime.now().isoformat(),
"current_phase": "complete"
}
)
عقد الإنهاء ومعالج الأخطاء
"""
عقدة الإنهاء - تضيف اللمسات النهائية للتقارير المعتمدة.
عقدة معالج الأخطاء - تنفذ منطق استعادة الأخطاء.
"""
@traceable(name="finalize", run_type="chain", tags=["finalize"])
def finalize(state: ResearchState) -> dict:
"""
إضافة اللمسات النهائية للتقرير المعتمد.
يتضمن الإنهاء:
- إضافة البيانات الوصفية
- توليد إحصائيات ملخصة
- إضافة الطوابع الزمنية
- التنسيق للإخراج
"""
metadata = f"""
---
## البيانات الوصفية للتقرير
- **تاريخ التوليد**: {datetime.now().isoformat()}
- **الاستعلام**: {state['query']}
- **المستندات المحللة**: {len(state.get('documents', []))}
- **التكرارات**: {state['iteration']}
- **تمت الموافقة**: {state.get('approved', False)}
- **ملاحظات الموافقة**: {state.get('approval_notes', 'غير متوفر')}
*تم التوليد بواسطة نظام البحث الإنتاجي*
*مدعوم من LangGraph*
---
"""
final_report = f"{state['report']}\n\n{metadata}"
message: Message = {
"from_agent": "finalize",
"to_agent": "system",
"message_type": "complete",
"content": "اكتمل سير عمل البحث بنجاح",
"timestamp": datetime.now().isoformat()
}
return {
"report": final_report,
"messages": [message],
"completed_at": datetime.now().isoformat(),
"current_phase": "complete"
}
@traceable(name="error_handler", run_type="chain", tags=["error-handling"])
def error_handler(state: ResearchState) -> dict:
"""
التعامل مع الأخطاء بمنطق إعادة المحاولة.
استراتيجيات التعامل مع الأخطاء:
1. إعادة محاولة العملية الفاشلة
2. التخطي إلى الخطوة التالية إن أمكن
3. طلب التدخل البشري للأخطاء الحرجة
4. التسجيل للتحليل لاحقاً
"""
if not state.get("errors"):
return {"next_worker": "supervisor"}
recent_error = state["errors"][-1]
error_count = len(state.get("errors", []))
# تمييز الخطأ على أنه تم التعامل معه
recent_error["handled"] = True
recent_error["handled_at"] = datetime.now().isoformat()
message: Message = {
"from_agent": "error_handler",
"to_agent": "supervisor",
"message_type": "error_handled",
"content": f"تم التعامل مع الخطأ من {recent_error.get('node', 'unknown')}",
"timestamp": datetime.now().isoformat()
}
# إذا كان هناك الكثير من الأخطاء، التصعيد للإنسان
if error_count >= 3:
escalation_message: Message = {
"from_agent": "error_handler",
"to_agent": "human",
"message_type": "escalation",
"content": f"حدثت أخطاء متعددة ({error_count}). يُوصى بالتدخل البشري.",
"timestamp": datetime.now().isoformat()
}
return {
"messages": [message, escalation_message],
"next_worker": "human_review"
}
return {
"messages": [message],
"next_worker": "supervisor"
}
بناء الرسم البياني
"""
بناء الرسم البياني - بناء سير العمل.
"""
def route_to_worker(state: ResearchState) -> str:
"""التوجيه من المشرف إلى العامل المناسب."""
return state.get("next_worker", "done")
def create_research_system() -> StateGraph:
"""
بناء الرسم البياني الكامل لنظام البحث.
"""
graph = StateGraph(ResearchState)
# إضافة جميع العقد
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher)
graph.add_node("analyzer", analyzer)
graph.add_node("writer", writer)
graph.add_node("human_review", human_review)
graph.add_node("finalize", finalize)
graph.add_node("error_handler", error_handler)
# المشرف يوجه للعمال باستخدام الحواف الشرطية
graph.add_conditional_edges(
"supervisor",
route_to_worker,
{
"researcher": "researcher",
"analyzer": "analyzer",
"writer": "writer",
"quality_check": "supervisor",
"human_review": "human_review",
"finalize": "finalize",
"error_handler": "error_handler",
"done": END
}
)
# العمال يعودون للمشرف للقرار التالي
graph.add_edge("researcher", "supervisor")
graph.add_edge("analyzer", "supervisor")
graph.add_edge("writer", "supervisor")
graph.add_edge("error_handler", "supervisor")
# الإنهاء ينهي سير العمل
graph.add_edge("finalize", END)
# تعيين نقطة الدخول
graph.set_entry_point("supervisor")
return graph
def get_development_app():
"""
إنشاء تطبيق للتطوير/الاختبار مع MemorySaver.
"""
graph = create_research_system()
checkpointer = MemorySaver()
return graph.compile(checkpointer=checkpointer)
def get_production_app():
"""
إنشاء تطبيق جاهز للإنتاج مع PostgresSaver.
"""
graph = create_research_system()
checkpointer = PostgresSaver.from_conn_string(
os.environ["DATABASE_URL"]
)
return graph.compile(checkpointer=checkpointer)
اختبار النظام
اختبارات الوحدة
"""
اختبارات الوحدة لنظام البحث الإنتاجي
اختبار المكونات الفردية بمعزل.
"""
import pytest
from unittest.mock import Mock, patch
class TestSupervisorLogic:
"""اختبار اتخاذ قرارات المشرف."""
def test_supervisor_routes_to_researcher_when_no_documents(self):
"""المشرف يجب أن يوجه للباحث عندما تكون المستندات فارغة."""
state = {
"query": "استعلام اختبار",
"documents": [],
"analysis": None,
"report": None,
"iteration": 0,
"max_iterations": 10,
"errors": [],
"approved": False,
"require_human_approval": True
}
result = supervisor(state)
assert result["next_worker"] == "researcher"
def test_supervisor_routes_to_analyzer_when_has_documents(self):
"""المشرف يجب أن يوجه للمحلل عندما توجد مستندات بدون تحليل."""
state = {
"query": "استعلام اختبار",
"documents": [{"content": "مستند اختبار", "source": "test"}],
"analysis": None,
"report": None,
"iteration": 1,
"max_iterations": 10,
"errors": [],
"approved": False,
"require_human_approval": True
}
result = supervisor(state)
assert result["next_worker"] == "analyzer"
def test_supervisor_respects_max_iterations(self):
"""المشرف يجب أن ينهي سير العمل عند الوصول للحد الأقصى."""
state = {
"query": "استعلام اختبار",
"documents": [],
"analysis": None,
"report": None,
"iteration": 10,
"max_iterations": 10,
"errors": [],
"approved": False,
"require_human_approval": True
}
result = supervisor(state)
assert result["next_worker"] == "done"
اختبارات التكامل
"""
اختبارات التكامل لنظام البحث الإنتاجي
اختبار سير العمل الكامل مع LLM محاكى.
"""
import pytest
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
from unittest.mock import patch, Mock
class TestResearchWorkflowIntegration:
"""اختبارات التكامل لسير العمل الكامل."""
@pytest.fixture
def app(self, mock_openai):
"""إنشاء تطبيق مُجمَّع مع MemorySaver."""
graph = create_research_system()
return graph.compile(checkpointer=MemorySaver())
def test_complete_workflow_with_approval(self, app):
"""اختبار سير العمل الكامل المنتهي بموافقة بشرية."""
config = {"configurable": {"thread_id": "test-complete-workflow"}}
# تشغيل حتى المراجعة البشرية
result = app.invoke(initial_state, config)
# يجب أن يكون لديه مستندات، تحليل، وتقرير
assert len(result["documents"]) > 0
assert result["analysis"] is not None
assert result["report"] is not None
assert result["approved"] is False
# استئناف مع الموافقة
final_result = app.invoke(
Command(resume={"action": "approve", "notes": "يبدو جيداً!"}),
config
)
assert final_result["approved"] is True
قائمة فحص النشر
التحقق قبل النشر
## قائمة فحص ما قبل النشر
### جودة الكود
- [ ] جميع اختبارات الوحدة ناجحة
- [ ] جميع اختبارات التكامل ناجحة
- [ ] اكتملت مراجعة الكود
- [ ] لا توجد أسرار أو مفاتيح API مضمنة
### البنية التحتية
- [ ] تم توفير قاعدة بيانات PostgreSQL
- [ ] تم تعيين متغير البيئة DATABASE_URL
- [ ] تم تكوين تجميع الاتصالات
### المراقبة
- [ ] LANGCHAIN_TRACING_V2=true معين
- [ ] LANGCHAIN_API_KEY مُكوَّن
- [ ] تم إنشاء مشروع LangSmith
### الأمان
- [ ] مفاتيح API في مدير الأسرار
- [ ] تحديد المعدل مُكوَّن
- [ ] التحقق من المدخلات في مكانه
أسئلة المقابلة
س1: اشرح بنية نظام البحث هذا.
الإجابة:
"هذا نمط مشرف متعدد الوكلاء مع خمسة مكونات رئيسية:
1. إدارة الحالة
class ResearchState(TypedDict):
# تراكم البيانات مع المخفضات
documents: Annotated[list[dict], operator.add]
messages: Annotated[list[dict], operator.add]
2. نمط المشرف المشرف هو المنسق المركزي الذي يقيم الحالة ويقرر العامل التالي.
3. عقد العمال كل عامل متخصص:
- الباحث: يجمع المعلومات
- المحلل: يستخرج الرؤى
- الكاتب: ينشئ التقارير
4. الإنسان في الحلقة
يستخدم interrupt() الأصلي من LangGraph:
decision = interrupt({...})
# التنفيذ يتوقف هنا حتى الاستئناف
5. استعادة الأخطاء معالج الأخطاء يسجل، يتعامل، ويصعّد عند الحاجة."
س2: كيف يمكّن التفتيش الإيقاف/الاستئناف في هذا النظام؟
الإجابة:
"التفتيش أساسي لنمط الإنسان في الحلقة:
- كل تنفيذ عقدة يحفظ نقطة تفتيش
- interrupt() يُحفز حفظ نقطة التفتيش
- الاستئناف يستخدم thread_id لإيجاد نقطة التفتيش
# بدء سير العمل
result = app.invoke(initial_state, config)
# يعود فوراً عند الوصول لـ interrupt()
# ساعات لاحقاً، الإنسان يوافق
result = app.invoke(Command(resume={'action': 'approve'}), config)
# يستمر من حيث توقف
المفتاح هو أن التفتيش يفصل التنفيذ عن التخزين."
النقاط الرئيسية
| المفهوم | التنفيذ | الفائدة |
|---|---|---|
| مخفضات الحالة | Annotated[list, operator.add] |
تراكم البيانات عبر التكرارات |
| نمط المشرف | عقدة توجيه مركزية | تنسيق عمال متعددين |
| التفتيش | PostgresSaver | استمرار الحالة لسير العمل الطويل |
| الإنسان في الحلقة | interrupt() + Command(resume=) |
إيقاف/استئناف أصلي |
| استعادة الأخطاء | عقدة معالج الأخطاء | تدهور لطيف |
| المراقبة | مزين @traceable |
تتبع LangSmith |
| الاختبار | MemorySaver + LLM محاكى | اختبارات سريعة وموثوقة |
إكمال الدورة
تهانينا على إكمال دورة LangGraph Deep Dive! لديك الآن المعرفة لبناء أنظمة متعددة الوكلاء جاهزة للإنتاج مع:
- إدارة الحالة: مخططات TypedDict مع المخفضات
- أنماط الرسم البياني: المشرف، التنفيذ المتوازي، الرسوم الفرعية
- الاستمرارية: التفتيش مع PostgresSaver
- التكامل البشري: إيقاف/استئناف أصلي مع المقاطعات
- النشر: Docker، Kubernetes، LangGraph Platform
- الاختبار: استراتيجيات اختبار شاملة
- التصحيح: التصور وتتبع LangSmith
الخطوات التالية:
- ابنِ نظامك متعدد الوكلاء الخاص
- استكشف LangGraph Platform للنشر المُدار
- ساهم في مجتمع LangGraph
- ابقَ محدثاً مع إصدارات LangGraph
أكمل اختبار الوحدة للحصول على شهادة إكمال الدورة!
:::