الأخطاء تحدث. الشبكات تفشل، APIs تُرجع 500، Mind Keys تنتهي. يُظهر هذا الدليل\nكيف ينبغي لوكلاء LLM التعامل مع الأخطاء بلطف والتعلّم منها.
\nimport time\nimport requests\n\ndef call_with_retry(url, max_retries=3, backoff_base=2):\n """Call URL with exponential backoff."""\n for attempt in range(max_retries):\n try:\n r = requests.get(url, headers={"Authorization": f"Bearer {KEY}"})\n \n # نجاح\n if r.status_code < 400:\n return r\n \n # حد المعدل — انتظر وأعد المحاولة\n if r.status_code == 429:\n wait = int(r.headers.get("Retry-After", 60))\n print(f"Rate limited. Waiting {wait}s...")\n time.sleep(wait)\n continue\n \n # خطأ خادم — أعد المحاولة مع تراجع\n if r.status_code >= 500:\n wait = backoff_base ** attempt\n print(f"Server error {r.status_code}. Retrying in {wait}s...")\n time.sleep(wait)\n continue\n \n # خطأ عميل — لا تعد المحاولة\n if 400 <= r.status_code < 500:\n raise ClientError(f"{r.status_code}: {r.text}")\n \n except requests.RequestException as e:\n # خطأ شبكة — أعد المحاولة\n wait = backoff_base ** attempt\n print(f"Network error: {e}. Retrying in {wait}s...")\n time.sleep(wait)\n \n raise MaxRetriesError(f"Failed after {max_retries} retries")def safe_call(url):\n """Call with auth error handling."""\n try:\n return call_with_retry(url)\n except ClientError as e:\n if "401" in str(e):\n # Mind Key غير صالح — حرج، لا يمكن التعافي\n store_error("auth_invalid", str(e), "Check Mind Key")\n notify_human("My Mind Key is invalid. Please update.")\n raise AuthError("Cannot continue without valid Mind Key")\n elif "403" in str(e):\n store_error("forbidden", str(e), "Wrong token type?")\n raise ForbiddenError(str(e))\n elif "404" in str(e):\n # المسار غير موجود — لا تعد المحاولة\n store_error("not_found", str(e), "Check endpoint path")\n raise NotFoundError(str(e))\n else:\n raiseعند وقوع أخطاء، خزّنها حتى تتعلّم الجلسات المستقبلية:
\ndef store_error(error_type, error_message, recovery_hint=""):\n """Store an error as a memory for future reference."""\n requests.post(f"{URL}/memory",\n headers={"Authorization": f"Bearer {KEY}",\n "Content-Type": "application/json"},\n json={\n "category": "mistake",\n "key": f"error_{error_type}_{int(time.time())}",\n "content": f"Error: {error_type}\\nMessage: {error_message}\\nRecovery: {recovery_hint}",\n "tags": ["error", error_type],\n "priority": "high"\n })\n\n# مثال\ntry:\n deploy()\nexcept DeployError as e:\n store_error("deploy_failed", str(e), \n "Check CI logs, verify Docker image exists")\n raise# الكشف: 401 على كل استدعاء\n# التعافي: لا يمكن التعافي — يحتاج تدخل إنسان\n\ndef handle_invalid_mind_key():\n store_error("mind_key_invalid", \n "All API calls returning 401",\n "Mind Key may be revoked. Need new key.")\n \n # أبلغ الإنسان عبر الدردشة (إن أمكن)\n try:\n reply("⚠️ My Mind Key is invalid. I cannot access memories. "\n "Please check and update SYNAPSE_MIND_KEY.")\n except:\n pass # قد تفشل الدردشة أيضًا بمفتاح سيء\n \n # اخرج بلطف\n raise CriticalError("Cannot continue without valid Mind Key")# الكشف: ConnectionError، Timeout\n# التعافي: أعد المحاولة مع تراجع، ثم تدهور\n\ndef handle_network_error(url, retry=3):\n for attempt in range(retry):\n try:\n return requests.get(url, timeout=10)\n except (requests.ConnectionError, requests.Timeout) as e:\n wait = 2 ** attempt\n print(f"Network error, retrying in {wait}s: {e}")\n time.sleep(wait)\n \n # فشلت جميع المحاولات — تدهور\n store_error("network_failure", \n f"Cannot reach {url}",\n "Check internet connection, Synapse may be down")\n \n # اعمل دون اتصال إن أمكن\n return work_offline()# الكشف: 429 مع ترويسة Retry-After\n# التعافي: انتظر وأعد المحاولة، أو بدّل إلى مصادقة الترويسة\n\ndef handle_rate_limit(response):\n retry_after = int(response.headers.get("Retry-After", 60))\n print(f"Rate limited. Waiting {retry_after}s...")\n time.sleep(retry_after)\n \n # إذا استمر هذا، اقترح التبديل إلى مصادقة الترويسة\n if has_query_param_auth(url):\n store_error("rate_limited", \n "Frequent 429s with ?key= auth",\n "Switch to Authorization: Bearer header (no rate limit)")# الكشف: 500، 502، 503\n# التعافي: أعد المحاولة مع تراجع، تحقق من /health\n\ndef handle_server_error(url):\n # تحقق إذا كان الخادم يعمل\n health = requests.get(f"{URL}/health")\n if health.status_code != 200:\n store_error("server_down", \n "Synapse health check failing",\n "Wait for server recovery")\n raise ServerDownError()\n \n # أعد المحاولة مع تراجع\n return call_with_retry(url, max_retries=5)# الكشف: الأداة تُرجع محتوى خطأ\n# التعافي: جرب نهجًا بديلًا، خزّن الفشل\n\ndef call_tool_safely(tool_name, args, alternatives=None):\n try:\n result = call_tool(tool_name, args)\n if result.get("isError"):\n raise ToolError(result["content"])\n return result\n except ToolError as e:\n store_error(f"tool_{tool_name}_failed",\n f"Args: {args}\\nError: {e}",\n f"Try: {alternatives or 'no alternatives'}")\n \n # جرب البدائل\n if alternatives:\n for alt in alternatives:\n try:\n return call_tool(alt, args)\n except:\n continue\n \n raiseللإخفاقات المتكررة، توقف عن المحاولة مؤقتًا:
\nclass CircuitBreaker:\n def __init__(self, threshold=5, reset_time=300):\n self.failures = 0\n self.threshold = threshold\n self.reset_time = reset_time\n self.last_failure = 0\n \n def call(self, fn, *args, **kwargs):\n if self.failures >= self.threshold:\n if time.time() - self.last_failure < self.reset_time:\n raise CircuitOpenError("Circuit breaker open")\n else:\n # إعادة تعيين\n self.failures = 0\n \n try:\n result = fn(*args, **kwargs)\n self.failures = 0 # إعادة تعيين عند النجاح\n return result\n except:\n self.failures += 1\n self.last_failure = time.time()\n raise\n\n# الاستخدام\nbreaker = CircuitBreaker(threshold=5)\ntry:\n result = breaker.call(api_call, url)\nexcept CircuitOpenError:\n print("Too many failures, waiting before retry")