"""Torch-free verbatim extract of the tokenization/confidence functions from the AUDITED laya==0.3.5 package (src/laya/common.py), so the plugin runtime venv needs no torch (deployment footprint claim). If the audited package changes, regenerate this file; do not hand-edit the logic. Extracted verbatim: QTYPES, QTYPE_NAMES, serialize_state, render_criterion, render_options, build_sequence, ece_score, confidence_from_probs, temp_bucket, TEMP_MIN, TEMP_MAX, clamp_temperature. """ import json import math from typing import Dict, List, Optional, Union import numpy as np QTYPES = {"choice": 0, "score": 1, "noul": 2} QTYPE_NAMES = {v: k for k, v in QTYPES.items()} def serialize_state(state: Union[str, dict, list]) -> str: if isinstance(state, str): return state return json.dumps(state, ensure_ascii=False) def render_criterion(value) -> str: """Render one criterion value as text. Strings pass through; anything structured (dict, list, number) becomes compact JSON, so a rubric reads as JSON rather than a Python repr. Without this a dict-valued criterion crashed `noul` outright and leaked `{'desc': ...}` into `choice` and `score` prompts. """ if isinstance(value, str): return value return json.dumps(value, ensure_ascii=False, separators=(", ", ": "), default=str) def render_options(q: Dict) -> List[str]: """Render option texts in label-index order. Noul is always [false, true].""" t, crit = q["t"], q.get("crit") if t == "choice": # only None/"" mean "no description"; 0 and False are legitimate criterion values return [k if v is None or v == "" else "%s: %s" % (k, render_criterion(v)) for k, v in crit.items()] if t == "score": return ["level %d: %s" % (i, render_criterion(c)) for i, c in enumerate(crit)] crit = crit or {} false_crit, true_crit = crit.get("false"), crit.get("true") return [ "false: " + (render_criterion(false_crit) if false_crit not in (None, "") else "no, the statement does not hold"), "true: " + (render_criterion(true_crit) if true_crit not in (None, "") else "yes, the statement holds"), ] def build_sequence( tok, state: Union[str, dict, list], q: Dict, max_len: int = 512, head_max_len: int = 192, option_order: Optional[List[int]] = None, truncate_left: bool = False, ): """Format: [CLS] instructions [SEP] [MASK] opt0 [MASK] opt1 ... [SEP] state [SEP].""" mask_tok = tok.mask_token opts = render_options(q) order = option_order if option_order is not None else list(range(len(opts))) ins = str(q["ins"]).replace(mask_tok, " ") head_ids = tok("%s question: %s" % (q["t"], ins), add_special_tokens=False)["input_ids"] opt_ids = [] for i in order: opt_ids.append( [tok.mask_token_id] + tok(" " + opts[i].replace(mask_tok, " "), add_special_tokens=False)["input_ids"][:48] ) opt_budget = head_max_len - sum(len(o) for o in opt_ids) if opt_budget < 16: per = max(4, (head_max_len - 16) // max(1, len(opt_ids))) opt_ids = [o[:per] for o in opt_ids] opt_budget = head_max_len - sum(len(o) for o in opt_ids) head_ids = head_ids[: max(8, opt_budget)] ids = [tok.cls_token_id] + head_ids + [tok.sep_token_id] markers = [] for o in opt_ids: markers.append(len(ids)) ids.extend(o) ids.append(tok.sep_token_id) room = max(0, max_len - len(ids) - 1) st = tok(serialize_state(state).replace(mask_tok, " "), add_special_tokens=False)["input_ids"] st = st[-room:] if truncate_left else st[:room] ids = ids + st + [tok.sep_token_id] return ids[:max_len], [m for m in markers if m < max_len] def ece_score(conf: np.ndarray, correct: np.ndarray, bins: int = 15) -> float: """Expected Calibration Error across confidence bins.""" if len(conf) == 0: return float("nan") edges = np.linspace(0, 1, bins + 1) e = 0.0 for lo, hi in zip(edges[:-1], edges[1:]): sel = (conf > lo) & (conf <= hi) if sel.any(): e += sel.mean() * abs(conf[sel].mean() - correct[sel].mean()) return float(e) def confidence_from_probs(p: np.ndarray, k: int) -> float: """Normalized Shannon entropy confidence: 1 - H(p) / log(k).""" if k < 2: return 1.0 p = p[:k] ent = -(p * np.log(np.clip(p, 1e-12, 1.0))).sum() return float(np.clip(1.0 - ent / math.log(k), 0.0, 1.0)) def temp_bucket(qtype: int, k: int) -> str: size = "2" if k <= 2 else "3-5" if k <= 5 else "6-10" if k <= 10 else "11+" return "%s:%s" % (QTYPE_NAMES[int(qtype)], size) # A fitted temperature below 1 sharpens the logits instead of softening them. The shipped # `choice:11+` bucket is 0.1006, which multiplies them ~10x: a 0.24 top probability is published as # 0.99, so a caller gating on confidence is told a coin flip is a certainty. No honest calibration # needs to sharpen this hard, so refuse to apply one that does. TEMP_MIN = 0.5 TEMP_MAX = 5.0 def clamp_temperature(t, lo: float = TEMP_MIN, hi: float = TEMP_MAX) -> float: """A usable temperature: `t` confined to [lo, hi], falling back to 1.0 if it is not a number.""" try: t = float(t) except (TypeError, ValueError): return 1.0 if t != t or t in (float("inf"), float("-inf")): # NaN / inf return 1.0 return min(hi, max(lo, t))