From 45f303c2f77bedf0ec0a005e9920fa0afc313906 Mon Sep 17 00:00:00 2001 From: Marcos Lohmann Date: Wed, 7 Oct 2026 18:24:10 -0300 Subject: [PATCH] =?UTF-8?q?Calibra=C3=A7=C3=A3o=20por=20temperatura,=20lim?= =?UTF-8?q?iares=20de=20absten=C3=A7=C3=A3o=20e=20congelamento=20(tarefas?= =?UTF-8?q?=2060=E2=80=9362)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- analysis/calibrate.py | 114 ++++++++++++++++++++++++++++++++++++++++ harness/evaluate.py | 8 ++- tests/test_calibrate.py | 38 ++++++++++++++ tools/__init__.py | 0 tools/freeze.py | 59 +++++++++++++++++++++ 5 files changed, 217 insertions(+), 2 deletions(-) create mode 100644 analysis/calibrate.py create mode 100644 tests/test_calibrate.py create mode 100644 tools/__init__.py create mode 100644 tools/freeze.py diff --git a/analysis/calibrate.py b/analysis/calibrate.py new file mode 100644 index 0000000..c7e1316 --- /dev/null +++ b/analysis/calibrate.py @@ -0,0 +1,114 @@ +"""Calibração e limiares de abstenção, só com a partição de calibração (tarefas 60–61; Q5, Q10). + +Para cada candidato × dispositivo: +- temperature scaling sobre log-probabilidades (um parâmetro T, minimizando NLL); +- o T só é aceito se reduzir o ECE em validação cruzada de 5 dobras por `group_id`; senão T = 1; +- limiar de abstenção sobre max(p) calibrado: a maior cobertura com taxa de rota errada com + confiança alta ≤ 2% (Q5). Abster-se = encaminhar para desambiguação. + +Uso: python -m analysis.calibrate reports/calibration → config/calibration.json +""" +import json +import math +import sys +from collections import defaultdict +from pathlib import Path + +from analysis.metrics import calibration, selective + +ROOT = Path(__file__).resolve().parents[1] +OUT = ROOT / "config" / "calibration.json" +MAX_CONFIDENT_WRONG = 0.02 +MIN_COVERAGE = 0.80 +GRID = [math.exp(x / 40) for x in range(-160, 121)] # T de ~0,018 a ~20 + + +def group_of(example_id: str) -> str: + return example_id.rsplit("-", 2)[0] # "--D1" + + +def apply_temperature(probs: dict, t: float) -> dict: + logits = {k: math.log(max(v, 1e-12)) / t for k, v in probs.items()} + m = max(logits.values()) + exp = {k: math.exp(v - m) for k, v in logits.items()} + z = sum(exp.values()) + return {k: v / z for k, v in exp.items()} + + +def calibrated_rows(rows: list[dict], t: float) -> list[dict]: + out = [] + for r in rows: + p = apply_temperature(r["probabilities"], t) + out.append({**r, "probabilities": p, "pred": max(p, key=p.get)}) + return out + + +def nll(rows: list[dict], t: float) -> float: + return -sum(math.log(max(apply_temperature(r["probabilities"], t)[r["gold"]], 1e-12)) for r in rows) / len(rows) + + +def fit_temperature(rows: list[dict]) -> float: + return min(GRID, key=lambda t: nll(rows, t)) + + +def cross_validated_ece(rows: list[dict], folds: int = 5) -> tuple[float, float]: + groups = sorted({group_of(r["id"]) for r in rows}) + fold_of = {g: i % folds for i, g in enumerate(groups)} + raw, cal = [], [] + for k in range(folds): + train = [r for r in rows if fold_of[group_of(r["id"])] != k] + test = [r for r in rows if fold_of[group_of(r["id"])] == k] + if not test or not train: + continue + t = fit_temperature(train) + raw.append(calibration(test)["ece"] * len(test)) + cal.append(calibration(calibrated_rows(test, t))["ece"] * len(test)) + n = len(rows) + return sum(raw) / n, sum(cal) / n + + +def choose_threshold(rows: list[dict]) -> dict: + confidences = sorted({round(max(r["probabilities"].values()), 6) for r in rows}) + best = selective(rows, 1.01) # abstém de tudo: cobertura 0 + for thr in confidences: + s = selective(rows, thr) + if s["confident_wrong_rate"] <= MAX_CONFIDENT_WRONG: + best = s + break # menor limiar que respeita o risco = maior cobertura + return best + + +def main(): + src = Path(sys.argv[1]) + result = {"method": "temperature_scaling", "fit_split": "calibration", + "max_confident_wrong_rate": MAX_CONFIDENT_WRONG, "min_coverage": MIN_COVERAGE, "candidates": {}} + for f in sorted(src.glob("predictions-*.jsonl")): + rows = [json.loads(l) for l in f.open()] + rows = [r for r in rows if r.get("probabilities")] + cand = json.loads((src / f.name.replace("predictions-", "summary-").replace(".jsonl", ".json")) + .read_text())["candidate"]["name"] + device = f.stem.rsplit("-", 1)[1] + t = fit_temperature(rows) + ece_cv_raw, ece_cv_cal = cross_validated_ece(rows) + accepted = ece_cv_cal < ece_cv_raw + final_t = t if accepted else 1.0 + cal_rows = calibrated_rows(rows, final_t) + thr = choose_threshold(cal_rows) + result["candidates"].setdefault(cand, {})[device] = { + "temperature": round(final_t, 4), "fitted_temperature": round(t, 4), "accepted": accepted, + "n": len(rows), "ece_raw": round(calibration(rows)["ece"], 4), + "ece_calibrated": round(calibration(cal_rows)["ece"], 4), + "ece_cv_raw": round(ece_cv_raw, 4), "ece_cv_calibrated": round(ece_cv_cal, 4), + "nll_raw": round(nll(rows, 1.0), 4), "nll_calibrated": round(nll(rows, final_t), 4), + "threshold": round(thr["threshold"], 6), "coverage": round(thr["coverage"] or 0, 4), + "selective_accuracy": round(thr["selective_accuracy"] or 0, 4), + "confident_wrong_rate": round(thr["confident_wrong_rate"] or 0, 4), + "meets_min_coverage": (thr["coverage"] or 0) >= MIN_COVERAGE, + } + print(cand, device, json.dumps(result["candidates"][cand][device])) + OUT.write_text(json.dumps(result, indent=2, ensure_ascii=False) + "\n") + print("→", OUT.relative_to(ROOT)) + + +if __name__ == "__main__": + main() diff --git a/harness/evaluate.py b/harness/evaluate.py index 82e13ec..bed7751 100644 --- a/harness/evaluate.py +++ b/harness/evaluate.py @@ -98,8 +98,12 @@ def main(): ap.add_argument("--state-budget", type=int, default=int(os.environ.get("DMB_STATE_BUDGET", "512"))) args = ap.parse_args() - if args.split == "test" and not FROZEN.is_file(): - raise SystemExit("partição de teste bloqueada: configuração ainda não congelada (tarefa 62)") + if args.split == "test": + if not FROZEN.is_file(): + raise SystemExit("partição de teste bloqueada: configuração ainda não congelada (tarefa 62)") + from tools.freeze import current_hashes + if current_hashes() != json.loads(FROZEN.read_text())["files"]: + raise SystemExit("partição de teste bloqueada: arquivos mudaram depois do congelamento") enforce_offline() from dmb.offline import self_test diff --git a/tests/test_calibrate.py b/tests/test_calibrate.py new file mode 100644 index 0000000..d9c23c3 --- /dev/null +++ b/tests/test_calibrate.py @@ -0,0 +1,38 @@ +import json +from pathlib import Path + +import pytest + +from analysis.calibrate import apply_temperature, choose_threshold, fit_temperature + +ROOT = Path(__file__).resolve().parents[1] + + +def test_temperature_preserves_ranking_and_sums_to_one(): + p = {"a": 0.7, "b": 0.2, "c": 0.1} + q = apply_temperature(p, 2.0) + assert sum(q.values()) == pytest.approx(1.0) + assert max(q, key=q.get) == "a" and q["a"] < p["a"] # T > 1 suaviza + + +def test_fit_softens_overconfident_model(): + rows = [{"gold": "a", "probabilities": {"a": 0.99, "b": 0.01}}] * 6 + \ + [{"gold": "b", "probabilities": {"a": 0.99, "b": 0.01}}] * 4 + assert fit_temperature(rows) > 1.0 + + +def test_threshold_respects_risk(): + rows = [{"gold": "a", "pred": "a", "probabilities": {"a": 0.9, "b": 0.1}}] * 98 + \ + [{"gold": "b", "pred": "a", "probabilities": {"a": 0.6, "b": 0.4}}] * 2 + \ + [{"gold": "b", "pred": "a", "probabilities": {"a": 0.95, "b": 0.05}}] * 1 + s = choose_threshold(rows) + assert s["confident_wrong_rate"] <= 0.02 + + +def test_frozen_files_unchanged(): + """Depois do congelamento (tarefa 62), nada do que foi congelado pode mudar.""" + frozen = ROOT / "config" / "frozen.json" + if not frozen.exists(): + pytest.skip("configuração ainda não congelada") + from tools.freeze import current_hashes + assert current_hashes() == json.loads(frozen.read_text())["files"] diff --git a/tools/__init__.py b/tools/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/freeze.py b/tools/freeze.py new file mode 100644 index 0000000..3b40efe --- /dev/null +++ b/tools/freeze.py @@ -0,0 +1,59 @@ +"""Congela a configuração antes da abertura do teste final (tarefa 62). + +Grava config/frozen.json com o SHA-256 da configuração, da calibração, das partições, do manifesto +de artefatos, do código dos adaptadores, do contrato e do executor, e com os ambientes publicados de +cada candidato. A presença desse arquivo libera a partição de teste em harness/evaluate.py; depois +do congelamento, prompts, opções, pesos, calibradores e limiares não mudam (os testes conferem). + +Uso: PYTHONPATH=infra/oci .venv/bin/python tools/freeze.py +""" +import hashlib +import json +import sys +from datetime import datetime, timezone +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +FROZEN = ROOT / "config" / "frozen.json" +TRACKED = ["config/benchmark.yaml", "config/calibration.json", "data/splits/MANIFEST.json", + "env/artifacts-manifest.json", "contracts/jev/schemas/request.schema.json", + "contracts/jev/schemas/response.schema.json", "harness/evaluate.py", "dmb/canonical.py", + "dmb/jev.py", "dmb/server.py", "dmb/adapter.py", "adapters/laya/adapter.py", + "adapters/semif/adapter.py", "adapters/gliner/adapter.py", "adapters/rizzo_flow/adapter.py"] + + +def sha(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def current_hashes() -> dict: + return {p: sha(ROOT / p) for p in TRACKED} + + +def env_snapshot() -> dict: + sys.path.insert(0, str(ROOT)) + from harness import objstore + out = {} + for cand in ("laya", "gliner", "semif", "rizzo_flow"): + tmp = Path("/tmp") / f"freeze-{cand}-latest.json" + out[cand] = json.loads(objstore.get("dmb-artefatos", f"envs/{cand}/latest.json", tmp).read_text()) + cuda = Path("/tmp/freeze-cuda-latest.json") + out["rizzo_flow_cuda_runtime"] = json.loads( + objstore.get("dmb-artefatos", "runtimes/rizzo_flow/cuda-latest.json", cuda).read_text()) + return out + + +def main(): + if FROZEN.exists(): + sys.exit("configuração já congelada: config/frozen.json existe") + missing = [p for p in TRACKED if not (ROOT / p).is_file()] + if missing: + sys.exit(f"faltam arquivos para congelar: {missing}") + frozen = {"frozen_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + "files": current_hashes(), "environments": env_snapshot()} + FROZEN.write_text(json.dumps(frozen, indent=2, ensure_ascii=False) + "\n") + print(f"congelado: {len(frozen['files'])} arquivos e {len(frozen['environments'])} ambientes") + + +if __name__ == "__main__": + main()