File size: 4,655 Bytes
bca45b0 | 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 | #!/usr/bin/env python3
"""Compare multiple audited CV-Bench runs at the paired sample level."""
from __future__ import annotations
import argparse
import json
import math
from itertools import combinations
from pathlib import Path
def _json(path: Path) -> dict:
return json.loads(path.read_text(encoding="utf-8"))
def _jsonl_by_id(path: Path) -> dict[str, dict]:
rows = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line]
result = {row["sample_id"]: row for row in rows}
if len(result) != len(rows):
raise AssertionError(f"Duplicate sample IDs: {path}")
return result
def _mcnemar_exact_p(improved: int, regressed: int) -> float:
discordant = improved + regressed
if discordant == 0:
return 1.0
tail = sum(math.comb(discordant, k) for k in range(min(improved, regressed) + 1))
return min(1.0, 2.0 * tail / (2**discordant))
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--eval-root", type=Path, required=True)
parser.add_argument("--experiment", default="stage2_covt_sam_vggt_covt4_v2")
parser.add_argument("--steps", type=int, nargs="+", required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
runs: dict[int, dict] = {}
results: dict[int, dict[str, dict]] = {}
for step in args.steps:
directory = args.eval_root / f"{args.experiment}-checkpoint-{step}"
validation = _json(directory / "validation_report.json")
if validation.get("assessment") != "ready_to_share":
raise AssertionError(f"Run has not passed validation: checkpoint-{step}")
metrics = _json(directory / "metrics.json")["metrics"]
swanlab = _json(directory / "swanlab.json")
results[step] = _jsonl_by_id(directory / "results.jsonl")
runs[step] = {
"correct": int(validation["correct_total"]),
"overall_accuracy": metrics["eval/cvbench_accuracy"],
"task_macro_accuracy": metrics["eval/cvbench_task_macro_accuracy"],
"2d_accuracy": metrics["eval/cvbench_2d_accuracy"],
"2d_count_accuracy": metrics["eval/cvbench_2d_count_accuracy"],
"2d_relation_accuracy": metrics["eval/cvbench_2d_relation_accuracy"],
"3d_accuracy": metrics["eval/cvbench_3d_accuracy"],
"3d_depth_accuracy": metrics["eval/cvbench_3d_depth_accuracy"],
"3d_distance_accuracy": metrics["eval/cvbench_3d_distance_accuracy"],
"parse_rate": metrics["eval/cvbench_parse_rate"],
"errors": int(metrics["eval/cvbench_errors"]),
"swanlab_run_id": swanlab["run_id"],
"swanlab_run_url": swanlab["run_url"],
}
reference_ids = set(results[args.steps[0]])
if len(reference_ids) != 500 or any(set(results[step]) != reference_ids for step in args.steps):
raise AssertionError("Checkpoint runs do not cover the same 500 samples")
pairwise = []
for earlier, later in combinations(args.steps, 2):
improved = sum(
not results[earlier][sample_id]["correct"] and results[later][sample_id]["correct"]
for sample_id in reference_ids
)
regressed = sum(
results[earlier][sample_id]["correct"] and not results[later][sample_id]["correct"]
for sample_id in reference_ids
)
pairwise.append(
{
"earlier_step": earlier,
"later_step": later,
"accuracy_delta_percentage_points": 100
* (runs[later]["overall_accuracy"] - runs[earlier]["overall_accuracy"]),
"improved_samples": improved,
"regressed_samples": regressed,
"net_correct_delta": improved - regressed,
"mcnemar_exact_p": _mcnemar_exact_p(improved, regressed),
}
)
payload = {
"assessment": "ready_to_share_with_subset_and_sample_size_caveat",
"population": "fixed CV-Bench-500: 250 2D + 250 3D",
"scoring": "greedy option exact-match; no LLM judge",
"runs": {str(step): runs[step] for step in args.steps},
"pairwise": pairwise,
"best_overall_step": max(args.steps, key=lambda step: runs[step]["overall_accuracy"]),
}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
if __name__ == "__main__":
main()
|