Buckets:
| """Shared pieces: prompt format, letter readout, answer parsing, metrics. | |
| An item is a dict with | |
| id, set, family, level, tier, qtype ("choice" | "noul" | "score"), | |
| evidence (str), criterion (str), options ([{"id": str|None, "desc": str|None}]), | |
| target ([float] over options, sums to 1), gold (int, argmax of target). | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import math | |
| import random | |
| import re | |
| import numpy as np | |
| SYSTEM = ("Apply the supplied criterion to the supplied evidence. Choose exactly one listed option. " | |
| "Your final answer must be only that option's uppercase letter, with nothing else.") | |
| LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" | |
| MAX_OPTIONS = len(LETTERS) | |
| CONTEXT = 32768 | |
| MAX_PROMPT_TOKENS = 16000 | |
| # ------------------------------------------------------------------ prompt | |
| def option_text(opt: dict) -> str: | |
| """'<id>: <text>' as in SemIf / OpenDecisions; whichever part exists otherwise.""" | |
| oid, desc = opt.get("id"), opt.get("desc") | |
| if oid is not None and desc: | |
| return f"{oid}: {desc}" | |
| return str(desc if desc else oid) | |
| def user_message(item: dict, order: list[int] | None = None) -> str: | |
| order = list(range(len(item["options"]))) if order is None else order | |
| payload = {"evidence": item["evidence"], "criterion": item["criterion"], | |
| "options": [{"letter": LETTERS[i], "description": option_text(item["options"][j])} | |
| for i, j in enumerate(order)]} | |
| return json.dumps(payload, ensure_ascii=False) | |
| def messages(item: dict, order: list[int] | None = None) -> list[dict]: | |
| return [{"role": "system", "content": SYSTEM}, {"role": "user", "content": user_message(item, order)}] | |
| def render(tok, item: dict, think: bool, order: list[int] | None = None) -> str: | |
| return tok.apply_chat_template(messages(item, order), tokenize=False, add_generation_prompt=True, | |
| enable_thinking=think) | |
| def shuffled_order(item: dict, rng: random.Random) -> list[int]: | |
| """Choice and yes/no options are shuffled; score levels keep their order.""" | |
| order = list(range(len(item["options"]))) | |
| if item["qtype"] != "score": | |
| rng.shuffle(order) | |
| return order | |
| def letter_ids(tok) -> list[int]: | |
| ids = [] | |
| for letter in LETTERS: | |
| enc = tok.encode(letter, add_special_tokens=False) | |
| if len(enc) != 1 or tok.decode(enc) != letter: | |
| raise ValueError(f"letter {letter!r} is not a single token") | |
| ids.append(enc[0]) | |
| return ids | |
| # ------------------------------------------------------------------ parsing a thinking-mode answer | |
| _LETTER_RE = re.compile(r"(?<![A-Za-z])([A-Z])(?![A-Za-z])") | |
| def split_chain(text: str) -> tuple[str | None, str]: | |
| """(chain, answer part). chain is None when the model never closed its think block.""" | |
| end = text.find("</think>") | |
| if end < 0: | |
| return None, "" | |
| chain = text[:end] | |
| if chain.startswith("<think>"): | |
| chain = chain[len("<think>"):] | |
| return chain.strip("\n"), text[end + len("</think>"):] | |
| def parse_letter(answer: str, k: int) -> int | None: | |
| """Index of the first standalone capital letter among the first k letters, else None.""" | |
| for m in _LETTER_RE.finditer(answer): | |
| idx = LETTERS.find(m.group(1)) | |
| if 0 <= idx < k: | |
| return idx | |
| return None | |
| # ------------------------------------------------------------------ metrics | |
| def softmax(x: np.ndarray, t: float = 1.0) -> np.ndarray: | |
| z = np.asarray(x, dtype=np.float64) / t | |
| z = z - z.max() | |
| e = np.exp(z) | |
| return e / e.sum() | |
| def ece(conf: np.ndarray, correct: np.ndarray, bins: int = 15) -> float: | |
| conf, correct = np.asarray(conf, float), np.asarray(correct, float) | |
| if len(conf) == 0: | |
| return float("nan") | |
| idx = np.minimum((conf * bins).astype(int), bins - 1) | |
| total = 0.0 | |
| for b in range(bins): | |
| m = idx == b | |
| if m.any(): | |
| total += m.mean() * abs(correct[m].mean() - conf[m].mean()) | |
| return float(total) | |
| def nll(probs: list[np.ndarray], gold: list[int]) -> float: | |
| return float(np.mean([-math.log(max(p[g], 1e-12)) for p, g in zip(probs, gold)])) | |
| def brier(probs: list[np.ndarray], gold: list[int]) -> float: | |
| out = [] | |
| for p, g in zip(probs, gold): | |
| y = np.zeros_like(p) | |
| y[g] = 1.0 | |
| out.append(float(((p - y) ** 2).sum())) | |
| return float(np.mean(out)) | |
| def auroc(score: np.ndarray, label: np.ndarray) -> float: | |
| """Probability that a random positive outranks a random negative (ties count half).""" | |
| score, label = np.asarray(score, float), np.asarray(label, bool) | |
| pos, neg = score[label], score[~label] | |
| if len(pos) == 0 or len(neg) == 0: | |
| return float("nan") | |
| order = np.argsort(np.concatenate([pos, neg]), kind="mergesort") | |
| ranks = np.empty(len(order)) | |
| allv = np.concatenate([pos, neg])[order] | |
| i = 0 | |
| while i < len(allv): # average ranks over ties | |
| j = i | |
| while j + 1 < len(allv) and allv[j + 1] == allv[i]: | |
| j += 1 | |
| ranks[order[i:j + 1]] = (i + j) / 2 + 1 | |
| i = j + 1 | |
| return float((ranks[:len(pos)].sum() - len(pos) * (len(pos) + 1) / 2) / (len(pos) * len(neg))) | |
| def fit_temperature(logits: list[np.ndarray], gold: list[int], lo: float = 0.05, hi: float = 20.0) -> float: | |
| """NLL-minimising temperature by golden-section search on log T.""" | |
| def loss(log_t): | |
| t = math.exp(log_t) | |
| return nll([softmax(z, t) for z in logits], gold) | |
| a, b = math.log(lo), math.log(hi) | |
| g = (math.sqrt(5) - 1) / 2 | |
| c, d = b - g * (b - a), a + g * (b - a) | |
| for _ in range(60): | |
| if loss(c) < loss(d): | |
| b = d | |
| else: | |
| a = c | |
| c, d = b - g * (b - a), a + g * (b - a) | |
| return math.exp((a + b) / 2) | |
| def paired_bootstrap(a: np.ndarray, b: np.ndarray, n: int = 1000, seed: int = 0, stat=np.mean): | |
| """Mean of a - b and its 95% interval, resampling items jointly.""" | |
| a, b = np.asarray(a, float), np.asarray(b, float) | |
| rng = np.random.default_rng(seed) | |
| idx = rng.integers(0, len(a), size=(n, len(a))) | |
| diffs = np.array([stat(a[i]) - stat(b[i]) for i in idx]) | |
| return float(stat(a) - stat(b)), float(np.percentile(diffs, 2.5)), float(np.percentile(diffs, 97.5)) | |
| def read_jsonl(path) -> list[dict]: | |
| with open(path, encoding="utf-8") as f: | |
| return [json.loads(line) for line in f if line.strip()] | |
| def write_jsonl(path, rows) -> None: | |
| with open(path, "w", encoding="utf-8") as f: | |
| for r in rows: | |
| f.write(json.dumps(r, ensure_ascii=False) + "\n") | |
Xet Storage Details
- Size:
- 6.62 kB
- Xet hash:
- eb1bdc1074b2702a1547f4ce2a0cd247c4cf792488cac7020cc127cdef892e89
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.