Gestion d'erreurs et recovery

Les erreurs sont inévitables

En automatisation visuelle, les erreurs sont la norme, pas l’exception. Une page qui charge lentement, un bouton qui a changé de place, une popup inattendue, un captcha — votre agent doit savoir gérer tout cela gracieusement.

Types d’erreurs

Retry avec backoff exponentiel

Pour les erreurs réseau et API, implémentez un retry intelligent :

import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1.0):
    """Exécute une fonction avec retry et backoff exponentiel."""
    for attempt in range(max_retries + 1):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries:
                raise

            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"  Erreur (tentative {attempt + 1}) : {e}")
            print(f"  Retry dans {delay:.1f}s...")
            time.sleep(delay)

Appliquez-le aux appels API :

def call_model_with_retry(client, messages, tools):
    """Appel API avec retry automatique."""
    def _call():
        return client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

    return retry_with_backoff(_call, max_retries=3, base_delay=2.0)

Détection de boucle infinie

Un agent peut tourner en rond s’il ne comprend pas ce qu’il voit. Détectez cela :

import hashlib

class LoopDetector:
    """Détecte quand l'agent tourne en boucle."""

    def __init__(self, max_repeats: int = 3):
        self.screenshot_hashes = []
        self.max_repeats = max_repeats

    def check(self, screenshot_bytes: bytes) -> bool:
        """Retourne True si une boucle est détectée."""
        h = hashlib.md5(screenshot_bytes).hexdigest()
        self.screenshot_hashes.append(h)

        # Vérifier les N derniers screenshots
        recent = self.screenshot_hashes[-self.max_repeats:]
        if len(recent) == self.max_repeats and len(set(recent)) == 1:
            return True  # Boucle détectée

        return False

# Utilisation dans la boucle
detector = LoopDetector(max_repeats=3)

for i in range(MAX_ITERATIONS):
    screenshot = page.screenshot()

    if detector.check(screenshot):
        print("Boucle détectée ! Arrêt de l'agent.")
        break

    # ... suite de la boucle

Gestion des popups inattendues

Les popups (consentement, newsletter, chat, notifications) peuvent bloquer l’agent. Ajoutez des instructions dans le system prompt :

system_prompt = """
Tu es un agent d'automatisation web.

GESTION DES OBSTACLES :
- Si une popup ou une bannière bloque l'écran, ferme-la d'abord
  (bouton X, "Fermer", "Non merci", "Dismiss")
- Si un captcha apparaît, ARRÊTE-TOI et signale-le
- Si une erreur 404/500 s'affiche, retourne à la page précédente
- Si la page demande une mise à jour ou une confirmation,
  refuse et continue ta tâche principale
"""

Système de checkpoints

Pour les workflows longs, sauvegardez l’état régulièrement :

import json
from datetime import datetime

class CheckpointManager:
    """Sauvegarde et restaure l'état du workflow."""

    def __init__(self, filepath: str = "checkpoint.json"):
        self.filepath = filepath

    def save(self, step_index: int, context: dict):
        """Sauvegarde un checkpoint."""
        state = {
            "step_index": step_index,
            "context": context,
            "timestamp": datetime.utcnow().isoformat(),
        }
        with open(self.filepath, "w") as f:
            json.dump(state, f, indent=2)
        print(f"  Checkpoint sauvegardé (étape {step_index})")

    def load(self) -> dict:
        """Charge le dernier checkpoint."""
        try:
            with open(self.filepath, "r") as f:
                return json.load(f)
        except FileNotFoundError:
            return None

    def clear(self):
        """Supprime le checkpoint (workflow terminé)."""
        import os
        if os.path.exists(self.filepath):
            os.remove(self.filepath)

Intégrez-le dans votre orchestrateur :

def resilient_workflow(workflow, page):
    """Workflow avec checkpoints et recovery."""
    ckpt = CheckpointManager()
    saved = ckpt.load()

    start_index = 0
    if saved:
        start_index = saved["step_index"]
        print(f"Reprise depuis l'étape {start_index}")

    for i, step in enumerate(workflow[start_index:], start=start_index):
        try:
            run_agent(step["task"], page=page)
            ckpt.save(i + 1, step)
        except Exception as e:
            print(f"Erreur à l'étape {i} : {e}")
            ckpt.save(i, step)  # Sauvegarder pour reprendre
            raise

    ckpt.clear()  # Workflow terminé
    print("Workflow terminé avec succès")

Alertes et escalade

Quand l’agent ne peut pas résoudre un problème seul, il doit alerter un humain :

def alert_human(reason: str, screenshot_path: str = None):
    """Envoie une alerte quand l'agent est bloqué."""
    import smtplib
    from email.message import EmailMessage

    msg = EmailMessage()
    msg["Subject"] = f"Agent bloqué : {reason}"
    msg["From"] = "[email protected]"
    msg["To"] = "[email protected]"
    msg.set_content(f"L'agent Computer Use est bloqué.\nRaison : {reason}")

    if screenshot_path:
        with open(screenshot_path, "rb") as f:
            msg.add_attachment(
                f.read(), maintype="image", subtype="png",
                filename="screenshot.png"
            )

    with smtplib.SMTP("smtp.company.com", 587) as s:
        s.starttls()
        s.login("[email protected]", "password")
        s.send_message(msg)

Points clés à retenir

• Implémentez du retry avec backoff exponentiel pour les erreurs réseau et API
• Détectez les boucles infinies en comparant les hashes des screenshots successifs
• Prévoyez la gestion des popups et obstacles dans le system prompt
• Utilisez des checkpoints pour reprendre un workflow interrompu
• Mettez en place des alertes pour les situations que l’agent ne peut pas résoudre seul