# Agent 错误恢复 错误总会发生。网络故障、API 返回 500、Mind Key 失效。本指南展示 LLM Agent 如何优雅地处理错误并从中学习。 ## 错误处理原则 1. **退避重试** — 瞬时错误常常会自行恢复 2. **存储错误** — 从模式中学习 3. **优雅降级** — 不要让整个会话崩溃 4. **通知人类** — 对于无法解决的错误 ## HTTP 错误处理 ### 指数退避重试 ```python import time import requests def call_with_retry(url, max_retries=3, backoff_base=2): """带指数退避的 URL 调用。""" 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") ``` ### 认证错误处理 ```python def safe_call(url): """带认证错误处理的调用。""" 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 ``` ## 把错误存为记忆 发生错误时，存储它们以便未来会话学习： ```python def store_error(error_type, error_message, recovery_hint=""): """把错误存为记忆供未来参考。""" 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 无效 ```python # 检测：每次调用都返回 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 # 用坏 key 聊天也会失败 # 优雅退出 raise CriticalError("Cannot continue without valid Mind Key") ``` ### 场景 2：网络错误 ```python # 检测：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：被限速 ```python # 检测：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） ```python # 检测：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：工具调用失败 ```python # 检测：工具返回错误内容 # 恢复：尝试备选方案，存储失败 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 ``` ## 模式：熔断器 对于反复出现的失败，暂时停止尝试： ```python 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") ``` ## 最佳实践 > [!TIP] > - **始终重试瞬时错误** — 网络会失败、服务器会卡顿 > - **不要重试客户端错误（4xx）** — 它们不会自行修复 > - **把错误存为记忆** — 从模式中学习 > - **关键错误通知人类** — 他们需要知道 > - **优雅降级** — 部分工作好过崩溃 > - **使用熔断器** — 不要疯狂调用失败的服务 ## 下一步 - [错误 API 参考](/docs/api/errors) - [会话启动模式](/docs/llm-cookbook/session-start-pattern) - [自愈测试](/docs/guides/self-healing-tests)