الحلقات التكرارية واكتشاف الدورات

لماذا الحلقات التكرارية ضرورية لوكلاء الذكاء الاصطناعي

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

احتاج وكيل دعم العملاء بالذكاء الاصطناعي في شركة SaaS للتعامل مع تذاكر معقدة تتطلب جولات متعددة من التحقيق. التنفيذ الأولي: سير عمل خطي يجري مكالمة LLM واحدة ويعود. النتيجة؟ 40% من التذاكر تطلبت تصعيداً يدوياً لأن الوكيل لم يستطع التكرار لإيجاد الحل الصحيح.

بعد إضافة الحلقات التكرارية مع اكتشاف الدورات المناسب وحدود التكرار، حل الوكيل 85% من التذاكر تلقائياً من خلال التكرار حتى يجد الإجابة الصحيحة أو يصل إلى حد الأمان.

هذا الدرس يعلمك: كيفية تنفيذ حلقات تكرارية آمنة للإنتاج في LangGraph، واكتشاف الدورات، وتعيين حدود التكرار، وتجنب خطأ الإنتاج رقم 1: الحلقات اللانهائية التي تستهلك رصيد API.

---

فهم دورات الرسم البياني في LangGraph

ما هي الدورة؟

تحدث الدورة عندما يمكن للعقدة الانتقال مرة أخرى إلى عقدة سابقة، مما يخلق حلقة:

research → analyze → decide
    ↑                    ↓
    └────────────────────┘

لماذا الدورات مفيدة:

• وكلاء "يفكرون مرة أخرى" عندما تكون النتائج غير مُرضية
• التنقيح التكراري (مسودة → مراجعة → تنقيح → مراجعة → انتهاء)
• أنماط البحث والتنقيح (بحث → تقييم → بحث مرة أخرى إذا لزم الأمر)

لماذا الدورات خطيرة:

• الحلقات اللانهائية تستنفد الذاكرة وميزانيات API
• بدون حدود، LLM مرتبك يمكن أن يتكرر للأبد
• أنظمة الإنتاج تحتاج توقفات صلبة

---

نمط الحلقة التكرارية الأساسي

تنفيذ LangGraph 1.0.5 (يناير 2026)

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

class ResearchState(TypedDict):
    """حالة لسير عمل البحث التكراري."""
    query: str
    research_results: Annotated[list[str], operator.add]
    analysis: Optional[str]
    is_satisfactory: bool

    # التحكم في التكرار
    iteration_count: int
    max_iterations: int

def research_node(state: ResearchState) -> dict:
    """تنفيذ تكرار البحث."""
    # محاكاة البحث (استبدل بمكالمة LLM فعلية)
    new_findings = f"نتيجة البحث #{state['iteration_count'] + 1}"

    return {
        "research_results": [new_findings],
        "iteration_count": state["iteration_count"] + 1
    }

def analyze_node(state: ResearchState) -> dict:
    """تحليل نتائج البحث وتحديد إذا كانت مُرضية."""
    # محاكاة التحليل (استبدل بمكالمة LLM فعلية)
    combined_results = "\n".join(state["research_results"])
    analysis = f"تحليل {len(state['research_results'])} نتيجة"

    # منطق القرار (في الإنتاج، استخدم LLM)
    is_satisfactory = len(state["research_results"]) >= 3

    return {
        "analysis": analysis,
        "is_satisfactory": is_satisfactory
    }

def should_continue(state: ResearchState) -> Literal["research", "end"]:
    """
    حافة شرطية: متابعة الحلقة أو الانتهاء؟
    حرج: تحقق دائماً من حد التكرار أولاً!
    """
    # فحص الأمان: دائماً فرض حد التكرار
    if state["iteration_count"] >= state["max_iterations"]:
        print(f"تم الوصول للحد الأقصى ({state['max_iterations']}) تكرارات")
        return "end"

    # منطق العمل: هل البحث مُرضٍ؟
    if state["is_satisfactory"]:
        return "end"

    # متابعة التكرار
    return "research"

# بناء الرسم البياني مع دورة
graph = StateGraph(ResearchState)
graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)

# تعيين نقطة الدخول
graph.set_entry_point("research")

# إنشاء الدورة: research -> analyze -> (research OR end)
graph.add_edge("research", "analyze")
graph.add_conditional_edges(
    "analyze",
    should_continue,
    {
        "research": "research",  # العودة للخلف
        "end": END               # الخروج
    }
)

# الترجمة
app = graph.compile()

# الاستدعاء مع حد الأمان
result = app.invoke({
    "query": "أحدث اتجاهات AI 2026",
    "research_results": [],
    "analysis": None,
    "is_satisfactory": False,
    "iteration_count": 0,
    "max_iterations": 5  # حد صلب
})

print(f"اكتمل في {result['iteration_count']} تكرارات")

---

نمط الإنتاج: حدود التكرار

لماذا الحدود غير قابلة للتفاوض

بدون حدود (حادثة إنتاجية):

# ❌ سيء: لا يوجد حد تكرار
def should_continue(state):
    if state["is_satisfactory"]:
        return "end"
    return "research"  # يمكن أن يتكرر للأبد!

# النتيجة: الوكيل أجرى 847 تكراراً في 3 ساعات
# التكلفة: $2,400 في مكالمات API قبل الإيقاف اليدوي

مع حدود (آمن للإنتاج):

# ✅ جيد: دائماً تحقق من الحد أولاً
def should_continue(state: ResearchState) -> str:
    # الأولوية 1: فرض الحد الصلب
    if state["iteration_count"] >= state["max_iterations"]:
        return "end"

    # الأولوية 2: منطق العمل
    if state["is_satisfactory"]:
        return "end"

    return "research"

نمط الحدود القابلة للتكوين

class ConfigurableLoopState(TypedDict):
    query: str
    results: Annotated[list[str], operator.add]
    iteration_count: int

    # حدود قابلة للتكوين
    max_iterations: int       # حد صلب (مثل، 10)
    min_iterations: int       # الحد الأدنى قبل التحقق من الرضا
    timeout_seconds: float    # حد قائم على الوقت
    start_time: str           # طابع زمني ISO

from datetime import datetime

def should_continue_with_config(state: ConfigurableLoopState) -> str:
    """فحص استمرار درجة الإنتاج."""

    # الفحص 1: حد التكرار الصلب
    if state["iteration_count"] >= state["max_iterations"]:
        print(f"تم الوصول للحد الصلب: {state['max_iterations']} تكرارات")
        return "end"

    # الفحص 2: حد قائم على الوقت
    start = datetime.fromisoformat(state["start_time"])
    elapsed = (datetime.now() - start).total_seconds()
    if elapsed > state["timeout_seconds"]:
        print(f"انتهت المهلة: {elapsed:.1f}s > {state['timeout_seconds']}s")
        return "end"

    # الفحص 3: الحد الأدنى من التكرارات قبل الخروج المبكر
    if state["iteration_count"] < state["min_iterations"]:
        return "continue"  # فرض N تكرارات على الأقل

    # الفحص 4: منطق العمل (فحص الرضا)
    if is_satisfactory(state["results"]):
        return "end"

    return "continue"

---

اكتشاف الدورات في الرسوم البيانية المعقدة

المشكلة: اكتشاف الحلقات اللانهائية

في الرسوم البيانية ذات الدورات المتعددة، اكتشاف الحلقات اللانهائية يتطلب تتبع الحالات المزارة:

from typing import TypedDict, Annotated
import operator
import hashlib
import json

class CycleAwareState(TypedDict):
    """حالة مع اكتشاف الدورات."""
    query: str
    results: Annotated[list[str], operator.add]
    iteration_count: int
    max_iterations: int

    # اكتشاف الدورات
    state_hashes: Annotated[list[str], operator.add]
    cycle_detected: bool

def compute_state_hash(state: dict, exclude_keys: list[str]) -> str:
    """
    حساب hash لحقول الحالة ذات الصلة لاكتشاف الحالات المتكررة.
    استبعاد عدادات التكرار والبيانات الوصفية.
    """
    relevant = {k: v for k, v in state.items() if k not in exclude_keys}
    state_str = json.dumps(relevant, sort_keys=True, default=str)
    return hashlib.sha256(state_str.encode()).hexdigest()[:16]

def cycle_aware_node(state: CycleAwareState) -> dict:
    """عقدة تكتشف الدورات."""
    # حساب hash للحالة الحالية (استبعاد العدادات)
    current_hash = compute_state_hash(
        state,
        exclude_keys=["iteration_count", "state_hashes", "cycle_detected"]
    )

    # التحقق من الدورة
    if current_hash in state["state_hashes"]:
        print(f"تم اكتشاف دورة! hash الحالة {current_hash} شوهد من قبل")
        return {
            "cycle_detected": True,
            "iteration_count": state["iteration_count"] + 1
        }

    # معالجة عادية
    new_result = process_query(state["query"])

    return {
        "results": [new_result],
        "state_hashes": [current_hash],
        "iteration_count": state["iteration_count"] + 1
    }

def should_continue_cycle_aware(state: CycleAwareState) -> str:
    """التحقق من الدورات قبل المتابعة."""
    if state["cycle_detected"]:
        print("الخروج من الدورة")
        return "end"

    if state["iteration_count"] >= state["max_iterations"]:
        return "end"

    return "continue"

---

نمط الإنتاج: حدود عمق التكرار

حماية LangGraph المدمجة

اعتباراً من LangGraph 1.0.5 (يناير 2026)، يمكنك تعيين حدود التكرار في وقت الترجمة:

from langgraph.graph import StateGraph

# بناء الرسم البياني الخاص بك
graph = StateGraph(ResearchState)
# ... إضافة العقد والحواف ...

# الترجمة مع حد التكرار
app = graph.compile(
    recursion_limit=25  # الحد الأقصى للخطوات قبل التوقف القسري
)

# هذا يمنع الحلقات اللانهائية على مستوى الإطار
try:
    result = app.invoke(initial_state)
except RecursionError as e:
    print(f"تم الوصول لحد التكرار: {e}")
    # التعامل بأمان

استراتيجية الحماية المتعددة الطبقات

# الطبقة 1: عداد تكرار مستوى الحالة (منطق العمل)
class State(TypedDict):
    iteration_count: int
    max_iterations: int  # مثل، 10

# الطبقة 2: حد التكرار في وقت الترجمة (أمان الإطار)
app = graph.compile(recursion_limit=50)  # 5x احتياط

# الطبقة 3: مهلة عند الاستدعاء (أمان البنية التحتية)
import asyncio

async def invoke_with_timeout(app, state, timeout_seconds=300):
    """الاستدعاء مع مهلة صلبة."""
    try:
        result = await asyncio.wait_for(
            app.ainvoke(state),
            timeout=timeout_seconds
        )
        return result
    except asyncio.TimeoutError:
        print(f"المهلة الصلبة بعد {timeout_seconds}s")
        return {"error": "Timeout", "partial_state": state}

---

مثال واقعي: وكيل بحث تكراري

from typing import TypedDict, Annotated, Optional, Literal
import operator
from langgraph.graph import StateGraph, END
from langchain_anthropic import ChatAnthropic

class IterativeResearchState(TypedDict):
    """حالة بحث تكراري جاهزة للإنتاج."""
    # الإدخال
    query: str

    # تراكم البحث
    sources: Annotated[list[str], operator.add]
    findings: Annotated[list[str], operator.add]

    # التركيب
    draft_report: Optional[str]
    quality_score: float  # 0.0 إلى 1.0
    quality_threshold: float  # مثل، 0.8

    # التحكم
    iteration_count: int
    max_iterations: int
    phase: Literal["research", "synthesize", "review", "done"]

    # قابلية المراقبة
    total_tokens: Annotated[int, operator.add]

# تهيئة LLM
llm = ChatAnthropic(model="claude-sonnet-4-20250514")

def research_node(state: IterativeResearchState) -> dict:
    """جمع المزيد من المصادر والنتائج."""
    prompt = f"""
    الاستعلام: {state['query']}

    النتائج الموجودة ({len(state['findings'])}):
    {chr(10).join(state['findings'][-5:])}  # آخر 5 نتائج

    اعثر على 2-3 نتائج جديدة غير مغطاة بالفعل. كن محدداً واستشهد بالمصادر.
    """

    response = llm.invoke(prompt)
    new_findings = parse_findings(response.content)

    return {
        "findings": new_findings,
        "sources": extract_sources(response.content),
        "total_tokens": response.usage_metadata.get("total_tokens", 0),
        "iteration_count": state["iteration_count"] + 1,
        "phase": "synthesize"
    }

def synthesize_node(state: IterativeResearchState) -> dict:
    """إنشاء أو تحديث مسودة التقرير."""
    prompt = f"""
    أنشئ تقريراً شاملاً من هذه النتائج:
    {chr(10).join(state['findings'])}

    المسودة السابقة (إن وجدت):
    {state['draft_report'] or 'لا يوجد'}

    اكتب تقريراً محسناً ومنظماً جيداً.
    """

    response = llm.invoke(prompt)

    return {
        "draft_report": response.content,
        "total_tokens": response.usage_metadata.get("total_tokens", 0),
        "phase": "review"
    }

def review_node(state: IterativeResearchState) -> dict:
    """تقييم جودة التقرير."""
    prompt = f"""
    قيّم هذا التقرير على مقياس من 0.0 إلى 1.0:

    {state['draft_report']}

    المعايير:
    - الشمولية (يغطي جميع الجوانب)
    - الدقة (صحيح وقائعياً)
    - الوضوح (مكتوب جيداً)
    - الدليل (يستشهد بالمصادر)

    أعد فقط رقماً بين 0.0 و 1.0.
    """

    response = llm.invoke(prompt)
    score = float(response.content.strip())

    return {
        "quality_score": score,
        "total_tokens": response.usage_metadata.get("total_tokens", 0),
        "phase": "done" if score >= state["quality_threshold"] else "research"
    }

def route_by_phase(state: IterativeResearchState) -> str:
    """التوجيه بناءً على المرحلة والحدود."""
    # دائماً تحقق من حد التكرار أولاً
    if state["iteration_count"] >= state["max_iterations"]:
        return "end"

    # التوجيه حسب المرحلة
    phase = state["phase"]
    if phase == "done":
        return "end"
    elif phase == "research":
        return "research"
    elif phase == "synthesize":
        return "synthesize"
    elif phase == "review":
        return "review"
    else:
        return "end"  # احتياط الأمان

# بناء الرسم البياني
graph = StateGraph(IterativeResearchState)
graph.add_node("research", research_node)
graph.add_node("synthesize", synthesize_node)
graph.add_node("review", review_node)

graph.set_entry_point("research")

# إضافة التوجيه الشرطي من كل عقدة
for node_name in ["research", "synthesize", "review"]:
    graph.add_conditional_edges(
        node_name,
        route_by_phase,
        {
            "research": "research",
            "synthesize": "synthesize",
            "review": "review",
            "end": END
        }
    )

# الترجمة مع حد الأمان
app = graph.compile(recursion_limit=30)

---

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

س1: "كيف تمنع الحلقات اللانهائية في سير عمل LangGraph التكراري؟"

إجابة قوية:

"أستخدم الحماية المتعددة الطبقات: أولاً، عدادات التكرار على مستوى الحالة مع حدود منطق العمل (مثل، حد أقصى 10 تكرارات). ثانياً، recursion_limit في LangGraph في وقت الترجمة كشبكة أمان للإطار (مثل، 50 خطوة). ثالثاً، مهلات async على مستوى الاستدعاء للحدود الزمنية الصلبة. دائماً أتحقق من حدود التكرار قبل منطق العمل في الحواف الشرطية لضمان فرض التوقفات الصلبة حتى لو كان منطق العمل به أخطاء."

س2: "متى تستخدم اكتشاف الدورات مقابل عدادات التكرار البسيطة؟"

الإجابة:

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

س3: "كيف تصحح مشاكل الحلقة اللانهائية في الإنتاج؟"

الإجابة:

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

---

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

• دائماً عيّن حدود التكرار في الحالة وتحقق منها أولاً في الحواف الشرطية
• استخدم recursion_limit في وقت الترجمة كشبكة أمان
• أضف حماية المهلة للاستدعاءات async
• اكتشاف الدورات يلتقط الوكلاء العالقين الذين يكررون نفس الحالة
• الحماية المتعددة الطبقات: حدود الحالة + حدود الإطار + مهلات البنية التحتية
• سجّل البيانات الوصفية للتكرار لتصحيح مشاكل الإنتاج

التالي: تعلم التفريع الشرطي مع 3+ فروع و Send API للتفريع المتوازي في الدرس 2.

:::