التعافي من الأخطاء للوكلاء

الأخطاء تحدث. الشبكات تفشل، APIs تُرجع 500، Mind Keys تنتهي. يُظهر هذا الدليل كيف ينبغي لوكلاء LLM التعامل مع الأخطاء بلطف والتعلّم منها.

مبادئ معالجة الأخطاء

1. أعد المحاولة مع تراجع — الأخطاء العابرة تُحل غالبًا
2. خزّن الخطأ — تعلّم من الأنماط
3. تدهور بلطف — لا تُعطّل الجلسة كاملة
4. أبلغ الإنسان — للأخطاء التي لا تستطيع حلها

معالجة أخطاء HTTP

إعادة المحاولة مع تراجع أُسي

import time
import requests

def call_with_retry(url, max_retries=3, backoff_base=2):
    """Call URL with exponential backoff."""
    for attempt in range(max_retries):
        try:
            r = requests.get(url, headers={"Authorization": f"Bearer {KEY}"})
            
            # نجاح
            if r.status_code < 400:
                return r
            
            # حد المعدل — انتظر وأعد المحاولة
            if r.status_code == 429:
                wait = int(r.headers.get("Retry-After", 60))
                print(f"Rate limited. Waiting {wait}s...")
                time.sleep(wait)
                continue
            
            # خطأ خادم — أعد المحاولة مع تراجع
            if r.status_code >= 500:
                wait = backoff_base ** attempt
                print(f"Server error {r.status_code}. Retrying in {wait}s...")
                time.sleep(wait)
                continue
            
            # خطأ عميل — لا تعد المحاولة
            if 400 <= r.status_code < 500:
                raise ClientError(f"{r.status_code}: {r.text}")
        
        except requests.RequestException as e:
            # خطأ شبكة — أعد المحاولة
            wait = backoff_base ** attempt
            print(f"Network error: {e}. Retrying in {wait}s...")
            time.sleep(wait)
    
    raise MaxRetriesError(f"Failed after {max_retries} retries")

معالجة خطأ المصادقة

def safe_call(url):
    """Call with auth error handling."""
    try:
        return call_with_retry(url)
    except ClientError as e:
        if "401" in str(e):
            # Mind Key غير صالح — حرج، لا يمكن التعافي
            store_error("auth_invalid", str(e), "Check Mind Key")
            notify_human("My Mind Key is invalid. Please update.")
            raise AuthError("Cannot continue without valid Mind Key")
        elif "403" in str(e):
            store_error("forbidden", str(e), "Wrong token type?")
            raise ForbiddenError(str(e))
        elif "404" in str(e):
            # المسار غير موجود — لا تعد المحاولة
            store_error("not_found", str(e), "Check endpoint path")
            raise NotFoundError(str(e))
        else:
            raise

تخزين الأخطاء كذكريات

عند وقوع أخطاء، خزّنها حتى تتعلّم الجلسات المستقبلية:

def store_error(error_type, error_message, recovery_hint=""):
    """Store an error as a memory for future reference."""
    requests.post(f"{URL}/memory",
        headers={"Authorization": f"Bearer {KEY}",
                 "Content-Type": "application/json"},
        json={
            "category": "mistake",
            "key": f"error_{error_type}_{int(time.time())}",
            "content": f"Error: {error_type}\nMessage: {error_message}\nRecovery: {recovery_hint}",
            "tags": ["error", error_type],
            "priority": "high"
        })

# مثال
try:
    deploy()
except DeployError as e:
    store_error("deploy_failed", str(e), 
                "Check CI logs, verify Docker image exists")
    raise

سيناريوهات أخطاء شائعة

السيناريو 1: Mind Key غير صالح

# الكشف: 401 على كل استدعاء
# التعافي: لا يمكن التعافي — يحتاج تدخل إنسان

def handle_invalid_mind_key():
    store_error("mind_key_invalid", 
                "All API calls returning 401",
                "Mind Key may be revoked. Need new key.")
    
    # أبلغ الإنسان عبر الدردشة (إن أمكن)
    try:
        reply("⚠️ My Mind Key is invalid. I cannot access memories. "
              "Please check and update SYNAPSE_MIND_KEY.")
    except:
        pass  # قد تفشل الدردشة أيضًا بمفتاح سيء
    
    # اخرج بلطف
    raise CriticalError("Cannot continue without valid Mind Key")

السيناريو 2: خطأ شبكة

# الكشف: ConnectionError، Timeout
# التعافي: أعد المحاولة مع تراجع، ثم تدهور

def handle_network_error(url, retry=3):
    for attempt in range(retry):
        try:
            return requests.get(url, timeout=10)
        except (requests.ConnectionError, requests.Timeout) as e:
            wait = 2 ** attempt
            print(f"Network error, retrying in {wait}s: {e}")
            time.sleep(wait)
    
    # فشلت جميع المحاولات — تدهور
    store_error("network_failure", 
                f"Cannot reach {url}",
                "Check internet connection, Synapse may be down")
    
    # اعمل دون اتصال إن أمكن
    return work_offline()

السيناريو 3: محدود بحد المعدل

# الكشف: 429 مع ترويسة Retry-After
# التعافي: انتظر وأعد المحاولة، أو بدّل إلى مصادقة الترويسة

def handle_rate_limit(response):
    retry_after = int(response.headers.get("Retry-After", 60))
    print(f"Rate limited. Waiting {retry_after}s...")
    time.sleep(retry_after)
    
    # إذا استمر هذا، اقترح التبديل إلى مصادقة الترويسة
    if has_query_param_auth(url):
        store_error("rate_limited", 
                    "Frequent 429s with ?key= auth",
                    "Switch to Authorization: Bearer header (no rate limit)")

السيناريو 4: خطأ خادم (5xx)

# الكشف: 500، 502، 503
# التعافي: أعد المحاولة مع تراجع، تحقق من /health

def handle_server_error(url):
    # تحقق إذا كان الخادم يعمل
    health = requests.get(f"{URL}/health")
    if health.status_code != 200:
        store_error("server_down", 
                    "Synapse health check failing",
                    "Wait for server recovery")
        raise ServerDownError()
    
    # أعد المحاولة مع تراجع
    return call_with_retry(url, max_retries=5)

السيناريو 5: فشل استدعاء أداة

# الكشف: الأداة تُرجع محتوى خطأ
# التعافي: جرب نهجًا بديلًا، خزّن الفشل

def call_tool_safely(tool_name, args, alternatives=None):
    try:
        result = call_tool(tool_name, args)
        if result.get("isError"):
            raise ToolError(result["content"])
        return result
    except ToolError as e:
        store_error(f"tool_{tool_name}_failed",
                    f"Args: {args}\nError: {e}",
                    f"Try: {alternatives or 'no alternatives'}")
        
        # جرب البدائل
        if alternatives:
            for alt in alternatives:
                try:
                    return call_tool(alt, args)
                except:
                    continue
        
        raise

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

للإخفاقات المتكررة، توقف عن المحاولة مؤقتًا:

class CircuitBreaker:
    def __init__(self, threshold=5, reset_time=300):
        self.failures = 0
        self.threshold = threshold
        self.reset_time = reset_time
        self.last_failure = 0
    
    def call(self, fn, *args, **kwargs):
        if self.failures >= self.threshold:
            if time.time() - self.last_failure < self.reset_time:
                raise CircuitOpenError("Circuit breaker open")
            else:
                # إعادة تعيين
                self.failures = 0
        
        try:
            result = fn(*args, **kwargs)
            self.failures = 0  # إعادة تعيين عند النجاح
            return result
        except:
            self.failures += 1
            self.last_failure = time.time()
            raise

# الاستخدام
breaker = CircuitBreaker(threshold=5)
try:
    result = breaker.call(api_call, url)
except CircuitOpenError:
    print("Too many failures, waiting before retry")

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

الخطوات التالية

• مرجع الأخطاء API
• نمط بدء الجلسة
• اختبارات ذاتية الإصلاح