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)

# Beispiel
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

# Beispiel
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

# Beispiel: 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

# Beispiel 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"""
        Extrahiere alle sachlichen Behauptungen aus dieser Antwort.
        Jede Behauptung soll ein einfacher, überprüfbarer Satz sein.

        Antwort: {answer}

        Behauptungen (eine pro Zeile):
        """

        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"""
        Wird die folgende Behauptung ausdrücklich durch den Kontext gestützt?

        Kontext: {context}

        Behauptung: {claim}

        Antworte nur mit "JA" oder "NEIN".
        """

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

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"""
        Generiere {n} verschiedene Fragen, die diese Antwort beantworten könnte.

        Antwort: {answer}

        Fragen (eine pro Zeile):
        """

        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

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

# Evaluationsdaten vorbereiten
eval_data = {
    "question": [
        "Wie lautet die Rückgaberichtlinie?",
        "Wie kann ich den Support kontaktieren?",
    ],
    "answer": [
        "Sie haben 30 Tage Zeit, um ein ungenutztes Produkt zurückzugeben.",
        "Sie können den Support per E-Mail an [email protected] kontaktieren.",
    ],
    "contexts": [
        ["Die Rückgaberichtlinie erlaubt Kunden, jedes ungenutzte Produkt innerhalb von 30 Tagen zurückzugeben."],
        ["Der Support ist per E-Mail an [email protected] oder telefonisch unter 01 23 45 67 89 erreichbar."],
    ],
    "ground_truth": [
        "Kunden können ungenutzte Produkte innerhalb von 30 Tagen zurückgeben.",
        "Der Support ist per E-Mail ([email protected]) und Telefon erreichbar.",
    ]
}

dataset = Dataset.from_dict(eval_data)

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

print(results)

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:
            # Fragen generieren
            questions = await self._generate_questions(doc["content"], questions_per_doc)

            for q in questions:
                # Erwartete Antwort generieren
                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]:
        """
        Generiert Fragen ausgehend vom Inhalt
        """
        prompt = f"""
        Generiere {n} vielfältige und realistische Fragen, die ein Nutzer stellen könnte
        und die dieses Dokument beantworten kann.

        Dokument:
        {content[:2000]}

        Fragen (eine pro Zeile, unterschiedliche Komplexität):
        """

        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:
        """
        Generiert die erwartete Antwort basierend auf dem Dokument
        """
        prompt = f"""
        Beantworte diese Frage ausschließlich anhand der Informationen im Dokument.

        Dokument: {content[:2000]}

        Frage: {question}

        Prägnante und sachliche Antwort:
        """

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

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

        # Aufgaben den Annotatoren zuweisen
        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})

        # Cohen's Kappa zur Übereinstimmung berechnen
        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"])

        # Übereinstimmung prüfen
        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
        }

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": {}
        }

        # Jedes Sample evaluieren
        for sample in self.dataset:
            # RAG ausführen
            rag_result = await self.rag.query(sample["question"])

            # Evaluieren
            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)

        # Metriken aggregieren
        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
            });

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

        # Alerts prüfen
        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.