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. Un script qui réussit neuf fois sur dix en démonstration échouera plusieurs fois par jour en production, et la différence entre un prototype et un système fiable tient entièrement à ce que vous prévoyez pour la dixième fois.

Types d’erreurs

Toutes les erreurs ne se traitent pas de la même façon, et le premier réflexe consiste à identifier à quelle famille appartient celle que vous observez. Un timeout réseau se rejoue tel quel ; un captcha, jamais. Le tableau suivant résume les cinq catégories que vous rencontrerez et la stratégie appropriée à chacune.

Retry avec backoff exponentiel

Pour les erreurs réseau et API, un simple retry immédiat aggrave souvent le problème : si vous êtes limité en débit, réessayer aussitôt vous fait rejeter à nouveau. Le backoff exponentiel double le délai à chaque tentative et y ajoute un bruit aléatoire, ce qui évite que plusieurs agents relancés au même instant ne se resynchronisent en vagues successives.

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)

L’appel au modèle est le premier candidat à cette protection. Vous encapsulez la requête dans une closure, puis vous la confiez au mécanisme de retry avec un délai de base plus généreux, car les erreurs 429 et 529 côté API mettent quelques secondes à se résorber.

def call_model_with_retry(client, messages, tools):
    """Appel API avec retry automatique."""
    def _call():
        return client.messages.create(
            model="claude-sonnet-5",
            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 : il clique sur un bouton inactif, reprend un screenshot identique, reclique, et consomme ainsi votre budget jusqu’à la limite d’itérations. Le symptôme est facile à mesurer, puisque l’écran ne change plus. En calculant l’empreinte de chaque capture et en comparant les trois dernières, vous détectez la situation en quelques secondes et vous arrêtez proprement plutôt que de payer cinquante tours à vide.

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 de consentement, d’inscription à la newsletter, de chat ou de notifications peuvent bloquer l’agent, qui cherche alors un bouton masqué par une surcouche et finit par cliquer au hasard. Plutôt que de traiter chaque cas dans votre code, donnez à l’agent une doctrine générale dans le system prompt. Elle couvre les quatre situations récurrentes : fermer les surcouches, s’arrêter net devant un captcha au lieu de tenter de le résoudre, revenir en arrière sur une page d’erreur, et décliner les demandes de mise à jour qui détournent de la tâche principale.

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

Sur un workflow long — deux cents factures à saisir, une heure d’exécution — une erreur à la centième ligne ne doit pas vous obliger à tout recommencer. Le gestionnaire de checkpoints sérialise après chaque étape l’indice atteint, le contexte courant et un horodatage. Il expose trois opérations symétriques : save écrit l’état, load le relit et renvoie None si aucun fichier n’existe, clear efface le fichier une fois le workflow mené à son terme, de sorte que la prochaine exécution reparte bien de zéro.

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égré à l’orchestrateur de la leçon précédente, le mécanisme devient transparent : au démarrage, on lit le checkpoint et on découpe la liste d’étapes à partir de l’indice sauvegardé. En cas d’exception, on enregistre l’étape fautive avant de relancer l’erreur, ce qui permet de corriger la cause puis de relancer le script sans perdre le travail déjà accompli.

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

Certains blocages ne se résolvent pas par la persévérance. Quand l’agent ne peut pas résoudre un problème seul, il doit alerter un humain, et lui donner de quoi comprendre immédiatement la situation. La fonction d’alerte envoie un message court indiquant la raison, et y joint la capture d’écran du moment : dans l’immense majorité des cas, l’opérateur diagnostique le problème en regardant l’image, sans avoir à rejouer le workflow.

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