الاختبار ومشروع التخرج
اختبار سير عمل LangGraph
5 دقيقة للقراءة
اختبار تطبيقات LangGraph يتطلب نهجاً منظماً يغطي العقد الفردية وانتقالات الحالة وتنفيذ الرسم البياني والتفاعلات المعقدة بين الوكلاء. يوفر هذا الدرس استراتيجيات اختبار شاملة لبناء أنظمة سير عمل موثوقة وجاهزة للإنتاج.
تأثير الاختبار الواقعي
دراسة حالة يناير 2026: وكيل تداول لشركة تكنولوجيا مالية اتخذ قراراً خاطئاً كلف 47,000 دولار. تحليل السبب الجذري كشف حالة حافة غير مختبرة: عندما تجاوز عدد المستندات 50، فشلت عقدة المحلل بصمت وأعادت تحليلاً فارغاً، الذي فسره المشرف خطأً كـ "لا توجد مشاكل." بعد تنفيذ استراتيجيات الاختبار في هذا الدرس، حقق الفريق صفر حوادث إنتاج خلال 8 أشهر تالية.
مقاييس عائد الاستثمار للاختبار:
قبل الاختبار الشامل:
- 3-5 حوادث إنتاج شهرياً
- 2-4 ساعات متوسط استعادة الحادث
- 15% من النشرات سببت مشاكل
بعد التنفيذ:
- 0 حوادث إنتاج في 8 أشهر
- الأخطاء تُكتشف في CI/CD قبل النشر
- ثقة النشر زادت إلى 99%
هرم استراتيجية الاختبار لـ LangGraph
┌─────────────────┐
│ اختبارات E2E │ ← قليلة، مكلفة، LLMs حقيقية
│ (الإنتاج) │
├─────────────────┤
│ التكامل │ ← أكثر، LLMs محاكاة
│ (رسم كامل) │
├─────────────────┤
│ اختبار الوحدة │ ← كثيرة، سريعة، بدون LLMs
│ (العقد) │
└─────────────────┘
توزيع الاختبار المستهدف:
- اختبارات الوحدة: 70% (سريعة، معزولة، حتمية)
- اختبارات التكامل: 25% (تبعيات محاكاة، تدفقات كاملة)
- اختبارات E2E: 5% (APIs حقيقية، اختبارات دخان فقط)
اختبار وحدة العقد الفردية
اختبارات الوحدة تتحقق من أن العقد الفردية تنتج تحديثات حالة صحيحة لمدخلات محددة. يجب أن تكون هذه الاختبارات سريعة وحتمية ومعزولة عن التبعيات الخارجية.
اختبار دوال العقد
"""
اختبارات وحدة لدوال عقد LangGraph.
كل عقدة دالة نقية: حالة -> تحديث حالة
اختبر بتوفير حالة وهمية والتحقق من التحديثات المُرجعة.
"""
import pytest
from unittest.mock import Mock, patch, MagicMock
from datetime import datetime
from typing import Any
# استيراد العقد
from src.nodes.researcher import researcher_node
from src.nodes.analyzer import analyzer_node
from src.nodes.writer import writer_node
from src.nodes.supervisor import supervisor_node
class TestResearcherNode:
"""مجموعة اختبارات لوظائف عقدة الباحث."""
def test_researcher_returns_documents(self):
"""تحقق من أن عقدة الباحث تضيف مستندات للحالة."""
# ترتيب - إنشاء الحالة الأولية
initial_state = {
"query": "اتجاهات AI في 2026",
"documents": [],
"iteration": 0,
"max_iterations": 10
}
# تنفيذ - تنفيذ العقدة
result = researcher_node(initial_state)
# تأكيد - التحقق من تحديثات الحالة
assert "documents" in result, "يجب أن يُرجع الباحث مستندات"
assert isinstance(result["documents"], list)
assert len(result["documents"]) > 0, "يجب إيجاد مستند واحد على الأقل"
assert result["iteration"] == 1, "يجب زيادة التكرار"
def test_researcher_incorporates_feedback(self):
"""تحقق من أن الباحث يستخدم التغذية الراجعة لتحسين البحث."""
state = {
"query": "تطبيقات التعلم الآلي",
"documents": [],
"iteration": 1,
"feedback": "ركز تحديداً على تطبيقات الرعاية الصحية",
"max_iterations": 10
}
result = researcher_node(state)
# تحقق من اعتبار التغذية الراجعة
assert len(result["documents"]) > 0
def test_researcher_handles_empty_query(self):
"""تحقق من أن الباحث يتعامل مع الحالة الحافة للاستعلام الفارغ."""
state = {
"query": "",
"documents": [],
"iteration": 0,
"max_iterations": 10
}
# يجب إما إرجاع مستندات فارغة أو رفع خطأ مناسب
with pytest.raises(ValueError, match="الاستعلام لا يمكن أن يكون فارغاً"):
researcher_node(state)
class TestSupervisorNode:
"""مجموعة اختبارات لمنطق توجيه المشرف."""
def test_supervisor_routes_to_researcher_when_no_documents(self):
"""المشرف يجب أن يطلب البحث عندما لا توجد مستندات."""
state = {
"documents": [],
"analysis": None,
"report": None,
"iteration": 0,
"max_iterations": 10
}
result = supervisor_node(state)
assert result["next_worker"] == "researcher"
def test_supervisor_routes_to_analyzer_when_documents_exist(self):
"""المشرف يجب أن يطلب التحليل عندما تكون المستندات جاهزة."""
state = {
"documents": [{"content": "مستند اختبار"}],
"analysis": None,
"report": None,
"iteration": 1,
"max_iterations": 10
}
result = supervisor_node(state)
assert result["next_worker"] == "analyzer"
def test_supervisor_routes_to_done_when_complete(self):
"""المشرف يجب أن ينهي عندما يُوافق على التقرير."""
state = {
"documents": [{"content": "مستند اختبار"}],
"analysis": "محتوى التحليل",
"report": "محتوى التقرير النهائي",
"approved": True,
"iteration": 4,
"max_iterations": 10
}
result = supervisor_node(state)
assert result["next_worker"] == "done"
def test_supervisor_respects_iteration_limit(self):
"""المشرف يجب أن يتوقف عند الحد الأقصى للتكرارات."""
state = {
"documents": [],
"analysis": None,
"report": None,
"iteration": 10,
"max_iterations": 10
}
result = supervisor_node(state)
assert result["next_worker"] == "done"
الاختبار مع التبعيات المحاكاة
"""
محاكاة التبعيات الخارجية للاختبار الحتمي.
LLMs و APIs وقواعد البيانات يجب أن تُحاكى لضمان:
- تنفيذ اختبار سريع
- نتائج حتمية
- لا تكاليف API خارجية أثناء الاختبار
"""
import pytest
from unittest.mock import Mock, patch, AsyncMock
@pytest.fixture
def mock_openai_llm():
"""
مثبت يوفر LLM OpenAI محاكى.
يُرجع استجابات متوقعة لاختبار منطق العقدة.
"""
with patch("src.nodes.analyzer.ChatOpenAI") as mock_class:
mock_instance = Mock()
# تكوين المحاكاة لإرجاع استجابة منظمة
mock_response = Mock()
mock_response.content = """
## ملخص التحليل
### المواضيع الرئيسية
1. الذكاء الاصطناعي يتقدم بسرعة
2. LLMs أصبحت سائدة
3. الأتمتة تحول الصناعات
### مستوى الثقة: عالي
"""
mock_instance.invoke.return_value = mock_response
mock_class.return_value = mock_instance
yield mock_instance
def test_analyzer_with_mocked_llm(mock_openai_llm):
"""اختبار عقدة المحلل باستخدام LLM محاكى."""
state = {
"query": "اتجاهات صناعة AI",
"documents": [
{"source": "تقرير", "content": "اعتماد AI ينمو"},
{"source": "أخبار", "content": "الشركات تستثمر في الأتمتة"}
],
"analysis": None,
"iteration": 1
}
result = analyzer_node(state)
# تحقق من استدعاء LLM
mock_openai_llm.invoke.assert_called_once()
# تحقق من معالجة الاستجابة بشكل صحيح
assert "ملخص التحليل" in result["analysis"]
@pytest.fixture
def mock_search_api():
"""مثبت لـ API بحث المستندات المحاكى."""
with patch("src.nodes.researcher.search_documents") as mock:
mock.return_value = [
{
"source": "arxiv",
"content": "ورقة بحثية عن معماريات المحولات",
"url": "https://arxiv.org/paper/123",
"timestamp": "2026-01-01"
},
{
"source": "أخبار",
"content": "OpenAI تعلن عن قدرات نموذج جديدة",
"url": "https://news.example.com/article",
"timestamp": "2026-01-02"
}
]
yield mock
def test_researcher_with_mocked_api(mock_search_api):
"""اختبار عقدة الباحث مع API بحث محاكى."""
state = {
"query": "معماريات المحولات",
"documents": [],
"iteration": 0,
"max_iterations": 10
}
result = researcher_node(state)
# تحقق من استدعاء API مع الاستعلام الصحيح
mock_search_api.assert_called_once()
# تحقق من إضافة المستندات بشكل صحيح
assert len(result["documents"]) == 2
اختبار التكامل للرسوم البيانية الكاملة
اختبارات التكامل تتحقق من أن الرسم البياني الكامل ينفذ بشكل صحيح، مع عمل جميع العقد معاً وتدفق الحالة بشكل صحيح بينها.
اختبار تنفيذ الرسم البياني الكامل
"""
اختبارات التكامل لسير عمل LangGraph الكاملة.
هذه الاختبارات تتحقق من:
- تجميع الرسم البياني بشكل صحيح
- تدفق الحالة بين العقد
- توجيه الحواف الشرطية بشكل صحيح
- الحالة النهائية صحيحة
"""
import pytest
from langgraph.checkpoint.memory import MemorySaver
from src.graphs.research_graph import create_research_graph
class TestResearchGraphIntegration:
"""مجموعة اختبارات التكامل للرسم البياني البحثي."""
@pytest.fixture
def compiled_graph(self):
"""إنشاء رسم بياني مجمع مع التفتيش في الذاكرة."""
graph = create_research_graph()
checkpointer = MemorySaver()
return graph.compile(checkpointer=checkpointer)
def test_full_graph_execution(self, compiled_graph):
"""اختبار تدفق الرسم البياني الكامل من البداية للنهاية."""
initial_input = {
"query": "تلخيص التطورات الأخيرة في سلامة AI",
"documents": [],
"analysis": None,
"report": None,
"iteration": 0,
"max_iterations": 10,
"approved": False,
"messages": []
}
config = {"configurable": {"thread_id": "integration-test-1"}}
result = compiled_graph.invoke(initial_input, config)
# تحقق من أن الحالة النهائية لديها الحقول المتوقعة
assert result["report"] is not None, "يجب إنتاج تقرير"
assert len(result["documents"]) > 0, "يجب جمع مستندات"
assert result["analysis"] is not None, "يجب إنتاج تحليل"
def test_graph_respects_iteration_limit(self, compiled_graph):
"""اختبار توقف الرسم البياني عند الحد الأقصى للتكرارات."""
initial_input = {
"query": "استعلام معقد قد يتكرر",
"documents": [],
"analysis": None,
"report": None,
"iteration": 0,
"max_iterations": 3, # حد منخفض للاختبار
"approved": False,
"messages": []
}
config = {"configurable": {"thread_id": "integration-test-3"}}
result = compiled_graph.invoke(initial_input, config)
assert result["iteration"] <= 3, "يجب احترام حد التكرار"
class TestStateTransitions:
"""اختبار انتقالات الحالة في الرسم البياني."""
def test_all_conditional_edge_paths(self):
"""تحقق من عمل جميع مسارات التوجيه الشرطي بشكل صحيح."""
from src.nodes.supervisor import supervisor_node
# المسار 1: لا مستندات -> باحث
state1 = {
"documents": [],
"analysis": None,
"report": None,
"iteration": 0,
"max_iterations": 10
}
result1 = supervisor_node(state1)
assert result1["next_worker"] == "researcher"
# المسار 2: لديه مستندات، لا تحليل -> محلل
state2 = {
"documents": [{"content": "مستند"}],
"analysis": None,
"report": None,
"iteration": 1,
"max_iterations": 10
}
result2 = supervisor_node(state2)
assert result2["next_worker"] == "analyzer"
# المسار 3: لديه تحليل، لا تقرير -> كاتب
state3 = {
"documents": [{"content": "مستند"}],
"analysis": "تحليل مفصل يلبي عتبة الجودة مع محتوى كافٍ",
"report": None,
"iteration": 2,
"max_iterations": 10
}
result3 = supervisor_node(state3)
assert result3["next_worker"] == "writer"
# المسار 4: كل شيء مكتمل -> انتهى
state4 = {
"documents": [{"content": "مستند"}],
"analysis": "تحليل",
"report": "تقرير",
"approved": True,
"iteration": 4,
"max_iterations": 10
}
result4 = supervisor_node(state4)
assert result4["next_worker"] == "done"
اختبار تدفقات الإنسان في الحلقة
"""
اختبارات لوظائف المقاطعة والاستئناف للإنسان في الحلقة.
"""
import pytest
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command
class TestHumanInTheLoop:
"""اختبار سير عمل الموافقة البشرية."""
@pytest.fixture
def approval_graph(self):
"""إنشاء رسم بياني يتطلب موافقة بشرية."""
from src.graphs.approval_graph import create_approval_graph
graph = create_approval_graph()
return graph.compile(checkpointer=MemorySaver())
def test_interrupt_pauses_execution(self, approval_graph):
"""تحقق من أن interrupt() يوقف الرسم البياني للإدخال البشري."""
config = {"configurable": {"thread_id": "interrupt-test"}}
result = approval_graph.invoke({
"document": "عقد يتطلب مراجعة",
"approved": False,
"review_notes": None
}, config)
# الحصول على الحالة للتحقق من الإيقاف
state = approval_graph.get_state(config)
# يجب أن يكون متوقفاً عند عقدة human_review
assert len(state.next) > 0, "يجب أن يكون الرسم البياني متوقفاً"
def test_resume_with_approval(self, approval_graph):
"""اختبار الاستئناف بعد موافقة الإنسان."""
config = {"configurable": {"thread_id": "resume-approve-test"}}
# الاستدعاء الأولي - سيتوقف للمراجعة
approval_graph.invoke({
"document": "عقد للمراجعة",
"approved": False,
"review_notes": None
}, config)
# استئناف مع الموافقة
result = approval_graph.invoke(
Command(resume={"action": "approve", "notes": "يبدو جيداً!"}),
config
)
assert result["approved"] is True
def test_resume_with_rejection(self, approval_graph):
"""اختبار الاستئناف بعد رفض الإنسان."""
config = {"configurable": {"thread_id": "resume-reject-test"}}
approval_graph.invoke({
"document": "عقد للمراجعة",
"approved": False,
"review_notes": None
}, config)
# استئناف مع الرفض
result = approval_graph.invoke(
Command(resume={"action": "reject", "reason": "الشروط غير مقبولة"}),
config
)
assert result["approved"] is False
تكوين الاختبار والمثبتات
"""
conftest.py - مثبتات pytest مشتركة لاختبار LangGraph.
"""
import pytest
import os
from unittest.mock import patch, Mock
from langgraph.checkpoint.memory import MemorySaver
@pytest.fixture(scope="session")
def mock_all_llms():
"""
مثبت على مستوى الجلسة لمحاكاة جميع استدعاءات LLM.
يوفر اختبارات سريعة وحتمية بدون تكاليف API.
"""
responses = {
"research": "وُجدت مستندات ذات صلة حول الموضوع",
"analyze": "التحليل يُظهر ثلاثة مواضيع رئيسية ناشئة",
"write": "# تقرير البحث\n\nملخص تنفيذي للنتائج..."
}
def create_mock_response(content):
mock = Mock()
mock.content = content
return mock
with patch("langchain_openai.ChatOpenAI") as mock_openai, \
patch("langchain_anthropic.ChatAnthropic") as mock_anthropic:
for mock_class in [mock_openai, mock_anthropic]:
instance = Mock()
instance.invoke.return_value = create_mock_response(responses["analyze"])
mock_class.return_value = instance
yield {"openai": mock_openai, "anthropic": mock_anthropic}
@pytest.fixture
def memory_checkpointer():
"""توفير حافظ نقاط تفتيش في الذاكرة جديد لكل اختبار."""
return MemorySaver()
@pytest.fixture
def test_config():
"""توليد تكوين اختبار فريد."""
import uuid
return {
"configurable": {
"thread_id": f"test-{uuid.uuid4().hex[:8]}"
}
}
أسئلة المقابلة
س: كيف تتعامل مع اختبار عقد LangGraph التي تستدعي LLMs خارجية؟
"حاكِ LLM على مستوى الفئة باستخدام مثبتات pytest. أنشئ استجابات محاكاة حتمية تغطي حالات النجاح والحالات الحافة وسيناريوهات الخطأ. هذا يضمن اختبارات سريعة وقابلة للتكرار بدون تكاليف API. للمسارات الحرجة، احتفظ أيضاً بمجموعة صغيرة من اختبارات التكامل مع LLMs حقيقية في CI/CD."
س: ما استراتيجيتك لاختبار منطق توجيه المشرف؟
"اختبر كل مسار شرطي بشكل مستقل. أنشئ مثبتات حالة تمثل كل نقطة قرار: مستندات فارغة، لديه مستندات، لديه تحليل، عتبات الجودة، حدود التكرار. تحقق من أن المشرف يُرجع next_worker الصحيح لكل حالة. استخدم اختبارات معلمية لتغطية جميع مجموعات التوجيه بشكل منهجي."
س: كيف تختبر تدفقات الإنسان في الحلقة في LangGraph؟
"استخدم حافظ نقاط تفتيش MemorySaver لتمكين فحص الحالة. استدعِ الرسم البياني، تحقق من توقفه عند المقاطعة مع السياق الصحيح. ثم استدعِ مرة أخرى مع Command(resume=data) للمتابعة. اختبر جميع مسارات الاستجابة البشرية: الموافقة، الرفض، طلب التغييرات. تحقق من تحديث الحالة بشكل صحيح بعد كل نوع استجابة."
س: ما مقاييس تغطية الاختبار التي تستهدفها لتطبيقات LangGraph الإنتاجية؟
"استهدف تغطية أسطر 80%+ مع التركيز على المسارات الحرجة: منطق توجيه المشرف (100%)، معالجات الأخطاء (100%)، مخفضات الحالة (100%). تغطية اختبار التكامل على جميع مسارات الرسم البياني. اختبارات دخان E2E لرحلات المستخدم الحرجة. أهم من نسبة التغطية: جودة الاختبار وتغطية الحالات الحافة."
النقاط الرئيسية
| طبقة الاختبار | التركيز | الأدوات | السرعة |
|---|---|---|---|
| اختبارات الوحدة | العقد الفردية، الدوال النقية | pytest، Mock | سريع (مللي ثانية) |
| التكامل | تدفقات الرسم البياني الكاملة، انتقالات الحالة | MemorySaver، المثبتات | متوسط (ثوان) |
| اختبارات E2E | APIs حقيقية، سيناريوهات الإنتاج | بيئة التجهيز، LLMs حقيقية | بطيء (دقائق) |
أنماط الاختبار الحرجة:
- حاكِ LLMs و APIs الخارجية لاختبارات الوحدة
- استخدم MemorySaver لاختبار التكامل
- اختبر جميع مسارات توجيه المشرف صراحة
- تحقق من دورات مقاطعة/استئناف الإنسان في الحلقة
- اختبر تراكم الحالة مع المخفضات
- غطِّ مسارات استعادة الأخطاء
التالي: تقنيات التصحيح والتصور
:::