pv-agent/agent/eval/evaluate.py

"""Goldset-Evaluation.

Modus 1 (offline, ohne Ollama): Retrieval-Metriken — Recall@k, Hit-Rate, MRR.
Modus 2 (--answers, benötigt Ollama): Zitier-Präzision (validated),
Verweigerungskorrektheit, erwartete Quelle zitiert, Latenz.

Kriterien laut Skill: Recall@8 > 0,9; Zitier-Präzision 100 %;
Verweigerungen korrekt.
"""
from __future__ import annotations

import json
import statistics
from pathlib import Path

import yaml

from ..config import Config
from ..retrieve import Retriever

GOLDSET_PATH = Path(__file__).parent / "goldset.yaml"


def load_goldset(path: Path = GOLDSET_PATH) -> list[dict]:
    data = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
    questions = data.get("questions", [])
    if not questions:
        raise SystemExit(f"goldset leer: {path}")
    return questions


def evaluate_retrieval(cfg: Config, questions: list[dict], k: int = 8) -> list[dict]:
    retriever = Retriever(cfg)
    rows: list[dict] = []
    try:
        for q in questions:
            if q.get("expect_refusal"):
                continue  # Verweigerungsfälle werden nur im Antwortmodus gemessen
            expected = set(q.get("expected_ids", []))
            if not expected:
                continue
            results = retriever.search(q["question"], n_entries=k)
            retrieved = [r.entry_id for r in results]
            hits = expected & set(retrieved)
            rank = next(
                (retrieved.index(e) + 1 for e in expected if e in retrieved), None
            )
            rows.append(
                {
                    "id": q["id"],
                    "question": q["question"],
                    "expected": sorted(expected),
                    "retrieved": retrieved,
                    "recall": len(hits) / len(expected),
                    "hit": bool(hits),
                    "mrr": (1.0 / rank) if rank else 0.0,
                }
            )
    finally:
        retriever.close()
    return rows


def summarize_retrieval(rows: list[dict]) -> dict:
    if not rows:
        return {"n": 0}
    return {
        "n": len(rows),
        "hit_rate": round(sum(r["hit"] for r in rows) / len(rows), 4),
        "mean_recall_at_k": round(statistics.mean(r["recall"] for r in rows), 4),
        "mrr": round(statistics.mean(r["mrr"] for r in rows), 4),
    }


def evaluate_answers(cfg: Config, questions: list[dict]) -> list[dict]:
    from ..generate import answer_question  # Ollama nötig — hier erst importieren
    rows: list[dict] = []
    for q in questions:
        result = answer_question(q["question"], cfg)
        expected = set(q.get("expected_ids", []))
        should_refuse = bool(q.get("expect_refusal"))
        rows.append(
            {
                "id": q["id"],
                "question": q["question"],
                "refused": result["refused"],
                "should_refuse": should_refuse,
                "refusal_correct": result["refused"] == should_refuse,
                "verified": result["verified"],
                "citations": result["citations"],
                "expected_cited": (
                    any(c in expected for c in result["citations"]) if expected else None
                ),
                "regenerations": result["regenerations"],
                "latency_ms": result["latency_ms"],
            }
        )
    return rows


def summarize_answers(rows: list[dict]) -> dict:
    if not rows:
        return {"n": 0}
    lat = [r["latency_ms"] for r in rows]
    return {
        "n": len(rows),
        "citation_precision": round(
            sum(r["verified"] for r in rows) / len(rows), 4
        ),
        "refusal_correct_rate": round(
            sum(r["refusal_correct"] for r in rows) / len(rows), 4
        ),
        "expected_cited_rate": round(
            sum(bool(r["expected_cited"]) for r in rows)
            / max(1, sum(1 for r in rows if r["expected_cited"] is not None)),
            4,
        ),
        "latency_ms_mean": round(statistics.mean(lat)),
        "latency_ms_p95": round(sorted(lat)[int(len(lat) * 0.95) - 1]),
        "regenerations_total": sum(r["regenerations"] for r in rows),
    }


def run_eval(cfg: Config, args) -> int:
    questions = load_goldset()
    if args.limit:
        questions = questions[: args.limit]
    mode = "retrieval + Antworten" if args.answers else "nur Retrieval (offline)"
    print(f"Goldset: {len(questions)} Fragen · Modus: {mode} · k={args.k}")
    print(f"Antwortmodell: {cfg.answer_model} · Embedding: {cfg.embed_model}")
    print("-" * 78)

    report: dict = {"mode": mode, "k": args.k, "model": cfg.answer_model}

    rows = evaluate_retrieval(cfg, questions, k=args.k)
    for r in rows:
        mark = "✓" if r["hit"] else "✗"
        print(
            f"{mark} {r['id']:8s} recall={r['recall']:.2f} "
            f"mrr={r['mrr']:.2f}  {r['question'][:56]}"
        )
        if not r["hit"]:
            print(f"   erwartet: {', '.join(r['expected'])}")
            print(f"   erhalten: {', '.join(r['retrieved'][:args.k])}")
    summary = summarize_retrieval(rows)
    report["retrieval"] = {"summary": summary, "rows": rows}
    print("-" * 78)
    print(f"Retrieval: n={summary['n']}  Hit-Rate={summary['hit_rate']}  "
          f"Recall@{args.k}={summary['mean_recall_at_k']}  MRR={summary['mrr']}")

    if args.answers:
        arows = evaluate_answers(cfg, questions)
        for r in arows:
            mark = "✓" if (r["verified"] and r["refusal_correct"]) else "✗"
            print(
                f"{mark} {r['id']:8s} refused={r['refused']} verified={r['verified']} "
                f"zit={','.join(r['citations'][:4])} {r['latency_ms']}ms"
            )
        asummary = summarize_answers(arows)
        report["answers"] = {"summary": asummary, "rows": arows}
        print("-" * 78)
        print(
            f"Antworten: n={asummary['n']}  Zitier-Präzision={asummary['citation_precision']}  "
            f"Verweigerung korrekt={asummary['refusal_correct_rate']}  "
            f"erwartete Quelle zitiert={asummary['expected_cited_rate']}"
        )
        print(
            f"Latenz: mean={asummary['latency_ms_mean']}ms  p95={asummary['latency_ms_p95']}ms  "
            f"Regenerierungen={asummary['regenerations_total']}"
        )

    if args.json_out:
        Path(args.json_out).parent.mkdir(parents=True, exist_ok=True)
        Path(args.json_out).write_text(
            json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8"
        )
        print(f"Report geschrieben: {args.json_out}")
    return 0