Journaux et observabilité

Une fois l'application en production, les utilisateurs s'en servent là où vous ne regardez pas. Un jour, quelqu'un signale « votre assistant a répondu une chose complètement fausse », ou la facture de fin de mois est trois fois plus élevée que prévu. Comment enquêter ?

Si vous n'avez que la question de l'utilisateur et la réponse finale, vous ne savez presque pas par où commencer : la recherche n'a-t-elle pas trouvé les bons documents ? Le modèle a-t-il mal lu ? Un outil a-t-il signalé une erreur ? Ou a-t-il dévié à une étape, pour ensuite s'égarer de plus en plus ?

L'observabilité (observability), c'est cela : le système laisse en fonctionnant assez de traces pour qu'on puisse reconstituer après coup ce qu'il a réellement fait. Pour une application d'IA, les traces les plus importantes sont chaque appel de modèle et chaque appel d'outil.

Ce qu'il faut enregistrer

Le call_llm de la leçon 4 du module 03 tenait déjà des comptes : une ligne JSON par appel, avec l'heure, le modèle, la durée, les tokens et le coût. C'est un bon début, mais insuffisant pour un agent. Pour répondre à une question, un agent appelle plusieurs fois le modèle et plusieurs outils, et ces enregistrements doivent pouvoir être reliés.

Chaque enregistrement a donc aussi besoin de :

• trace_id : tous les enregistrements d'une même question partagent un numéro. Grâce à lui, on retrouve tout le déroulé d'une question.
• span_id et parent_id : le numéro propre de chaque enregistrement et celui de son parent. Par exemple, « répondre à la question » est le parent, sous lequel se trouvent trois appels de modèle et trois appels d'outils.
• kind et name : le type d'enregistrement (tâche, appel de modèle, appel d'outil) et son nom précis (quel modèle, quel outil).
• Un résumé de l'entrée et de la sortie : les paramètres de l'outil, les quelques dizaines de premiers caractères du contenu renvoyé, les outils que le modèle a demandé d'appeler.
• Durée, tokens, coût, statut : succès, erreur, ou outil ayant renvoyé un message d'erreur.

Ce vocabulaire (trace, span) vient du traçage des systèmes distribués. Les outils open source existants, comme Langfuse (auto-hébergeable), et le standard OpenTelemetry utilisent les mêmes notions. Écrivons d'abord nous-mêmes le plus simple, pour comprendre ce qu'il fait.

Un traceur en quelques dizaines de lignes

import json
import threading
import time
import uuid
from contextlib import contextmanager
from pathlib import Path

_lock = threading.Lock()
_local = threading.local()  # 记录当前线程正在进行的 span，用来自动找到上级


class Tracer:
    def __init__(self, path):
        self.path = Path(path)

    def _write(self, record):
        with _lock, self.path.open("a") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")

    @contextmanager
    def span(self, kind, name, **attrs):
        """用法：with tracer.span("llm", "chat") as s: ...; s["tokens"] = 100"""
        parent = getattr(_local, "current", None)
        record = {
            "trace_id": parent["trace_id"] if parent else uuid.uuid4().hex[:12],
            "span_id": uuid.uuid4().hex[:8],
            "parent_id": parent["span_id"] if parent else None,
            "kind": kind,
            "name": name,
            "start": time.strftime("%Y-%m-%d %H:%M:%S"),
            **attrs,
        }
        _local.current = record
        start = time.time()
        try:
            yield record
            record.setdefault("status", "ok")
        except Exception as e:
            record["status"] = "error"
            record["error"] = f"{type(e).__name__}: {e}"
            raise
        finally:
            record["ms"] = round((time.time() - start) * 1000)
            _local.current = parent
            self._write(record)

On l'utilise avec une instruction with :

with tracer.span("tool", "grep_docs", args={"keyword": "timeout"}) as s:
    result = grep_docs("timeout")
    s["result_chars"] = len(result)

À la fin du bloc with, le traceur calcule automatiquement la durée et écrit une ligne JSON. Si une exception survient dans le bloc, elle est aussi enregistrée, avec le statut error, puis relancée normalement.

parent_id est trouvé automatiquement : le traceur mémorise avec threading.local() « quel span est en cours ». Un appel de modèle lancé à l'intérieur du span « répondre à la question » a naturellement « répondre à la question » pour parent. Chaque thread a ses propres enregistrements, si bien que plusieurs questions traitées en parallèle ne se mélangent pas.

Le brancher sur l'agent

Inutile de modifier le code de l'agent de la leçon 2 du module 05 ; il suffit d'ajouter une couche autour. Pour les appels de modèle :

class TracedModel(agent_loop.RealModel):
    """每次调用模型，记一个 llm 类型的 span。"""

    def __call__(self, messages, tools):
        with tracer.span("llm", self.model, input_messages=len(messages)) as s:
            message, usage = super().__call__(messages, tools)
            s.update(prompt_tokens=usage.prompt_tokens, completion_tokens=usage.completion_tokens,
                     cost=round(cost(usage), 6), tool_calls=[c.function.name for c in message.tool_calls or []])
            return message, usage

Pour les appels d'outils :

def traced(name, fn):
    """包装一个工具函数，每次调用记一个 tool 类型的 span。"""
    def wrapper(**kwargs):
        with tracer.span("tool", name, args=kwargs) as s:
            result = fn(**kwargs)
            s["result_chars"] = len(result)
            s["result_preview"] = result[:80]
            if result.startswith("错误"):
                s["status"] = "tool_error"
            return result
    return wrapper


for name, t in agent_loop.TOOLS.items():
    t["fn"] = traced(name, t["fn"])

Quand un outil renvoie « Erreur : … », la fonction elle-même ne lève pas d'exception (le module 05 l'a dit : les erreurs deviennent des observations transmises au modèle) ; il faut donc la marquer spécialement tool_error, sinon tout semble normal dans le journal.

Au niveau le plus extérieur, chaque question ouvre un span de type task :

    for q in QUESTIONS:
        with tracer.span("task", "answer_question", question=q) as s:
            answer, stats = agent_loop.run_agent(model, q, verbose=False)
            s["answer_preview"] = (answer or "")[:60]

Trois questions, puis seulement le journal

Trois questions, dont la dernière contient un nom de fichier volontairement erroné (le bon est timeouts.md). Une fois l'exécution terminée, le programme ne lit que le fichier journal pour analyser, comme lors d'une enquête après coup (code complet dans code/06-production/traced_agent.py) :

日志共 25 条记录，最后一条：
{"trace_id": "173ee49bde3e", "span_id": "f9e2c3a5", "parent_id": null, "kind": "task", "name": "answer_question", "start": "2026-09-14 23:07:41", "question": "advanced/timeout.md 里写了什么？", "answer_preview": "`advanced/timeouts.md` 的内容如下（文件共 71 行；注意文件名是复数 timeouts，没有 `", "status": "ok", "ms": 5729}

「httpx 的超时分成哪几种？」 4006 毫秒，3 次模型调用，3 次工具调用，0.00104 美元
「httpx 怎么上传文件？」 5352 毫秒，3 次模型调用，6 次工具调用，0.00210 美元
「advanced/timeout.md 里写了什么？」 5729 毫秒，4 次模型调用，3 次工具调用，0.00135 美元

最慢的一步：llm deepseek-flash，3120 毫秒
工具报错次数：无

Les trois questions totalisent 25 enregistrements. Regroupés par trace_id, la durée, le nombre d'appels et le coût de chaque question se lisent d'un coup d'œil. « Comment téléverser un fichier ? » a appelé 6 fois des outils, deux fois plus que les autres questions, pour un coût deux fois plus élevé ; ses enregistrements montrent qu'il a cherché successivement les trois mots-clés files=, multipart et upload, puis lu trois fichiers. L'étape la plus lente est un appel de modèle, un peu plus de 3 secondes ; les appels d'outils ne prennent que quelques millisecondes.

Je m'attendais à ce que la troisième question laisse une erreur d'outil « fichier inexistant », mais non : l'agent a d'abord appelé list_docs pour voir quels fichiers existaient, a trouvé que le bon nom était timeouts.md, l'a lu directement, et a même rappelé à l'utilisateur dans sa réponse que « le nom du fichier est au pluriel ». C'est aussi la valeur d'un journal : il enregistre ce qui s'est réellement passé, et non ce que vous imaginiez.

Enquêter avec le journal

Avec un tel journal, voici comment enquêter sur les problèmes courants :

Un fichier JSONL est simple et fiable, s'analyse en quelques lignes de Python, et convient aux projets personnels et aux produits débutants. Quand le volume grandit, on peut envoyer ces enregistrements à un outil dédié (comme Langfuse), qui fournit une interface pour parcourir chaque trace, chercher par critères et tracer des graphiques. Les notions sont les mêmes.

Attention à la vie privée

Le journal enregistre les questions des utilisateurs, les paramètres des outils, le début des réponses. Si un utilisateur écrit dans sa question son numéro de téléphone, un mot de passe ou des informations internes de son entreprise, tout cela entre dans le journal.

• N'enregistrer que le nécessaire. Le code ci-dessus n'enregistre que les 60 premiers caractères de la réponse et les 80 premiers du résultat d'outil, pas le contenu complet.
• Filtrer les informations sensibles. Avant d'écrire dans le journal, remplacer ce qui ressemble à une clé, un numéro de téléphone ou un numéro de pièce d'identité. La leçon d'après la prochaine, « garde-fous », écrit une fonction de détection simple.
• Fixer une durée de conservation. Ne gardez pas les journaux indéfiniment, par exemple seulement 30 jours.
• Contrôler les accès. Qui peut lire les journaux peut voir ce que les utilisateurs ont demandé.

Exercices

1. Lancez traced_agent.py, puis écrivez vous-même quelques lignes qui lisent traces.jsonl et trouvent, pour chaque question, l'appel de modèle avec le plus de tokens d'entrée.
2. Ajoutez à TracedModel un champ qui enregistre le nombre total de caractères d'entrée de chaque appel, et observez comment le contexte de l'agent grandit d'étape en étape.
3. Provoquez volontairement une erreur d'outil (par exemple en modifiant temporairement read_doc pour qu'il renvoie toujours « Erreur : … » pour un certain fichier), et vérifiez que les statistiques du journal la détectent.

Auto-test

1. À quoi servent respectivement trace_id, span_id et parent_id ?

trace_id relie tous les enregistrements d'une même tâche (par exemple la réponse à une question) ; span_id est le numéro propre de chaque enregistrement ; parent_id désigne l'enregistrement parent. Avec ces trois champs, on peut reconstituer à partir du journal la structure complète d'une tâche : ce qui a été fait d'abord, et ce qui a été appelé à l'intérieur.

2. Quand un outil renvoie « Erreur : ce fichier n'existe pas », la fonction ne lève pas d'exception. Comment le repérer dans le journal ?

Le code qui enveloppe l'outil doit vérifier la valeur renvoyée et, s'il s'agit d'un message d'erreur, marquer spécialement le statut, par exemple tool_error. Sinon, le statut de cet appel est normal dans le journal, et il manquera dans les statistiques d'erreurs.

3. Pourquoi ne pas enregistrer dans le journal les questions et réponses complètes des utilisateurs ?

Les saisies des utilisateurs peuvent contenir des informations personnelles, des mots de passe ou des secrets d'entreprise ; si le journal fuit ou tombe sous les yeux de personnes non autorisées, cela pose un problème de vie privée. N'enregistrez que ce qui est nécessaire pour enquêter, filtrez les informations sensibles avant l'écriture, et fixez une durée de conservation et des droits d'accès.