Buckets:
| """Shared fixed-CV-Bench contract for external Zebra-CoT checkpoints.""" | |
| from __future__ import annotations | |
| import json | |
| import re | |
| from collections import Counter | |
| from pathlib import Path | |
| from uniar_adapter.eval_cv_bench import OFFICIAL_QUESTION_EXTENSION, _letter, parse_choice | |
| TASKS = ("Count", "Relation", "Depth", "Distance") | |
| def read_jsonl(path: Path) -> list[dict]: | |
| return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line] | |
| def write_json(path: Path, value: object) -> None: | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") | |
| def fixed_records(data_dir: Path, selection_path: Path) -> list[dict]: | |
| selection = json.loads(selection_path.read_text(encoding="utf-8")) | |
| ordered = [index for task in TASKS for index in selection[task]] | |
| if len(ordered) != 20 or len(set(ordered)) != 20: | |
| raise ValueError("expected exactly 20 unique fixed CV-Bench indices") | |
| records = {row["idx"]: row for row in read_jsonl(data_dir / "prepared/records.jsonl")} | |
| missing = [index for index in ordered if index not in records] | |
| if missing: | |
| raise ValueError(f"fixed CV-Bench records missing: {missing}") | |
| return [records[index] for index in ordered] | |
| def evaluation_prompt(record: dict) -> str: | |
| return f"{record['prompt']}\n{OFFICIAL_QUESTION_EXTENSION}" | |
| def gold_letter(record: dict) -> str: | |
| letter = _letter(record["answer"]) | |
| if letter is None: | |
| raise ValueError(f"unparseable gold answer for CV-Bench index {record['idx']}") | |
| return letter | |
| def score(record: dict, raw_output: str) -> dict: | |
| final_fields = re.findall( | |
| r"(?is)(?:final\s+answer|answer)\s*(?:is|:|=)\s*([^\n<]+)", raw_output | |
| ) | |
| prediction = None | |
| method = "unparsed" | |
| if final_fields: | |
| candidate = re.sub(r"\s+", " ", final_fields[-1]).strip().rstrip(".:,; ") | |
| compact = re.sub(r"[^a-z0-9]+", " ", candidate.lower()).strip() | |
| for index, choice in enumerate(record["choices"]): | |
| choice_compact = re.sub(r"[^a-z0-9]+", " ", str(choice).lower()).strip() | |
| if compact == choice_compact: | |
| prediction, method = chr(ord("A") + index), "final_choice_text" | |
| break | |
| if prediction is None: | |
| prediction, method = parse_choice(raw_output, record["choices"]) | |
| gold = gold_letter(record) | |
| return { | |
| "prediction": prediction, | |
| "gold": gold, | |
| "correct": prediction == gold, | |
| "parse_method": method, | |
| } | |
| def summary(rows: list[dict]) -> dict: | |
| task_counts = Counter(row["task"] for row in rows) | |
| correct_by_task = Counter(row["task"] for row in rows if row["score"]["correct"]) | |
| stops = Counter(row["stop_reason"] for row in rows) | |
| return { | |
| "samples": len(rows), | |
| "correct": sum(row["score"]["correct"] for row in rows), | |
| "accuracy": sum(row["score"]["correct"] for row in rows) / len(rows) if rows else 0.0, | |
| "by_task": { | |
| task: { | |
| "samples": task_counts[task], | |
| "correct": correct_by_task[task], | |
| "accuracy": correct_by_task[task] / task_counts[task] if task_counts[task] else 0.0, | |
| } | |
| for task in TASKS | |
| }, | |
| "generated_images": sum(len(row["generated_images"]) for row in rows), | |
| "stop_reasons": dict(sorted(stops.items())), | |
| } | |
Xet Storage Details
- Size:
- 3.46 kB
- Xet hash:
- 6cab036e8917a26c56efdfa1158bfbb947dd622ef42e3ae555982104006130bb
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.