File size: 9,905 Bytes
fa82a9d | 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 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 | #!/usr/bin/env python3
"""Score the deterministic (metamorphic) sets: Official criterion vs Repair.
Currently implements the two prompt-side families:
``scene-synonym``
Official criterion (VBench scene): every word of the scene phrase must appear as a
substring of the Tag2Text caption. A same-scene synonym therefore fails even though
a human would call the scene supported -- this is the invariance the repair targets.
Two official readings are reported: the whole prompt words, and only the scene noun.
``action-swap``
Deterministic text-side interventions. ``class_swap`` must move the verdict
(sensitivity); ``out_of_vocabulary`` must become ``other`` and therefore must not be
scored as the original class (protocol check).
Repair labels are produced by the project models (``--scene-model`` / ``--adapters``);
without them the script reports the official half only.
Usage::
uv run --no-sync python scripts/score_deterministic.py --family scene-synonym
uv run --no-sync python scripts/score_deterministic.py --family scene-synonym \
--base-model /home/winbeau/models/Qwen3-0.6B --scene-model runs/formal/v4-8b/scene --limit 40
"""
from __future__ import annotations
import argparse
from collections import Counter
import json
import os
from pathlib import Path
import sys
from typing import Any, Sequence
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
from vbench_prompts_compile import records as R # noqa: E402
SCENE_STOPWORDS = {"a", "an", "the", "of", "in", "on", "at", "with", "and"}
def official_scene_pass(scene_key: str, caption: str) -> bool:
"""VBench scene rule: every word of the key appears in the caption."""
words = [word for word in scene_key.split(" ") if word]
return bool(words) and all(word in caption for word in words)
def scene_noun(prompt: str) -> str:
words = [word for word in prompt.split() if word.lower() not in SCENE_STOPWORDS]
return words[-1] if words else prompt
def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--family", default="scene-synonym")
parser.add_argument("--data", default=None, help="path to the deterministic set JSONL")
parser.add_argument("--output", default=None)
parser.add_argument("--limit", type=int, default=None)
parser.add_argument("--base-model", default=None)
parser.add_argument("--scene-model", default=None)
parser.add_argument("--adapters", action="append", default=[], help="task=path (for the action family)")
parser.add_argument("--allow-download", action="store_true")
return parser.parse_args(argv)
def score_scene_synonym(rows: list[dict[str, Any]]) -> dict[str, Any]:
paired: list[dict[str, Any]] = []
for row in rows:
caption = row["caption"]
original, variant = row["original_prompt"], row["variant_prompt"]
paired.append(
{
**row,
"official_original": official_scene_pass(original, caption),
"official_variant": official_scene_pass(variant, caption),
"official_noun_original": official_scene_pass(scene_noun(original), caption),
"official_noun_variant": official_scene_pass(scene_noun(variant), caption),
"repair_original": None,
"repair_variant": None,
}
)
return {"rows": paired}
def attach_repair(rows: list[dict[str, Any]], args: argparse.Namespace) -> None:
if not args.scene_model or not args.base_model:
return
os.environ.setdefault("HF_HUB_OFFLINE", "1")
os.environ.setdefault("TRANSFORMERS_OFFLINE", "1")
from vbench_prompts_compile.inference import SceneVerifier
path = Path(args.scene_model)
if not path.is_absolute():
path = ROOT / path
if (path / "adapter_config.json").exists():
verifier = SceneVerifier(base_model_path=args.base_model, adapter_path=str(path), local_files_only=not args.allow_download)
else:
verifier = SceneVerifier(model_path=str(path), local_files_only=not args.allow_download)
for row in rows:
row["repair_original"] = verifier.predict(row["original_prompt"], row["caption"], max_new_tokens=8, stop_at_newline=True).value
row["repair_variant"] = verifier.predict(row["variant_prompt"], row["caption"], max_new_tokens=8, stop_at_newline=True).value
def summarize_scene_synonym(rows: list[dict[str, Any]]) -> dict[str, Any]:
total = len(rows)
official_orig = sum(row["official_original"] for row in rows)
official_var = sum(row["official_variant"] for row in rows)
false_fail = sum(1 for row in rows if row["official_original"] and not row["official_variant"])
noun_orig = sum(row["official_noun_original"] for row in rows)
noun_var = sum(row["official_noun_variant"] for row in rows)
summary: dict[str, Any] = {
"family": "scene-synonym",
"pairs": total,
"official_full_prompt": {"original_pass": official_orig, "variant_pass": official_var, "false_failures": false_fail},
"official_scene_noun_only": {"original_pass": noun_orig, "variant_pass": noun_var, "false_failures": sum(1 for row in rows if row["official_noun_original"] and not row["official_noun_variant"])},
"by_pair": {},
}
for pair in sorted({row["pair"] for row in rows}):
subset = [row for row in rows if row["pair"] == pair]
summary["by_pair"][pair] = {
"n": len(subset),
"official_false_failures": sum(1 for row in subset if row["official_original"] and not row["official_variant"]),
}
if rows and rows[0]["repair_original"] is not None:
stable = sum(1 for row in rows if row["repair_original"] == row["repair_variant"])
supported = sum(1 for row in rows if row["repair_original"] == "supported" and row["repair_variant"] == "supported")
summary["repair"] = {
"label_stability": stable / total,
"supported_both": supported,
"supported_original": sum(1 for row in rows if row["repair_original"] == "supported"),
"supported_variant": sum(1 for row in rows if row["repair_variant"] == "supported"),
"label_counts": dict(Counter(row["repair_original"] for row in rows)),
}
return summary
def score_action_swap(rows: list[dict[str, Any]], args: argparse.Namespace) -> dict[str, Any]:
"""Repair-side check of the action interventions (official side needs UMT on the GPU host).
* class_swap: the parsed action set must change to the new class (sensitivity);
* out_of_vocabulary: the parsed action must be exactly ["other"] (protocol), never
the original class and never a guessed K400 class.
"""
counters: Counter[str] = Counter()
results: list[dict[str, Any]] = []
if args.adapters:
os.environ.setdefault("HF_HUB_OFFLINE", "1")
os.environ.setdefault("TRANSFORMERS_OFFLINE", "1")
from vbench_prompts_compile.inference import AdapterRouter
from vbench_prompts_compile import training as T
adapters = {}
for item in args.adapters:
task, _, path = item.partition("=")
adapters[task] = path
router = AdapterRouter(base_model_path=args.base_model, adapters=adapters, local_files_only=not args.allow_download)
else:
router = None
for row in rows:
entry = dict(row)
if router is not None:
original = router.predict("action", T.user_text({"task": "action", "input": {"prompt": row["original_prompt"]}}), max_new_tokens=64)
variant = router.predict("action", T.user_text({"task": "action", "input": {"prompt": row["variant_prompt"]}}), max_new_tokens=64)
entry["repair_original"] = (original.value or {}).get("actions")
entry["repair_variant"] = (variant.value or {}).get("actions")
if row["kind"] == "class_swap":
moved = bool(entry["repair_variant"]) and row["variant_action"] in entry["repair_variant"]
counters["class_swap_moved" if moved else "class_swap_not_moved"] += 1
else:
ok = entry["repair_variant"] == ["other"]
counters["oov_reported_other" if ok else "oov_not_other"] += 1
if entry["repair_original"] == ["other"]:
counters["original_already_other"] += 1
results.append(entry)
return {"rows": results, "counters": dict(counters)}
def main(argv: Sequence[str] | None = None) -> int:
args = parse_args(argv)
data_path = Path(args.data) if args.data else ROOT / "data" / "deterministic" / f"{args.family}.jsonl"
rows = [json.loads(line) for line in data_path.read_text(encoding="utf-8").splitlines() if line.strip()]
if args.limit:
rows = rows[: args.limit]
if args.family == "scene-synonym":
payload = score_scene_synonym(rows)
attach_repair(payload["rows"], args)
summary = summarize_scene_synonym(payload["rows"])
elif args.family == "action-swap":
payload = score_action_swap(rows, args)
summary = {"family": "action-swap", "pairs": len(rows), "repair": payload["counters"]}
else:
print(f"family {args.family!r} is not implemented yet", file=sys.stderr)
return 2
out = Path(args.output) if args.output else ROOT / "output" / "deterministic" / f"{args.family}-summary.json"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps({"summary": summary, "rows": payload["rows"]}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(json.dumps(summary, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|