Spaces:
Sleeping
Sleeping
| """ | |
| 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) | |