Monitoring et logging

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.

Les trois niveaux de logging

Niveau 1 : Actions

Chaque action exécutée par l’agent doit être tracée :

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 = ""):
    """Logge 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 :

def log_model_response(response, iteration: int):
    """Logge 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 :

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

Implémentez un collecteur de métriques :

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 :

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

Configurez des alertes basées sur les seuils :

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

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)