VLM 读图评测：先审计评分器，再判断读错还是算错

"""CPU evaluation-protocol experiment; NO image/OCR/VLM inference.

Python >=3.10, standard library only. Synthetic predictions are fault injections.
The local metrics are explicit diagnostic policies, not official submissions.
"""
import argparse
from collections import defaultdict
from decimal import Decimal, InvalidOperation
import json
from pathlib import Path
import random
import re
import sys

ROOT = Path(__file__).resolve().parent


def normalize(s):
    return " ".join(s.lower().split())


def distance(a, b):
    row = list(range(len(b) + 1))
    for i, x in enumerate(a, 1):
        new = [i]
        for j, y in enumerate(b, 1):
            new.append(min(new[-1] + 1, row[j] + 1, row[j - 1] + (x != y)))
        row = new
    return row[-1]


def anls(golds, pred):
    """Normalized strings, strict distance < .5; empty prediction is failure."""
    p = normalize(pred)
    if not p:
        return 0.0
    scores = []
    for gold in golds:
        g = normalize(gold)
        n = max(len(g), len(p))
        d = distance(g, p)
        scores.append(1 - d / n if 2 * d < n else 0.0)
    return max(scores)


def number(text):
    """Only one finite decimal, optionally %. No unit deletion/regex extraction."""
    s = text.strip()
    if not re.fullmatch(r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)%?", s):
        return None
    try:
        return Decimal(s.rstrip("%")) / (100 if s.endswith("%") else 1)
    except InvalidOperation:
        return None


def numeric(golds, pred, tolerance="0.05"):
    p = number(pred)
    if p is None:
        return 0.0
    for gold in golds:
        g = number(gold)
        if g is None:
            raise ValueError("Non-numeric gold in numeric task")
        if abs(p - g) <= Decimal(tolerance) * abs(g):
            return 1.0
    return 0.0


def generate(seed=72):
    """Twenty synthetic document SPECS, not raster images; split by document."""
    rng = random.Random(seed)
    records = []
    for doc in range(20):
        a, b = rng.randrange(120, 201), rng.randrange(20, 81)
        doc_id = f"doc_{doc:02d}"
        split = "dev" if doc < 5 else "test"
        identifier = f"AB{rng.randrange(100000, 1000000)}"
        spec = {"identifier": identifier, "series": {"cyan": a, "amber": b},
                "unit": "count", "percent_question_unit": "percent"}
        questions = [
            ("ocr", "Copy the document identifier.", identifier, identifier[:-1] + ("0" if identifier[-1] != "0" else "1"), "character_substitution"),
            ("lookup", "What is the cyan value?", str(a), str(b), "series_binding"),
            ("difference", "What is cyan minus amber?", str(a-b), str(a+b), "operator_swap"),
            ("percentage", "Cyan as a percent of total? Round to two decimals.",
             f"{100*a/(a+b):.2f}%", f"{100*a/(a+b):.2f}", "percent_unit_drop"),
        ]
        for task, question, answer, faulty, cause in questions:
            records.append({"question_id": f"{doc_id}_{task}", "doc_id": doc_id,
                            "split": split, "task": task, "question": question,
                            "answers": [answer], "spec": spec,
                            "synthetic_fault_answer": faulty, "injected_cause": cause})
    return records


def predictions(records, mode):
    rows = []
    for r in records:
        g = r["answers"][0]
        if mode == "gold_fixture":
            p = g
        elif mode == "mixed_faults":
            p = r["synthetic_fault_answer"]
        elif mode == "format_prefix":
            p = "The answer is " + g
        else:
            factor = Decimal("1.04" if mode == "numeric_4pct" else "1.06")
            p = g if r["task"] == "ocr" else format(number(g) * factor, "f")
        rows.append({"question_id": r["question_id"], "answer": p,
                     "provenance": "synthetic_fault_injection", "mode": mode})
    return rows


def evaluate(records, predictions_):
    if not records:
        raise ValueError("Empty evaluation set")
    ids = [r["question_id"] for r in records]
    pids = [p["question_id"] for p in predictions_]
    if len(set(ids)) != len(ids) or len(set(pids)) != len(pids):
        raise ValueError("Duplicate question_id")
    if set(ids) != set(pids):
        raise ValueError("Missing or extra predictions; never silently drop failures")
    if any(not isinstance(p["answer"], str) for p in predictions_):
        raise ValueError("Answers must be strings")
    lookup = {p["question_id"]: p["answer"] for p in predictions_}
    totals = defaultdict(list)
    output = []
    for r in records:
        p = lookup[r["question_id"]]
        em = float(any(normalize(p) == normalize(g) for g in r["answers"]))
        totals["exact_match_all"].append(em)
        row = {"question_id": r["question_id"], "task": r["task"],
               "answers": r["answers"], "prediction": p, "exact_match": em}
        if r["task"] == "ocr":
            score = anls(r["answers"], p)
            row["anls"] = score
            totals["anls_ocr"].append(score)
        else:
            row["relaxed"] = numeric(r["answers"], p)
            row["numeric_exact"] = numeric(r["answers"], p, "0")
            totals["relaxed_numeric"].append(row["relaxed"])
            totals["numeric_exact"].append(row["numeric_exact"])
        output.append(row)
    summary = {k: {"mean": sum(v)/len(v), "n": len(v)} for k, v in totals.items()}
    return summary, output


def dump(path, obj):
    path.write_text(json.dumps(obj, ensure_ascii=False, indent=2, allow_nan=False) + "\n")


def read_jsonl(path):
    return [json.loads(line) for line in Path(path).read_text().splitlines() if line.strip()]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--out", type=Path, default=ROOT / "results")
    parser.add_argument("--gold", type=Path, help="External gold JSONL using the same schema")
    parser.add_argument("--predictions", type=Path, help="External predictions JSONL")
    args = parser.parse_args()
    args.out.mkdir(parents=True, exist_ok=True)
    if bool(args.gold) != bool(args.predictions):
        parser.error("--gold and --predictions must be supplied together")
    if args.gold:
        summary, rows = evaluate(read_jsonl(args.gold), read_jsonl(args.predictions))
        dump(args.out / "external_summary.json", summary)
        dump(args.out / "external_rows.json", rows)
        print(json.dumps(summary, indent=2))
        return
    records = generate()
    dev_docs = {r["doc_id"] for r in records if r["split"] == "dev"}
    test_docs = {r["doc_id"] for r in records if r["split"] == "test"}
    assert not dev_docs & test_docs
    for split in ["dev", "test"]:
        subset = [r for r in records if r["split"] == split]
        (args.out / f"{split}.jsonl").write_text("".join(json.dumps(r) + "\n" for r in subset))
    test = [r for r in records if r["split"] == "test"]
    report = {"python": sys.version, "seed": 72, "model_executed": False,
              "data": {"specs": 20, "dev_questions": 20, "test_questions": 60,
                       "numeric_test_questions": 45, "ocr_test_questions": 15}, "modes": {}}
    for mode in ["gold_fixture", "numeric_4pct", "numeric_6pct", "mixed_faults", "format_prefix"]:
        preds = predictions(test, mode)
        summary, rows = evaluate(test, preds)
        report["modes"][mode] = summary
        dump(args.out / f"{mode}_rows.json", rows)
        (args.out / f"{mode}_predictions.jsonl").write_text("".join(json.dumps(r)+"\n" for r in preds))
    dump(args.out / "summary.json", report)
    print("CHECKPOINT seed=72 spec_shape=(20,2) test_records=60 OCR=15 numeric=45")
    print("No images, OCR system, neural model, or benchmark evaluation executed.")
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()