Spaces:
Running
Running
File size: 5,398 Bytes
d30d906 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 | """Validate development/gold separation and menu labels."""
from __future__ import annotations
import argparse
import json
import sys
from collections import Counter, defaultdict
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from config import DATA_DIR, RAW_DIR
from core.query_data import normalize_query_text
DEV_PATH = DATA_DIR / "eval_queries_dev.json"
GOLD_PATH = DATA_DIR / "eval_queries_gold.json"
NO_MATCH_DEV_PATH = DATA_DIR / "eval_queries_no_match_dev.json"
QUERY_POOL_PATH = DATA_DIR / "query_index_pool.json"
def load_json(path: Path) -> list:
with path.open("r", encoding="utf-8") as f:
return json.load(f)
def answer_ids(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 validate_rows(name: str, rows: list, valid_menu_ids: set[str]) -> list[str]:
errors = []
labels_by_query: dict[str, set[tuple[str, ...]]] = defaultdict(set)
seen_rows = set()
for index, item in enumerate(rows, 1):
query = normalize_query_text(item.get("query", ""))
ids = answer_ids(item)
no_match = bool(item.get("no_match", False))
if not query:
errors.append(f"{name}[{index}]: empty query")
continue
if no_match and ids:
errors.append(f"{name}[{index}]: no_match row has expected_ids")
if not no_match and not ids:
errors.append(f"{name}[{index}]: positive row has no expected_ids")
unknown = ids - valid_menu_ids
if unknown:
errors.append(f"{name}[{index}]: unknown menu IDs {sorted(unknown)}")
label = ("__NO_MATCH__",) if no_match else tuple(sorted(ids))
labels_by_query[query].add(label)
row_key = (query, label)
if row_key in seen_rows:
errors.append(f"{name}[{index}]: duplicate row '{query}'")
seen_rows.add(row_key)
for query, labels in labels_by_query.items():
if len(labels) > 1:
errors.append(
f"{name}: conflicting labels for '{query}' -> {sorted(labels)}"
)
return errors
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--require-gold", action="store_true")
args = parser.parse_args()
menus = load_json(RAW_DIR / "real_menus.json")
valid_menu_ids = {str(menu["menu_id"]) for menu in menus}
dev = load_json(DEV_PATH)
errors = validate_rows("dev", dev, valid_menu_ids)
print(f"dev: {len(dev)} rows")
print(f"dev sources: {dict(Counter(x.get('source', 'unknown') for x in dev))}")
dev_queries = {normalize_query_text(x.get("query", "")) for x in dev}
if NO_MATCH_DEV_PATH.exists():
no_match_dev = load_json(NO_MATCH_DEV_PATH)
errors.extend(validate_rows("no-match-dev", no_match_dev, valid_menu_ids))
invalid_rows = [
index
for index, item in enumerate(no_match_dev, 1)
if not item.get("no_match", False) or answer_ids(item)
]
if invalid_rows:
errors.append(
"no-match-dev contains positive rows at indexes "
f"{invalid_rows}"
)
no_match_queries = {
normalize_query_text(x.get("query", "")) for x in no_match_dev
}
overlap = (dev_queries & no_match_queries) - {""}
if overlap:
errors.append(f"dev/no-match-dev query overlap: {len(overlap)}")
print(f"no-match-dev: {len(no_match_dev)} rows")
else:
print(f"no-match-dev: not created ({NO_MATCH_DEV_PATH.name})")
if GOLD_PATH.exists():
gold = load_json(GOLD_PATH)
errors.extend(validate_rows("gold", gold, valid_menu_ids))
gold_queries = {normalize_query_text(x.get("query", "")) for x in gold}
overlap = (dev_queries & gold_queries) - {""}
if overlap:
errors.append(f"dev/gold query overlap: {len(overlap)}")
if NO_MATCH_DEV_PATH.exists():
overlap = (no_match_queries & gold_queries) - {""}
if overlap:
errors.append(f"no-match-dev/gold query overlap: {len(overlap)}")
if QUERY_POOL_PATH.exists():
pool_queries = {
normalize_query_text(x.get("query", ""))
for x in load_json(QUERY_POOL_PATH)
}
leaked = (gold_queries & pool_queries) - {""}
if leaked:
errors.append(f"gold/query-index pool overlap: {len(leaked)}")
llm_rows = sum(
str(item.get("source", "")).lower() == "llm"
for item in gold
)
if llm_rows:
errors.append(f"gold contains LLM-authored rows: {llm_rows}")
print(f"gold: {len(gold)} rows")
print(
"gold sources: "
f"{dict(Counter(x.get('source', 'unknown') for x in gold))}"
)
elif args.require_gold:
errors.append(f"gold set missing: {GOLD_PATH}")
else:
print(f"gold: not created yet ({GOLD_PATH.name})")
if errors:
print("\nVALIDATION FAILED")
for error in errors:
print(f"- {error}")
raise SystemExit(1)
print("\nVALIDATION PASSED")
if __name__ == "__main__":
main()
|