Pipeline d'Évaluation RAG Automatisé : Détectez les Régressions Avant vos Utilisateurs

TL;DR

83% des équipes RAG découvrent les régressions par les plaintes utilisateurs. Ce guide vous montre comment construire un pipeline d'évaluation automatisé qui détecte les problèmes avant la production : golden datasets managés, métriques RAGAS automatisées (faithfulness, relevancy, context recall), seuils d'alerte et intégration GitHub Actions. Avec un pipeline bien configuré, vous détectez 95% des régressions avant qu'elles n'atteignent les utilisateurs.

Pourquoi un pipeline d'évaluation automatisé

Le problème des évaluations manuelles

Les causes de régression en RAG

┌────────────────────────────────────────────────────┐
│              CAUSES DE RÉGRESSION RAG                │
├────────────────────────────────────────────────────┤
│                                                    │
│  📄 Changement de documents                        │
│  ├─ Nouveaux documents mal formatés                │
│  ├─ Documents supprimés qui étaient pertinents     │
│  └─ Mise à jour qui casse le chunking              │
│                                                    │
│  🔧 Changement de configuration                    │
│  ├─ Nouveau modèle d'embedding                     │
│  ├─ Changement de chunk size                       │
│  ├─ Modification du prompt                         │
│  └─ Mise à jour du reranker                        │
│                                                    │
│  🤖 Changement de modèle LLM                       │
│  ├─ Nouvelle version du modèle                     │
│  ├─ Changement de provider                         │
│  └─ Modification de température/top-p              │
│                                                    │
│  📊 Dégradation silencieuse                         │
│  ├─ Drift des données                              │
│  ├─ Index vectoriel fragmenté                      │
│  └─ Changement API embedding provider              │
└────────────────────────────────────────────────────┘

Architecture du pipeline d'évaluation

Vue d'ensemble

┌─────────────┐     ┌──────────────┐     ┌──────────────┐
│   Golden     │     │    RAG       │     │  Évaluation  │
│   Dataset    │────▶│   Pipeline   │────▶│  Automatique │
│              │     │              │     │              │
│ - Questions  │     │ - Retrieval  │     │ - RAGAS      │
│ - Réponses   │     │ - Generation │     │ - DeepEval   │
│ - Contextes  │     │              │     │ - Custom     │
└─────────────┘     └──────────────┘     └──────┬───────┘
                                                 │
                                        ┌────────┴────────┐
                                        │                 │
                                  ┌─────┴─────┐    ┌──────┴──────┐
                                  │  Seuils    │    │  Dashboard  │
                                  │  & Alertes │    │  Langfuse   │
                                  └─────┬─────┘    └─────────────┘
                                        │
                                  ┌─────┴─────┐
                                  │  CI/CD    │
                                  │  Gate     │
                                  └───────────┘

Étape 1 : Construire le golden dataset

Structure du golden dataset

DEVELOPERpython
from dataclasses import dataclass, field
from typing import List, Optional
import json

@dataclass
class GoldenExample:
    """Un exemple du golden dataset."""
    id: str
    question: str
    expected_answer: str
    expected_contexts: List[str]
    category: str  # "factual", "comparison", "multi-hop", "opinion"
    difficulty: str  # "easy", "medium", "hard"
    tags: List[str] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)

@dataclass
class GoldenDataset:
    """Dataset d'évaluation complet."""
    version: str
    created_at: str
    examples: List[GoldenExample]

    def to_json(self, path: str):
        with open(path, "w") as f:
            json.dump(self.__dict__, f, indent=2, default=str)

    @classmethod
    def from_json(cls, path: str):
        with open(path) as f:
            data = json.load(f)
        examples = [GoldenExample(**ex) for ex in data["examples"]]
        return cls(
            version=data["version"],
            created_at=data["created_at"],
            examples=examples
        )

Générer le golden dataset automatiquement

DEVELOPERpython
import anthropic

client = anthropic.Anthropic()

def generate_golden_examples(
    documents: list,
    num_questions: int = 50,
    categories: list = None
) -> list:
    """Génère des exemples de test à partir des documents."""
    if categories is None:
        categories = ["factual", "comparison", "multi-hop"]

    examples = []

    for doc in documents:
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2000,
            messages=[{
                "role": "user",
                "content": f"""À partir du document suivant, génère
{num_questions // len(documents)} paires question/réponse pour
tester un système RAG.

Pour chaque paire :
- question : naturelle, comme un vrai utilisateur poserait
- expected_answer : la réponse correcte et complète
- context : le passage exact du document qui contient la réponse
- category : {', '.join(categories)}
- difficulty : easy, medium, ou hard

Retourne un JSON : [{{"question": "...", "expected_answer": "...",
"context": "...", "category": "...", "difficulty": "..."}}]

Document :
{doc['text'][:5000]}"""
            }]
        )

        batch = json.loads(response.content[0].text)
        for item in batch:
            examples.append(GoldenExample(
                id=f"gen_{len(examples)}",
                question=item["question"],
                expected_answer=item["expected_answer"],
                expected_contexts=[item["context"]],
                category=item["category"],
                difficulty=item["difficulty"],
                tags=[doc.get("source", "unknown")]
            ))

    return examples

Taille recommandée du golden dataset

Étape 2 : Implémenter les métriques

Avec RAGAS

DEVELOPERpython
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    answer_correctness,
)
from datasets import Dataset

def evaluate_rag_with_ragas(
    golden_dataset: GoldenDataset,
    rag_pipeline,
    metrics=None
):
    """Évalue le pipeline RAG avec RAGAS."""
    if metrics is None:
        metrics = [
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall,
            answer_correctness,
        ]

    # Exécuter le RAG sur chaque question
    questions = []
    answers = []
    contexts = []
    ground_truths = []

    for example in golden_dataset.examples:
        result = rag_pipeline.query(example.question)

        questions.append(example.question)
        answers.append(result["answer"])
        contexts.append(result["retrieved_contexts"])
        ground_truths.append(example.expected_answer)

    # Créer le dataset RAGAS
    eval_dataset = Dataset.from_dict({
        "question": questions,
        "answer": answers,
        "contexts": contexts,
        "ground_truth": ground_truths,
    })

    # Évaluer
    results = evaluate(
        eval_dataset,
        metrics=metrics,
    )

    return results

# Utilisation
results = evaluate_rag_with_ragas(golden_dataset, my_rag_pipeline)
print(f"Faithfulness: {results['faithfulness']:.3f}")
print(f"Answer Relevancy: {results['answer_relevancy']:.3f}")
print(f"Context Recall: {results['context_recall']:.3f}")
print(f"Context Precision: {results['context_precision']:.3f}")

Avec DeepEval

DEVELOPERpython
from deepeval import evaluate
from deepeval.metrics import (
    FaithfulnessMetric,
    AnswerRelevancyMetric,
    ContextualRecallMetric,
    ContextualPrecisionMetric,
    HallucinationMetric,
)
from deepeval.test_case import LLMTestCase

def evaluate_with_deepeval(
    golden_dataset: GoldenDataset,
    rag_pipeline
):
    """Évalue le pipeline RAG avec DeepEval."""
    test_cases = []

    for example in golden_dataset.examples:
        result = rag_pipeline.query(example.question)

        test_case = LLMTestCase(
            input=example.question,
            actual_output=result["answer"],
            expected_output=example.expected_answer,
            retrieval_context=result["retrieved_contexts"],
            context=example.expected_contexts,
        )
        test_cases.append(test_case)

    # Métriques
    metrics = [
        FaithfulnessMetric(threshold=0.8),
        AnswerRelevancyMetric(threshold=0.7),
        ContextualRecallMetric(threshold=0.7),
        ContextualPrecisionMetric(threshold=0.7),
        HallucinationMetric(threshold=0.2),
    ]

    # Évaluer
    results = evaluate(test_cases, metrics)
    return results

Métriques custom avec pytest

DEVELOPERpython
# tests/test_rag_quality.py
import pytest
import json
from pathlib import Path

# Charger le golden dataset
GOLDEN_DATASET = GoldenDataset.from_json("tests/golden_dataset.json")
RAG_PIPELINE = init_rag_pipeline()

# Seuils de qualité
THRESHOLDS = {
    "faithfulness": 0.85,
    "answer_relevancy": 0.80,
    "context_recall": 0.75,
    "context_precision": 0.70,
    "answer_correctness": 0.75,
    "latency_p95_ms": 3000,
    "hallucination_rate": 0.05,
}

class TestRAGQuality:
    """Tests de qualité RAG automatisés."""

    @pytest.fixture(autouse=True)
    def setup(self):
        """Exécuter le RAG sur le golden dataset."""
        self.results = []
        for example in GOLDEN_DATASET.examples:
            result = RAG_PIPELINE.query(example.question)
            self.results.append({
                "example": example,
                "result": result
            })

    def test_faithfulness_above_threshold(self):
        """Les réponses sont fidèles aux sources."""
        scores = [
            compute_faithfulness(r["result"]["answer"],
                                 r["result"]["retrieved_contexts"])
            for r in self.results
        ]
        avg_score = sum(scores) / len(scores)
        assert avg_score >= THRESHOLDS["faithfulness"], \
            f"Faithfulness {avg_score:.3f} < {THRESHOLDS['faithfulness']}"

    def test_context_recall_above_threshold(self):
        """Les bons documents sont récupérés."""
        scores = []
        for r in self.results:
            retrieved = r["result"]["retrieved_contexts"]
            expected = r["example"].expected_contexts
            recall = compute_context_recall(retrieved, expected)
            scores.append(recall)

        avg_score = sum(scores) / len(scores)
        assert avg_score >= THRESHOLDS["context_recall"], \
            f"Context Recall {avg_score:.3f} < {THRESHOLDS['context_recall']}"

    def test_no_hallucination_spike(self):
        """Pas plus de 5% de hallucinations."""
        hallucination_count = sum(
            1 for r in self.results
            if is_hallucination(
                r["result"]["answer"],
                r["result"]["retrieved_contexts"]
            )
        )
        rate = hallucination_count / len(self.results)
        assert rate <= THRESHOLDS["hallucination_rate"], \
            f"Hallucination rate {rate:.3f} > {THRESHOLDS['hallucination_rate']}"

    def test_latency_within_bounds(self):
        """Latence P95 sous 3 secondes."""
        latencies = [r["result"]["latency_ms"] for r in self.results]
        latencies.sort()
        p95 = latencies[int(len(latencies) * 0.95)]
        assert p95 <= THRESHOLDS["latency_p95_ms"], \
            f"P95 latency {p95}ms > {THRESHOLDS['latency_p95_ms']}ms"

    def test_no_regression_by_category(self):
        """Pas de régression par catégorie de question."""
        category_scores = {}
        for r in self.results:
            cat = r["example"].category
            score = compute_answer_quality(
                r["result"]["answer"],
                r["example"].expected_answer
            )
            category_scores.setdefault(cat, []).append(score)

        # Vérifier chaque catégorie
        baseline = load_baseline_scores()
        for cat, scores in category_scores.items():
            avg = sum(scores) / len(scores)
            if cat in baseline:
                assert avg >= baseline[cat] - 0.05, \
                    f"Regression in {cat}: {avg:.3f} vs baseline {baseline[cat]:.3f}"

Étape 3 : Intégration CI/CD avec GitHub Actions

Workflow GitHub Actions

DEVELOPERyaml
# .github/workflows/rag-evaluation.yml
name: RAG Evaluation Pipeline

on:
  push:
    paths:
      - 'backend/rag/**'
      - 'backend/prompts/**'
      - 'backend/config/**'
  pull_request:
    branches: [main, develop]
  schedule:
    - cron: '0 6 * * 1'  # Tous les lundis à 6h

jobs:
  evaluate:
    runs-on: ubuntu-latest
    timeout-minutes: 30

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install ragas deepeval pytest langfuse
          pip install -r backend/requirements.txt

      - name: Run RAG evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          QDRANT_URL: ${{ secrets.QDRANT_URL }}
          LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SECRET_KEY }}
        run: |
          pytest tests/test_rag_quality.py \
            --tb=long \
            --json-report \
            --json-report-file=eval_results.json \
            -v

      - name: Check thresholds
        run: |
          python scripts/check_eval_thresholds.py \
            --results eval_results.json \
            --baseline tests/baseline_scores.json

      - name: Upload results to Langfuse
        if: always()
        run: |
          python scripts/upload_eval_to_langfuse.py \
            --results eval_results.json

      - name: Comment PR with results
        if: github.event_name == 'pull_request'
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const results = JSON.parse(
              fs.readFileSync('eval_results.json', 'utf8')
            );
            const body = formatEvalResults(results);
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: body
            });

      - name: Alert on regression
        if: failure()
        run: |
          python scripts/send_regression_alert.py \
            --results eval_results.json \
            --channel slack \
            --webhook ${{ secrets.SLACK_WEBHOOK }}

Script de vérification des seuils

DEVELOPERpython
# scripts/check_eval_thresholds.py
import json
import sys
import argparse

def check_thresholds(results_path: str, baseline_path: str):
    """Vérifie que les résultats respectent les seuils."""
    with open(results_path) as f:
        results = json.load(f)

    with open(baseline_path) as f:
        baseline = json.load(f)

    failures = []
    warnings = []

    for metric, value in results["metrics"].items():
        threshold = baseline["thresholds"].get(metric)
        previous = baseline["previous_scores"].get(metric)

        if threshold and value < threshold:
            failures.append(
                f"FAIL: {metric} = {value:.3f} "
                f"(threshold: {threshold:.3f})"
            )
        elif previous and value < previous - 0.03:
            warnings.append(
                f"WARN: {metric} = {value:.3f} "
                f"(previous: {previous:.3f}, delta: "
                f"{value - previous:.3f})"
            )

    # Afficher le rapport
    print("=" * 60)
    print("RAG EVALUATION REPORT")
    print("=" * 60)

    for metric, value in results["metrics"].items():
        status = "PASS"
        for f in failures:
            if metric in f:
                status = "FAIL"
        print(f"  {metric}: {value:.3f} [{status}]")

    if warnings:
        print("\nWarnings:")
        for w in warnings:
            print(f"  {w}")

    if failures:
        print(f"\n{len(failures)} FAILURES detected!")
        for f in failures:
            print(f"  {f}")
        sys.exit(1)

    print("\nAll metrics within thresholds.")
    sys.exit(0)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--results", required=True)
    parser.add_argument("--baseline", required=True)
    args = parser.parse_args()
    check_thresholds(args.results, args.baseline)

Étape 4 : Dashboard et monitoring

Comparatif des outils d'évaluation

Intégration Langfuse

DEVELOPERpython
from langfuse import Langfuse

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

def log_evaluation_to_langfuse(
    results: dict,
    run_id: str,
    metadata: dict = None
):
    """Enregistre les résultats d'évaluation dans Langfuse."""
    for example_result in results["examples"]:
        trace = langfuse.trace(
            name="rag-evaluation",
            metadata={
                "run_id": run_id,
                "category": example_result["category"],
                **(metadata or {})
            }
        )

        # Log du retrieval
        trace.span(
            name="retrieval",
            input=example_result["question"],
            output=example_result["retrieved_contexts"],
            metadata={
                "num_contexts": len(example_result["retrieved_contexts"]),
                "context_recall": example_result["context_recall"],
            }
        )

        # Log de la génération
        generation = trace.generation(
            name="generation",
            input=example_result["question"],
            output=example_result["answer"],
            metadata={
                "faithfulness": example_result["faithfulness"],
                "relevancy": example_result["relevancy"],
            }
        )

        # Scores
        trace.score(
            name="faithfulness",
            value=example_result["faithfulness"],
        )
        trace.score(
            name="answer_relevancy",
            value=example_result["relevancy"],
        )
        trace.score(
            name="context_recall",
            value=example_result["context_recall"],
        )

    langfuse.flush()

Métriques à tracker dans le dashboard

Étape 5 : Détection de régressions avancée

Détection statistique

DEVELOPERpython
import numpy as np
from scipy import stats

def detect_regression(
    current_scores: list,
    baseline_scores: list,
    alpha: float = 0.05,
    min_delta: float = 0.03
) -> dict:
    """Détecte les régressions statistiquement significatives."""
    current = np.array(current_scores)
    baseline = np.array(baseline_scores)

    # Test de Welch (t-test pour échantillons inégaux)
    t_stat, p_value = stats.ttest_ind(
        baseline, current,
        equal_var=False,
        alternative='greater'  # baseline > current = régression
    )

    delta = np.mean(baseline) - np.mean(current)

    is_regression = (
        p_value < alpha and
        delta > min_delta
    )

    return {
        "is_regression": is_regression,
        "p_value": p_value,
        "delta": delta,
        "current_mean": np.mean(current),
        "baseline_mean": np.mean(baseline),
        "confidence": 1 - p_value,
        "effect_size": delta / np.std(baseline)  # Cohen's d
    }

Alerting intelligent

DEVELOPERpython
def evaluate_and_alert(rag_pipeline, golden_dataset, config):
    """Évalue le RAG et envoie des alertes si nécessaire."""
    results = evaluate_rag_with_ragas(golden_dataset, rag_pipeline)
    baseline = load_baseline()

    alerts = []

    for metric, value in results.items():
        threshold = config["thresholds"].get(metric)
        previous = baseline.get(metric)

        # Alerte critique : sous le seuil absolu
        if threshold and value < threshold:
            alerts.append({
                "level": "critical",
                "metric": metric,
                "value": value,
                "threshold": threshold,
                "message": (
                    f"{metric} dropped to {value:.3f}, "
                    f"below critical threshold {threshold:.3f}"
                )
            })

        # Alerte warning : régression par rapport au baseline
        elif previous and value < previous - 0.03:
            alerts.append({
                "level": "warning",
                "metric": metric,
                "value": value,
                "previous": previous,
                "message": (
                    f"{metric} decreased from {previous:.3f} to "
                    f"{value:.3f} (delta: {value - previous:+.3f})"
                )
            })

    # Envoyer les alertes
    if alerts:
        send_alerts(alerts, config)

    # Mettre à jour le baseline si tout est OK
    if not any(a["level"] == "critical" for a in alerts):
        update_baseline(results)

    return results, alerts

Bonnes pratiques

Gestion du golden dataset

1. Versionner le golden dataset dans git (comme le code)
2. Mettre à jour quand les documents changent significativement
3. Stratifier par catégorie, difficulté et source
4. Valider humainement un échantillon régulièrement
5. Ne jamais optimiser pour le golden dataset (overfitting)

Pipeline robuste

Fréquence d'évaluation

FAQ

Combien d'exemples dans le golden dataset ?

Minimum 30 pour des résultats statistiquement significatifs. L'idéal est 100-200 exemples stratifiés par catégorie de question. Au-delà de 500, les coûts d'évaluation augmentent sans gain proportionnel en confiance. Commencez avec 50 et augmentez progressivement. Voir notre guide sur les métriques d'évaluation RAG pour le détail des métriques.

RAGAS ou DeepEval, lequel choisir ?

RAGAS est plus mature, mieux documenté et la référence académique. DeepEval offre plus de métriques (14+), une intégration CI/CD native et un dashboard. Pour un projet simple, RAGAS suffit. Pour un pipeline de production avec monitoring, DeepEval + Langfuse est plus complet. Les deux sont open-source et compatibles avec pytest.

Comment gérer les questions sans bonne réponse ?

Incluez dans votre golden dataset des questions auxquelles le RAG ne devrait pas répondre (hors scope, information absente). Ajoutez une métrique "refusal accuracy" : le RAG doit refuser de répondre ou dire "je ne sais pas" plutôt que d'halluciner. C'est souvent la métrique la plus difficile à maintenir.

Le pipeline d'évaluation est-il cher ?

Pour 100 exemples avec RAGAS (qui utilise GPT-4o-mini par défaut pour l'évaluation) : de quelques centimes à quelques dollars par run selon le modèle juge choisi. Avec DeepEval et un modèle local, c'est quasi gratuit. Le coût principal est le temps de run (~5-15 minutes pour 100 exemples). C'est négligeable comparé au coût d'une régression en production. Voir notre guide sur l'optimisation des coûts RAG.

Comment tester les changements de prompt ?

Utilisez l'évaluation A/B : exécutez le golden dataset avec l'ancien et le nouveau prompt, puis comparez les métriques. Le nouveau prompt doit être statistiquement meilleur (pas juste marginal) pour justifier le changement. Intégrez cette comparaison dans votre PR review. Voir notre guide sur le prompt engineering RAG.

---

Un pipeline d'évaluation RAG automatisé est votre filet de sécurité contre les régressions silencieuses. Investir quelques heures dans sa mise en place vous économisera des jours de debugging et protégera la confiance de vos utilisateurs. Testez Ailog pour bénéficier d'un monitoring qualité intégré sur votre chatbot RAG.