localRAG/eval_retrieval.py

#!/usr/bin/env python3
"""Bewertet die Retrieval-Qualität gegen einen Satz Referenzfragen. Kein LLM beteiligt.

Das Skript stellt jede Frage aus eval/goldset.csv an den Vektorspeicher und prüft,
ob die erwartete Fundstelle unter den ersten k Treffern auftaucht. Ergebnis sind
Recall@k und MRR, ausgegeben auf dem Bildschirm und in eval/baseline.txt.

Aus dem Projektordner starten:  python eval_retrieval.py

Fragen, Index und Bericht lassen sich auf der Kommandozeile überschreiben, z. B.
    python eval_retrieval.py --goldset eval/goldset_kafka.csv --index ./vectorstore_docling \
        --report eval/kafka_docling.txt
"""
import argparse
import csv
import sys
import unicodedata
from datetime import datetime
from pathlib import Path

from local_rag import (
    CHUNK_OVERLAP,
    CHUNK_SIZE,
    QUERY_ADDITIONAL_INSTRUCTIONS,
    LocalRAG,
)

# --- Einstellungen -----------------------------------------------------------

# Datei mit den Referenzfragen (Spalten siehe eval/goldset.csv).
GOLDSET_PATH = Path("eval/goldset.csv")

# Wohin der Bericht geschrieben wird. Diese Datei gehört ins Repository.
REPORT_PATH = Path("eval/baseline.txt")

# Muss mit server.py übereinstimmen, sonst wird ein anderer Index gemessen.
VECTORSTORE_PATH = "./vectorstore_docling_hdr"

# Für diese k-Werte wird Recall@k berechnet.
K_VALUES = [5, 10, 20]

# Den Zusatztext aus local_rag.QUERY_ADDITIONAL_INSTRUCTIONS an jede Frage anhängen?
# Einmal mit True und einmal mit False laufen lassen zeigt, ob er tatsächlich hilft.
USE_QUERY_INSTRUCTIONS = False

# PyPDFLoader zählt Seiten ab 0, Menschen ab 1. Bei 1 bedeutet Seite 42 in der
# CSV also die Seite, die im PDF-Betrachter als 42 angezeigt wird.
PAGE_OFFSET = 1

# Ein Chunk kann über einen Seitenumbruch reichen: so viele Seiten Abweichung
# gelten noch als Treffer. 0 verlangt die exakte Seite.
PAGE_TOLERANCE = 1

RETRIEVAL_SETUP = ""   # wird in main() gesetzt und im Bericht vermerkt
INDEX_INFO = {}        # Einstellungen, mit denen der Index gebaut wurde (index_info.json)

# -----------------------------------------------------------------------------


def read_goldset(path):
    """Liest die Referenzfragen. Erwartet die Spalten frage, korpus,
    erwartete_quelle, erwartete_seite, anmerkung (die letzten beiden optional)."""
    if not path.exists():
        sys.exit(
            f"Keine Referenzfragen gefunden: {path}\n"
            "Lege die Datei an (Vorlage liegt im Repository) und trage echte Fragen ein."
        )

    rows = []
    with path.open(encoding="utf-8-sig", newline="") as handle:
        # Deutsches Excel speichert CSV oft mit Semikolon statt Komma.
        first_line = handle.readline()
        delimiter = ";" if first_line.count(";") > first_line.count(",") else ","
        handle.seek(0)

        for line_no, row in enumerate(csv.DictReader(handle, delimiter=delimiter), start=2):
            question = (row.get("frage") or "").strip()
            source = (row.get("erwartete_quelle") or "").strip()
            if not question or not source:
                print(f"  Zeile {line_no} übersprungen: frage oder erwartete_quelle fehlt")
                continue

            page_raw = (row.get("erwartete_seite") or "").strip()
            page = None
            if page_raw:
                try:
                    page = int(page_raw)
                except ValueError:
                    print(f"  Zeile {line_no}: '{page_raw}' ist keine Seitenzahl, wird ignoriert")

            rows.append({
                "question": question,
                "corpus": (row.get("korpus") or "ohne").strip() or "ohne",
                "source": source,
                "page": page,
            })

    if not rows:
        sys.exit(f"{path} enthält keine verwertbaren Zeilen.")
    return rows


def same_file(found, expected):
    """Gleiche Datei? Groß-/Kleinschreibung, Unicode-Form (macOS speichert Umlaute zerlegt)
    und Unterstriche, Punkte oder Leerzeichen am Ende des Namens spielen keine Rolle."""
    def norm(name):
        stem = unicodedata.normalize("NFC", Path(name).stem.lower())
        return stem.rstrip("_. ")
    return norm(found) == norm(expected)


def is_hit(doc, expected_source, expected_page):
    """Passt dieser Treffer zur erwarteten Fundstelle? Dateiname ohne Pfad,
    Groß-/Kleinschreibung egal; Seite nur, wenn eine angegeben wurde."""
    if not same_file(doc.metadata.get("source", ""), expected_source):
        return False

    if expected_page is None:
        return True

    found_page = doc.metadata.get("page")
    if found_page is None:
        return False
    return abs(int(found_page) - (expected_page - PAGE_OFFSET)) <= PAGE_TOLERANCE


def evaluate(rag, goldset, max_k):
    """Stellt jede Frage und merkt sich den Rang des ersten korrekten Treffers."""
    results = []
    for i, item in enumerate(goldset, start=1):
        query = item["question"]
        if USE_QUERY_INSTRUCTIONS:
            query = f"{query}\n\n{QUERY_ADDITIONAL_INSTRUCTIONS}"

        # Dieselbe Suche wie im Chat: Vektorsuche, ggf. Stichwortsuche und Reranking
        docs = [doc for doc, _ in rag.retrieve(query, k=max_k)]

        rank = None
        for position, doc in enumerate(docs, start=1):
            if is_hit(doc, item["source"], item["page"]):
                rank = position
                break

        results.append({**item, "rank": rank})
        marker = f"Rang {rank}" if rank else "nicht gefunden"
        print(f"  [{i}/{len(goldset)}] {marker:<15} {item['question'][:60]}")

    return results


def recall_at(results, k):
    """Anteil der Fragen, deren Fundstelle unter den ersten k Treffern liegt."""
    if not results:
        return 0.0
    found = sum(1 for r in results if r["rank"] is not None and r["rank"] <= k)
    return found / len(results)


def mrr(results):
    """Mean Reciprocal Rank: belohnt Treffer weit oben. 1.0 = immer Platz 1."""
    if not results:
        return 0.0
    return sum(1 / r["rank"] for r in results if r["rank"] is not None) / len(results)


def build_report(results, embedding_model):
    """Baut den Berichtstext, der auf dem Bildschirm und in der Datei landet."""
    line = "=" * 68
    out = [
        line,
        f"Retrieval-Auswertung — {datetime.now():%Y-%m-%d %H:%M}",
        line,
        "",
        "Konfiguration",
        f"  Vektorspeicher      {VECTORSTORE_PATH}",
        f"  Embedding-Modell    {embedding_model}",
        f"  Chunk-Größe         {INDEX_INFO.get('chunk_size', CHUNK_SIZE)} (Overlap {INDEX_INFO.get('chunk_overlap', CHUNK_OVERLAP)})",
        f"  Loader              {INDEX_INFO.get('pdf_loader', '?')}, Kopfzeilen {'an' if INDEX_INFO.get('chunk_headers') else 'aus'}",
        f"  Query-Zusatztext    {'an' if USE_QUERY_INSTRUCTIONS else 'aus'}",
        f"  Suche               {RETRIEVAL_SETUP}",
        f"  Seitentoleranz      ±{PAGE_TOLERANCE}",
        f"  Fragen              {len(results)}",
        "",
        "Ergebnis",
    ]

    for k in K_VALUES:
        value = recall_at(results, k)
        found = sum(1 for r in results if r["rank"] is not None and r["rank"] <= k)
        out.append(f"  Recall@{k:<12} {value:.2f}   ({found} von {len(results)})")
    out.append(f"  MRR{'':<16} {mrr(results):.2f}")

    corpora = sorted({r["corpus"] for r in results})
    if len(corpora) > 1:
        out += ["", "Nach Korpus"]
        smallest_k = min(K_VALUES)
        for corpus in corpora:
            subset = [r for r in results if r["corpus"] == corpus]
            wort = "Frage" if len(subset) == 1 else "Fragen"
            out.append(
                f"  {corpus:<12} Recall@{smallest_k} {recall_at(subset, smallest_k):.2f}"
                f"   MRR {mrr(subset):.2f}   ({len(subset)} {wort})"
            )

    out += ["", "Einzelne Fragen (Rang = Position des ersten korrekten Treffers)"]
    for r in sorted(results, key=lambda r: (r["rank"] is None, r["rank"] or 0)):
        rank = f"{r['rank']:>3}" if r["rank"] is not None else "  –"
        out.append(f"  Rang {rank}  {r['corpus']:<10} {r['question']}")

    nowhere = [r for r in results if r["rank"] is None]
    if nowhere:
        out += [
            "",
            f"{len(nowhere)} Frage(n) ohne Treffer unter den ersten {max(K_VALUES)}.",
            "Prüfen: Steht die Fundstelle wirklich dort? Stimmt der Dateiname?",
            "Ist das Dokument überhaupt im Index (rag.list_documents())?",
        ]

    out.append("")
    return "\n".join(out)


def main():
    global GOLDSET_PATH, VECTORSTORE_PATH, REPORT_PATH, USE_QUERY_INSTRUCTIONS
    parser = argparse.ArgumentParser(description="Retrieval-Auswertung ohne LLM")
    parser.add_argument("--goldset", type=Path, default=GOLDSET_PATH)
    parser.add_argument("--index", default=VECTORSTORE_PATH)
    parser.add_argument("--report", type=Path, default=REPORT_PATH)
    parser.add_argument("--mode", choices=["dense", "hybrid", "union"], default=None,
                        help="Suchmodus, Standard aus local_rag.RETRIEVAL_MODE")
    parser.add_argument("--no-rerank", action="store_true", help="Cross-Encoder abschalten")
    parser.add_argument("--query-instructions", action="store_true", default=USE_QUERY_INSTRUCTIONS,
                        help="Zusatztext aus local_rag.QUERY_ADDITIONAL_INSTRUCTIONS anhängen")
    args = parser.parse_args()
    GOLDSET_PATH, VECTORSTORE_PATH, REPORT_PATH = args.goldset, args.index, args.report
    USE_QUERY_INSTRUCTIONS = args.query_instructions

    print(f"Lese Referenzfragen aus {GOLDSET_PATH} ...")
    goldset = read_goldset(GOLDSET_PATH)
    print(f"{len(goldset)} Frage(n) geladen.\n")

    # Lädt Embeddings und Vektorspeicher genau wie die Chat-Anwendung, damit
    # gemessen wird, was dort auch tatsächlich passiert. Es wird kein LLM
    # aufgerufen — nur die Ähnlichkeitssuche im Vektorspeicher.
    options = {}
    if args.mode:
        options["retrieval_mode"] = args.mode
    if args.no_rerank:
        options["reranker_model"] = None
    rag = LocalRAG(vectorstore_path=VECTORSTORE_PATH, **options)
    if rag.vectorstore is None:
        sys.exit(
            f"\nKein Vektorspeicher unter {VECTORSTORE_PATH} gefunden.\n"
            "Erst den Index bauen (python add_pdfs.py), dann erneut auswerten."
        )

    print(f"\nStelle {len(goldset)} Frage(n) an den Vektorspeicher ...")
    global RETRIEVAL_SETUP, INDEX_INFO
    info_file = Path(VECTORSTORE_PATH) / "index_info.json"
    if info_file.exists():
        import json
        INDEX_INFO = json.loads(info_file.read_text(encoding="utf-8"))
    RETRIEVAL_SETUP = f"{rag.retrieval_mode}, Reranker {rag.reranker_model or 'aus'}"
    results = evaluate(rag, goldset, max_k=max(K_VALUES))

    report = build_report(results, rag.embeddings.model_name)
    print("\n" + report)

    REPORT_PATH.parent.mkdir(parents=True, exist_ok=True)
    REPORT_PATH.write_text(report, encoding="utf-8")
    print(f"Bericht gespeichert: {REPORT_PATH}")


if __name__ == "__main__":
    main()