# Odzyskiwanie po błędach dla agentów Błędy się zdarzają. Sieci zawodzą, API zwracają 500, Mind Key wygasają. Ten przewodnik pokazuje, jak agenty LLM powinny obsługiwać błędy z gracją i uczyć się na nich. ## Zasady obsługi błędów 1. **Ponowienie z wycofywaniem** — błędy przejściowe często ustępują 2. **Zapisanie błędu** — nauka z wzorców 3. **Skradzona degradacja** — nie zawieszać całej sesji 4. **Powiadomienie człowieka** — o błędach, których nie da się rozwiązać ## Obsługa błędów HTTP ### Ponowienie z wykładniczym wycofywaniem ```python 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}"}) # Sukces if r.status_code < 400: return r # Limit żądań — odczekać i ponowić if r.status_code == 429: wait = int(r.headers.get("Retry-After", 60)) print(f"Rate limited. Waiting {wait}s...") time.sleep(wait) continue # Błąd serwera — ponowienie z wycofywaniem if r.status_code >= 500: wait = backoff_base ** attempt print(f"Server error {r.status_code}. Retrying in {wait}s...") time.sleep(wait) continue # Błąd klienta — nie ponawiać if 400 <= r.status_code < 500: raise ClientError(f"{r.status_code}: {r.text}") except requests.RequestException as e: # Błąd sieci — ponowienie wait = backoff_base ** attempt print(f"Network error: {e}. Retrying in {wait}s...") time.sleep(wait) raise MaxRetriesError(f"Failed after {max_retries} retries") ``` ### Obsługa błędów autoryzacji ```python 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 nieprawidłowy — krytyczne, nie odzyskać 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): # Ścieżka nie istnieje — nie ponawiać store_error("not_found", str(e), "Check endpoint path") raise NotFoundError(str(e)) else: raise ``` ## Zapisywanie błędów jako wspomnień Gdy występują błędy, zapisywać je, aby przyszłe sesje mogły się uczyć: ```python 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" }) # Przykład try: deploy() except DeployError as e: store_error("deploy_failed", str(e), "Check CI logs, verify Docker image exists") raise ``` ## Typowe scenariusze błędów ### Scenariusz 1: nieprawidłowy Mind Key ```python # Wykrycie: 401 przy każdym wywołaniu # Odzyskiwanie: nie można odzyskać — potrzebna interwencja człowieka def handle_invalid_mind_key(): store_error("mind_key_invalid", "All API calls returning 401", "Mind Key may be revoked. Need new key.") # Powiadomienie człowieka przez czat (jeśli możliwe) try: reply("⚠️ My Mind Key is invalid. I cannot access memories. " "Please check and update SYNAPSE_MIND_KEY.") except: pass # Czat może też zawieść z błędnym kluczem # Grzeczne wyjście raise CriticalError("Cannot continue without valid Mind Key") ``` ### Scenariusz 2: błąd sieci ```python # Wykrycie: ConnectionError, Timeout # Odzyskiwanie: ponowienie z wycofywaniem, a następnie degradacja 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) # Wszystkie ponowienia zawiodły — degradacja store_error("network_failure", f"Cannot reach {url}", "Check internet connection, Synapse may be down") # Praca offline, jeśli możliwa return work_offline() ``` ### Scenariusz 3: limit żądań ```python # Wykrycie: 429 z nagłówkiem Retry-After # Odzyskiwanie: odczekać i ponowić lub przełączyć się na autoryzację nagłówkiem 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) # Jeśli to się powtarza, zasugerować przejście na autoryzację nagłówkiem if has_query_param_auth(url): store_error("rate_limited", "Frequent 429s with ?key= auth", "Switch to Authorization: Bearer header (no rate limit)") ``` ### Scenariusz 4: błąd serwera (5xx) ```python # Wykrycie: 500, 502, 503 # Odzyskiwanie: ponowienie z wycofywaniem, sprawdzenie /health def handle_server_error(url): # Sprawdzenie, czy serwer działa 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() # Ponowienie z wycofywaniem return call_with_retry(url, max_retries=5) ``` ### Scenariusz 5: wywołanie narzędzia nie udaje się ```python # Wykrycie: narzędzie zwraca treść błędu # Odzyskiwanie: spróbować alternatywnego podejścia, zapisać błąd 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'}") # Spróbowanie alternatyw if alternatives: for alt in alternatives: try: return call_tool(alt, args) except: continue raise ``` ## Wzorzec: wyłącznik Dla powtarzających się błędów tymczasowo przestać próbować: ```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: # Reset self.failures = 0 try: result = fn(*args, **kwargs) self.failures = 0 # Reset przy sukcesie return result except: self.failures += 1 self.last_failure = time.time() raise # Użycie breaker = CircuitBreaker(threshold=5) try: result = breaker.call(api_call, url) except CircuitOpenError: print("Too many failures, waiting before retry") ``` ## Najlepsze praktyki > [!TIP] > - **Zawsze ponawiać błędy przejściowe** — sieci zawodzą, serwery czkają > - **Nie ponawiać błędów klienta (4xx)** — same się nie naprawią > - **Zapisywać błędy jako wspomnienia** — uczyć się z wzorców > - **Powiadamiać ludzi o krytycznych błędach** — muszą wiedzieć > - **Degradować z gracją** — częściowa praca lepsza niż zawieszenie > - **Używać wyłączników** — nie bombardować usługi, która zawodzi ## Następne kroki - [Dokumentacja błędów API](/docs/api/errors) - [Wzorzec początku sesji](/docs/llm-cookbook/session-start-pattern) - [Testy samonaprawiające](/docs/guides/self-healing-tests)