evaluate.py

"""NetTutor AI evaluation harness (roadmap section 10).

Runs the fixed question set in data/eval_questions.json and reports:

  * Retrieval accuracy@k  - does a chunk from the expected source file appear
                            in the top-k retrieved chunks?
  * Topic routing accuracy - does the graph classify/route to the expected
                             topic family?
  * Quiz quality           - does each generated quiz have 3 questions spanning
                             easy / medium / applied difficulty?
  * Latency                - wall-clock time from question to final answer.

Usage:
    python -m scripts.evaluate
    python -m scripts.evaluate --k 5 --limit 20 --json results.json

The harness runs fully offline with the hashing embedder + template LLM, and
unchanged against the production sentence-transformer + Anthropic stack.
"""
from __future__ import annotations

import argparse
import json
import statistics
import time
from pathlib import Path
from typing import Dict, List

from app.config import settings
from app.retriever import get_retriever
from app.graph import run_tutor
from app.quiz import generate_quiz

EVAL_PATH = Path(__file__).resolve().parent.parent / "data" / "eval_questions.json"

# Topic families that should be considered equivalent when scoring routing,
# because they live in the same source notes / are sub-topics of each other.
TOPIC_FAMILIES = {
    "subnetting": {"subnetting"},
    "tcp_ip": {"tcp_ip", "packet_flow", "arp_l2"},
    "routing": {"routing", "switching_stp", "arp_l2"},
    "switching_stp": {"switching_stp", "routing", "arp_l2"},
    "linux": {"linux"},
    "iot": {"iot"},
    "packet_flow": {"packet_flow", "tcp_ip", "routing"},
    "arp_l2": {"arp_l2", "routing", "tcp_ip"},
}


def _load_questions(limit: int | None) -> List[Dict]:
    data = json.loads(EVAL_PATH.read_text())
    qs = data["questions"]
    return qs[:limit] if limit else qs


def _retrieval_hit(result, expected_source: str, k: int) -> bool:
    for c in result.chunks[:k]:
        if expected_source in c.source:
            return True
    return False


def _quiz_ok(quiz) -> bool:
    if len(quiz.questions) != 3:
        return False
    diffs = {q.difficulty for q in quiz.questions}
    return {"easy", "medium", "applied"} <= diffs


def evaluate(k: int = 5, limit: int | None = None) -> Dict:
    retriever = get_retriever()
    questions = _load_questions(limit)

    retr_hits = 0
    topic_hits = 0
    quiz_hits = 0
    latencies: List[float] = []
    rows: List[Dict] = []

    for item in questions:
        q = item["q"]
        exp_source = item["source"]
        exp_topic = item["topic"]

        # --- retrieval accuracy@k (retriever in isolation) ---
        rresult = retriever.retrieve(q, top_k=max(k, settings.TOP_K))
        r_hit = _retrieval_hit(rresult, exp_source, k)
        retr_hits += int(r_hit)

        # --- full pipeline: topic routing + latency + quiz ---
        t0 = time.perf_counter()
        out = run_tutor(q, make_quiz=True)
        dt = time.perf_counter() - t0
        latencies.append(dt)

        got_topic = out.get("topic", "")
        family = TOPIC_FAMILIES.get(exp_topic, {exp_topic})
        t_hit = got_topic in family
        topic_hits += int(t_hit)

        quiz = generate_quiz(got_topic or exp_topic, out.get("answer", ""),
                             rresult.context_block())
        q_ok = _quiz_ok(quiz)
        quiz_hits += int(q_ok)

        rows.append({
            "question": q,
            "expected_topic": exp_topic,
            "got_topic": got_topic,
            "retrieval_hit": r_hit,
            "topic_hit": t_hit,
            "quiz_ok": q_ok,
            "latency_s": round(dt, 4),
        })

    n = len(questions)
    summary = {
        "n_questions": n,
        "k": k,
        "retrieval_accuracy": round(retr_hits / n, 4),
        "topic_routing_accuracy": round(topic_hits / n, 4),
        "quiz_quality": round(quiz_hits / n, 4),
        "latency_avg_s": round(statistics.mean(latencies), 4),
        "latency_p95_s": round(sorted(latencies)[int(0.95 * (n - 1))], 4),
        "latency_max_s": round(max(latencies), 4),
        "embedder": type(getattr(get_retriever(), "_embedder", None)).__name__,
        "llm_enabled": settings.llm_enabled,
    }
    return {"summary": summary, "rows": rows}


def _print_report(report: Dict) -> None:
    s = report["summary"]
    print("\n" + "=" * 60)
    print("  NetTutor AI - Evaluation Report")
    print("=" * 60)
    print(f"  Questions evaluated : {s['n_questions']}")
    print(f"  LLM enabled         : {s['llm_enabled']}")
    print("-" * 60)
    print(f"  Retrieval accuracy@{s['k']:<2}: {s['retrieval_accuracy']:.1%}")
    print(f"  Topic routing acc.   : {s['topic_routing_accuracy']:.1%}")
    print(f"  Quiz quality         : {s['quiz_quality']:.1%}")
    print("-" * 60)
    print(f"  Latency  avg / p95 / max : "
          f"{s['latency_avg_s']:.3f}s / {s['latency_p95_s']:.3f}s / {s['latency_max_s']:.3f}s")
    print("=" * 60 + "\n")


def main() -> None:
    ap = argparse.ArgumentParser(description="Evaluate NetTutor AI retrieval + pipeline.")
    ap.add_argument("--k", type=int, default=5, help="top-k for retrieval accuracy")
    ap.add_argument("--limit", type=int, default=None, help="evaluate only first N questions")
    ap.add_argument("--json", type=str, default=None, help="write full results to this JSON path")
    args = ap.parse_args()

    report = evaluate(k=args.k, limit=args.limit)
    _print_report(report)

    if args.json:
        Path(args.json).write_text(json.dumps(report, indent=2))
        print(f"Full results written to {args.json}")


if __name__ == "__main__":
    main()