Bewertung von RAG-Systemen: Metriken und Methoden

Warum Evaluierung wichtig ist

Ohne Messung können Sie nicht:

• Wissen, ob Änderungen die Leistung verbessern
• Fehlerarten identifizieren
• Hyperparameter optimieren
• Kosten gegenüber Stakeholdern rechtfertigen
• SLA-Qualitätsanforderungen erfüllen

Kernaussage : RAG hat mehrere Komponenten (retrieval, generation), die jeweils eine Bewertung benötigen.

Bewertungsebenen

Auf Komponenten-Ebene

Bewerten Sie die einzelnen Teile:

• Qualität des Retrieval
• Qualität der Generation
• Effizienz des chunking

Ende-zu-Ende

Bewerten Sie die gesamte Pipeline:

• Genauigkeit der Antwort
• Benutzerzufriedenheit
• Erfolgsrate bei der Aufgabenerfüllung

Beides ist notwendig

Komponentenmetriken diagnostizieren Probleme. Ende-zu-Ende-Metriken messen die Business-Auswirkung.

Retrieval-Metriken

Precision@k

Anteil der abgerufenen Dokumente, die relevant sind.

DEVELOPERpython
def precision_at_k(retrieved, relevant, k):
    """
    retrieved: List of retrieved document IDs
    relevant: Set of relevant document IDs
    k: Number of top results to consider
    """
    top_k = set(retrieved[:k])
    relevant_retrieved = top_k & relevant

    return len(relevant_retrieved) / k if k > 0 else 0

Beispiel :

Retrieved top 5: [doc1, doc2, doc3, doc4, doc5]
Relevant: {doc1, doc3, doc8}

Precision@5 = 2/5 = 0.4

Interpretation :

• Höher ist besser
• Misst Präzision
• Berücksichtigt Recall nicht

Recall@k

Anteil der relevanten Dokumente, die abgerufen wurden.

DEVELOPERpython
def recall_at_k(retrieved, relevant, k):
    """
    What fraction of relevant docs did we find?
    """
    top_k = set(retrieved[:k])
    relevant_retrieved = top_k & relevant

    return len(relevant_retrieved) / len(relevant) if relevant else 0

Beispiel :

Retrieved top 5: [doc1, doc2, doc3, doc4, doc5]
Relevant: {doc1, doc3, doc8}

Recall@5 = 2/3 ≈ 0.67

Interpretation :

• Höher ist besser
• Misst Abdeckung
• Schwerer zu optimieren als Precision

F1@k

Harmonisches Mittel von Precision und Recall.

DEVELOPERpython
def f1_at_k(retrieved, relevant, k):
    p = precision_at_k(retrieved, relevant, k)
    r = recall_at_k(retrieved, relevant, k)

    if p + r == 0:
        return 0

    return 2 * (p * r) / (p + r)

Zu verwenden wenn :

• Ein Ausgleich zwischen Precision und Recall erforderlich ist
• Eine einzige Metrik für Optimierung gewünscht ist

Mean Reciprocal Rank (MRR)

Durchschnitt der reziproken Ränge des ersten relevanten Ergebnisses.

DEVELOPERpython
def reciprocal_rank(retrieved, relevant):
    """
    Rank of first relevant document
    """
    for i, doc_id in enumerate(retrieved, 1):
        if doc_id in relevant:
            return 1 / i
    return 0

def mrr(queries_results, queries_relevant):
    """
    Average across multiple queries
    """
    rr_scores = [
        reciprocal_rank(retrieved, relevant)
        for retrieved, relevant in zip(queries_results, queries_relevant)
    ]

    return sum(rr_scores) / len(rr_scores)

Beispiel :

Query 1: First relevant at position 2 → RR = 1/2 = 0.5
Query 2: First relevant at position 1 → RR = 1/1 = 1.0
Query 3: First relevant at position 5 → RR = 1/5 = 0.2

MRR = (0.5 + 1.0 + 0.2) / 3 = 0.57

Interpretation :

• Legt den Fokus auf die Qualität des Rankings
• Betrachtet nur das erste relevante Ergebnis
• Gut für Frage-Antwort-Szenarien

NDCG@k (Normalized Discounted Cumulative Gain)

Berücksichtigt graduelle Relevanz und Position.

DEVELOPERpython
import numpy as np
from sklearn.metrics import ndcg_score

def calculate_ndcg(retrieved, relevance_scores, k):
    """
    relevance_scores: Dict mapping doc_id to relevance (0-3 typical)
    """
    # Get scores for retrieved docs
    scores = [relevance_scores.get(doc_id, 0) for doc_id in retrieved[:k]]

    # Calculate ideal ranking (best possible)
    ideal_scores = sorted(relevance_scores.values(), reverse=True)[:k]

    # NDCG
    return ndcg_score([ideal_scores], [scores])

Beispiel :

Retrieved: [doc1, doc2, doc3]
Scores:    [2,    3,    1]     (votre système)
Ideal:     [3,    2,    1]     (classement parfait)

NDCG calcule à quel point vous êtes proche de l'idéal

Zu verwenden wenn :

• Mehrere Relevanzstufen vorhanden sind (nicht nur binär)
• Die Position wichtig ist (erstes Ergebnis ist wichtiger)
• Unternehmens-/Enterprise-Suche

Hit Rate@k

Haben wir mindestens ein relevantes Dokument abgerufen?

DEVELOPERpython
def hit_rate_at_k(retrieved, relevant, k):
    top_k = set(retrieved[:k])
    return 1 if len(top_k & relevant) > 0 else 0

Zu verwenden für :

• Minimale Brauchbarkeit (haben wir etwas Nützliches erhalten?)
• Aggregation über Anfragen hinweg für die Erfolgsquote

Generation-Metriken

Treue / Verankerung

Ist die Antwort durch den retrieved context gestützt?

LLM-as-Judge :

DEVELOPERpython
def evaluate_faithfulness(answer, context, llm):
    prompt = f"""Cette réponse est-elle fidèle au contexte ? Réponds seulement oui ou non.

Contexte: {context}

Réponse: {answer}

La réponse est-elle soutenue par le contexte ?"""

    response = llm.generate(prompt, max_tokens=5)
    return 1 if 'yes' in response.lower() else 0

Warum das wichtig ist :

• Erkennt Halluzinationen
• Stellt sicher, dass Antworten in Fakten verankert sind
• Kritisch für stark regulierte / risikobehaftete Anwendungen

Relevanz der Antwort

Adressiert die Antwort die Frage?

DEVELOPERpython
def evaluate_relevance(question, answer, llm):
    prompt = f"""Cette réponse aborde-t-elle la question ? Note de 1-5.

Question: {question}

Réponse: {answer}

Pertinence (1-5):"""

    score = int(llm.generate(prompt, max_tokens=5))
    return score / 5  # Normaliser à 0-1

Präzision des Kontexts

Ist der abgerufene Kontext relevant?

DEVELOPERpython
def context_precision(retrieved_chunks, question, llm):
    """
    Are the retrieved chunks relevant to the question?
    """
    relevant_count = 0

    for chunk in retrieved_chunks:
        prompt = f"""Ce contexte est-il pertinent pour la question ?

Question: {question}

Contexte: {chunk}

Pertinent ? (yes/no)"""

        response = llm.generate(prompt, max_tokens=5)
        if 'yes' in response.lower():
            relevant_count += 1

    return relevant_count / len(retrieved_chunks)

Recall des Kontexts

Sind alle notwendigen Informationen im abgerufenen Kontext vorhanden?

DEVELOPERpython
def context_recall(ground_truth_answer, retrieved_context, llm):
    """
    Does the context contain all info needed for the ground truth answer?
    """
    prompt = f"""Cette réponse peut-elle être dérivée du contexte ?

Contexte: {retrieved_context}

Réponse: {ground_truth_answer}

Toutes les informations sont-elles présentes ? (yes/no)"""

    response = llm.generate(prompt, max_tokens=5)
    return 1 if 'yes' in response.lower() else 0

Frameworks für automatisierte Bewertung

RAGAS

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

# Prepare dataset
dataset = {
    'question': [q1, q2, q3],
    'answer': [a1, a2, a3],
    'contexts': [c1, c2, c3],  # List of retrieved chunks
    'ground_truth': [gt1, gt2, gt3]
}

# Evaluate
result = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall
    ]
)

print(result)
# {
#   'faithfulness': 0.92,
#   'answer_relevancy': 0.87,
#   'context_precision': 0.81,
#   'context_recall': 0.89
# }

TruLens

DEVELOPERpython
from trulens_eval import TruChain, Feedback, Tru

# Initialize
tru = Tru()

# Define feedback functions
f_groundedness = Feedback(groundedness_llm).on_output()
f_answer_relevance = Feedback(answer_relevance_llm).on_input_output()
f_context_relevance = Feedback(context_relevance_llm).on_input()

# Wrap RAG chain
tru_rag = TruChain(
    rag_chain,
    app_id='my_rag_v1',
    feedbacks=[f_groundedness, f_answer_relevance, f_context_relevance]
)

# Use normally - metrics auto-collected
result = tru_rag.run(query)

# View dashboard
tru.run_dashboard()

DeepEval

DEVELOPERpython
from deepeval import evaluate
from deepeval.metrics import HallucinationMetric, AnswerRelevancyMetric
from deepeval.test_case import LLMTestCase

# Create test case
test_case = LLMTestCase(
    input="What is the capital of France?",
    actual_output="The capital of France is Paris.",
    context=["France is a country in Europe.", "Paris is the capital of France."]
)

# Define metrics
metrics = [
    HallucinationMetric(threshold=0.9),
    AnswerRelevancyMetric(threshold=0.8)
]

# Evaluate
evaluate([test_case], metrics)

Test-Set erstellen

Manuelle Kuration

DEVELOPERpython
test_cases = [
    {
        'query': 'How do I reset my password?',
        'ground_truth_answer': 'Click "Forgot Password" on the login page...',
        'relevant_docs': {'doc_123', 'doc_456'},
        'difficulty': 'easy'
    },
    {
        'query': 'What are the differences between plans?',
        'ground_truth_answer': 'Premium includes...',
        'relevant_docs': {'doc_789'},
        'difficulty': 'medium'
    },
    # ... more test cases
]

Best Practices :

• Vielfältige Anfragetypen (einfach, komplex, mehrdeutig)
• Unterschiedliche Schwierigkeitsgrade
• Echte Nutzeranfragen
• Randfälle
• Mindestens 50–100 Testfälle

Synthetische Generierung

DEVELOPERpython
def generate_test_cases(documents, llm, num_cases=50):
    test_cases = []

    for doc in random.sample(documents, num_cases):
        prompt = f"""Génère une question à laquelle on peut répondre en utilisant ce document.

Document: {doc}

Question:"""

        question = llm.generate(prompt)

        prompt_answer = f"""Réponds à cette question en utilisant le document.

Document: {doc}

Question: {question}

Réponse:"""

        answer = llm.generate(prompt_answer)

        test_cases.append({
            'query': question,
            'ground_truth_answer': answer,
            'relevant_docs': {doc['id']},
            'source': 'synthetic'
        })

    return test_cases

Extraktion von Nutzeranfragen

DEVELOPERpython
# Extract from logs
def extract_queries_from_logs(log_file, sample_size=100):
    # Parse logs
    queries = parse_query_logs(log_file)

    # Filter for quality
    queries = [q for q in queries if len(q.split()) >= 3]  # Not too short

    # Sample diverse queries
    return random.sample(queries, sample_size)

A/B-Tests

Experiment-Konfiguration

DEVELOPERpython
class ABTest:
    def __init__(self, control_system, treatment_system):
        self.control = control_system
        self.treatment = treatment_system
        self.results = {'control': [], 'treatment': []}

    def run_query(self, query, user_id):
        # Assign to variant (50/50 split)
        variant = 'treatment' if hash(user_id) % 2 else 'control'
        system = self.treatment if variant == 'treatment' else self.control

        # Get answer
        answer = system.query(query)

        # Log result
        self.results[variant].append({
            'query': query,
            'answer': answer,
            'timestamp': time.time()
        })

        return answer, variant

    def analyze(self):
        # Compare metrics between variants
        control_metrics = calculate_metrics(self.results['control'])
        treatment_metrics = calculate_metrics(self.results['treatment'])

        return {
            'control': control_metrics,
            'treatment': treatment_metrics,
            'lift': calculate_lift(control_metrics, treatment_metrics)
        }

Zu verfolgende Metriken

Qualität :

• Genauigkeit der Antwort
• Nutzerbewertungen (Daumen hoch/runter)
• Rate Folgefragen

Engagement :

• Sitzungsdauer
• Anfragen pro Sitzung
• Erfolgsquote bei Aufgaben

Business :

• Conversion-Rate
• Support-Ticket-Reduktion (Deflection)
• Kundenzufriedenheit (CSAT)

Kontinuierliche Bewertung

Monitoring-Pipeline

DEVELOPERpython
class RAGMonitor:
    def __init__(self, rag_system, test_set):
        self.system = rag_system
        self.test_set = test_set
        self.history = []

    def run_evaluation(self):
        results = []

        for test_case in self.test_set:
            # Run RAG
            answer, contexts = self.system.query(test_case['query'])

            # Calculate metrics
            metrics = {
                'precision@5': precision_at_k(contexts, test_case['relevant_docs'], 5),
                'faithfulness': evaluate_faithfulness(answer, contexts),
                'relevance': evaluate_relevance(test_case['query'], answer)
            }

            results.append(metrics)

        # Aggregate
        aggregated = aggregate_metrics(results)

        # Save history
        self.history.append({
            'timestamp': time.time(),
            'metrics': aggregated
        })

        # Alert if degradation
        if self.detect_degradation(aggregated):
            self.send_alert(aggregated)

        return aggregated

    def detect_degradation(self, current_metrics, threshold=0.05):
        if not self.history:
            return False

        previous = self.history[-1]['metrics']

        for metric, value in current_metrics.items():
            if value < previous[metric] - threshold:
                return True

        return False

Geplante Evaluation

DEVELOPERpython
import schedule

def daily_evaluation():
    monitor = RAGMonitor(rag_system, test_set)
    results = monitor.run_evaluation()

    # Log to monitoring system
    metrics_logger.log(results)

    # Update dashboard
    update_dashboard(results)

# Run daily at 2 AM
schedule.every().day.at("02:00").do(daily_evaluation)

while True:
    schedule.run_pending()
    time.sleep(60)

Menschliche Bewertung

Bewertungsoberfläche

DEVELOPERpython
def collect_human_ratings(test_cases, rag_system):
    ratings = []

    for test_case in test_cases:
        # Generate answer
        answer, contexts = rag_system.query(test_case['query'])

        # Show to human rater
        print(f"Query: {test_case['query']}")
        print(f"Answer: {answer}")
        print(f"Contexts: {contexts}")

        # Collect ratings
        correctness = int(input("Correctness (1-5): "))
        completeness = int(input("Completeness (1-5): "))
        conciseness = int(input("Conciseness (1-5): "))

        ratings.append({
            'query': test_case['query'],
            'correctness': correctness,
            'completeness': completeness,
            'conciseness': conciseness
        })

    return ratings

Inter-Rater-Reliabilität

DEVELOPERpython
from sklearn.metrics import cohen_kappa_score

def calculate_agreement(rater1_scores, rater2_scores):
    """
    Cohen's Kappa for inter-rater agreement
    """
    kappa = cohen_kappa_score(rater1_scores, rater2_scores)

    if kappa > 0.8:
        return "Strong agreement"
    elif kappa > 0.6:
        return "Moderate agreement"
    else:
        return "Weak agreement - review rating criteria"

Kosten der Bewertung

Kosten für LLM-basierte Metriken

DEVELOPERpython
def estimate_evaluation_cost(num_test_cases, metrics_per_case=3):
    # GPT-4 pricing (example)
    cost_per_1k_tokens = 0.03  # Input
    tokens_per_evaluation = 500  # Typical

    total_evaluations = num_test_cases * metrics_per_case
    total_tokens = total_evaluations * tokens_per_evaluation

    cost = (total_tokens / 1000) * cost_per_1k_tokens

    return cost

# Example
cost = estimate_evaluation_cost(100)  # 4,50 $ pour 100 cas de test

Optimierung

• Ergebnisse cachen für unveränderte Ausgaben
• Kleinere Modelle verwenden (GPT-3.5 vs GPT-4) für bestimmte Metriken
• Batch-Auswertungen
• Seltener ausführen (täglich statt bei jedem PR)

Best Practices

1. Diverses Test-Set : Alle Anfragetypen und Schwierigkeitsgrade abdecken
2. Im Zeitverlauf verfolgen : Metriken überwachen, während sich das System entwickelt
3. Komponenten + E2E : Sowohl einzelne Teile als auch das Gesamtsystem bewerten
4. Echte Anfragen : Echte Nutzeranfragen in das Test-Set aufnehmen
5. Automatisieren : Bewertung bei jeder Änderung ausführen
6. Menschliche Validierung : Periodische manuelle Überprüfung automatischer Metriken
7. Business-Metriken : Qualität mit Geschäftsergebnissen verknüpfen

Nächste Schritte

Mit der Evaluierung in place verlagert sich der Fokus auf das Deployment von RAG-Systemen in Produktion. Der nächste Leitfaden behandelt Produktions-Deployment, Skalierung, Monitoring und operative Überlegungen.