""" scripts/evaluate.py -------------------- Run evaluation sets against the RAG pipeline and produce results. .venv/bin/python scripts/evaluate.py catan .venv/bin/python scripts/evaluate.py catan --retriever tfidf .venv/bin/python scripts/evaluate.py monopoly --retriever random Outputs: data/outputs/{game}_results.csv (dense — default) data/outputs/{game}_tfidf_results.csv (tfidf) data/outputs/{game}_random_results.csv (random) After running, fill in the 'correctness_score' column: 1 = fully correct, 0.5 = partially correct, 0 = wrong """ import csv import json import os import sys from pathlib import Path from model import query_rag, DEFAULT_K, DEFAULT_RETRIEVER, RETRIEVERS K = DEFAULT_K ABSTAIN_PHRASES = [ "don't specify", "doesn't specify", "not specify", "rules don't", "rules do not", "not addressed", "not mentioned", "not explicit", "not state", "silent", "i don't know", "cannot answer", "unclear", "no rule", "not covered", ] # --------------------------------------------------------------------------- # Evaluation helpers # --------------------------------------------------------------------------- def check_retrieval(retrieved: list[dict], source_keywords: list[str]) -> dict: """Keyword-based retrieval quality check against chunk titles.""" retrieved_titles = [c["title"].lower() for c in retrieved] hits = sum( 1 for kw in source_keywords if any(kw.lower() in t for t in retrieved_titles) ) coverage = hits / len(source_keywords) if source_keywords else 0.0 return { "hits": hits, "expected": len(source_keywords), "coverage": round(coverage, 3), "precision_at_k": 1.0 if hits > 0 else 0.0, } def check_hallucination(answer: str, sub_category: str) -> str: """For Category C questions, detect abstention vs. potential hallucination.""" if sub_category != "C": return "N/A" abstained = any(p in answer.lower() for p in ABSTAIN_PHRASES) return "ABSTAINED" if abstained else "POTENTIAL_HALLUCINATION" # --------------------------------------------------------------------------- # Main evaluation loop # --------------------------------------------------------------------------- def run_evaluation(game: str, retriever: str = DEFAULT_RETRIEVER, k: int = K) -> list[dict]: eval_path = f"data/raw/{game}_eval.json" # Output filename: {game}_results.csv for dense, {game}_{retriever}_results.csv for others if retriever == "dense": results_path = f"data/outputs/{game}_results.csv" else: results_path = f"data/outputs/{game}_{retriever}_results.csv" if not os.path.exists(eval_path): print(f"ERROR: Eval data not found at '{eval_path}'", file=sys.stderr) sys.exit(1) Path("data/outputs").mkdir(parents=True, exist_ok=True) with open(eval_path) as f: data = json.load(f) all_questions = ( data.get("correctness_questions", []) + data.get("stress_test_questions", []) ) results = [] total = len(all_questions) print(f"Running evaluation — game='{game}', retriever='{retriever}', k={k}, n={total}") print("=" * 70) for i, q in enumerate(all_questions, 1): qid = q["id"] question = q["question"] expected = q["expected_answer"] keywords = q.get("source_keywords", []) category = q["category"] sub_cat = q.get("sub_category", "") print(f"\n[{i}/{total}] {qid} ({category}/{sub_cat or '-'})") print(f" Q: {question[:80]}{'...' if len(question) > 80 else ''}") rag_result = query_rag(question, game=game, k=k, retriever=retriever) answer = rag_result["answer"] retrieved = rag_result["retrieved"] ret = check_retrieval(retrieved, keywords) halluc = check_hallucination(answer, sub_cat) print(f" Retrieved: {[c['title'] for c in retrieved]}") print(f" Retrieval hits: {ret['hits']}/{ret['expected']}") print(f" Expected: {expected[:100]}{'...' if len(expected) > 100 else ''}") print(f" Model: {answer[:100]}{'...' if len(answer) > 100 else ''}") if halluc != "N/A": print(f" Hallucination check: {halluc}") results.append({ "id": qid, "category": category, "sub_category": sub_cat, "retriever": retriever, "question": question, "expected_answer": expected, "model_answer": answer, "retrieved_chunks": " | ".join(c["title"] for c in retrieved), "retrieval_hits": ret["hits"], "retrieval_expected": ret["expected"], "retrieval_coverage": ret["coverage"], "precision_at_k": ret["precision_at_k"], "hallucination_check": halluc, "correctness_score": "", "notes": "", }) fieldnames = list(results[0].keys()) with open(results_path, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(results) print(f"\n\nResults saved to '{results_path}'") _print_summary(results, k, retriever) return results # --------------------------------------------------------------------------- # Summary statistics # --------------------------------------------------------------------------- def _print_summary(results: list[dict], k: int = K, retriever: str = DEFAULT_RETRIEVER): def group(cat, sub=None): return [r for r in results if r["category"] == cat and (sub is None or r["sub_category"] == sub)] correctness = group("correctness") stress_a = group("stress", "A") stress_b = group("stress", "B") stress_c = group("stress", "C") def avg_prec(g): return (sum(r["precision_at_k"] for r in g) / len(g) * 100) if g else 0 def avg_cov(g): return (sum(r["retrieval_coverage"] for r in g) / len(g) * 100) if g else 0 halluc_rate = ( sum(1 for r in stress_c if r["hallucination_check"] == "POTENTIAL_HALLUCINATION") / len(stress_c) * 100 if stress_c else 0 ) print("\n" + "=" * 60) print(f"EVALUATION SUMMARY (retriever={retriever}, k={k})") print("=" * 60) print(f"\nRetrieval Precision@{k} (≥1 expected chunk retrieved):") print(f" Correctness Qs ({len(correctness):2d}): {avg_prec(correctness):.0f}%") print(f" Stress A — multi-sec ({len(stress_a):2d}): {avg_prec(stress_a):.0f}%") print(f" Stress B — edge case ({len(stress_b):2d}): {avg_prec(stress_b):.0f}%") print(f" Stress C — unanswer ({len(stress_c):2d}): {avg_prec(stress_c):.0f}%") print(f" Overall ({len(results):2d}): {avg_prec(results):.0f}%") print(f"\nRetrieval Coverage (fraction of required chunks found):") print(f" Stress A — multi-sec ({len(stress_a):2d}): {avg_cov(stress_a):.0f}%") print(f" Overall ({len(results):2d}): {avg_cov(results):.0f}%") print(f"\nHallucination Rate (Category C — gave definitive answer when rules are silent):") abstained = sum(1 for r in stress_c if r["hallucination_check"] == "ABSTAINED") print(f" Appropriately abstained: {abstained}/{len(stress_c)}") print(f" Potential hallucinations: {len(stress_c) - abstained}/{len(stress_c)} ({halluc_rate:.0f}%)") print(f"\n{'─'*60}") print(f" Fill in 'correctness_score' in the results CSV:") print(f" 1 = correct, 0.5 = partially correct, 0 = wrong") print(f"{'─'*60}\n") # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Evaluate RAG pipeline for a board game.") parser.add_argument("game", help="Game to evaluate (e.g. catan, monopoly)") parser.add_argument( "--retriever", choices=RETRIEVERS, default=DEFAULT_RETRIEVER, help=f"Retrieval strategy (default: {DEFAULT_RETRIEVER})", ) args = parser.parse_args() if not os.path.exists(f"models/{args.game}/chunks.json"): print(f"ERROR: Index not found for '{args.game}'. Run: .venv/bin/python scripts/build_features.py {args.game}") sys.exit(1) run_evaluation(args.game, retriever=args.retriever)