FluidInference/verdict-coreml

1"""Verdict's pinned formatting, calibration and typed output contract.2 3The rendering and decoding match Verdict-open-jev/core/{formatting,engine_encoder}.py4at 30f15564821626ca5c1ad5b2638c4eb7078787dd. The encoder is loaded from5heman10x/rlcd-modernbert-151m at the revision in assets.lock.json.6"""7 8from __future__ import annotations9 10import json11import math12from dataclasses import dataclass13from pathlib import Path14from typing import Any15 16import numpy as np17 18ABSTAIN_ID = "__insufficient_evidence__"19ABSTAIN_DESCRIPTION = "insufficient evidence"20MAX_SUBSTANTIVE_OPTIONS = 2421MAX_CANDIDATES = 2522 23 24@dataclass(frozen=True)25class Request:26    kind: str27    text: str28    labels: tuple[str, ...]29    ids: tuple[str, ...]30    values: tuple[float, ...] = ()31 32 33def build_request(context: str, question: dict[str, Any]) -> Request:34    """Render one native typed query without dropping trained abstention."""35    kind = question["type"]36    if kind == "choice":37        options = question["options"]38        if not 1 <= len(options) <= MAX_SUBSTANTIVE_OPTIONS:39            raise ValueError("Verdict supports 1 to 24 substantive choice options")40        labels = tuple(f"It is {option['description']}" for option in options)41        ids = tuple(option["id"] for option in options)42        query = question["question"]43        text = f"Question: {query}\n\nContext:\n{context}"44        values: tuple[float, ...] = ()45    elif kind == "score":46        levels = question["levels"]47        if not 1 <= len(levels) <= MAX_SUBSTANTIVE_OPTIONS:48            raise ValueError("Verdict supports 1 to 24 substantive score levels")49        labels = tuple(f"{level['description']} (Value: {level['value']})" for level in levels)50        ids = tuple(level["id"] for level in levels)51        values = tuple(float(level["value"]) for level in levels)52        query = question["question"]53        text = f"Question: {query}\n\nContext:\n{context}"54    elif kind == "noul":55        proposition = question["proposition"]56        labels = (f"true: {proposition}", f"false: not {proposition}")57        ids = ("true", "false")58        values = ()59        text = f"Context:\n{context}\n\nEvaluate proposition: {proposition}"60        query = ""61    else:62        raise ValueError(f"unknown Verdict question type: {kind}")63    labels += (ABSTAIN_DESCRIPTION,)64    ids += (ABSTAIN_ID,)65    if len(ids) != len(set(ids)):66        raise ValueError("Verdict candidate IDs must be unique and cannot use the abstention ID")67    prefix = "".join(f"<<LABEL>>{label}" for label in labels)68    return Request(kind, f"{prefix}<<SEP>>{text}", labels, ids, values)69 70 71def temperature(calibrator: dict[str, Any], count: int) -> float:72    per_k = calibrator.get("per_k", {})73    value = float(per_k.get(str(count), calibrator["temperature"]))74    if not math.isfinite(value) or value <= 0:75        raise ValueError("invalid trained temperature")76    return value77 78 79def decode(logits: np.ndarray, request: Request, calibrator: dict[str, Any]) -> dict[str, Any]:80    """Apply the released per-K calibration and preserve native abstention."""81    k = len(request.ids)82    z = np.asarray(logits, dtype=np.float64).reshape(-1)[:k]83    if len(z) != k or not np.isfinite(z).all():84        raise ValueError("invalid Verdict logits")85    z = z / temperature(calibrator, k)86    p = np.exp(z - z.max())87    p /= p.sum()88    distribution = dict(zip(request.ids, map(float, p)))89    selected_id = request.ids[int(p.argmax())]90    abstained = selected_id == ABSTAIN_ID91    entropy = -sum(float(value) * math.log(float(value)) for value in p if value > 1e-12)92    concentration = max(0.0, min(1.0, 1.0 - entropy / math.log(k))) if k > 1 else 1.093    result: dict[str, Any] = {94        "type": request.kind,95        "selected_id": selected_id,96        "selected_probability": distribution[selected_id],97        "probabilities": distribution,98        "concentration": concentration,99        "is_abstention": abstained,100        "p_abstain": distribution[ABSTAIN_ID],101        "calibration_status": "calibrated_for_scope",102    }103    if request.kind == "choice":104        result["choice"] = selected_id105    elif request.kind == "noul":106        substantive = distribution["true"] + distribution["false"]107        result["noul"] = distribution["true"] / substantive if substantive and not abstained else None108        result["selected_outcome"] = selected_id109    else:110        substantive = sum(distribution[key] for key in request.ids[:-1])111        result["score"] = (112            sum(value * distribution[key] / substantive for key, value in zip(request.ids[:-1], request.values))113            if substantive and not abstained114            else None115        )116        result["selected_level_id"] = selected_id117        result["selected_value"] = (118            request.values[request.ids.index(selected_id)] if not abstained else None119        )120    return result121 122 123def load_calibrator(source: Path) -> dict[str, Any]:124    return json.loads((source / "calibrator.json").read_text())125