Monitoring et logging — Computer Use et Automatisation

Observer votre agent en production

Un agent Computer Use en production est une boîte noire si vous ne mettez pas en place un monitoring adapté. Contrairement à une API classique où vous loggez les requêtes et réponses, ici vous devez aussi tracer les actions visuelles, les screenshots et les décisions du modèle. La différence est concrète : quand un workflow échoue à trois heures du matin, la question qui vous intéresse n’est pas « quelle exception a été levée », mais « qu’est-ce que l’agent avait sous les yeux, et pourquoi a-t-il cliqué là ». Un journal qui ne répond pas à cette question ne vous servira à rien le lendemain matin.

Les trois niveaux de logging

L’instrumentation se construit donc sur trois plans complémentaires : ce que l’agent a fait, ce qu’il a pensé avant de le faire, et ce qu’il voyait au moment de décider. Chacun s’écrit dans un canal différent parce qu’il n’a ni le même volume ni la même durée de vie utile.

Niveau 1 : Actions

Chaque action exécutée par l’agent doit être tracée. Le format retenu ci-dessous préfixe systématiquement le numéro d’itération et le nom de l’étape en cours, ce qui vous permet ensuite de filtrer un journal de plusieurs milliers de lignes sur la seule séquence qui a mal tourné. Notez le traitement réservé à l’action type : on enregistre la longueur du texte saisi, jamais son contenu. Un agent qui remplit un formulaire de connexion tape un mot de passe, et la règle de la leçon précédente s’applique ici sans exception.

import logging
from datetime import datetime

logger = logging.getLogger("computer_use")
logger.setLevel(logging.INFO)

handler = logging.FileHandler("agent.log")
formatter = logging.Formatter(
    "%(asctime)s | %(levelname)s | %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)


def log_action(action: dict, iteration: int, step: str = ""):
    """Loggé une action Computer Use."""
    action_type = action.get("action", "unknown")

    if action_type == "click":
        coord = action.get("coordinate", [0, 0])
        logger.info(
            f"[iter={iteration}] [{step}] "
            f"CLICK at ({coord[0]}, {coord[1]})"
        )
    elif action_type == "type":
        text = action.get("text", "")
        # Ne pas logger le texte complet (données sensibles)
        logger.info(
            f"[iter={iteration}] [{step}] "
            f"TYPE ({len(text)} chars)"
        )
    elif action_type == "scroll":
        direction = action.get("direction", "down")
        logger.info(
            f"[iter={iteration}] [{step}] SCROLL {direction}"
        )
    elif action_type == "key":
        key = action.get("key", "")
        logger.info(f"[iter={iteration}] [{step}] KEY {key}")

Niveau 2 : Décisions du modèle

Loggez le raisonnement du modèle pour comprendre pourquoi il a choisi telle action. Un clic à des coordonnées inattendues ne veut rien dire seul ; accompagné de la phrase qui l’a précédé — « le bouton d’envoi semble masqué par la bannière de cookies » — il devient un diagnostic. Le texte est tronqué à deux cents caractères, largement suffisant pour saisir l’intention sans faire exploser le volume du fichier. La consommation de tokens est journalisée à la même cadence : une itération qui coûte soudain trois fois plus que les précédentes signale presque toujours un écran chargé d’images ou un historique de conversation qui n’a pas été élagué.

def log_model_response(response, iteration: int):
    """Loggé la réponse du modèle (sans les images)."""
    for block in response.content:
        if hasattr(block, "text") and block.text:
            logger.info(
                f"[iter={iteration}] REASONING: "
                f"{block.text[:200]}..."
            )
        elif block.type == "tool_use":
            logger.info(
                f"[iter={iteration}] TOOL: {block.name} "
                f"action={block.input.get('action')}"
            )

    # Tokens consommés
    logger.info(
        f"[iter={iteration}] TOKENS: "
        f"in={response.usage.input_tokens} "
        f"out={response.usage.output_tokens}"
    )

Niveau 3 : Screenshots

Sauvegardez les screenshots pour le debug post-mortem, mais attention à l’espace disque : une capture pèse quelques centaines de kilo-octets et un workflow d’une heure en produit plusieurs centaines. La classe ci-dessous résout le problème par une rotation simple. Le nom de fichier commence par un horodatage, ce qui fait coïncider l’ordre alphabétique et l’ordre chronologique ; le tri de _cleanup suffit alors à identifier les plus anciennes captures et à les supprimer jusqu’à retomber sous le plafond de cent fichiers. Le champ label vous permet d’annoter les moments intéressants, par exemple erreur ou avant_validation, et de les retrouver d’un coup d’œil dans le dossier.

import os
from pathlib import Path

class ScreenshotLogger:
    """Sauvegarde les screenshots avec rotation."""

    def __init__(self, base_dir: str = "logs/screenshots", max_files: int = 100):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)
        self.max_files = max_files
        self.counter = 0

    def save(self, screenshot_bytes: bytes, iteration: int, label: str = ""):
        """Sauvegarde un screenshot avec un nom descriptif."""
        ts = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
        filename = f"{ts}_iter{iteration:03d}_{label}.png"
        filepath = self.base_dir / filename

        with open(filepath, "wb") as f:
            f.write(screenshot_bytes)

        self.counter += 1
        self._cleanup()

        return filepath

    def _cleanup(self):
        """Supprime les vieux screenshots."""
        files = sorted(self.base_dir.glob("*.png"))
        while len(files) > self.max_files:
            files[0].unlink()
            files.pop(0)

Métriques à surveiller

Taux

de succès par workflow

Durée

moyenne par étape

Coût

en tokens par exécution

Erreurs

par type et fréquence

Ces quatre indicateurs ne valent que comparés à eux-mêmes dans le temps : un workflow qui passe de quatre-vingt-quinze à quatre-vingts pour cent de succès vous apprend qu’une interface a bougé, bien avant que le premier utilisateur ne se plaigne. Implémentez donc un collecteur de métriques attaché à chaque exécution. La dataclasse WorkflowMetrics accumule les tokens itération après itération, garde la trace des erreurs avec leur étape et leur horodatage, et expose deux propriétés calculées : la durée, qui fonctionne même si le workflow n’est pas encore terminé, et une estimation de coût fondée sur les tarifs approximatifs de Claude Sonnet, soit trois dollars par million de tokens en entrée et quinze en sortie. La méthode summary produit le dictionnaire compact que vous archiverez à chaque exécution.

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class WorkflowMetrics:
    """Collecte les métriques d'un workflow."""
    workflow_id: str
    start_time: datetime = field(default_factory=datetime.utcnow)
    end_time: datetime = None
    iterations: int = 0
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    errors: list = field(default_factory=list)
    steps_completed: int = 0
    steps_total: int = 0

    def record_iteration(self, response):
        self.iterations += 1
        self.total_input_tokens += response.usage.input_tokens
        self.total_output_tokens += response.usage.output_tokens

    def record_error(self, error: str, step: str):
        self.errors.append({
            "step": step,
            "error": error,
            "timestamp": datetime.utcnow().isoformat(),
        })

    def finalize(self):
        self.end_time = datetime.utcnow()

    @property
    def duration_seconds(self) -> float:
        end = self.end_time or datetime.utcnow()
        return (end - self.start_time).total_seconds()

    @property
    def estimated_cost_usd(self) -> float:
        # Tarifs approximatifs Claude Sonnet
        input_cost = self.total_input_tokens / 1_000_000 * 3.0
        output_cost = self.total_output_tokens / 1_000_000 * 15.0
        return input_cost + output_cost

    def summary(self) -> dict:
        return {
            "workflow_id": self.workflow_id,
            "duration_s": round(self.duration_seconds, 1),
            "iterations": self.iterations,
            "tokens": self.total_input_tokens + self.total_output_tokens,
            "cost_usd": round(self.estimated_cost_usd, 4),
            "success": len(self.errors) == 0,
            "errors": len(self.errors),
            "steps": f"{self.steps_completed}/{self.steps_total}",
        }

Dashboard de monitoring

Pour un monitoring en temps réel, envoyez les métriques vers un service de monitoring plutôt que de les laisser dormir dans un fichier. La fonction ci-dessous poste le résumé horodaté sur un endpoint HTTP, avec deux précautions qui comptent autant que l’envoi lui-même : un délai maximal de cinq secondes, et une exception rattrapée qui se contente d’un avertissement. Un collecteur de métriques indisponible ne doit jamais faire tomber le workflow qu’il observe — l’inverse serait la définition d’une supervision contre-productive.

import json
import requests

def push_metrics(metrics: WorkflowMetrics, endpoint: str):
    """Envoie les métriques vers un service de monitoring."""
    payload = metrics.summary()
    payload["timestamp"] = datetime.utcnow().isoformat()

    try:
        requests.post(
            endpoint,
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=5,
        )
    except requests.RequestException as e:
        logger.warning(f"Erreur push métriques : {e}")

Alertes automatiques

Un tableau de bord que personne ne regarde n’est qu’une archive. Configurez donc des alertes basées sur les seuils, de manière que l’anomalie vienne à vous. Les quatre conditions retenues ici couvrent les dérives classiques : un coût supérieur à deux dollars, une durée qui dépasse trois minutes, plus de trente itérations, ou plus de deux erreurs sur la même exécution. Ces valeurs sont des points de départ, pas des vérités : calibrez-les sur les mesures de vos propres exécutions réussies, en plaçant le seuil nettement au-dessus de votre médiane observée. Une alerte qui se déclenche tous les jours sera ignorée dès la deuxième semaine.

def check_alerts(metrics: WorkflowMetrics):
    """Vérifie les seuils et déclenche des alertes."""
    alerts = []

    if metrics.estimated_cost_usd > 2.0:
        alerts.append(f"Coût élevé : ${metrics.estimated_cost_usd:.2f}")

    if metrics.duration_seconds > 180:
        alerts.append(f"Durée excessive : {metrics.duration_seconds:.0f}s")

    if metrics.iterations > 30:
        alerts.append(f"Trop d'itérations : {metrics.iterations}")

    if len(metrics.errors) > 2:
        alerts.append(f"Erreurs multiples : {len(metrics.errors)}")

    for alert in alerts:
        logger.warning(f"ALERTE : {alert}")
        # Envoyer notification (email, Slack, etc.)

    return alerts

Prenez l’habitude d’appeler check_alerts juste après finalize, puis push_metrics : vous obtenez à chaque exécution une trace archivée, un signal envoyé au tableau de bord, et une notification uniquement lorsque quelque chose mérite votre attention.

Tarifs relevés le 5 août 2026 — les prix évoluent régulièrement : avant tout calcul de budget, vérifiez la grille en vigueur sur la tarification officielle OpenAI.

Points clés à retenir

Loggez les actions, les décisions du modèle et les screenshots sur trois niveaux séparés
Surveillez le taux de succès, la durée, le coût en tokens et les erreurs
Implémentez une rotation des screenshots pour éviter de saturer le disque
Mettez en place des alertes automatiques sur les seuils critiques
Ne loggez jamais de données sensibles (mots de passe, tokens, données personnelles)