File size: 5,010 Bytes
dbe0ec7 d734701 dbe0ec7 d734701 dbe0ec7 d734701 dbe0ec7 d734701 dbe0ec7 d734701 dbe0ec7 d734701 dbe0ec7 | 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 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """Verdict's pinned formatting, calibration and typed output contract.
The rendering and decoding match Verdict-open-jev/core/{formatting,engine_encoder}.py
at 30f15564821626ca5c1ad5b2638c4eb7078787dd. The encoder is loaded from
heman10x/rlcd-modernbert-151m at the revision in assets.lock.json.
"""
from __future__ import annotations
import json
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import numpy as np
ABSTAIN_ID = "__insufficient_evidence__"
ABSTAIN_DESCRIPTION = "insufficient evidence"
MAX_SUBSTANTIVE_OPTIONS = 24
MAX_CANDIDATES = 25
@dataclass(frozen=True)
class Request:
kind: str
text: str
labels: tuple[str, ...]
ids: tuple[str, ...]
values: tuple[float, ...] = ()
def build_request(context: str, question: dict[str, Any]) -> Request:
"""Render one native typed query without dropping trained abstention."""
kind = question["type"]
if kind == "choice":
options = question["options"]
if not 1 <= len(options) <= MAX_SUBSTANTIVE_OPTIONS:
raise ValueError("Verdict supports 1 to 24 substantive choice options")
labels = tuple(f"It is {option['description']}" for option in options)
ids = tuple(option["id"] for option in options)
query = question["question"]
text = f"Question: {query}\n\nContext:\n{context}"
values: tuple[float, ...] = ()
elif kind == "score":
levels = question["levels"]
if not 1 <= len(levels) <= MAX_SUBSTANTIVE_OPTIONS:
raise ValueError("Verdict supports 1 to 24 substantive score levels")
labels = tuple(f"{level['description']} (Value: {level['value']})" for level in levels)
ids = tuple(level["id"] for level in levels)
values = tuple(float(level["value"]) for level in levels)
query = question["question"]
text = f"Question: {query}\n\nContext:\n{context}"
elif kind == "noul":
proposition = question["proposition"]
labels = (f"true: {proposition}", f"false: not {proposition}")
ids = ("true", "false")
values = ()
text = f"Context:\n{context}\n\nEvaluate proposition: {proposition}"
query = ""
else:
raise ValueError(f"unknown Verdict question type: {kind}")
labels += (ABSTAIN_DESCRIPTION,)
ids += (ABSTAIN_ID,)
if len(ids) != len(set(ids)):
raise ValueError("Verdict candidate IDs must be unique and cannot use the abstention ID")
prefix = "".join(f"<<LABEL>>{label}" for label in labels)
return Request(kind, f"{prefix}<<SEP>>{text}", labels, ids, values)
def temperature(calibrator: dict[str, Any], count: int) -> float:
per_k = calibrator.get("per_k", {})
value = float(per_k.get(str(count), calibrator["temperature"]))
if not math.isfinite(value) or value <= 0:
raise ValueError("invalid trained temperature")
return value
def decode(logits: np.ndarray, request: Request, calibrator: dict[str, Any]) -> dict[str, Any]:
"""Apply the released per-K calibration and preserve native abstention."""
k = len(request.ids)
z = np.asarray(logits, dtype=np.float64).reshape(-1)[:k]
if len(z) != k or not np.isfinite(z).all():
raise ValueError("invalid Verdict logits")
z = z / temperature(calibrator, k)
p = np.exp(z - z.max())
p /= p.sum()
distribution = dict(zip(request.ids, map(float, p)))
selected_id = request.ids[int(p.argmax())]
abstained = selected_id == ABSTAIN_ID
entropy = -sum(float(value) * math.log(float(value)) for value in p if value > 1e-12)
concentration = max(0.0, min(1.0, 1.0 - entropy / math.log(k))) if k > 1 else 1.0
result: dict[str, Any] = {
"type": request.kind,
"selected_id": selected_id,
"selected_probability": distribution[selected_id],
"probabilities": distribution,
"concentration": concentration,
"is_abstention": abstained,
"p_abstain": distribution[ABSTAIN_ID],
"calibration_status": "calibrated_for_scope",
}
if request.kind == "choice":
result["choice"] = selected_id
elif request.kind == "noul":
substantive = distribution["true"] + distribution["false"]
result["noul"] = distribution["true"] / substantive if substantive and not abstained else None
result["selected_outcome"] = selected_id
else:
substantive = sum(distribution[key] for key in request.ids[:-1])
result["score"] = (
sum(value * distribution[key] / substantive for key, value in zip(request.ids[:-1], request.values))
if substantive and not abstained
else None
)
result["selected_level_id"] = selected_id
result["selected_value"] = (
request.values[request.ids.index(selected_id)] if not abstained else None
)
return result
def load_calibrator(source: Path) -> dict[str, Any]:
return json.loads((source / "calibrator.json").read_text())
|