أنماط استرداد الأخطاء: Try-Catch والمسارات البديلة وقواطع الدائرة — الغوص العميق في LangGraph: سير عمل الذكاء الاصطناعي الإنتاجية ذات الحالة — Nerd Level Tech

لماذا استرداد الأخطاء حاسم في الإنتاج

حادثة إنتاجية حقيقية (يناير 2026):

تعطل نظام دعم العملاء متعدد الوكلاء في شركة تجارة إلكترونية 23 مرة في يوم واحد خلال الجمعة السوداء. السبب الجذري: حدود معدل Claude API أثارت فشلاً متتالياً عبر جميع الوكلاء. تراكمت تذاكر العملاء، مما كلف 150 ألف دولار في الإيرادات المفقودة.

بعد تنفيذ أنماط استرداد الأخطاء المناسبة—عقد try-catch، والمسارات البديلة، والتراجع الأسي، وقواطع الدائرة—تعامل نفس النظام مع 10 أضعاف حركة المرور في يوم الاثنين السيبراني مع صفر أعطال.

هذا الدرس يعلمك: أنماط استرداد الأخطاء المعززة للإنتاج التي تمنع الفشل المتتالي وتبقي سير عمل LangGraph يعمل 24/7.

الركائز الثلاث لاسترداد الأخطاء

عقد Try-Catch: عزل الفشل حتى لا يعطل الرسم البياني
المسارات البديلة: توفير بدائل متدهورة ولكن وظيفية
قواطع الدائرة: منع الفشل المتتالي للخدمات الخارجية

نمط عقدة Try-Catch

المشكلة: الاستثناءات غير المعالجة

# ❌ سيء: يعطل الرسم البياني بالكامل
def risky_node(state: dict) -> dict:
    result = call_external_api()  # يمكن أن يرمي!
    return {"result": result}

# مهلة API واحدة = تعطل سير العمل بالكامل
# نقطة التفتيش مفقودة، يجب إعادة التشغيل من الصفر

الحل: غلاف Try-Catch

from typing import TypedDict, Annotated, Optional, Literal
import operator
import traceback

class RobustState(TypedDict):
    """حالة مع معالجة أخطاء شاملة."""
    # البيانات الأساسية
    query: str
    documents: Annotated[list[str], operator.add]
    analysis: Optional[str]

    # تتبع الأخطاء
    error_message: Optional[str]
    error_type: Optional[str]
    error_traceback: Optional[str]
    failed_node: Optional[str]

    # إدارة إعادة المحاولة
    retry_count: int
    max_retries: int

def try_research_node(state: RobustState) -> dict:
    """
    عقدة البحث مغلفة في try-catch.
    لا تتعطل أبداً، تعيد دائماً تحديث الحالة.
    """
    try:
        # عملية محفوفة بالمخاطر
        documents = search_documents(state["query"])

        # نجاح: مسح أي أخطاء سابقة
        return {
            "documents": documents,
            "error_message": None,
            "error_type": None,
            "failed_node": None
        }

    except TimeoutError as e:
        return {
            "error_message": f"انتهت مهلة البحث: {str(e)}",
            "error_type": "timeout",
            "error_traceback": traceback.format_exc(),
            "failed_node": "research",
            "retry_count": state["retry_count"] + 1
        }

    except RateLimitError as e:
        return {
            "error_message": f"تم تحديد المعدل: {str(e)}",
            "error_type": "rate_limit",
            "failed_node": "research",
            "retry_count": state["retry_count"] + 1
        }

    except Exception as e:
        return {
            "error_message": f"خطأ غير متوقع: {str(e)}",
            "error_type": "unknown",
            "error_traceback": traceback.format_exc(),
            "failed_node": "research",
            "retry_count": state["retry_count"] + 1
        }

def route_after_research(state: RobustState) -> Literal["continue", "retry", "fallback"]:
    """التوجيه بناءً على حالة الخطأ."""
    # حالة النجاح
    if not state.get("error_message"):
        return "continue"

    # التحقق من إمكانية إعادة المحاولة
    if state["retry_count"] < state["max_retries"]:
        error_type = state.get("error_type", "unknown")

        # إعادة المحاولة فقط للأخطاء العابرة
        if error_type in ["timeout", "rate_limit", "connection_error"]:
            return "retry"

    # تجاوز الحد الأقصى لإعادة المحاولات أو خطأ غير قابل لإعادة المحاولة
    return "fallback"

نمط الإنتاج: Try-Catch المعتمد على المزخرف

from functools import wraps
from typing import Callable

def catch_errors(node_name: str):
    """مزخرف لإضافة معالجة الأخطاء لأي عقدة."""
    def decorator(func: Callable):
        @wraps(func)
        def wrapper(state: dict) -> dict:
            try:
                return func(state)
            except Exception as e:
                return {
                    "error_message": str(e),
                    "error_type": type(e).__name__,
                    "error_traceback": traceback.format_exc(),
                    "failed_node": node_name,
                    "retry_count": state.get("retry_count", 0) + 1
                }
        return wrapper
    return decorator

# تطبيق على أي عقدة
@catch_errors("research")
def research_node(state: RobustState) -> dict:
    documents = search_documents(state["query"])
    return {"documents": documents}

@catch_errors("analysis")
def analysis_node(state: RobustState) -> dict:
    analysis = analyze_documents(state["documents"])
    return {"analysis": analysis}

المسارات البديلة: التدهور الآمن

المشكلة: ماذا يحدث بعد فشل إعادة المحاولات؟

بدون بدائل، إعادة المحاولات المستنفدة تتركك بلا شيء:

المستخدم يرى خطأ عام
لا توجد نتائج جزئية
يتوقف سير العمل تماماً

الحل: بدائل متعددة المستويات

from typing import TypedDict, Optional, Literal

class FallbackState(TypedDict):
    query: str
    primary_result: Optional[str]
    fallback_result: Optional[str]
    cached_result: Optional[str]
    error_message: Optional[str]
    fallback_level: int  # 0=رئيسي، 1=بديل، 2=مخزن مؤقتاً، 3=ثابت

def primary_processor(state: FallbackState) -> dict:
    """المسار الرئيسي: استخدم أحدث نموذج."""
    try:
        result = call_claude_opus(state["query"])
        return {
            "primary_result": result,
            "fallback_level": 0
        }
    except Exception as e:
        return {"error_message": str(e)}

def alternative_processor(state: FallbackState) -> dict:
    """البديل 1: استخدم نموذج أرخص/أسرع."""
    try:
        result = call_claude_haiku(state["query"])  # أسرع، أرخص
        return {
            "fallback_result": result,
            "fallback_level": 1,
            "error_message": None
        }
    except Exception as e:
        return {"error_message": str(e)}

def cached_processor(state: FallbackState) -> dict:
    """البديل 2: إرجاع نتيجة مماثلة من الذاكرة المخبأة."""
    cache_key = compute_query_hash(state["query"])
    cached = cache.get(cache_key)

    if cached:
        return {
            "cached_result": cached,
            "fallback_level": 2,
            "error_message": None
        }

    return {"error_message": "لا توجد نتيجة مخبأة متاحة"}

def static_fallback(state: FallbackState) -> dict:
    """البديل 3: إرجاع رد ثابت."""
    return {
        "fallback_result": """
عذراً، أواجه صعوبات تقنية.
إليك ما يمكنني إخبارك به عن استفسارك:

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

في هذه الأثناء، قد تجد هذه الموارد مفيدة:
- التوثيق: https://docs.example.com
- الأسئلة الشائعة: https://example.com/faq
        """.strip(),
        "fallback_level": 3,
        "error_message": "جميع الخدمات غير متاحة، استخدم البديل الثابت"
    }

def route_fallback(state: FallbackState) -> Literal["done", "try_alternative", "try_cache", "static"]:
    """التوجيه عبر سلسلة البدائل."""
    if state.get("primary_result") or state.get("fallback_result"):
        return "done"

    level = state.get("fallback_level", -1)
    error = state.get("error_message")

    if not error:
        return "done"

    # الانتقال إلى المستوى البديل التالي
    if level < 1:
        return "try_alternative"
    elif level < 2:
        return "try_cache"
    else:
        return "static"

التراجع الأسي مع التشويش

لماذا فشل إعادة المحاولات الأساسية

# ❌ سيء: يضرب الـ API
def bad_retry(state):
    for i in range(5):
        try:
            return call_api()
        except:
            time.sleep(1)  # تأخير ثابت = API يبقى محملاً زيادة

نمط الإنتاج: التراجع الأسي

import time
import random
from typing import TypedDict, Optional

class RetryState(TypedDict):
    task: str
    result: Optional[str]
    attempt: int
    max_attempts: int
    base_delay: float  # ثواني
    max_delay: float   # الحد الأقصى للتأخير

def calculate_backoff(attempt: int, base: float, max_delay: float) -> float:
    """
    حساب التأخير مع التراجع الأسي والتشويش.
    الصيغة: min(base * 2^attempt + random(0,1), max_delay)
    """
    exponential = base * (2 ** attempt)
    jitter = random.uniform(0, 1)  # منع القطيع المزدحم
    return min(exponential + jitter, max_delay)

def retry_with_backoff(state: RetryState) -> dict:
    """التنفيذ مع إعادة محاولة التراجع الأسي."""
    attempt = state["attempt"]

    if attempt > 0:
        # الانتظار قبل إعادة المحاولة (ليس في المحاولة الأولى)
        delay = calculate_backoff(
            attempt - 1,
            state.get("base_delay", 1.0),
            state.get("max_delay", 60.0)
        )
        print(f"المحاولة {attempt}: الانتظار {delay:.2f}s قبل إعادة المحاولة")
        time.sleep(delay)

    try:
        result = perform_operation(state["task"])
        return {
            "result": result,
            "attempt": attempt + 1,
            "error_message": None
        }
    except Exception as e:
        return {
            "error_message": str(e),
            "attempt": attempt + 1
        }

# مثال تسلسل التراجع (base=1s):
# المحاولة 1: فوري
# المحاولة 2: ~1-2s تأخير (1*2^0 + تشويش)
# المحاولة 3: ~2-3s تأخير (1*2^1 + تشويش)
# المحاولة 4: ~4-5s تأخير (1*2^2 + تشويش)
# المحاولة 5: ~8-9s تأخير (1*2^3 + تشويش)

نمط قاطع الدائرة

المشكلة: الفشل المتتالي

عندما تتعطل خدمة خارجية:

جميع الطلبات تنتهي مهلتها (بطيء)
المهلات تستنفد مجموعات الخيوط
الذاكرة تمتلئ بالطلبات المعلقة
النظام بأكمله يصبح غير مستجيب

الحل: قاطع الدائرة

from typing import TypedDict, Literal, Optional
from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # التشغيل العادي
    OPEN = "open"          # حظر الطلبات
    HALF_OPEN = "half_open"  # اختبار التعافي

class CircuitBreakerState(TypedDict):
    """حالة لنمط قاطع الدائرة."""
    query: str
    result: Optional[str]

    # تتبع قاطع الدائرة
    circuit_status: str  # closed, open, half_open
    consecutive_failures: int
    last_failure_time: Optional[str]

    # التكوين
    failure_threshold: int    # فتح بعد N فشل
    recovery_timeout: int     # ثواني قبل المحاولة مرة أخرى
    half_open_max_calls: int  # مكالمات الاختبار في نصف مفتوح

class CircuitBreaker:
    """تنفيذ قاطع الدائرة الإنتاجي."""

    def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failures = 0
        self.last_failure_time = None
        self.half_open_calls = 0

    def can_execute(self) -> tuple[bool, str]:
        """التحقق مما إذا كان يجب متابعة الطلب."""
        if self.state == CircuitState.CLOSED:
            return True, "closed"

        if self.state == CircuitState.OPEN:
            # التحقق من مرور مهلة التعافي
            if self.last_failure_time:
                elapsed = (datetime.now() - self.last_failure_time).total_seconds()
                if elapsed >= self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_calls = 0
                    return True, "half_open"
            return False, "open"

        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_calls < 3:  # السماح بمكالمات اختبار محدودة
                self.half_open_calls += 1
                return True, "half_open"
            return False, "half_open_limit"

        return False, "unknown"

    def record_success(self):
        """تسجيل مكالمة ناجحة."""
        self.failures = 0
        self.state = CircuitState.CLOSED
        self.half_open_calls = 0

    def record_failure(self):
        """تسجيل مكالمة فاشلة."""
        self.failures += 1
        self.last_failure_time = datetime.now()

        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"الدائرة فُتحت بعد {self.failures} فشل")

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

س1: "كيف تمنع فشل API واحد من تعطل سير عمل LangGraph بالكامل؟"

إجابة قوية:

"أغلف جميع المكالمات الخارجية في كتل try-catch داخل العقد. بدلاً من رفع الاستثناءات، تعيد العقد تحديثات حالة الخطأ مثل {error_message, error_type, failed_node}. حافة شرطية ثم توجه إما لإعادة المحاولة (للأخطاء العابرة مثل المهلات) أو البديل (للفشل المستمر). الرسم البياني لا يتعطل أبداً—يتدهور بأمان. أستخدم أيضاً نمط المزخرف لإضافة معالجة الأخطاء بشكل موحد عبر جميع العقد."

س2: "متى تستخدم التراجع الأسي مقابل تأخيرات إعادة المحاولة الثابتة؟"

الإجابة:

"التأخيرات الثابتة تسبب 'القطيع المزدحم'—عندما يتعافى API، جميع الطلبات المنتظرة تضربه في وقت واحد. التراجع الأسي يوزع حمل إعادة المحاولة عبر الوقت: 1s، 2s، 4s، 8s. أضيف التشويش (عشوائي 0-1s) لزيادة توزيع إعادة المحاولات عبر العملاء. هذا يعطي الـ API وقتاً للتعافي. أستخدم التراجع الأسي لحدود المعدل وأخطاء التحميل الزائد، لكن قد أستخدم تأخيرات ثابتة أقصر لمشاكل الشبكة حيث أتوقع التعافي الفوري."

س3: "اشرح نمط قاطع الدائرة ومتى تستخدمه في LangGraph."

الإجابة:

"قاطع الدائرة يتتبع الفشل المتتالي لخدمة خارجية. عندما يتجاوز الفشل عتبة (مثل 5)، 'يفتح' ويرفض الطلبات فوراً بدلاً من الانتظار للمهلات. بعد فترة تعافي (مثل 60s)، يدخل حالة 'نصف مفتوح' ويسمح ببضع طلبات اختبار. إذا نجحت، 'يغلق' ويستأنف التشغيل العادي. أستخدمه لـ APIs الخارجية التي قد تتعطل—يمنع الفشل المتتالي حيث طلبات انتظار المهلة تستنفد الموارد. في LangGraph، أتتبع حالة الدائرة في حالة الرسم البياني وأتحقق منها قبل إجراء مكالمات API."

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

Try-catch في كل عقدة: لا تدع الاستثناءات تعطل الرسم البياني
حقول حالة الخطأ: تتبع error_message، error_type، failed_node
التراجع الأسي مع التشويش: يمنع القطيع المزدحم
بدائل متعددة المستويات: رئيسي → بديل → مخبأ → ثابت
قاطع الدائرة: يفتح بعد N فشل، يختبر التعافي بعد المهلة
التوجيه حسب نوع الخطأ: الأخطاء القابلة لإعادة المحاولة تعيد المحاولة، الأخرى تذهب للبديل

الوحدة 2 اكتملت! لا تنسَ إجراء اختبار الوحدة 2 لاختبار معرفتك. التالي: الوحدة 3 - نقاط التفتيش والاستمرارية حيث ستتعلم جعل سير عملك قابلاً للاستئناف وجاهزاً للإنتاج.

:::