5 mars 2026

Débogage et Observabilité dans les Systèmes d'Agents Autonomes

Un agent autonome qui échoue silencieusement est pire qu’aucun agent. Quand une fonction traditionnelle lève une exception, vous obtenez une trace de pile. Quand un agent prend un mauvais chemin sur vingt appels d’outils et trois invocations du modèle, vous obtenez une mauvaise réponse — sans explication évidente.

Déboguer des agents nécessite un modèle mental différent. Le système n’exécute pas un chemin déterministe ; il prend une série de décisions. L’observabilité signifie capturer ces décisions — pas seulement les entrées et sorties, mais le raisonnement qui les relie.

Pourquoi le Débogage Traditionnel Échoue pour les Agents

La journalisation standard capture ce qui s’est passé. L’observabilité des agents nécessite de capturer le pourquoi — ce que le modèle a conclu, quel outil il a choisi et pourquoi, et à partir de quel état intermédiaire il travaillait.

Les modes d’échec sont aussi différents :

Hallucination silencieuse : L’agent produit une mauvaise réponse avec confiance sans signaler d’incertitude.
Dérive décisionnelle : Chaque étape semble raisonnable localement, mais la séquence s’éloigne de l’objectif.
Mauvaise utilisation d’outils : L’agent appelle le bon outil avec des paramètres subtilement incorrects.
Boucles infinies : L’agent reste bloqué à réessayer une approche qui échoue.
Empoisonnement de contexte : Une mauvaise sortie à une étape précoce corrompt tout le raisonnement ultérieur.

Aucun de ces cas ne produit d’exception. Ils produisent un comportement incorrect qui n’est visible que lorsqu’on reconstruit la trace d’exécution complète.

Journalisation Structurée pour les Décisions d’Agents

La première étape consiste à envelopper chaque interaction de l’agent dans des logs structurés. Ne journalisez pas les réponses brutes de l’API — journalisez des événements sémantiques.

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any
import anthropic

client = anthropic.Anthropic()

@dataclass
class AgentEvent:
    trace_id: str
    step: int
    event_type: str      # "llm_call", "tool_call", "tool_result", "decision", "error"
    model: str | None
    input_tokens: int | None
    output_tokens: int | None
    latency_ms: float | None
    content: dict[str, Any]
    timestamp: float

def log_event(event: AgentEvent):
    print(json.dumps(asdict(event)))  # Remplacez par votre destination de logs

class TracedAgent:
    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.step = 0
        self.tools = []

    def add_tool(self, name: str, description: str, input_schema: dict):
        self.tools.append({
            "name": name,
            "description": description,
            "input_schema": input_schema
        })

    def call(self, messages: list[dict], system: str = "") -> str:
        self.step += 1
        start = time.monotonic()

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            system=system,
            tools=self.tools,
            messages=messages
        )

        latency_ms = (time.monotonic() - start) * 1000

        log_event(AgentEvent(
            trace_id=self.trace_id,
            step=self.step,
            event_type="llm_call",
            model="claude-opus-4-6",
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            latency_ms=latency_ms,
            content={
                "stop_reason": response.stop_reason,
                "text_blocks": [b.text for b in response.content if b.type == "text"],
                "tool_calls": [
                    {"name": b.name, "input": b.input}
                    for b in response.content if b.type == "tool_use"
                ]
            },
            timestamp=time.time()
        ))

        return response

Chaque appel LLM émet maintenant un événement structuré avec l’ID de trace, le numéro d’étape, les comptages de tokens, la latence et la décision du modèle.

Construire une Trace Complète

Une seule ligne de log ne suffit pas — vous avez besoin de la trace d’exécution complète qui relie chaque décision à son résultat :

from typing import Callable

def run_traced_agent(
    task: str,
    tools: dict[str, Callable],
    tool_schemas: list[dict],
    system: str,
    max_steps: int = 20,
) -> dict:
    agent = TracedAgent()
    for schema in tool_schemas:
        agent.add_tool(**schema)

    messages = [{"role": "user", "content": task}]
    trace = {"trace_id": agent.trace_id, "task": task, "steps": []}
    step_count = 0

    while step_count < max_steps:
        step_count += 1
        response = agent.call(messages, system=system)

        step_record = {
            "step": step_count,
            "stop_reason": response.stop_reason,
            "model_output": [],
            "tool_results": []
        }

        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    step_record["model_output"].append(block.text)
            trace["steps"].append(step_record)
            trace["final_answer"] = step_record["model_output"][-1] if step_record["model_output"] else ""
            break

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                step_record["model_output"].append({
                    "tool": block.name,
                    "input": block.input
                })

                tool_fn = tools.get(block.name)
                if not tool_fn:
                    result = f"Erreur : outil inconnu '{block.name}'"
                else:
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"Erreur d'outil : {e}"

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result)
                })
                step_record["tool_results"].append({
                    "tool": block.name,
                    "result_preview": str(result)[:200]
                })

        trace["steps"].append(step_record)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

    else:
        trace["error"] = f"nombre maximum d'étapes dépassé ({max_steps})"

    return trace

Détection de Boucles

Les boucles infinies sont un mode d’échec courant. Détectez-les en prenant l’empreinte du pattern d’invocation d’outils de chaque appel LLM :

def detect_loop(trace: dict, window: int = 4) -> bool:
    steps = trace["steps"]
    if len(steps) < window:
        return False

    def step_signature(step: dict) -> str:
        tools_called = sorted(
            t["tool"] if isinstance(t, dict) else t
            for t in step.get("model_output", [])
            if isinstance(t, dict) and "tool" in t
        )
        return "|".join(tools_called)

    recent = [step_signature(s) for s in steps[-window:]]
    if len(set(recent)) == 1 and recent[0]:
        return True

    if len(steps) >= 4:
        pattern = [step_signature(s) for s in steps[-4:]]
        if pattern[0] == pattern[2] and pattern[1] == pattern[3]:
            return True

    return False

Métriques à Suivre en Production

from collections import Counter

def compute_trace_metrics(trace: dict) -> dict:
    steps = trace["steps"]
    errors = [s for s in steps if "error" in s]

    tool_calls_by_name: Counter = Counter()
    for step in steps:
        for output in step.get("model_output", []):
            if isinstance(output, dict) and "tool" in output:
                tool_calls_by_name[output["tool"]] += 1

    return {
        "trace_id": trace["trace_id"],
        "total_steps": len(steps),
        "error_steps": len(errors),
        "tool_call_distribution": dict(tool_calls_by_name),
        "completed": "final_answer" in trace,
        "loop_detected": detect_loop(trace),
    }

Signaux clés pour les alertes :

Taux de boucles > 5% des exécutions — l’agent reste bloqué
Taux d’erreur par outil > seuil — un outil est cassé
Nombre moyen d’étapes en hausse — les tâches deviennent plus difficiles
Latence p99 en pic — un endpoint du modèle est lent

Intégration OpenTelemetry

Pour les équipes utilisant déjà OpenTelemetry, émettez les traces de l’agent comme des spans :

from opentelemetry import trace as otel_trace

tracer = otel_trace.get_tracer("agent")

def run_with_otel(task: str, tools: dict, tool_schemas: list, system: str):
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("agent.task", task[:200])

        agent = TracedAgent()
        for schema in tool_schemas:
            agent.add_tool(**schema)

        messages = [{"role": "user", "content": task}]

        for step in range(20):
            with tracer.start_as_current_span(f"agent.step.{step}") as step_span:
                response = agent.call(messages, system=system)
                step_span.set_attribute("llm.stop_reason", response.stop_reason)
                step_span.set_attribute("llm.input_tokens", response.usage.input_tokens)

                if response.stop_reason == "end_turn":
                    break

Rédaction des DCP dans les Logs

Les logs d’agents contiennent souvent des données sensibles. Avant de les émettre vers un système externe, rédigez-les :

import re

PII_PATTERNS = [
    (re.compile(r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b'), '[EMAIL]'),
    (re.compile(r'\b\d{2}[-.]?\d{2}[-.]?\d{2}[-.]?\d{2}[-.]?\d{2}\b'), '[TÉLÉPHONE]'),
    (re.compile(r'\bsk-[a-zA-Z0-9]{20,}\b'), '[CLÉ_API]'),
]

def redact(text: str) -> str:
    for pattern, replacement in PII_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

Les Trois Métriques les Plus Importantes

Taux de complétion des tâches — quelle fraction des exécutions atteint final_answer vs. max_steps ou une erreur. Établissez une baseline par type de tâche.

Coût en tokens par tâche — somme de input_tokens + output_tokens sur toutes les étapes. Une augmentation de 20% du coût sans changement du taux de complétion signale généralement une dégradation du prompt.

Taux d’erreur des outils — error_steps / total_steps. Les pics dans cette métrique pointent directement vers un outil ou une API cassée.

L’observabilité dans les systèmes d’agents n’est pas optionnelle — c’est la différence entre un système sur lequel vous pouvez itérer et un que vous pouvez seulement redémarrer quand il tombe en panne. Commencez avec des événements structurés et des IDs de trace. Ajoutez la détection de boucles. Publiez des métriques. L’investissement se rembourse la première fois que vous avez une panne en production et pouvez reconstruire exactement ce qui s’est passé.