الدرس 14 من 20

معالجة الأخطاء والتعافي

التدهور السلس

3 دقيقة للقراءة

عندما تفشل أجزاء من وكيلك، يجب أن يتدهور النظام بسلاسة—الحفاظ على وظائف جزئية بدلاً من الفشل الكامل.

استراتيجيات التدهور

الاستراتيجية متى تستخدم مثال
استجابات احتياطية فشل API إرجاع استجابة مخزنة أو افتراضية
وظائف مخفضة أداة غير متوفرة تخطي الميزات الاختيارية
مسارات بديلة فشل الطريقة الأساسية جرب نهجاً احتياطياً
تسليم للبشر إخفاقات حرجة التصعيد للدعم البشري

تنفيذ الاحتياطيات

class ResilientAgent:
    def __init__(self, primary_llm, fallback_llm):
        self.primary = primary_llm
        self.fallback = fallback_llm
        self.cache = ResponseCache()

    async def generate(self, prompt, context):
        # جرب LLM الأساسي
        try:
            return await self.primary.generate(prompt)
        except (RateLimitError, ServiceError):
            pass

        # جرب LLM الاحتياطي
        try:
            return await self.fallback.generate(prompt)
        except Exception:
            pass

        # جرب الذاكرة المؤقتة
        cached = self.cache.get_similar(prompt)
        if cached:
            return f"[استجابة مخزنة] {cached}"

        # الملاذ الأخير: رسالة فشل صادقة
        return "أواجه صعوبات تقنية. يرجى المحاولة مرة أخرى قريباً."

سلاسل احتياطية للأدوات

class ToolWithFallbacks:
    def __init__(self, tools_by_priority):
        self.tools = tools_by_priority  # [أساسي، احتياطي1، احتياطي2]

    def execute(self, query):
        errors = []

        for tool in self.tools:
            try:
                result = tool.run(query)
                return {"success": True, "result": result, "tool": tool.name}
            except Exception as e:
                errors.append(f"{tool.name}: {e}")
                continue

        return {
            "success": False,
            "error": "فشلت جميع الأدوات",
            "details": errors
        }

# مثال: البحث مع احتياطيات
search_chain = ToolWithFallbacks([
    GoogleSearch(),      # أساسي
    BingSearch(),        # احتياطي 1
    DuckDuckGoSearch()   # احتياطي 2
])

أعلام الميزات للتدهور

class FeatureManager:
    def __init__(self):
        self.features = {
            "web_search": True,
            "code_execution": True,
            "image_generation": True,
            "file_operations": True
        }
        self.health_checks = {}

    def check_health(self):
        """التحقق الدوري وتعطيل الميزات غير السليمة"""
        for feature, checker in self.health_checks.items():
            try:
                is_healthy = checker()
                self.features[feature] = is_healthy
            except:
                self.features[feature] = False

    def is_available(self, feature):
        return self.features.get(feature, False)

# الاستخدام في الوكيل
if feature_manager.is_available("web_search"):
    result = web_search(query)
else:
    result = "البحث على الويب غير متاح مؤقتاً. استخدام المعرفة المخزنة."

التواصل مع المستخدم

تواصل دائماً بوضوح حول الحالات المتدهورة:

def format_degraded_response(response, degradation_info):
    """إضافة شفافية حول القيود"""

    warnings = []

    if degradation_info.get("using_fallback"):
        warnings.append("استخدام خدمة احتياطية")

    if degradation_info.get("cached"):
        warnings.append(f"بناءً على بيانات مخزنة من {degradation_info['cache_date']}")

    if degradation_info.get("limited_tools"):
        warnings.append("بعض الميزات غير متوفرة مؤقتاً")

    if warnings:
        disclaimer = "\n---\n⚠️ " + " | ".join(warnings)
        return response + disclaimer

    return response

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

منع الإخفاقات المتتالية:

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = "closed"  # مغلق، مفتوح، نصف مفتوح
        self.last_failure = None

    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure > self.reset_timeout:
                self.state = "half-open"
            else:
                raise CircuitOpenError("الخدمة غير متوفرة")

        try:
            result = func(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.threshold:
                self.state = "open"
            raise

أفضل الممارسات

افعل لا تفعل
تواصل بوضوح حول القيود إخفاء الإخفاقات عن المستخدمين
سجل جميع أحداث التدهور التبديل الصامت للاحتياطيات
حدد مهلات معقولة الانتظار إلى ما لا نهاية
اختبر مسارات الاحتياط بانتظام افتراض أن النسخ الاحتياطية تعمل

التالي: تعلم تقنيات التحقق لمنع المخرجات السيئة من الوصول للمستخدمين. :::

اختبار

الوحدة 4: معالجة الأخطاء والتعافي

خذ الاختبار