"""Exact-ID evaluation for development or sealed gold data.""" from __future__ import annotations import argparse import json import sys from concurrent.futures import ThreadPoolExecutor from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from config import DATA_DIR, DEFAULT_APP_VERSION, SIMILARITY_THRESHOLD from core.search_engine import MenuSearchEngine from core.version_paths import raw_menus_path def answers(item: dict) -> set[str]: values = item.get("expected_ids") if values is None and item.get("menu_id"): values = [item["menu_id"]] return {str(value).strip() for value in (values or []) if str(value).strip()} def summarize(ranks: list[int | None]) -> dict[str, float]: total = len(ranks) if not total: return {} return { "acc1": sum(rank is not None and rank <= 1 for rank in ranks) / total, "acc3": sum(rank is not None and rank <= 3 for rank in ranks) / total, "acc5": sum(rank is not None and rank <= 5 for rank in ranks) / total, "mrr5": sum(1.0 / rank for rank in ranks if rank) / total, } def print_metrics(label: str, ranks: list[int | None]) -> None: metrics = summarize(ranks) if not metrics: return print( f"{label}: n={len(ranks)} " f"Acc@1={metrics['acc1']:.1%} " f"Acc@3={metrics['acc3']:.1%} " f"Acc@5={metrics['acc5']:.1%} " f"MRR@5={metrics['mrr5']:.3f}" ) def parse_thresholds(value: str | None, fallback: float) -> list[float]: if not value: return [fallback] thresholds = sorted({float(item.strip()) for item in value.split(",")}) if any(value < 0 or value > 1 for value in thresholds): raise SystemExit("thresholds must be between 0 and 1") return thresholds def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--split", choices=("dev", "gold"), default="dev") parser.add_argument("--app-version", default=DEFAULT_APP_VERSION) parser.add_argument("--eval-path") parser.add_argument("--negative-path") parser.add_argument("--no-hyde", action="store_true") parser.add_argument( "--hyde-workers", type=int, default=1, help="parallel HyDE cache warm-up workers for evaluation only", ) parser.add_argument( "--serving-path", action="store_true", help="evaluate the MCP/API agent path with query rewriting disabled", ) parser.add_argument("--threshold", type=float, default=SIMILARITY_THRESHOLD) parser.add_argument( "--threshold-grid", help="comma-separated query-level thresholds; retrieval runs only once", ) parser.add_argument( "--output", help="optional JSON path for metrics and per-query predictions", ) args = parser.parse_args() default_name = ( "eval_queries_gold.json" if args.split == "gold" else "eval_queries_dev.json" ) path = Path(args.eval_path) if args.eval_path else DATA_DIR / default_name if not path.exists(): raise SystemExit(f"evaluation file missing: {path}") rows = json.loads(path.read_text(encoding="utf-8")) negative_path = ( Path(args.negative_path) if args.negative_path else DATA_DIR / "eval_queries_no_match_dev.json" ) if args.split == "dev" and negative_path.exists(): rows.extend(json.loads(negative_path.read_text(encoding="utf-8"))) available_menu_ids = { item["menu_id"] for item in json.loads( raw_menus_path(args.app_version).read_text(encoding="utf-8") ) } eligible_rows = [] unavailable_positive = 0 for item in rows: if item.get("no_match", False): eligible_rows.append(item) continue if answers(item) & available_menu_ids: eligible_rows.append(item) else: unavailable_positive += 1 rows = eligible_rows engine = MenuSearchEngine.get_instance(args.app_version) if not args.no_hyde: if not engine.hyde_gen.available: raise SystemExit( "HyDE evaluation requested, but the configured LLM backend is unavailable." ) unique_queries = list(dict.fromkeys(item["query"] for item in rows)) print( f"HyDE cache warm-up: queries={len(unique_queries)} " f"workers={args.hyde_workers}" ) if args.hyde_workers > 1: with ThreadPoolExecutor(max_workers=args.hyde_workers) as pool: list(pool.map(engine.hyde_gen.generate, unique_queries)) else: for query in unique_queries: engine.hyde_gen.generate(query) if not engine.hyde_gen.available: raise SystemExit( "HyDE generation failed during warm-up; no HyDE metrics were produced." ) run_agent = None if args.serving_path: from core.agent import run_agent as agent_runner run_agent = agent_runner retrieved = [] for item in rows: if run_agent: result = run_agent( item["query"], top_n=5, threshold=0.0, use_hyde=not args.no_hyde, use_reranker=False, allow_rewrite=False, app_version=args.app_version, ) hits = result.get("search_results", []) else: hits = engine.search( item["query"], top_n=5, threshold=0.0, use_hyde=not args.no_hyde, use_reranker=False, ) retrieved.append( ( item, [hit.get("menu_id", "") for hit in hits], float(hits[0].get("confidence", 0.0)) if hits else 0.0, ) ) thresholds = parse_thresholds(args.threshold_grid, args.threshold) print( f"split={args.split} app_version={args.app_version} rows={len(rows)} " f"unavailable_positive={unavailable_positive}" ) print(f"path={'serving' if args.serving_path else 'retrieval'}") print(f"HyDE={'OFF' if args.no_hyde else 'ON'}") threshold_reports = [] for threshold in thresholds: positive_ranks: list[int | None] = [] ranks_by_source: dict[str, list[int | None]] = {} ranks_by_group: dict[str, list[int | None]] = {} negative_total = 0 negative_rejected = 0 prediction_rows = [] for item, raw_predicted, top_confidence in retrieved: predicted = raw_predicted if top_confidence >= threshold else [] if item.get("no_match", False): negative_total += 1 negative_rejected += int(not predicted) prediction_rows.append( { "id": item.get("id"), "query": item["query"], "no_match": True, "predicted_ids": predicted, "top_confidence": top_confidence, } ) continue expected = answers(item) rank = next( ( index for index, menu_id in enumerate(predicted, 1) if menu_id in expected ), None, ) positive_ranks.append(rank) source = str(item.get("source") or "unknown") ranks_by_source.setdefault(source, []).append(rank) group = str(item.get("evaluation_group") or "ungrouped") ranks_by_group.setdefault(group, []).append(rank) prediction_rows.append( { "id": item.get("id"), "query": item["query"], "expected_ids": sorted(expected), "predicted_ids": predicted, "rank": rank, "top_confidence": top_confidence, "source": source, "evaluation_group": group, } ) print(f"\nthreshold={threshold:.2f}") print_metrics("all", positive_ranks) for source in sorted(ranks_by_source): print_metrics(source, ranks_by_source[source]) for group in sorted(ranks_by_group): print_metrics(f"group:{group}", ranks_by_group[group]) if negative_total: print( f"No-match rejection={negative_rejected / negative_total:.1%} " f"({negative_rejected}/{negative_total})" ) threshold_reports.append( { "threshold": threshold, "all": summarize(positive_ranks), "by_source": { source: summarize(source_ranks) for source, source_ranks in sorted(ranks_by_source.items()) }, "by_evaluation_group": { group: summarize(group_ranks) for group, group_ranks in sorted(ranks_by_group.items()) }, "no_match": { "rejected": negative_rejected, "total": negative_total, "rejection_rate": ( negative_rejected / negative_total if negative_total else None ), }, "predictions": prediction_rows, } ) if args.output: output_path = Path(args.output) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text( json.dumps( { "eval_path": str(path), "app_version": args.app_version, "serving_path": args.serving_path, "hyde": not args.no_hyde, "eligible_rows": len(rows), "unavailable_positive": unavailable_positive, "thresholds": threshold_reports, }, ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) print(f"report={output_path}") if __name__ == "__main__": main()