त्रुटियाँ होती रहती हैं। नेटवर्क विफल होते हैं, APIs 500 लौटाते हैं, Mind Keys समाप्त हो जाते हैं। यह गाइड दिखाती है कि 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 # Success\n if r.status_code < 400:\n return r\n \n # Rate limited — wait and retry\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 # Server error — retry with backoff\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 # Client error — don't retry\n if 400 <= r.status_code < 500:\n raise ClientError(f"{r.status_code}: {r.text}")\n \n except requests.RequestException as e:\n # Network error — retry\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 invalid — critical, can't recover\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 # Path doesn't exist — don't retry\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# Example\ntry:\n deploy()\nexcept DeployError as e:\n store_error("deploy_failed", str(e), \n "Check CI logs, verify Docker image exists")\n raise# Detection: 401 on every call\n# Recovery: Cannot recover — need human intervention\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 # Notify human via chat (if possible)\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 # Chat may also fail with bad key\n \n # Exit gracefully\n raise CriticalError("Cannot continue without valid Mind Key")# Detection: ConnectionError, Timeout\n# Recovery: Retry with backoff, then degrade\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 # All retries failed — degrade\n store_error("network_failure", \n f"Cannot reach {url}",\n "Check internet connection, Synapse may be down")\n \n # Work offline if possible\n return work_offline()# Detection: 429 with Retry-After header\n# Recovery: Wait and retry, or switch to header auth\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 # If this keeps happening, suggest switching to header auth\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)")# Detection: 500, 502, 503\n# Recovery: Retry with backoff, check /health\n\ndef handle_server_error(url):\n # Check if server is up\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 # Retry with backoff\n return call_with_retry(url, max_retries=5)# Detection: Tool returns error content\n# Recovery: Try alternative approach, store failure\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 # Try alternatives\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 # Reset\n self.failures = 0\n \n try:\n result = fn(*args, **kwargs)\n self.failures = 0 # Reset on success\n return result\n except:\n self.failures += 1\n self.last_failure = time.time()\n raise\n\n# Usage\nbreaker = CircuitBreaker(threshold=5)\ntry:\n result = breaker.call(api_call, url)\nexcept CircuitOpenError:\n print("Too many failures, waiting before retry")