Monitoring et logging

Leçon corrigée le 3 novembre 2026 d’après le guide Computer use d’OpenAI et la documentation de la Responses API : la journalisation lit désormais les actions d’un computer_call et les éléments de response.output (messages avec leur phase, résumés de raisonnement, appels), au lieu des blocs tool_use de l’API d’Anthropic ; les tarifs de Claude Sonnet codés en dur sont remplacés par ceux du modèle utilisé, et datetime.utcnow(), déprécié depuis Python 3.12, par datetime.now(timezone.utc). Sources : Computer use · Tarification.

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é. Les actions arrivent sous la forme des objets du computer_call, avec les mêmes champs que dans votre gestionnaire (x, y, scroll_x, keys…). 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

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

handler = logging.FileHandler("agent.log")
handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
handler.addFilter(SensitiveFilter())  # le filtre de la leçon 12
logger.addHandler(handler)


def log_action(action, iteration: int, step: str = ""):
    """Journalise une action d'un computer_call."""
    prefixe = f"[iter={iteration}] [{step}]"
    match action.type:
        case "click" | "double_click" | "move":
            logger.info("%s %s at (%s, %s)", prefixe, action.type.upper(), action.x, action.y)
        case "type":
            # Jamais le texte lui-même : seulement sa longueur
            logger.info("%s TYPE (%d chars)", prefixe, len(action.text))
        case "scroll":
            logger.info("%s SCROLL dx=%s dy=%s", prefixe, action.scroll_x, action.scroll_y)
        case "keypress":
            logger.info("%s KEYPRESS %s", prefixe, "+".join(action.keys))
        case "drag":
            logger.info("%s DRAG (%d points)", prefixe, len(action.path))
        case _:
            logger.info("%s %s", prefixe, action.type.upper())

Niveau 2 : Décisions du modèle

Loggez ce que le modèle dit de ses choix 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. Deux sources de texte existent dans une réponse de la Responses API. Les messages intermédiaires que le modèle écrit avant d’agir portent phase: "commentary", la réponse finale phase: "final_answer". Et si le modèle utilisé prend en charge les résumés de raisonnement, vous les obtenez en les demandant (reasoning={"summary": "auto"}) : ils arrivent dans des éléments de type reasoning.

La consommation de tokens est journalisée à la même cadence, et elle mérite d’être suivie de près : même avec previous_response_id, tous les tokens d’entrée des réponses précédentes de la chaîne sont refacturés en entrée, captures comprises. Le coût d’un tour augmente donc à mesure que la session s’allonge.

def log_model_response(response, iteration: int):
    """Journalise les notes, résumés et appels d'une réponse (sans les images)."""
    for item in response.output:
        if item.type == "reasoning":
            for resume in item.summary:
                logger.info("[iter=%d] REASONING: %s", iteration, resume.text[:200])
        elif item.type == "message":
            texte = "".join(c.text for c in item.content if c.type == "output_text")
            logger.info("[iter=%d] MESSAGE (%s): %s", iteration, item.phase or "-", texte[:200])
        elif item.type == "computer_call":
            actions = ", ".join(a.type for a in item.actions or [])
            logger.info("[iter=%d] COMPUTER_CALL %s: %s", iteration, item.call_id, actions)
        elif item.type == "function_call":
            logger.info("[iter=%d] FUNCTION_CALL %s", iteration, item.name)

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

Niveau 3 : Screenshots

Sauvegardez les screenshots pour le debug post-mortem, mais attention à l’espace disque : une capture PNG pèse souvent plusieurs centaines de kilo-octets, et un long workflow en produit des 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.

from datetime import datetime, timezone
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.now(timezone.utc).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

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 du modèle utilisé, que vous reprenez de la page de tarification d’OpenAI plutôt que de les coder en dur. La méthode summary produit le dictionnaire compact que vous archiverez à chaque exécution.

from dataclasses import dataclass, field
from datetime import datetime, timezone

def maintenant() -> datetime:
    return datetime.now(timezone.utc)

@dataclass
class WorkflowMetrics:
    """Collecte les métriques d'un workflow."""
    workflow_id: str
    prix_entree_par_million: float  # tarifs du modèle utilisé
    prix_sortie_par_million: float
    start_time: datetime = field(default_factory=maintenant)
    end_time: datetime | None = 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": maintenant().isoformat(),
        })

    def finalize(self):
        self.end_time = maintenant()

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

    @property
    def estimated_cost_usd(self) -> float:
        # Estimation : le cache de prompts et un environnement hébergé ne sont pas comptés
        input_cost = self.total_input_tokens / 1_000_000 * self.prix_entree_par_million
        output_cost = self.total_output_tokens / 1_000_000 * self.prix_sortie_par_million
        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 requests  # pip install requests

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

    try:
        requests.post(
            endpoint,
            json=payload,  # requests pose lui-même l'en-tête 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.

Points clés à retenir

• Loggez les actions, les messages et résumés du modèle, et les captures sur trois niveaux séparés
• Surveillez le taux de succès, la durée, le coût en tokens et les erreurs ; l’historique chaîné est refacturé en entrée à chaque tour
• 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)