Observer un agent Python

L'agent construit dans cet article de parcours enchaîne des tours de décision et exécute désormais ses tools en parallèle via asyncio.gather(). Tant qu'il répond juste, ça n'a pas d'importance. Le jour où un utilisateur dit « ça a mis 8 secondes pour rien », ou pire, où la réponse est incohérente, il faut pouvoir rejouer exactement ce qui s'est passé tour par tour — quel tool a été appelé, avec quels arguments, combien de temps il a pris. C'est le traçage structuré : enregistrer chaque étape dans un format exploitable par une machine plutôt que dans un message de log en texte libre, le tout corrélé par un identifiant commun à toute la session.

Le pourquoi de ce besoin — et ce qu'il faut tracer dans l'absolu — est détaillé dans observabilité d'un agent. Cet article-ci s'occupe du comment, concrètement, sur la boucle Python de ce parcours : instrumenter run_agent et call_tool pour produire une trace exploitable, y compris quand plusieurs tools tournent en même temps.

Une ligne de log par évènement, pas par run

Le piège le plus courant est de logger une seule ligne à la fin de tout l'appel, avec un résumé. Ça suffit pour savoir qu'un appel a duré 3,2 secondes ; ça ne dit rien sur où ce temps est parti. La bonne granularité est l'évènement : un tour de modèle, un appel de tool — chacun avec sa propre durée, écrit dès qu'il se termine.

import json
import logging
import time
import uuid

logger = logging.getLogger("agent.trace")


def log_event(run_id: str, turn: int, event: str, **fields) -> None:
    logger.info(json.dumps({
        "ts": time.time(),
        "run_id": run_id,
        "turn": turn,
        "event": event,
        **fields,
    }, default=str))

default=str évite que json.dumps plante sur un objet non sérialisable glissé par erreur dans les fields (une exception, un Decimal…) — il le convertit en chaîne plutôt que de faire tomber tout le run pour un souci de logging.

Instrumenter les tools, même exécutés en parallèle

Le call_tool de l'article précédent capture déjà les exceptions pour ne pas casser asyncio.gather(). Chronométrer chaque appel s'ajoute au même endroit, avec time.perf_counter() avant et après :

async def call_tool(block, run_id: str, turn: int) -> str:
    handler = TOOLS_BY_NAME[block.name]
    start = time.perf_counter()
    status = "ok"
    try:
        result = await handler(**block.input)
        return result
    except Exception as exc:
        status = "error"
        result = f"erreur: {exc}"
        return result
    finally:
        log_event(
            run_id, turn, "tool_call",
            tool=block.name, args=block.input, status=status,
            duration_ms=round((time.perf_counter() - start) * 1000),
        )

Le détail qui compte : chaque coroutine mesure sa propre durée, indépendamment des autres. Que trois tools tournent en séquence ou ensemble via asyncio.gather() ne change rien à la mesure individuelle de chacun — seul le temps total du tour, lui, dépend de la concurrence. C'est justement ce qui permet de distinguer « ce tool a mis 2 secondes » de « le tour a mis 2 secondes » : sans cette mesure par coroutine, la seconde information écrase la première.

Instrumenter la boucle de tours

Le tour de modèle lui-même se chronomètre pareil, avec en prime les tokens consommés — disponibles sur response.usage côté SDK Anthropic :

async def run_agent(question: str, client, max_turns: int = 6) -> str:
    run_id = str(uuid.uuid4())
    messages: list[dict[str, Any]] = [{"role": "user", "content": question}]

    for turn in range(max_turns):
        start = time.perf_counter()
        response = await client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1024,
            tools=[t.to_api() for t in TOOLS],
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})
        log_event(
            run_id, turn, "model_turn",
            stop_reason=response.stop_reason,
            tokens_in=response.usage.input_tokens,
            tokens_out=response.usage.output_tokens,
            duration_ms=round((time.perf_counter() - start) * 1000),
        )

        if response.stop_reason != "tool_use":
            return "".join(b.text for b in response.content if b.type == "text")

        tool_blocks = [b for b in response.content if b.type == "tool_use"]
        outputs = await asyncio.gather(
            *(call_tool(b, run_id, turn) for b in tool_blocks)
        )

        tool_results = [
            {"type": "tool_result", "tool_use_id": block.id, "content": output}
            for block, output in zip(tool_blocks, outputs)
        ]
        messages.append({"role": "user", "content": tool_results})

    log_event(run_id, max_turns, "max_turns_reached")
    return "budget de tours dépassé sans réponse finale"

Rien dans la logique métier de la boucle n'a changé par rapport à la version précédente : run_id et les appels à log_event s'ajoutent autour, sans toucher au flux de décision.

Ce que chaque champ apporte

Ne pas logger n'importe quoi

args et le résultat d'un tool proviennent en partie de ce que l'utilisateur a demandé, et un résultat de tool peut être volumineux ou sensible. Deux réflexes évitent les mauvaises surprises : tronquer toute valeur au-delà d'une longueur raisonnable avant de la sérialiser, et traiter le contenu d'un tool externe (une réponse d'API tierce, un contenu de page web) comme non fiable plutôt que de le journaliser tel quel — la même prudence que celle décrite dans injection de prompt : défense s'applique aussi à ce qu'on écrit dans ses propres logs.

FAQ

Pourquoi chronométrer chaque tool séparément si asyncio.gather() les lance ensemble ?

Parce que le temps du tour et le temps de chaque tool sont deux informations différentes. asyncio.gather() fait que le tour dure environ le temps du plus lent des tools, mais sans mesure individuelle par coroutine, impossible de savoir lequel des trois a été le plus lent — la seule donnée globale ne permet pas de distinguer un tool systématiquement lent d'un tool ponctuellement en échec réseau.

Faut-il utiliser une librairie de tracing (OpenTelemetry) plutôt que du logging stdlib ?

Pas forcément pour démarrer. Une ligne JSON par évènement, corrélée par run_id, se lit déjà sans outillage particulier et s'ingère telle quelle dans n'importe quel collecteur de logs. Passer à OpenTelemetry a du sens quand il faut corréler ces traces avec d'autres services, mais le format et les champs à capturer restent les mêmes.

Où envoyer ces logs en production ?

Sur la sortie standard, comme n'importe quel log applicatif — c'est au système qui fait tourner le processus (conteneur, service systemd) de les rediriger vers un collecteur. Écrire soi-même dans un fichier ou une base dédiée ajoute une gestion de rotation et de rétention qu'un agrégateur de logs standard fait déjà.