Tracer une application Gemini

Première version de l'assistant d'Arvea, première semaine : un conseiller signale des réponses coupées au milieu d'une phrase. Les journaux applicatifs montrent la question et la réponse, rien d'autre. Il manque ce qui explique la panne : combien de tokens le modèle a consommés pour réfléchir, pourquoi il s'est arrêté, avec quel prompt. Une trace Langfuse bien construite répond à ces trois questions sans relancer l'appel.

Le mécanisme : le modèle de données de Langfuse

Le SDK Python v4 est construit sur OpenTelemetry : chaque observation est un span OTel, exporté en OTLP/HTTP vers langfuse-web. Langfuse v4 écrit ensuite chaque observation une seule fois, dans une table où user_id et session_id sont recopiés sur chaque ligne. Conséquence pratique : ces attributs doivent être posés au moment où l'observation est créée, pas ajoutés après coup sur la trace.

flowchart TB
  T["Trace (trace_id)"] --> S["span repondre-conseiller<br/>input : la question"]
  S --> G["generation gemini<br/>model, input, output<br/>usage_details, prompt v1"]
  P["propagate_attributes<br/>user_id, session_id, tags"] -.->|"copiés sur chaque observation du bloc"| S
  P -.-> G
  M["Définition de prix<br/>gemini-3.5-flash-lite"] -.->|"le worker calcule le coût"| G

Trois façons de créer une génération coexistent :

Pour l'application directe, ce projet choisit la troisième : l'appel Gemini est à vous, autant en contrôler la trace.

La mise en œuvre

Le code complet est dans appli_gemini.py. L'essentiel :

@observe(name="repondre-conseiller", capture_input=False)
def repondre(langfuse, question, *, user_id, session_id, label=None, client=None) -> Reponse:
    langfuse.update_current_span(input={"question": question})
    prompt = charger_prompt(langfuse, label or SETTINGS.prompt_label)
    fiches = rechercher(question)
    ...
    with propagate_attributes(user_id=user_id, session_id=session_id,
                              tags=["appli-gemini"], trace_name="repondre-conseiller"):
        with langfuse.start_as_current_observation(
            as_type="generation", name="gemini", model=SETTINGS.model, prompt=prompt,
            input={"system": systeme, "user": contenu}, model_parameters=parametres,
        ) as generation:
            resultat = client_gemini().models.generate_content(model=SETTINGS.model, contents=contenu,
                config=types.GenerateContentConfig(system_instruction=systeme, **parametres))
            fin = resultat.candidates[0].finish_reason if resultat.candidates else None
            generation.update(
                output=resultat.text,
                usage_details=usage_gemini(resultat.usage_metadata),
                metadata={"model_version": resultat.model_version, "finish_reason": str(fin)},
            )
            if fin == types.FinishReason.MAX_TOKENS:
                generation.update(level="WARNING", status_message="réponse tronquée (MAX_TOKENS)")

Quatre décisions sont dans ces lignes.

1. Traduire l'usage sans double comptage. Gemini renvoie prompt_token_count (qui inclut les tokens servis depuis le cache de contexte), candidates_token_count (la réponse visible) et thoughts_token_count (la réflexion, facturée au prix de la sortie). Langfuse calcule un coût par clé d'usage :

def usage_gemini(u) -> dict[str, int]:
    if u is None:
        return {}
    cache = u.cached_content_token_count or 0
    usage = {"input": (u.prompt_token_count or 0) - cache, "output": u.candidates_token_count or 0,
             "output_reasoning_tokens": u.thoughts_token_count or 0, "input_cached_tokens": cache}
    return {k: v for k, v in usage.items() if v}

Ce découpage n'a de sens que si chaque clé a un prix : la leçon 6 déclare les quatre.

2. Le modèle demandé, pas le modèle servi. Pendant la rédaction, une requête à gemini-3.5-flash est revenue avec model_version = gemini-3.6-flash. Si vous enregistrez le modèle servi comme model, la définition de prix ne correspond plus et le coût tombe à zéro. Le modèle demandé va dans model, le modèle servi dans les métadonnées.

3. Signaler la troncature. finish_reason == MAX_TOKENS passe la génération au niveau WARNING : un filtre sur ce niveau dans l'interface liste toutes les réponses coupées.

4. Fermer proprement. Le SDK envoie les spans par lots en arrière-plan. Un script court qui se termine sans langfuse.flush() (ou shutdown()) perd ses dernières traces. __main__.py appelle shutdown() en fin de commande ; une application web le ferait à l'arrêt du serveur.

Résultat obtenu

make run du 2026-10-15 (extrait, application directe) :

[appli Gemini | prompt v1] Le conseiller ne doit jamais saisir un RIB dicté par téléphone. Le client doit
envoyer son nouveau RIB directement depuis son espace adhérent. La modification sera prise en compte sous 5 jours ouvrés.

La génération lue par l'API Observations v2 (/api/public/v2/observations?fields=core,basic,model,usage,prompt) :

gemini  gemini-3.5-flash-lite  usage {'input': 196, 'output': 47, 'total': 243}
        cost  {'input': 5.88e-05, 'output': 0.0001175, 'total': 0.0001763}

Capture réelle : la génération porte le modèle, le coût (0,000188 USD), le lien vers conseiller-reponse - v1, et l'entrée exacte envoyée au modèle.

Ce qui casse en production

La règle de décision

Instrumentez à la main l'appel de modèle que vous écrivez vous-même, avec le gestionnaire de contexte : vous contrôlez usage, modèle et prompt. Réservez @observe aux fonctions dont les arguments sont des données simples, et passez capture_input=False dès qu'un argument est un objet technique (client, connexion, session HTTP). Surveillez la proportion de générations en WARNING et la part des tokens de réflexion dans le coût.

Pratique

Exercice. Ajoutez à repondre une fonction de masquage qui remplace tout numéro de sécurité sociale français (13 chiffres, éventuellement espacés, plus une clé de 2 chiffres) par [NIR] avant envoi à Langfuse, sans modifier ce qui est envoyé à Gemini.

import re
from langfuse import Langfuse

NIR = re.compile(r"\b[12]\s?\d{2}\s?\d{2}\s?\d{2}\s?\d{3}\s?\d{3}(\s?\d{2})?\b")

def masquer(*, data, **_):
    if isinstance(data, str):
        return NIR.sub("[NIR]", data)
    if isinstance(data, dict):
        return {k: masquer(data=v) for k, v in data.items()}
    if isinstance(data, list):
        return [masquer(data=v) for v in data]
    return data

langfuse = Langfuse(mask=masquer)

• Je sais distinguer trace, observation, génération, session et utilisateur.
• Je sais enregistrer une génération Gemini avec un usage qui sépare réponse, réflexion et cache.
• Je sais empêcher @observe de capturer un objet sensible.
• Je sais repérer une réponse tronquée sans relire les conversations.