typedecide.evaluation.readout

The ScoreFn seam and the torch-backed readout. Background: The typed logit readout.

typedecide.evaluation.readout

Turning a causal language model into a probability distribution over options.

The readout is the whole trick of this library: instead of asking the model to write an answer and then parsing whatever it wrote, we build the prompt so that the very next token must be an answer letter, run one forward pass, and read the softmax over just those letters' logits. The output is a real distribution over the options that were listed, so it can be averaged over orderings and calibrated -- neither of which is possible with generated text.

Everything that touches torch or transformers is imported inside the function that needs it, so import typedecide stays cheap and the base install stays light.

The seam that matters for testing is ScoreFn: anything that maps a batch of ScoreRequest to a batch of probability vectors is a scorer. runner.evaluate builds a TorchReadout; a test injects a fake and exercises the same debiasing code with no model at all.

SLOT_CACHE_LIMIT module-attribute

SLOT_CACHE_LIMIT = 2048

ScoreFn module-attribute

ScoreFn = Callable[
    [Sequence[ScoreRequest]], list[list[float]]
]

ScoreRequest dataclass

ScoreRequest(state: str, criterion: Criterion)

One prompt to score: a state, and a criterion in the exact option order to present. The returned probabilities are indexed by position in this criterion, so a caller that reordered the options must map them back to option ids itself.

state instance-attribute

state: str

criterion instance-attribute

criterion: Criterion

prompt property

prompt: str

TorchReadout

TorchReadout(
    model_id: str,
    *,
    adapter: Path | None = None,
    device: str | None = None,
    batch_size: int = 8,
    seed: int = 0,
    max_length: int | None = None
)

A ScoreFn backed by a local transformers model.

Construction loads the weights, so build one and reuse it across a whole eval rather than per decision.

Source code in src/typedecide/evaluation/readout.py

def __init__(
    self,
    model_id: str,
    *,
    adapter: Path | None = None,
    device: str | None = None,
    batch_size: int = 8,
    seed: int = 0,
    max_length: int | None = None,
) -> None:
    torch = _require("torch")
    transformers = _require("transformers")

    if batch_size < 1:
        raise ConfigError(f"batch_size must be at least 1, got {batch_size}.")

    # Deterministic even though a readout does not sample: a run that cannot be
    # reproduced is not evidence.
    torch.manual_seed(seed)

    self.model_id = model_id
    self.adapter = adapter
    self.batch_size = batch_size
    self.seed = seed
    self.max_length = max_length
    self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")

    self.tokenizer = transformers.AutoTokenizer.from_pretrained(model_id)
    dtype = torch.float32 if self.device == "cpu" else torch.bfloat16
    # No `trust_remote_code`: see `from_pretrained_with_dtype`.
    model = from_pretrained_with_dtype(
        transformers.AutoModelForCausalLM.from_pretrained, model_id, dtype
    )
    if adapter is not None:
        peft = _require("peft")
        model = peft.PeftModel.from_pretrained(model, str(adapter))
        model = model.merge_and_unload()
    self.model = model.to(self.device).eval()

    pad = self.tokenizer.pad_token_id
    if pad is None:
        pad = self.tokenizer.eos_token_id
    if pad is None:
        raise ConfigError(
            f"Tokenizer for {model_id!r} has neither a pad nor an eos token, so batched "
            "scoring cannot pad. Set tokenizer.pad_token before evaluating, or use "
            "batch_size=1."
        )
    self.pad_id: int = int(pad)
    self._slot_cache: dict[str, list[int]] = {}

model_id instance-attribute

model_id = model_id

adapter instance-attribute

adapter = adapter

batch_size instance-attribute

batch_size = batch_size

seed instance-attribute

seed = seed

max_length instance-attribute

max_length = max_length

device instance-attribute

device = device or (
    "cuda" if torch.cuda.is_available() else "cpu"
)

tokenizer instance-attribute

tokenizer = transformers.AutoTokenizer.from_pretrained(
    model_id
)

model instance-attribute

model = model.to(self.device).eval()

pad_id instance-attribute

pad_id: int = int(pad)

score

score(
    requests: Sequence[ScoreRequest],
) -> list[list[float]]

Probability over each request's options, in the order the request lists them.

Source code in src/typedecide/evaluation/readout.py

def score(self, requests: Sequence[ScoreRequest]) -> list[list[float]]:
    """Probability over each request's options, in the order the request lists them."""
    rows = list(requests)
    out: list[list[float]] = []
    for start in range(0, len(rows), self.batch_size):
        out.extend(self._score_batch(rows[start : start + self.batch_size]))
    return out

load_readout

load_readout(
    model_id: str,
    *,
    adapter: Path | None = None,
    device: str | None = None,
    batch_size: int = 8,
    seed: int = 0,
    max_length: int | None = None
) -> TorchReadout

Load a model and return it as a ScoreFn.

Source code in src/typedecide/evaluation/readout.py

def load_readout(
    model_id: str,
    *,
    adapter: Path | None = None,
    device: str | None = None,
    batch_size: int = 8,
    seed: int = 0,
    max_length: int | None = None,
) -> TorchReadout:
    """Load a model and return it as a `ScoreFn`."""
    return TorchReadout(
        model_id,
        adapter=adapter,
        device=device,
        batch_size=batch_size,
        seed=seed,
        max_length=max_length,
    )