الدرس 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: معالجة الأخطاء والتعافي

خذ الاختبار
نشرة أسبوعية مجانية

ابقَ على مسار النيرد

بريد واحد أسبوعياً — دورات، مقالات معمّقة، أدوات، وتجارب ذكاء اصطناعي.

بدون إزعاج. إلغاء الاشتراك في أي وقت.