native_reference.py

"""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())