File size: 2,867 Bytes
97c5704 | 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 | """Temperature scaling of option scores, and tables for the typed answers' probabilities.
- `calibration.json` holds a global temperature and optional per-type ones; scaling keeps an argmax but can move a
score's median or expected level.
- A typed probability goes through its table, a monotone map fitted on held-out rows: `found:text`,
`found:number`, `found:date`, `found:box`, `box` (a box's confidence), `found:boxes` (P(count >= k)), `boxes`
(each listed box's confidence), `multiple` (each option's P(yes)), `estimate` (a quantile level, read backwards by
`Calibration.level`).
"""
from __future__ import annotations
import json
from bisect import bisect_left
from dataclasses import dataclass, field
from pathlib import Path
from .prompt import Noul, Question, Score, cardinality
def bucket(kind: str, k: int) -> str:
if kind == "noul":
return "noul"
if kind == "score":
return f"score:{k}"
if k <= 3:
return "choice:small"
if k <= 8:
return "choice:mid"
if k <= 26:
return "choice:large"
return "choice:inventory"
def question_bucket(q: Question) -> str:
kind = "noul" if isinstance(q, Noul) else "score" if isinstance(q, Score) else "choice"
return bucket(kind, cardinality(q))
def through(value: float, knots: list[list[float]] | None) -> float:
"""`value` through increasing (raw, calibrated) knots, linear between them, flat past the ends."""
if not knots:
return value
index = bisect_left([raw for raw, _ in knots], value)
if index == 0:
return knots[0][1]
if index == len(knots):
return knots[-1][1]
(a, pa), (b, pb) = knots[index - 1], knots[index]
return pa + (value - a) * (pb - pa) / (b - a)
@dataclass
class Calibration:
temperatures: dict[str, float] = field(default_factory=dict)
default: float = 1.0
counts: dict[str, int] = field(default_factory=dict)
tables: dict[str, list[list[float]]] = field(default_factory=dict)
def temperature(self, q: Question) -> float:
return self.temperatures.get(question_bucket(q), self.default)
def apply(self, q: Question, scores: list[float]) -> list[float]:
t = self.temperature(q)
return scores if t == 1.0 else [s / t for s in scores]
def through(self, name: str, value: float) -> float:
return through(value, self.tables.get(name))
def level(self, name: str, share: float) -> float:
"""The level whose observed share is `share`: the table read backwards (both columns increase)."""
return through(share, [[y, x] for x, y in self.tables.get(name) or ()])
@classmethod
def load(cls, path: Path | str) -> "Calibration":
d = json.loads(Path(path).read_text())
return cls(d["temperatures"], d.get("default", 1.0), d.get("counts", {}), d.get("tables", {}))
|