Ein RAG-System bewerten : Metriken und Methodiken

Ein RAG-System kann auf den ersten Blick korrekt funktionieren — wie kann man jedoch sicher sein, dass es wirklich den Erwartungen entspricht? Sorgfältige Evaluation ist der Schlüssel, um von einem Prototyp zu einem produktionsreifen System zu gelangen. Dieser Leitfaden stellt Ihnen Metriken, Frameworks und Methodiken vor, um Ihr RAG zu messen und zu verbessern.

Warum Evaluation kritisch ist

Das Problem subjektiver Bewertungen

Ohne objektive Metriken reduziert sich die Evaluation eines RAG oft auf:

• „Sieht gut aus“ (Confirmation Bias)
• Einige manuelle Tests mit günstigen Fällen
• Nutzerfeedback erst spät in Produktion

Dieser Ansatz verdeckt schwerwiegende Probleme:

• Subtile, aber häufige Halluzinationen
• Antworten, die für bestimmte Fragekategorien am Thema vorbeigehen
• Progressive Verschlechterung der Qualität nach Updates

Die Dimensionen der Evaluation

Ein RAG-System sollte in mehreren Dimensionen bewertet werden:

Metriken für retrieval

Recall@k

Misst den Anteil der relevanten Dokumente, die unter den k ersten Ergebnissen gefunden wurden.

DEVELOPERpython
def recall_at_k(retrieved_ids: list[str], relevant_ids: list[str], k: int) -> float:
    """
    Recall@k : Anteil der abgerufenen relevanten Dokumente

    Args:
        retrieved_ids: IDs der abgerufenen Dokumente (nach Score geordnet)
        relevant_ids: IDs der tatsächlich relevanten Dokumente
        k: Anzahl der zu betrachtenden Ergebnisse

    Returns:
        Wert zwischen 0 und 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)

Interpretation :

• Recall@5 von 0.8+ : Hervorragend, die meisten relevanten Dokumente werden gefunden
• Recall@5 von 0.5-0.8 : Akzeptabel, aber Verbesserung möglich
• Recall@5 < 0.5 : Problematisch, viele relevante Dokumente fehlen

Precision@k

Misst den Anteil relevanter Dokumente unter den k abgerufenen.

DEVELOPERpython
def precision_at_k(retrieved_ids: list[str], relevant_ids: list[str], k: int) -> float:
    """
    Precision@k : Anteil der abgerufenen Dokumente, die relevant sind
    """
    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)

Misst die durchschnittliche Position des ersten relevanten Dokuments.

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

    Args:
        queries_results: Liste von (abgerufene_dokumente, relevante_dokumente)

    Returns:
        MRR-Wert zwischen 0 und 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 Anfragen
results = [
    (["doc1", "doc2", "doc3"], ["doc1"]),     # Erstes = relevant -> 1/1
    (["doc4", "doc1", "doc2"], ["doc1"]),     # Zweites = relevant -> 1/2
    (["doc5", "doc6", "doc7"], ["doc8"]),     # Kein relevantes -> 0
]
print(f"MRR: {mrr(results)}")  # (1 + 0.5 + 0) / 3 = 0.5

NDCG (Normalized Discounted Cumulative Gain)

Berücksichtigt sowohl die Reihenfolge ALS AUCH abgestufte Relevanzscores.

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

    # Logarithmus Basis 2, Position 1-indexiert
    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 : Vergleich mit dem idealen DCG
    """
    dcg = dcg_at_k(relevances, k)

    # Ideales DCG : Relevanzen absteigend sortiert
    ideal_relevances = sorted(relevances, reverse=True)
    idcg = dcg_at_k(ideal_relevances, k)

    return dcg / idcg if idcg > 0 else 0.0

# Exemple mit abgestuften Relevanzscores (0=nicht relevant, 1=gering, 2=sehr)
relevances = [2, 0, 1, 1, 0]  # Dokument 1 sehr relevant, doc 2 nicht, usw.
print(f"NDCG@5: {ndcg_at_k(relevances, 5):.3f}")

Metriken für die Generation

Faithfulness (Treue)

Misst, ob die Antwort dem bereitgestellten Kontext treu ist, ohne zu halluzinieren.

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

    async def evaluate(
        self,
        question: str,
        context: str,
        answer: str
    ) -> dict:
        """
        Bewertet die Treue der Antwort zum Kontext
        """
        # Schritt 1 : Extrahiere die Behauptungen aus der Antwort
        claims = await self._extract_claims(answer)

        # Schritt 2 : Überprüfe jede Behauptung gegen den Kontext
        verification_results = []
        for claim in claims:
            is_supported = await self._verify_claim(claim, context)
            verification_results.append({
                "claim": claim,
                "supported": is_supported
            })

        # Score berechnen
        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]:
        """
        Extrahiert die faktischen Behauptungen aus der Antwort
        """
        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:
        """
        Prüft, ob eine Behauptung durch den Kontext gestützt wird
        """
        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"

Hinweis: Die in prompt-Strings enthaltenen Instruktionen wurden im Code belassen.

Answer Relevancy (Relevanz)

Misst, ob die Antwort tatsächlich die gestellte Frage beantwortet.

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:
        """
        Bewertet die Relevanz der Antwort in Bezug auf die Frage
        """
        # Methode 1 : Generiere Fragen aus der Antwort
        generated_questions = await self._generate_questions(answer)

        # Methode 2 : Berechne die Similarität mit der Originalfrage
        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]:
        """
        Generiert Fragen, die durch die Antwort beantwortet werden könnten
        """
        prompt = f"""
        Génère {n} questions différentes auxqu'elles 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))

Beachte: Prompt-Strings wurden unverändert gelassen.

Context Recall

Misst, ob der abgerufene Kontext die notwendigen Informationen zur Beantwortung enthält.

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

    async def evaluate(
        self,
        question: str,
        context: str,
        ground_truth: str
    ) -> dict:
        """
        Bewertet, ob der Kontext die Informationen der erwarteten Antwort enthält
        """
        # Extrahiere die Fakten aus der erwarteten Antwort
        gt_facts = await self._extract_facts(ground_truth)

        # Überprüfe das Vorhandensein jedes Fakts im Kontext
        attributions = []
        for fact in gt_facts:
            present = await self._check_presence(fact, context)
            attributions.append({
                "fact": fact,
                "present_in_context": present
            })

        # Score = Anteil der vorhandenen Fakten
        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) ist das Referenz-Framework für die RAG-Evaluation.

Installation und Konfiguration

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)

Hinweis: Die Beispieldaten im Codeblock wurden unverändert belassen.

Interpretation der RAGAS-Ergebnisse

Erstellung eines Evaluations-Datasets

Automatische Generierung von Fragen

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]:
        """
        Generiert ein Evaluations-Dataset aus Dokumenten
        """
        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)

Hinweis: Prompt-Strings im Code wurden unverändert belassen.

Menschliche Validierung

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

    async def create_validation_batch(
        self,
        eval_samples: list[dict],
        annotators: list[str]
    ) -> str:
        """
        Erstellt einen Validierungs-Batch für Annotatoren
        """
        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:
        """
        Sammelt die Annotationen und berechnet die Inter-Annotator-Agreement
        """
        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
        }

Kommentare und Strings im Code wurden entsprechend den Regeln belassen oder übersetzt.

Automatisierte Evaluations-Pipeline

CI/CD-Integration

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:
        """
        Führt eine vollständige Evaluation aus
        """
        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:
        """
        Aggregiert die Metriken über alle 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:
        """
        Prüft, ob die Ergebnisse die definierten Schwellenwerte erreichen
        """
        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

GitHub Actions Konfiguration

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

on:
  pull_request:
    paths:
      - 'rag/**'
      - 'prompts/**'
  schedule:
    - cron: '0 6 * * *'  # Täglich um 6 Uhr

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
            });

Hinweis: Kommentarinhalte in Skripten wurden nicht verändert.

Monitoring in Produktion

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):
        """
        Loggt eine RAG-Interaktion für das 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):
        """
        Prüft Alert-Bedingungen
        """
        # Hohe Latenz
        if interaction["latency_ms"] > 5000:
            await self.alerts.send("HIGH_LATENCY", {
                "latency_ms": interaction["latency_ms"],
                "query": interaction["query"][:100]
            })

        # Keine Quellen gefunden
        if not interaction["sources"]:
            await self.alerts.send("NO_SOURCES", {
                "query": interaction["query"]
            })

    async def get_daily_metrics(self) -> dict:
        """
        Tägliche Metriken für das 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}}
            )
        }

Evaluations-Checkliste

Vor dem Deployment

• Evaluations-Dataset mit 100+ repräsentativen Fragen
• Ground truth von Fachexperten validiert
• Schwellenwerte für jede Metrik definiert
• CI/CD-Pipeline konfiguriert

Kontinuierlich

• Monitoring der Latenzmetriken
• Alerts bei Verschlechterungen
• Nutzerfeedback gesammelt und analysiert
• Wöchentliche Evaluierungen für neue Fälle

Für weiterführende Informationen

• Introduction au RAG - Grundlagen verstehen
• Fondamentaux du Retrieval - Retrieval verbessern
• Génération RAG - Antworten optimieren

Vereinfachte Evaluation mit Ailog

Eine robuste Evaluations-Pipeline für RAG aufzubauen erfordert Zeit und Expertise. Mit Ailog erhalten Sie integrierte Evaluations-Tools:

• Dashboard qualité mit Echtzeitmetriken
• Évaluation automatique bei jedem Deployment
• Feedback loop integriert für kontinuierliche Verbesserung
• Alertes bei Performance-Verschlechterungen
• Historique der Evaluierungen für langfristiges Tracking

Découvrez Ailog und messen Sie die Qualität Ihres RAG einfach.