Observabilité RAG : Le Dashboard qui Détecte les Problèmes Avant vos Utilisateurs

TL;DR

Le monitoring classique (uptime, latence HTTP) est insuffisant pour un RAG en production. Vous devez tracker la pertinence du retrieval, la qualité des réponses, le taux d'hallucinations et la satisfaction utilisateur. Ce guide compare les outils d'observabilité RAG (LangSmith, Langfuse, Phoenix/Arize, W&B), montre comment tracer chaque étape du pipeline et configurer des alertes qui détectent les problèmes avant vos utilisateurs.

Pourquoi le monitoring RAG est différent

Les métriques classiques ne suffisent pas

Un RAG peut retourner un status 200 avec une latence de 500ms et quand même donner une réponse catastrophique. Voici ce que le monitoring API classique ne détecte pas :

Résultat : le monitoring classique ne détecte que 2 problèmes sur 8. Les 6 autres nécessitent une observabilité spécifique au RAG.

Les 4 piliers de l'observabilité RAG

┌─────────────┐  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│  Retrieval   │  │  Génération  │  │  Latence     │  │  Utilisateur │
│  Quality     │  │  Quality     │  │  & Coûts     │  │  Satisfaction│
├─────────────┤  ├─────────────┤  ├─────────────┤  ├─────────────┤
│ Relevance    │  │ Faithfulness │  │ P50/P95/P99  │  │ Thumbs up/  │
│ Recall       │  │ Hallucination│  │ Token usage  │  │ down         │
│ MRR/NDCG     │  │ Completeness │  │ Cost/query   │  │ Reformulation│
│ Empty results│  │ Toxicity     │  │ Cache hit    │  │ Escalation   │
└─────────────┘  └─────────────┘  └─────────────┘  └─────────────┘

Les métriques essentielles à tracker

Métriques de retrieval

DEVELOPERpython
from dataclasses import dataclass
from typing import Optional

@dataclass
class RetrievalMetrics:
    """Métriques de qualité du retrieval."""

    # Nombre de documents récupérés
    num_docs_retrieved: int
    # Score de pertinence moyen des documents
    avg_relevance_score: float
    # Le meilleur document est-il pertinent ?
    top_doc_relevant: bool
    # Taux de résultats vides
    empty_results: bool
    # Temps de recherche (ms)
    retrieval_latency_ms: float
    # Source des documents (pour détecter les biais)
    doc_sources: list[str]

    @property
    def is_healthy(self) -> bool:
        return (
            not self.empty_results
            and self.avg_relevance_score > 0.7
            and self.retrieval_latency_ms < 500
        )

Métriques de génération

DEVELOPERpython
@dataclass
class GenerationMetrics:
    """Métriques de qualité de la génération."""

    # Faithfulness : la réponse est-elle fidèle aux documents ?
    faithfulness_score: float  # 0.0 - 1.0
    # La réponse contient-elle des hallucinations ?
    hallucination_detected: bool
    # Complétude de la réponse
    completeness_score: float  # 0.0 - 1.0
    # Tokens utilisés (input + output)
    input_tokens: int
    output_tokens: int
    # Temps de génération (ms)
    generation_latency_ms: float
    # Coût estimé ($)
    estimated_cost_usd: float

    @property
    def is_healthy(self) -> bool:
        return (
            self.faithfulness_score > 0.8
            and not self.hallucination_detected
            and self.generation_latency_ms < 3000
        )

Métriques utilisateur

DEVELOPERpython
@dataclass
class UserMetrics:
    """Métriques de satisfaction utilisateur."""

    # Feedback explicite (thumbs up/down)
    user_feedback: Optional[str]  # "positive" | "negative" | None
    # L'utilisateur a-t-il reformulé sa question ?
    query_reformulated: bool
    # L'utilisateur a-t-il escaladé vers un humain ?
    escalated_to_human: bool
    # Durée de la session (secondes)
    session_duration_seconds: float
    # Nombre de messages dans la conversation
    message_count: int

Dashboard de synthèse

Tracing du pipeline RAG

Architecture de tracing

Chaque requête RAG doit être tracée étape par étape :

DEVELOPERpython
import time
import uuid
from contextlib import contextmanager

class RAGTracer:
    """Trace chaque étape du pipeline RAG."""

    def __init__(self, trace_backend):
        self.backend = trace_backend

    @contextmanager
    def trace_request(self, user_id: str, query: str):
        trace_id = str(uuid.uuid4())
        trace = {
            "trace_id": trace_id,
            "user_id": user_id,
            "query": query,
            "started_at": time.time(),
            "steps": [],
        }

        yield trace

        trace["total_duration_ms"] = (
            (time.time() - trace["started_at"]) * 1000
        )
        self.backend.save_trace(trace)

    @contextmanager
    def trace_step(self, trace: dict, step_name: str):
        step = {
            "name": step_name,
            "started_at": time.time(),
            "metadata": {},
        }

        yield step

        step["duration_ms"] = (time.time() - step["started_at"]) * 1000
        trace["steps"].append(step)


# Utilisation dans le pipeline
tracer = RAGTracer(backend=langfuse_backend)

async def process_rag_query(user_id: str, query: str):
    with tracer.trace_request(user_id, query) as trace:

        # Étape 1 : Embedding de la question
        with tracer.trace_step(trace, "query_embedding") as step:
            embedding = await embed_query(query)
            step["metadata"]["model"] = "text-embedding-3-small"
            step["metadata"]["dimensions"] = len(embedding)

        # Étape 2 : Recherche vectorielle
        with tracer.trace_step(trace, "vector_search") as step:
            docs = await search_vectors(embedding, top_k=5)
            step["metadata"]["num_results"] = len(docs)
            step["metadata"]["avg_score"] = avg_score(docs)

        # Étape 3 : Reranking
        with tracer.trace_step(trace, "reranking") as step:
            ranked_docs = await rerank(query, docs)
            step["metadata"]["top_score"] = ranked_docs[0].score

        # Étape 4 : Génération LLM
        with tracer.trace_step(trace, "llm_generation") as step:
            response = await generate(query, ranked_docs)
            step["metadata"]["model"] = "gpt-4o"
            step["metadata"]["input_tokens"] = response.usage.input
            step["metadata"]["output_tokens"] = response.usage.output

        return response.text

Visualisation d'une trace

Trace: abc-123 | Durée totale: 2340ms | Status: OK
├─ query_embedding     [45ms]  model=text-embedding-3-small
├─ vector_search       [120ms] results=5, avg_score=0.82
├─ reranking           [380ms] model=cohere-rerank-v3.5, top=0.94
├─ llm_generation      [1780ms] model=gpt-4o, tokens=1250/340
│   ├─ input_tokens:  1250
│   ├─ output_tokens: 340
│   └─ cost: $0.023
└─ total_cost: $0.028

Comparaison des outils d'observabilité

Tableau comparatif détaillé

Recommandation par cas d'usage

Implémentation avec Langfuse

Setup et instrumentation

DEVELOPERpython
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

# Initialisation
langfuse = Langfuse(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="https://cloud.langfuse.com"  # ou self-hosted
)

@observe()
async def rag_pipeline(query: str, user_id: str) -> str:
    """Pipeline RAG complet avec tracing Langfuse."""

    # Trace l'embedding
    langfuse_context.update_current_observation(
        name="query_embedding",
        metadata={"model": "text-embedding-3-small"}
    )
    embedding = await embed_query(query)

    # Trace la recherche
    with langfuse_context.observe(name="vector_search") as span:
        docs = await search_vectors(embedding, top_k=5)
        span.update(
            metadata={
                "num_results": len(docs),
                "avg_score": sum(d.score for d in docs) / len(docs)
            }
        )

    # Trace la génération
    with langfuse_context.observe(
        name="llm_generation",
        model="gpt-4o"
    ) as generation:
        response = await generate_response(query, docs)
        generation.update(
            usage={
                "input": response.usage.prompt_tokens,
                "output": response.usage.completion_tokens,
            },
            metadata={"temperature": 0.1}
        )

    # Score automatique
    langfuse_context.score_current_trace(
        name="relevance",
        value=compute_relevance(query, docs),
        comment="Relevance score automatique"
    )

    return response.text


@observe()
async def embed_query(query: str) -> list[float]:
    """Embedding avec tracing automatique."""
    result = await openai.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    return result.data[0].embedding

Évaluations automatiques avec Langfuse

DEVELOPERpython
from langfuse import Langfuse

langfuse = Langfuse()

# Évaluation de la fidélité (faithfulness)
async def evaluate_faithfulness(
    trace_id: str,
    query: str,
    response: str,
    documents: list[str]
):
    """Évalue si la réponse est fidèle aux documents."""
    eval_prompt = f"""
    Question: {query}
    Documents: {documents}
    Réponse: {response}

    La réponse est-elle entièrement supportée par les documents ?
    Score de 0.0 (hallucination totale) à 1.0 (parfaitement fidèle).
    Réponds uniquement avec le score numérique.
    """

    result = await openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": eval_prompt}],
        temperature=0
    )

    score = float(result.choices[0].message.content.strip())

    langfuse.score(
        trace_id=trace_id,
        name="faithfulness",
        value=score,
        comment="Auto-eval par GPT-4o-mini"
    )

    return score

Implémentation avec LangSmith

Setup et tracing

DEVELOPERpython
import os
from langsmith import traceable
from langsmith.run_helpers import get_current_run_tree

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "lsv2_..."
os.environ["LANGCHAIN_PROJECT"] = "rag-production"

@traceable(name="rag_pipeline")
async def rag_pipeline(query: str, user_id: str) -> str:
    """Pipeline RAG avec tracing LangSmith."""

    embedding = await embed_query(query)
    docs = await search_vectors(embedding)
    response = await generate_response(query, docs)

    # Attacher des métadonnées
    run = get_current_run_tree()
    if run:
        run.extra["metadata"] = {
            "user_id": user_id,
            "num_docs": len(docs),
            "model": "gpt-4o",
        }

    return response


@traceable(name="vector_search")
async def search_vectors(embedding: list[float]) -> list:
    """Recherche vectorielle avec tracing."""
    results = await qdrant.search(
        collection="documents",
        query_vector=embedding,
        limit=5
    )
    return results


# Évaluation avec LangSmith
from langsmith.evaluation import evaluate

results = evaluate(
    rag_pipeline,
    data="rag-test-dataset",
    evaluators=[
        "relevance",
        "faithfulness",
        "helpfulness",
    ],
    experiment_prefix="v2.1-gpt4o",
)

Configuration des alertes

Alertes critiques à configurer

DEVELOPERpython
class RAGAlertManager:
    """Gestionnaire d'alertes pour le monitoring RAG."""

    def __init__(self, notification_backend):
        self.backend = notification_backend
        self.thresholds = {
            "hallucination_rate": 0.15,      # Max 15%
            "empty_results_rate": 0.10,      # Max 10%
            "p95_latency_ms": 5000,          # Max 5s
            "avg_relevance_score": 0.5,      # Min 0.5
            "error_rate": 0.05,              # Max 5%
            "cost_per_query_usd": 0.15,      # Max $0.15
            "negative_feedback_rate": 0.30,  # Max 30%
        }

    async def check_metrics(self, window_minutes: int = 60):
        metrics = await self.get_aggregated_metrics(window_minutes)

        alerts = []

        if metrics["hallucination_rate"] > self.thresholds["hallucination_rate"]:
            alerts.append({
                "severity": "critical",
                "metric": "hallucination_rate",
                "value": metrics["hallucination_rate"],
                "message": (
                    f"Taux d'hallucination à "
                    f"{metrics['hallucination_rate']:.1%} "
                    f"(seuil: {self.thresholds['hallucination_rate']:.1%})"
                ),
                "action": "Vérifier le prompt et les documents récents"
            })

        if metrics["avg_relevance_score"] < self.thresholds["avg_relevance_score"]:
            alerts.append({
                "severity": "warning",
                "metric": "avg_relevance_score",
                "value": metrics["avg_relevance_score"],
                "message": (
                    f"Relevance score à "
                    f"{metrics['avg_relevance_score']:.2f} "
                    f"(seuil: {self.thresholds['avg_relevance_score']:.2f})"
                ),
                "action": "Vérifier les embeddings et la base vectorielle"
            })

        if metrics["p95_latency_ms"] > self.thresholds["p95_latency_ms"]:
            alerts.append({
                "severity": "warning",
                "metric": "p95_latency_ms",
                "value": metrics["p95_latency_ms"],
                "message": (
                    f"Latence P95 à {metrics['p95_latency_ms']}ms "
                    f"(seuil: {self.thresholds['p95_latency_ms']}ms)"
                ),
                "action": "Vérifier le cache et les performances LLM"
            })

        for alert in alerts:
            await self.backend.send_alert(alert)

        return alerts

Intégration Slack/Discord pour les alertes

DEVELOPERpython
import httpx

class SlackAlertBackend:
    """Envoie les alertes RAG vers Slack."""

    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    async def send_alert(self, alert: dict):
        emoji = {
            "critical": "🚨",
            "warning": "⚠️",
            "info": "ℹ️"
        }

        payload = {
            "blocks": [
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": f"{emoji[alert['severity']]} "
                                f"RAG Alert: {alert['metric']}"
                    }
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": (
                            f"*Message:* {alert['message']}\n"
                            f"*Action:* {alert['action']}"
                        )
                    }
                }
            ]
        }

        async with httpx.AsyncClient() as client:
            await client.post(self.webhook_url, json=payload)

Debugging avec les traces

Identifier la source d'un problème

Quand un utilisateur signale une mauvaise réponse, les traces permettent de diagnostiquer rapidement :

DEVELOPERpython
async def debug_bad_response(trace_id: str):
    """Analyse une trace pour identifier le problème."""
    trace = await langfuse.get_trace(trace_id)

    report = []

    # 1. Vérifier le retrieval
    search_step = find_step(trace, "vector_search")
    if search_step["metadata"]["num_results"] == 0:
        report.append("PROBLÈME: Aucun document trouvé")
    elif search_step["metadata"]["avg_score"] < 0.5:
        report.append("PROBLÈME: Documents peu pertinents")

    # 2. Vérifier le reranking
    rerank_step = find_step(trace, "reranking")
    if rerank_step and rerank_step["metadata"]["top_score"] < 0.3:
        report.append("PROBLÈME: Reranking n'a pas amélioré")

    # 3. Vérifier la génération
    gen_step = find_step(trace, "llm_generation")
    if gen_step["metadata"]["output_tokens"] < 20:
        report.append("PROBLÈME: Réponse trop courte")
    if gen_step["duration_ms"] > 5000:
        report.append("ATTENTION: Génération très lente")

    # 4. Vérifier les scores
    scores = trace.get("scores", {})
    if scores.get("faithfulness", 1.0) < 0.5:
        report.append("PROBLÈME: Hallucination détectée")

    return report

Bonnes pratiques

Checklist de mise en production

Phase 1 : Instrumentation (Jour 1)
  ✅ Tracer chaque étape du pipeline
  ✅ Logger les métriques de base (latence, tokens, coûts)
  ✅ Capturer le feedback utilisateur

Phase 2 : Évaluations (Semaine 1)
  ✅ Configurer l'évaluation automatique de la fidélité
  ✅ Créer un dataset de test (50+ questions/réponses)
  ✅ Baseline des métriques

Phase 3 : Alertes (Semaine 2)
  ✅ Configurer les alertes critiques (hallucinations, erreurs)
  ✅ Intégrer Slack/Discord
  ✅ Définir les runbooks de réponse

Phase 4 : Optimisation continue (Mois 1+)
  ✅ A/B testing des prompts
  ✅ Analyse des requêtes problématiques
  ✅ Amélioration continue basée sur les données

Pour aller plus loin

• Évaluation et métriques RAG : le guide parent sur l'évaluation
• Réduire la latence RAG : optimisation des performances
• Détection des hallucinations : techniques spécifiques
• Guardrails RAG : sécurité et prompt injections
• Cache RAG intelligent : optimisation des coûts

FAQ

Quel outil d'observabilité choisir pour commencer ?

Si vous utilisez LangChain, LangSmith est le choix naturel avec son intégration native. Si vous avez des contraintes RGPD ou voulez du self-hosted, Langfuse est excellent et open source. Pour un budget zéro, Phoenix (Arize) offre un dashboard local puissant. Notre recommandation pour les entreprises françaises : Langfuse self-hosted pour le contrôle total des données.

Combien de traces dois-je conserver ?

En production, conservez au minimum 30 jours de traces pour détecter les tendances. Pour le debugging, les 7 derniers jours sont généralement suffisants. Sur Langfuse self-hosted, la rétention est illimitée (limitée uniquement par votre stockage). Sur LangSmith gratuit, vous êtes limité à 14 jours et 5000 traces/mois.

Comment mesurer la qualité sans évaluation humaine ?

Utilisez le LLM-as-Judge : un modèle léger (GPT-4o-mini) évalue chaque réponse sur la fidélité, la pertinence et la complétude. Cette approche coûte environ $0.002 par évaluation et corrèle à 85% avec l'évaluation humaine selon les benchmarks récents. Combinez avec le feedback implicite (reformulations, escalades) pour une vue complète.

À quelle fréquence dois-je vérifier mon dashboard ?

Quotidiennement pendant le premier mois, puis hebdomadairement une fois les alertes configurées. Les alertes automatiques doivent couvrir les scénarios critiques. Planifiez une revue approfondie mensuelle pour analyser les tendances et identifier les améliorations structurelles.

Ailog propose-t-il un dashboard d'observabilité intégré ?

Oui. Ailog intègre un dashboard d'observabilité qui affiche en temps réel la qualité des réponses, les métriques de retrieval, les coûts et le feedback utilisateur. Vous pouvez configurer des alertes personnalisées et exporter les données pour une analyse approfondie. L'hébergement en France garantit la conformité RGPD de vos données d'observabilité.