Récupération d'erreurs pour les agents

Les erreurs arrivent. Les réseaux tombent, les API renvoient des 500, les Mind Keys expirent. Ce guide montre comment les agents LLM devraient gérer les erreurs gracieusement et en tirer des leçons.

Principes de gestion des erreurs

1. Réessayer avec backoff — les erreurs transitoires se résolvent souvent
2. Stocker l'erreur — apprendre des schémas
3. Dégrader gracieusement — ne pas crasher toute la session
4. Notifier l'humain — pour les erreurs que vous ne pouvez pas résoudre

Gestion des erreurs HTTP

Réessayer avec backoff exponentiel

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}"})
            
            # Succès
            if r.status_code < 400:
                return r
            
            # Limité en débit — attendre et réessayer
            if r.status_code == 429:
                wait = int(r.headers.get("Retry-After", 60))
                print(f"Rate limited. Waiting {wait}s...")
                time.sleep(wait)
                continue
            
            # Erreur serveur — réessayer avec backoff
            if r.status_code >= 500:
                wait = backoff_base ** attempt
                print(f"Server error {r.status_code}. Retrying in {wait}s...")
                time.sleep(wait)
                continue
            
            # Erreur client — ne pas réessayer
            if 400 <= r.status_code < 500:
                raise ClientError(f"{r.status_code}: {r.text}")
        
        except requests.RequestException as e:
            # Erreur réseau — réessayer
            wait = backoff_base ** attempt
            print(f"Network error: {e}. Retrying in {wait}s...")
            time.sleep(wait)
    
    raise MaxRetriesError(f"Failed after {max_retries} retries")

Gestion des erreurs d'auth

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 invalide — critique, impossible à récupérer
            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):
            # Le chemin n'existe pas — ne pas réessayer
            store_error("not_found", str(e), "Check endpoint path")
            raise NotFoundError(str(e))
        else:
            raise

Stocker les erreurs en mémoire

Quand des erreurs se produisent, stockez-les pour que les futures sessions puissent apprendre :

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"
        })

# Exemple
try:
    deploy()
except DeployError as e:
    store_error("deploy_failed", str(e), 
                "Check CI logs, verify Docker image exists")
    raise

Scénarios d'erreur courants

Scénario 1 : Mind Key invalide

# Détection : 401 sur chaque appel
# Récupération : Impossible — nécessite une intervention humaine

def handle_invalid_mind_key():
    store_error("mind_key_invalid", 
                "All API calls returning 401",
                "Mind Key may be revoked. Need new key.")
    
    # Notifier l'humain via chat (si possible)
    try:
        reply("⚠️ My Mind Key is invalid. I cannot access memories. "
              "Please check and update SYNAPSE_MIND_KEY.")
    except:
        pass  # Le chat peut aussi échouer avec une mauvaise clé
    
    # Sortir gracieusement
    raise CriticalError("Cannot continue without valid Mind Key")

Scénario 2 : erreur réseau

# Détection : ConnectionError, Timeout
# Récupération : Réessayer avec backoff, puis dégrader

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)
    
    # Tous les essais échoués — dégrader
    store_error("network_failure", 
                f"Cannot reach {url}",
                "Check internet connection, Synapse may be down")
    
    # Travailler hors ligne si possible
    return work_offline()

Scénario 3 : limité en débit

# Détection : 429 avec en-tête Retry-After
# Récupération : Attendre et réessayer, ou passer à l'auth par en-tête

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)
    
    # Si cela continue, suggérer de passer à l'auth par en-tête
    if has_query_param_auth(url):
        store_error("rate_limited", 
                    "Frequent 429s with ?key= auth",
                    "Switch to Authorization: Bearer header (no rate limit)")

Scénario 4 : erreur serveur (5xx)

# Détection : 500, 502, 503
# Récupération : Réessayer avec backoff, vérifier /health

def handle_server_error(url):
    # Vérifier si le serveur fonctionne
    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()
    
    # Réessayer avec backoff
    return call_with_retry(url, max_retries=5)

Scénario 5 : échec d'appel d'outil

# Détection : L'outil renvoie un contenu d'erreur
# Récupération : Essayer une approche alternative, stocker l'échec

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'}")
        
        # Essayer des alternatives
        if alternatives:
            for alt in alternatives:
                try:
                    return call_tool(alt, args)
                except:
                    continue
        
        raise

Schéma : disjoncteur

Pour les échecs répétés, arrêter d'essayer temporairement :

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:
                # Réinitialiser
                self.failures = 0
        
        try:
            result = fn(*args, **kwargs)
            self.failures = 0  # Réinitialiser sur succès
            return result
        except:
            self.failures += 1
            self.last_failure = time.time()
            raise

# Utilisation
breaker = CircuitBreaker(threshold=5)
try:
    result = breaker.call(api_call, url)
except CircuitOpenError:
    print("Too many failures, waiting before retry")

Bonnes pratiques

Prochaines étapes

• Référence API Erreurs
• Schéma de début de session
• Tests auto-réparables