Fel inträffar. Nätverk går ner, API:er returnerar 500, Mind Keys löper ut.\nDen här guiden visar hur LLM-agenter ska hantera fel elegant och lära sig av dem.
\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 raiseNär fel inträffar, lagra dem så framtida sessioner kan lära:
\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 raiseFör upprepade fel, sluta försöka tillfälligt:
\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")