File size: 4,563 Bytes
7845694 | 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 | """Recompute output-blind stability reports for direct logits and reranker."""
import argparse
from collections import defaultdict
import json
from pathlib import Path
import statistics
import evaluate
def read(path: Path):
return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
def indexed(rows):
result = {row["id"]: row for row in rows}
if len(result) != len(rows):
raise ValueError("Duplicate IDs")
return result
def chosen(row):
return row["option_ids"][max(range(len(row["probabilities"])), key=row["probabilities"].__getitem__)]
def distribution(row):
return dict(zip(row["option_ids"], row["probabilities"]))
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--gold", type=Path, required=True)
parser.add_argument("--perturbations", type=Path, required=True)
parser.add_argument("--direct-base", type=Path, required=True)
parser.add_argument("--direct-perturbations", type=Path, required=True)
parser.add_argument("--reranker-base", type=Path, required=True)
parser.add_argument("--reranker-perturbations", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
gold, perturb_gold = read(args.gold), read(args.perturbations)
original_ids = {row["id"] for row in gold if row["provenance"]["variant"] == "original"}
missing_gold = [row for row in gold if row["provenance"]["variant"] == "missing"]
if len(original_ids) != 36 or len(missing_gold) != 36 or len(perturb_gold) != 108:
raise ValueError("Unexpected stability populations")
systems = {
"direct_logits": (args.direct_base, args.direct_perturbations),
"reranker": (args.reranker_base, args.reranker_perturbations),
}
result = {
"meaning": "Stability under output-blind project-authored variants; accuracy against unchanged labels plus semantic-ID-aligned probability movement.",
"systems": {},
}
variants = defaultdict(list)
for row in perturb_gold:
variants[row["provenance"]["variant"]].append(row)
for name, (base_path, perturb_path) in systems.items():
base, perturb = indexed(read(base_path)), indexed(read(perturb_path))
if set(perturb) != {row["id"] for row in perturb_gold} or not original_ids <= set(base):
raise ValueError(f"Incomplete {name} stability outputs")
report = {
"base_original": evaluate.evaluate(
[row for row in gold if row["id"] in original_ids],
[base[row["id"]] for row in gold if row["id"] in original_ids],
),
"variants": {},
}
for variant, variant_gold in variants.items():
shifts, flips = [], []
for row in variant_gold:
candidate = perturb[row["id"]]
reference = base[row["provenance"]["base_id"]]
left, right = distribution(reference), distribution(candidate)
if set(left) != set(right):
raise ValueError("Semantic option IDs changed")
shifts.append(max(abs(left[key] - right[key]) for key in left))
if chosen(reference) != chosen(candidate):
flips.append({"base_id": row["provenance"]["base_id"], "variant_id": row["id"],
"from": chosen(reference), "to": chosen(candidate)})
report["variants"][variant] = {
"evaluation": evaluate.evaluate(variant_gold, [perturb[row["id"]] for row in variant_gold]),
"argmax_flips": len(flips),
"flip_rows": flips,
"mean_max_probability_movement": statistics.mean(shifts),
"max_probability_movement": max(shifts),
}
missing_predictions = [base[row["id"]] for row in missing_gold]
aligned = evaluate.align(missing_gold, missing_predictions)
report["missing_evidence"] = {
"evaluation": evaluate.evaluate(missing_gold, missing_predictions),
"confident_non_insufficient_at_0_8": sum(
row["predicted_id"] != "insufficient" and row["confidence"] is not None and row["confidence"] >= .8
for row in aligned
),
}
result["systems"][name] = report
with args.output.open("x") as destination:
destination.write(json.dumps(result, indent=2, allow_nan=False) + "\n")
if __name__ == "__main__":
main()
|