Make building training loop calibration-gated
This commit is contained in:
Jens
@@ -29,6 +29,45 @@ def write_json(path: Path, value: dict[str, Any]) -> None:
|
||||
temporary.replace(path)
|
||||
|
||||
|
||||
def select_calibration_threshold(report: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Choose a threshold without consulting test or background evidence."""
|
||||
eligible = [item for item in report["sweeps"] if item["pure_empty_false_positives"] == 0]
|
||||
if not eligible:
|
||||
eligible = report["sweeps"]
|
||||
return max(
|
||||
eligible,
|
||||
key=lambda item: (
|
||||
min(region["f1"] for region in item["regions"].values()),
|
||||
item["aggregate"]["f1"],
|
||||
-item["pure_empty_false_positives"],
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def calibration_failures(
|
||||
chosen: dict[str, Any],
|
||||
*,
|
||||
min_aggregate_f1: float,
|
||||
min_region_f1: float,
|
||||
min_region_precision: float,
|
||||
min_region_recall: float,
|
||||
max_pure_empty_fp: int,
|
||||
) -> list[str]:
|
||||
failures: list[str] = []
|
||||
if chosen["aggregate"]["f1"] < min_aggregate_f1:
|
||||
failures.append("calibration_aggregate_f1_below_gate")
|
||||
for region, values in chosen["regions"].items():
|
||||
if values["f1"] < min_region_f1:
|
||||
failures.append(f"calibration_{region}_f1_below_gate")
|
||||
if values["precision"] < min_region_precision:
|
||||
failures.append(f"calibration_{region}_precision_below_gate")
|
||||
if values["recall"] < min_region_recall:
|
||||
failures.append(f"calibration_{region}_recall_below_gate")
|
||||
if chosen["pure_empty_false_positives"] > max_pure_empty_fp:
|
||||
failures.append("calibration_pure_empty_false_positive_gate_failed")
|
||||
return failures
|
||||
|
||||
|
||||
def training_command(
|
||||
yolo: str,
|
||||
*,
|
||||
@@ -112,6 +151,11 @@ def main() -> int:
|
||||
parser.add_argument("--translate", type=float, default=0.1)
|
||||
parser.add_argument("--seed", type=int, default=20260731)
|
||||
parser.add_argument("--yolo", default="yolo")
|
||||
parser.add_argument("--min-aggregate-f1", type=float, default=0.55)
|
||||
parser.add_argument("--min-region-f1", type=float, default=0.45)
|
||||
parser.add_argument("--min-region-precision", type=float, default=0.5)
|
||||
parser.add_argument("--min-region-recall", type=float, default=0.4)
|
||||
parser.add_argument("--max-pure-empty-fp", type=int, default=0)
|
||||
parser.add_argument("--dry-run", action="store_true")
|
||||
args = parser.parse_args()
|
||||
if args.iterations < 1:
|
||||
@@ -174,11 +218,7 @@ def main() -> int:
|
||||
shutil.copy2(best, candidate)
|
||||
|
||||
reports: dict[str, Path] = {}
|
||||
for role, summary in (
|
||||
("calibration", args.calibration_summary),
|
||||
("test", args.test_summary),
|
||||
("background", args.background_summary),
|
||||
):
|
||||
for role, summary in (("calibration", args.calibration_summary),):
|
||||
report = iteration_dir / f"{role}.json"
|
||||
reports[role] = report
|
||||
run(
|
||||
@@ -203,22 +243,64 @@ def main() -> int:
|
||||
iteration_dir / f"{role}.log",
|
||||
)
|
||||
assessment = iteration_dir / "assessment.json"
|
||||
run(
|
||||
[
|
||||
sys.executable,
|
||||
str(scripts_dir / "assess_belgium_building_training_iteration.py"),
|
||||
"--calibration",
|
||||
str(reports["calibration"]),
|
||||
"--test",
|
||||
str(reports["test"]),
|
||||
"--background",
|
||||
str(reports["background"]),
|
||||
"--output",
|
||||
str(assessment),
|
||||
],
|
||||
iteration_dir / "assessment.log",
|
||||
allowed={0, 2},
|
||||
calibration = json.loads(reports["calibration"].read_text(encoding="utf-8"))
|
||||
chosen = select_calibration_threshold(calibration)
|
||||
failures = calibration_failures(
|
||||
chosen,
|
||||
min_aggregate_f1=args.min_aggregate_f1,
|
||||
min_region_f1=args.min_region_f1,
|
||||
min_region_precision=args.min_region_precision,
|
||||
min_region_recall=args.min_region_recall,
|
||||
max_pure_empty_fp=args.max_pure_empty_fp,
|
||||
)
|
||||
if failures:
|
||||
write_json(
|
||||
assessment,
|
||||
{
|
||||
"schema_version": 1,
|
||||
"status": "continue_training_loop",
|
||||
"phase": "calibration_rejected",
|
||||
"threshold_selection_source": "calibration_only",
|
||||
"selected_threshold": chosen["threshold"],
|
||||
"calibration": chosen,
|
||||
"test": None,
|
||||
"background": None,
|
||||
"failures": failures,
|
||||
},
|
||||
)
|
||||
else:
|
||||
for role, summary in (
|
||||
("test", args.test_summary),
|
||||
("background", args.background_summary),
|
||||
):
|
||||
report = iteration_dir / f"{role}.json"
|
||||
reports[role] = report
|
||||
run(
|
||||
[
|
||||
sys.executable,
|
||||
str(scripts_dir / "evaluate_belgium_building_candidate.py"),
|
||||
"--model", str(candidate),
|
||||
"--summary", str(summary),
|
||||
"--corpus-manifest", str(args.corpus_manifest),
|
||||
"--output", str(report),
|
||||
"--device", "cuda:0",
|
||||
"--max-det", str(args.max_det),
|
||||
"--imgsz", str(args.imgsz),
|
||||
],
|
||||
iteration_dir / f"{role}.log",
|
||||
)
|
||||
run(
|
||||
[
|
||||
sys.executable,
|
||||
str(scripts_dir / "assess_belgium_building_training_iteration.py"),
|
||||
"--calibration", str(reports["calibration"]),
|
||||
"--test", str(reports["test"]),
|
||||
"--background", str(reports["background"]),
|
||||
"--output", str(assessment),
|
||||
],
|
||||
iteration_dir / "assessment.log",
|
||||
allowed={0, 2},
|
||||
)
|
||||
decision = json.loads(assessment.read_text(encoding="utf-8"))
|
||||
record = {
|
||||
"iteration": index,
|
||||
|
||||
Reference in New Issue
Block a user