#!/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()