معالجة الأخطاء والتعافي
التدهور السلس
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
أفضل الممارسات
| افعل | لا تفعل |
|---|---|
| تواصل بوضوح حول القيود | إخفاء الإخفاقات عن المستخدمين |
| سجل جميع أحداث التدهور | التبديل الصامت للاحتياطيات |
| حدد مهلات معقولة | الانتظار إلى ما لا نهاية |
| اختبر مسارات الاحتياط بانتظام | افتراض أن النسخ الاحتياطية تعمل |
التالي: تعلم تقنيات التحقق لمنع المخرجات السيئة من الوصول للمستخدمين. :::