Évaluer un système RAG : Métriques et méthodologies

Un système RAG peut sembler fonctionner correctement en apparence, mais comment savoir s'il répond vraiment aux attentes ? L'évaluation rigoureuse est la clé pour passer d'un prototype à un produit de qualité production. Ce guide vous présente les métriques, frameworks et méthodologies pour mesurer et améliorer votre RAG.

Pourquoi l'évaluation est critique

Le problème des évaluations subjectives

Sans métriques objectives, l'évaluation d'un RAG se résume souvent à :

• "Ça a l'air correct" (biais de confirmation)
• Quelques tests manuels sur des cas favorables
• Retours utilisateurs tardifs en production

Cette approche masque des problèmes graves :

• Hallucinations subtiles mais fréquentes
• Réponses hors sujet pour certaines catégories de questions
• Dégradation progressive de la qualité après mises à jour

Les dimensions de l'évaluation

Un système RAG doit être évalué sur plusieurs axes :

Métriques de retrieval

Recall@k

Mesure la proportion de documents pertinents retrouvés parmi les k premiers résultats.

DEVELOPERpython
def recall_at_k(retrieved_ids: list[str], relevant_ids: list[str], k: int) -> float:
    """
    Recall@k : Proportion des documents pertinents récupérés

    Args:
        retrieved_ids: IDs des documents récupérés (ordonnés par score)
        relevant_ids: IDs des documents réellement pertinents
        k: Nombre de résultats à considérer

    Returns:
        Score entre 0 et 1
    """
    if not relevant_ids:
        return 0.0

    retrieved_k = set(retrieved_ids[:k])
    relevant_set = set(relevant_ids)

    return len(retrieved_k & relevant_set) / len(relevant_set)

# Exemple
retrieved = ["doc1", "doc3", "doc5", "doc2", "doc7"]
relevant = ["doc1", "doc2", "doc4"]

print(f"Recall@3: {recall_at_k(retrieved, relevant, 3)}")  # 0.33 (1/3)
print(f"Recall@5: {recall_at_k(retrieved, relevant, 5)}")  # 0.67 (2/3)

Interprétation :

• Recall@5 de 0.8+ : Excellent, la plupart des documents pertinents sont récupérés
• Recall@5 de 0.5-0.8 : Acceptable, mais amélioration possible
• Recall@5 < 0.5 : Problématique, beaucoup de documents pertinents manqués

Precision@k

Mesure la proportion de documents pertinents parmi les k récupérés.

DEVELOPERpython
def precision_at_k(retrieved_ids: list[str], relevant_ids: list[str], k: int) -> float:
    """
    Precision@k : Proportion des documents récupérés qui sont pertinents
    """
    if k == 0:
        return 0.0

    retrieved_k = set(retrieved_ids[:k])
    relevant_set = set(relevant_ids)

    return len(retrieved_k & relevant_set) / k

# Exemple
print(f"Precision@3: {precision_at_k(retrieved, relevant, 3)}")  # 0.33 (1/3)
print(f"Precision@5: {precision_at_k(retrieved, relevant, 5)}")  # 0.40 (2/5)

MRR (Mean Reciprocal Rank)

Mesure la position moyenne du premier document pertinent.

DEVELOPERpython
def mrr(queries_results: list[tuple[list[str], list[str]]]) -> float:
    """
    Mean Reciprocal Rank

    Args:
        queries_results: Liste de (documents_récupérés, documents_pertinents)

    Returns:
        Score MRR entre 0 et 1
    """
    reciprocal_ranks = []

    for retrieved, relevant in queries_results:
        relevant_set = set(relevant)

        for i, doc_id in enumerate(retrieved):
            if doc_id in relevant_set:
                reciprocal_ranks.append(1 / (i + 1))
                break
        else:
            reciprocal_ranks.append(0)

    return sum(reciprocal_ranks) / len(reciprocal_ranks) if reciprocal_ranks else 0

# Exemple : 3 requêtes
results = [
    (["doc1", "doc2", "doc3"], ["doc1"]),     # Premier = pertinent -> 1/1
    (["doc4", "doc1", "doc2"], ["doc1"]),     # Deuxième = pertinent -> 1/2
    (["doc5", "doc6", "doc7"], ["doc8"]),     # Aucun pertinent -> 0
]
print(f"MRR: {mrr(results)}")  # (1 + 0.5 + 0) / 3 = 0.5

NDCG (Normalized Discounted Cumulative Gain)

Prend en compte l'ordre ET les scores de pertinence gradués.

DEVELOPERpython
import numpy as np

def dcg_at_k(relevances: list[float], k: int) -> float:
    """
    Discounted Cumulative Gain
    """
    relevances = np.array(relevances[:k])
    if len(relevances) == 0:
        return 0.0

    # Logarithme base 2, position 1-indexée
    discounts = np.log2(np.arange(2, len(relevances) + 2))
    return np.sum(relevances / discounts)

def ndcg_at_k(relevances: list[float], k: int) -> float:
    """
    Normalized DCG : Compare au DCG idéal
    """
    dcg = dcg_at_k(relevances, k)

    # DCG idéal : relevances triées décroissant
    ideal_relevances = sorted(relevances, reverse=True)
    idcg = dcg_at_k(ideal_relevances, k)

    return dcg / idcg if idcg > 0 else 0.0

# Exemple avec scores de pertinence gradués (0=non pertinent, 1=peu, 2=très)
relevances = [2, 0, 1, 1, 0]  # Document 1 très pertinent, doc 2 non, etc.
print(f"NDCG@5: {ndcg_at_k(relevances, 5):.3f}")

Métriques de génération

Faithfulness (Fidélité)

Mesure si la réponse est fidèle au contexte fourni, sans hallucination.

DEVELOPERpython
class FaithfulnessEvaluator:
    def __init__(self, llm):
        self.llm = llm

    async def evaluate(
        self,
        question: str,
        context: str,
        answer: str
    ) -> dict:
        """
        Évalue la fidélité de la réponse au contexte
        """
        # Étape 1 : Extraire les affirmations de la réponse
        claims = await self._extract_claims(answer)

        # Étape 2 : Vérifier chaque affirmation contre le contexte
        verification_results = []
        for claim in claims:
            is_supported = await self._verify_claim(claim, context)
            verification_results.append({
                "claim": claim,
                "supported": is_supported
            })

        # Calculer le score
        supported_count = sum(1 for r in verification_results if r["supported"])
        score = supported_count / len(claims) if claims else 1.0

        return {
            "score": score,
            "claims": verification_results,
            "total_claims": len(claims),
            "supported_claims": supported_count
        }

    async def _extract_claims(self, answer: str) -> list[str]:
        """
        Extrait les affirmations factuelles de la réponse
        """
        prompt = f"""
        Extrais toutes les affirmations factuelles de cette réponse.
        Chaque affirmation doit être une phrase simple et vérifiable.

        Réponse : {answer}

        Affirmations (une par ligne) :
        """

        response = await self.llm.generate(prompt, temperature=0)
        return [line.strip() for line in response.split("\n") if line.strip()]

    async def _verify_claim(self, claim: str, context: str) -> bool:
        """
        Vérifie si une affirmation est supportée par le contexte
        """
        prompt = f"""
        L'affirmation suivante est-elle explicitement supportée par le contexte ?

        Contexte : {context}

        Affirmation : {claim}

        Réponds uniquement par "OUI" ou "NON".
        """

        response = await self.llm.generate(prompt, temperature=0)
        return response.strip().upper() == "OUI"

Answer Relevancy (Pertinence)

Mesure si la réponse répond effectivement à la question posée.

DEVELOPERpython
class RelevancyEvaluator:
    def __init__(self, llm, embedding_model):
        self.llm = llm
        self.embedder = embedding_model

    async def evaluate(
        self,
        question: str,
        answer: str
    ) -> dict:
        """
        Évalue la pertinence de la réponse par rapport à la question
        """
        # Méthode 1 : Générer des questions à partir de la réponse
        generated_questions = await self._generate_questions(answer)

        # Méthode 2 : Calculer la similarité avec la question originale
        similarities = []
        question_embedding = self.embedder.encode(question)

        for gen_q in generated_questions:
            gen_embedding = self.embedder.encode(gen_q)
            sim = self._cosine_similarity(question_embedding, gen_embedding)
            similarities.append(sim)

        score = sum(similarities) / len(similarities) if similarities else 0

        return {
            "score": score,
            "generated_questions": generated_questions,
            "similarities": similarities
        }

    async def _generate_questions(self, answer: str, n: int = 3) -> list[str]:
        """
        Génère des questions auxquelles la réponse pourrait répondre
        """
        prompt = f"""
        Génère {n} questions différentes auxquelles cette réponse pourrait répondre.

        Réponse : {answer}

        Questions (une par ligne) :
        """

        response = await self.llm.generate(prompt, temperature=0.5)
        return [line.strip() for line in response.split("\n") if line.strip()][:n]

    def _cosine_similarity(self, a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

Context Recall

Mesure si le contexte récupéré contient les informations nécessaires pour répondre.

DEVELOPERpython
class ContextRecallEvaluator:
    def __init__(self, llm):
        self.llm = llm

    async def evaluate(
        self,
        question: str,
        context: str,
        ground_truth: str
    ) -> dict:
        """
        Évalue si le contexte contient les informations de la réponse attendue
        """
        # Extraire les faits de la réponse attendue
        gt_facts = await self._extract_facts(ground_truth)

        # Vérifier la présence de chaque fait dans le contexte
        attributions = []
        for fact in gt_facts:
            present = await self._check_presence(fact, context)
            attributions.append({
                "fact": fact,
                "present_in_context": present
            })

        # Score = proportion de faits présents
        present_count = sum(1 for a in attributions if a["present_in_context"])
        score = present_count / len(gt_facts) if gt_facts else 1.0

        return {
            "score": score,
            "attributions": attributions,
            "total_facts": len(gt_facts),
            "facts_in_context": present_count
        }

Framework RAGAS

RAGAS (Retrieval Augmented Generation Assessment) est le framework de référence pour l'évaluation RAG.

Installation et configuration

DEVELOPERpython
# pip install ragas

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

# Préparer les données d'évaluation
eval_data = {
    "question": [
        "Quelle est la politique de retour ?",
        "Comment contacter le support ?",
    ],
    "answer": [
        "Vous avez 30 jours pour retourner un produit non utilisé.",
        "Vous pouvez contacter le support par email à [email protected].",
    ],
    "contexts": [
        ["La politique de retour permet aux clients de retourner tout produit non utilisé dans un délai de 30 jours."],
        ["Le support est joignable par email à [email protected] ou par téléphone au 01 23 45 67 89."],
    ],
    "ground_truth": [
        "Les clients peuvent retourner les produits non utilisés sous 30 jours.",
        "Le support est disponible par email ([email protected]) et téléphone.",
    ]
}

dataset = Dataset.from_dict(eval_data)

# Évaluer
results = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_recall,
        context_precision,
    ]
)

print(results)

Interpréter les résultats RAGAS

Création d'un dataset d'évaluation

Génération automatique de questions

DEVELOPERpython
class EvalDatasetGenerator:
    def __init__(self, llm):
        self.llm = llm

    async def generate_from_documents(
        self,
        documents: list[dict],
        questions_per_doc: int = 3
    ) -> list[dict]:
        """
        Génère un dataset d'évaluation à partir de documents
        """
        eval_data = []

        for doc in documents:
            # Générer des questions
            questions = await self._generate_questions(doc["content"], questions_per_doc)

            for q in questions:
                # Générer la réponse attendue
                ground_truth = await self._generate_answer(q, doc["content"])

                eval_data.append({
                    "question": q,
                    "ground_truth": ground_truth,
                    "source_doc_id": doc["id"],
                    "source_content": doc["content"]
                })

        return eval_data

    async def _generate_questions(self, content: str, n: int) -> list[str]:
        """
        Génère des questions à partir du contenu
        """
        prompt = f"""
        Génère {n} questions diverses et réalistes qu'un utilisateur pourrait poser
        et auxquelles ce document peut répondre.

        Document :
        {content[:2000]}

        Questions (une par ligne, variées en complexité) :
        """

        response = await self.llm.generate(prompt, temperature=0.7)
        return [line.strip().lstrip("0123456789.- ") for line in response.split("\n") if line.strip()][:n]

    async def _generate_answer(self, question: str, content: str) -> str:
        """
        Génère la réponse attendue basée sur le document
        """
        prompt = f"""
        Réponds à cette question en utilisant uniquement les informations du document.

        Document : {content[:2000]}

        Question : {question}

        Réponse concise et factuelle :
        """

        return await self.llm.generate(prompt, temperature=0)

Validation humaine

DEVELOPERpython
class HumanValidation:
    def __init__(self, db):
        self.db = db

    async def create_validation_batch(
        self,
        eval_samples: list[dict],
        annotators: list[str]
    ) -> str:
        """
        Crée un batch de validation pour les annotateurs
        """
        batch_id = str(uuid.uuid4())

        for sample in eval_samples:
            await self.db.insert("validation_tasks", {
                "batch_id": batch_id,
                "sample_id": sample["id"],
                "question": sample["question"],
                "rag_answer": sample["answer"],
                "context": sample["context"],
                "status": "pending",
                "assigned_to": None
            })

        # Assigner aux annotateurs
        await self._assign_tasks(batch_id, annotators)

        return batch_id

    async def collect_annotations(self, batch_id: str) -> dict:
        """
        Collecte les annotations et calcule l'accord inter-annotateurs
        """
        tasks = await self.db.find("validation_tasks", {"batch_id": batch_id})

        # Calculer le Cohen's Kappa pour l'accord
        annotations = {}
        for task in tasks:
            sample_id = task["sample_id"]
            if sample_id not in annotations:
                annotations[sample_id] = []
            if task.get("annotation"):
                annotations[sample_id].append(task["annotation"])

        # Vérifier l'accord
        agreement_scores = []
        for sample_id, annots in annotations.items():
            if len(annots) >= 2:
                agreement = self._calculate_agreement(annots)
                agreement_scores.append(agreement)

        return {
            "batch_id": batch_id,
            "total_samples": len(eval_samples),
            "completed": len([a for a in annotations.values() if len(a) >= 2]),
            "average_agreement": sum(agreement_scores) / len(agreement_scores) if agreement_scores else 0
        }

Pipeline d'évaluation automatisée

CI/CD intégration

DEVELOPERpython
import json
from datetime import datetime

class RAGEvaluationPipeline:
    def __init__(self, rag_system, evaluator, dataset_path: str):
        self.rag = rag_system
        self.evaluator = evaluator
        self.dataset = self._load_dataset(dataset_path)

    def _load_dataset(self, path: str) -> list[dict]:
        with open(path) as f:
            return json.load(f)

    async def run_evaluation(self, version: str = None) -> dict:
        """
        Exécute une évaluation complète
        """
        results = {
            "version": version or datetime.now().isoformat(),
            "timestamp": datetime.now().isoformat(),
            "samples": [],
            "metrics": {}
        }

        # Évaluer chaque sample
        for sample in self.dataset:
            # Exécuter le RAG
            rag_result = await self.rag.query(sample["question"])

            # Évaluer
            sample_result = {
                "question": sample["question"],
                "ground_truth": sample["ground_truth"],
                "rag_answer": rag_result["answer"],
                "retrieved_docs": rag_result["sources"],
                "scores": {}
            }

            # Faithfulness
            faithfulness = await self.evaluator.faithfulness(
                sample["question"],
                rag_result["context"],
                rag_result["answer"]
            )
            sample_result["scores"]["faithfulness"] = faithfulness["score"]

            # Relevancy
            relevancy = await self.evaluator.relevancy(
                sample["question"],
                rag_result["answer"]
            )
            sample_result["scores"]["relevancy"] = relevancy["score"]

            results["samples"].append(sample_result)

        # Agréger les métriques
        results["metrics"] = self._aggregate_metrics(results["samples"])

        return results

    def _aggregate_metrics(self, samples: list[dict]) -> dict:
        """
        Agrège les métriques sur tous les samples
        """
        metrics = {}

        for metric_name in ["faithfulness", "relevancy"]:
            scores = [s["scores"].get(metric_name, 0) for s in samples]
            metrics[metric_name] = {
                "mean": sum(scores) / len(scores),
                "min": min(scores),
                "max": max(scores),
                "std": np.std(scores)
            }

        return metrics

    async def check_thresholds(self, results: dict, thresholds: dict) -> bool:
        """
        Vérifie si les résultats passent les seuils définis
        """
        passed = True

        for metric, threshold in thresholds.items():
            actual = results["metrics"].get(metric, {}).get("mean", 0)
            if actual < threshold:
                print(f"FAIL: {metric} = {actual:.3f} < {threshold}")
                passed = False
            else:
                print(f"PASS: {metric} = {actual:.3f} >= {threshold}")

        return passed

Configuration GitHub Actions

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

on:
  pull_request:
    paths:
      - 'rag/**'
      - 'prompts/**'
  schedule:
    - cron: '0 6 * * *'  # Daily at 6am

jobs:
  evaluate:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

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

      - name: Install dependencies
        run: pip install -r requirements-eval.txt

      - name: Run evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: python scripts/run_evaluation.py

      - name: Check thresholds
        run: python scripts/check_thresholds.py --results eval_results.json

      - name: Upload results
        uses: actions/upload-artifact@v3
        with:
          name: eval-results
          path: eval_results.json

      - name: Comment PR
        if: github.event_name == 'pull_request'
        uses: actions/github-script@v6
        with:
          script: |
            const results = require('./eval_results.json');
            const body = `## RAG Evaluation Results

            | Metric | Score | Threshold | Status |
            |--------|-------|-----------|--------|
            | Faithfulness | ${results.metrics.faithfulness.mean.toFixed(3)} | 0.80 | ${results.metrics.faithfulness.mean >= 0.8 ? '✅' : '❌'} |
            | Relevancy | ${results.metrics.relevancy.mean.toFixed(3)} | 0.75 | ${results.metrics.relevancy.mean >= 0.75 ? '✅' : '❌'} |
            `;

            github.rest.issues.createComment({
              owner: context.repo.owner,
              repo: context.repo.repo,
              issue_number: context.issue.number,
              body: body
            });

Monitoring en production

DEVELOPERpython
class ProductionMonitor:
    def __init__(self, analytics_db, alert_service):
        self.db = analytics_db
        self.alerts = alert_service

    async def log_interaction(self, interaction: dict):
        """
        Log une interaction RAG pour monitoring
        """
        await self.db.insert("rag_interactions", {
            "timestamp": datetime.now(),
            "query": interaction["query"],
            "answer": interaction["answer"],
            "sources": interaction["sources"],
            "latency_ms": interaction["latency_ms"],
            "user_id": interaction.get("user_id"),
            "session_id": interaction.get("session_id")
        })

        # Vérifier les alertes
        await self._check_alerts(interaction)

    async def _check_alerts(self, interaction: dict):
        """
        Vérifie les conditions d'alerte
        """
        # Latence élevée
        if interaction["latency_ms"] > 5000:
            await self.alerts.send("HIGH_LATENCY", {
                "latency_ms": interaction["latency_ms"],
                "query": interaction["query"][:100]
            })

        # Pas de sources trouvées
        if not interaction["sources"]:
            await self.alerts.send("NO_SOURCES", {
                "query": interaction["query"]
            })

    async def get_daily_metrics(self) -> dict:
        """
        Métriques quotidiennes pour dashboard
        """
        today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)

        return {
            "total_queries": await self.db.count(
                "rag_interactions",
                {"timestamp": {"$gte": today}}
            ),
            "avg_latency_ms": await self.db.avg(
                "rag_interactions",
                "latency_ms",
                {"timestamp": {"$gte": today}}
            ),
            "zero_result_rate": await self._zero_result_rate(today),
            "unique_users": await self.db.distinct_count(
                "rag_interactions",
                "user_id",
                {"timestamp": {"$gte": today}}
            )
        }

Checklist d'évaluation

Avant le déploiement

• Dataset d'évaluation de 100+ questions représentatives
• Ground truth validé par des experts métier
• Seuils définis pour chaque métrique
• Pipeline CI/CD configuré

En continu

• Monitoring des métriques de latence
• Alertes sur les dégradations
• Feedback utilisateur collecté et analysé
• Évaluations hebdomadaires sur nouveaux cas

Pour aller plus loin

• Introduction au RAG - Comprendre les fondamentaux
• Fondamentaux du Retrieval - Améliorer la recherche
• Génération RAG - Optimiser les réponses

Évaluation simplifiée avec Ailog

Mettre en place un pipeline d'évaluation RAG robuste demande du temps et de l'expertise. Avec Ailog, bénéficiez d'outils d'évaluation intégrés :

• Dashboard qualité avec métriques temps réel
• Évaluation automatique sur chaque déploiement
• Feedback loop intégré pour amélioration continue
• Alertes sur dégradation des performances
• Historique des évaluations pour tracking long terme

Découvrez Ailog et mesurez la qualité de votre RAG facilement.