odm_mini.py

"""Inference-only loader for ODM Mini v1.

The released repository contains only the trained DecisionHead. The frozen
Qwen/Qwen3-0.6B backbone is downloaded separately by ``from_pretrained``.
"""

from __future__ import annotations

import json
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Mapping, Sequence

import torch
from huggingface_hub import hf_hub_download
from safetensors.torch import load_file
from torch import nn
from transformers import AutoModel, AutoTokenizer
from transformers.cache_utils import DynamicCache


DEFAULT_REPO_ID = "samatv256/mini-Jev"
DEFAULT_QUESTION = "What is the best next action for the agent?"


def canonical_state(state: Any) -> str:
    if isinstance(state, str):
        return state.rstrip()
    return json.dumps(state, sort_keys=True, ensure_ascii=False)


def format_state_prefix(state: Any) -> str:
    return f"STATE:\n{canonical_state(state)}\n\n"


def format_candidate_suffix_prefix(question: str, candidate_id: str) -> str:
    question = question.strip() if question and question.strip() else DEFAULT_QUESTION
    return f"QUESTION:\n{question}\n\nCANDIDATE:\n{str(candidate_id).strip()}\n\nDESCRIPTION:\n"


def mean_pool_description(
    hidden_states: torch.Tensor, description_mask: torch.Tensor
) -> torch.Tensor:
    mask = description_mask.to(
        dtype=hidden_states.dtype, device=hidden_states.device
    ).unsqueeze(-1)
    return (hidden_states * mask).sum(dim=1) / mask.sum(dim=1).clamp_min(1.0)


class DecisionHead(nn.Module):
    """Linear(1024, 256) -> GELU -> Linear(256, 1)."""

    def __init__(self, hidden_size: int = 1024, inner_size: int = 256) -> None:
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(hidden_size, inner_size),
            nn.GELU(),
            nn.Linear(inner_size, 1),
        )

    def forward(self, hidden: torch.Tensor) -> torch.Tensor:
        return self.net(hidden.to(self.net[0].weight.dtype)).squeeze(-1)


@dataclass(frozen=True)
class DecisionCandidate:
    id: str
    description: str

    def __post_init__(self) -> None:
        if not self.id.strip():
            raise ValueError("candidate id must not be empty")
        if not self.description.strip():
            raise ValueError("candidate description must not be empty")


CandidateLike = str | Mapping[str, Any] | DecisionCandidate


def normalize_candidates(candidates: Sequence[CandidateLike]) -> list[DecisionCandidate]:
    normalized: list[DecisionCandidate] = []
    for item in candidates:
        if isinstance(item, DecisionCandidate):
            candidate = item
        elif isinstance(item, Mapping):
            candidate = DecisionCandidate(
                id=str(item["id"]), description=str(item["description"])
            )
        elif isinstance(item, str):
            candidate = DecisionCandidate(id=item, description=item)
        else:
            raise TypeError(f"unsupported candidate type: {type(item)!r}")
        normalized.append(candidate)
    if not normalized:
        raise ValueError("candidates list must not be empty")
    ids = [candidate.id for candidate in normalized]
    if len(ids) != len(set(ids)):
        raise ValueError("candidate ids must be unique within a decision")
    return normalized


@dataclass(frozen=True)
class ChoiceResponse:
    selected: str
    probabilities: dict[str, float]
    confidence: float
    decision_margin: float
    latency_ms: float


def _resolve_file(
    model_id_or_path: str | Path,
    filename: str,
    *,
    revision: str | None,
    token: str | bool | None,
) -> Path:
    local_path = Path(model_id_or_path)
    if local_path.is_dir():
        artifact = local_path / filename
        if not artifact.is_file():
            raise FileNotFoundError(f"missing release artifact: {artifact}")
        return artifact
    return Path(
        hf_hub_download(
            repo_id=str(model_id_or_path),
            filename=filename,
            revision=revision,
            token=token,
        )
    )


def replicate_cache(source_cache: DynamicCache, repeats: int) -> DynamicCache:
    new_cache = DynamicCache()
    for layer in source_cache.layers:
        keys = layer.keys.clone() if repeats == 1 else layer.keys.repeat(repeats, 1, 1, 1)
        values = (
            layer.values.clone()
            if repeats == 1
            else layer.values.repeat(repeats, 1, 1, 1)
        )
        new_cache.update(keys, values, len(new_cache.layers))
    return new_cache


class ODMMiniModel(nn.Module):
    """Frozen Qwen3-0.6B plus the trained ODM Mini DecisionHead."""

    def __init__(
        self,
        backbone: nn.Module,
        tokenizer: Any,
        head: DecisionHead,
        *,
        hidden_size: int = 1024,
        head_inner_size: int = 256,
        max_state_length: int = 4096,
        max_suffix_length: int = 256,
        candidate_batch_size: int = 32,
        temperature: float = 1.0,
    ) -> None:
        super().__init__()
        self.backbone = backbone
        self.tokenizer = tokenizer
        self.head = head
        self.hidden_size = hidden_size
        self.head_inner_size = head_inner_size
        self.max_state_length = max_state_length
        self.max_suffix_length = max_suffix_length
        self.candidate_batch_size = candidate_batch_size
        self.temperature = float(temperature)
        self.escalation_threshold = 0.85
        self.backbone.eval()
        for parameter in self.backbone.parameters():
            parameter.requires_grad = False

    @classmethod
    def from_pretrained(
        cls,
        model_id_or_path: str | Path = DEFAULT_REPO_ID,
        *,
        revision: str | None = None,
        device: str | torch.device | None = None,
        dtype: torch.dtype | None = None,
        token: str | bool | None = None,
        candidate_batch_size: int = 32,
    ) -> "ODMMiniModel":
        device = device or ("cuda:0" if torch.cuda.is_available() else "cpu")
        dtype = dtype or (torch.bfloat16 if torch.cuda.is_available() else torch.float32)
        config_path = _resolve_file(
            model_id_or_path, "config.json", revision=revision, token=token
        )
        weights_path = _resolve_file(
            model_id_or_path, "model.safetensors", revision=revision, token=token
        )
        config = json.loads(config_path.read_text(encoding="utf-8"))

        base_model = str(config["base_model"])
        tokenizer = AutoTokenizer.from_pretrained(base_model)
        backbone = AutoModel.from_pretrained(base_model, dtype=dtype)
        backbone.to(device)
        backbone.eval()

        hidden_size = int(config["hidden_size"])
        actual_hidden_size = int(getattr(backbone.config, "hidden_size", hidden_size))
        if actual_hidden_size != hidden_size:
            raise ValueError(
                f"backbone hidden size {actual_hidden_size} does not match release config {hidden_size}"
            )

        head = DecisionHead(
            hidden_size=hidden_size,
            inner_size=int(config["head_inner_size"]),
        )
        head.load_state_dict(load_file(str(weights_path), device="cpu"), strict=True)
        head.to(device=device, dtype=torch.float32)
        head.eval()

        return cls(
            backbone=backbone,
            tokenizer=tokenizer,
            head=head,
            hidden_size=hidden_size,
            head_inner_size=int(config["head_inner_size"]),
            max_state_length=int(config["max_state_length"]),
            max_suffix_length=int(config["max_suffix_length"]),
            candidate_batch_size=candidate_batch_size,
            temperature=float(config["temperature"]),
        )

    @property
    def device(self) -> torch.device:
        return next(self.backbone.parameters()).device

    @torch.inference_mode()
    def encode_state(self, state: Any) -> tuple[DynamicCache, int]:
        prefix_ids = self.tokenizer.encode(
            format_state_prefix(state),
            add_special_tokens=False,
            truncation=True,
            max_length=self.max_state_length,
        )
        if not prefix_ids:
            prefix_ids = [self.tokenizer.pad_token_id or 0]
        input_ids = torch.tensor([prefix_ids], dtype=torch.long, device=self.device)
        output = self.backbone(input_ids=input_ids, use_cache=True)
        return output.past_key_values, len(prefix_ids)

    @torch.inference_mode()
    def score_candidates_cached(
        self,
        past_key_values: DynamicCache,
        prefix_length: int,
        candidates: Sequence[DecisionCandidate],
        question: str = DEFAULT_QUESTION,
        candidate_batch_size: int | None = None,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        batch_size_limit = candidate_batch_size or self.candidate_batch_size
        all_representations: list[torch.Tensor] = []

        for start in range(0, len(candidates), batch_size_limit):
            batch = candidates[start : start + batch_size_limit]
            suffixes: list[list[int]] = []
            description_spans: list[tuple[int, int]] = []
            for candidate in batch:
                suffix_prefix = format_candidate_suffix_prefix(question, candidate.id)
                prefix_ids = self.tokenizer.encode(
                    suffix_prefix,
                    add_special_tokens=False,
                    truncation=True,
                    max_length=self.max_suffix_length // 2,
                )
                description_ids = self.tokenizer.encode(
                    candidate.description.strip(),
                    add_special_tokens=False,
                    truncation=True,
                    max_length=max(1, self.max_suffix_length - len(prefix_ids)),
                )
                suffix = prefix_ids + description_ids
                suffixes.append(suffix)
                description_spans.append((len(prefix_ids), len(suffix)))

            max_length = max(len(suffix) for suffix in suffixes)
            pad_id = int(self.tokenizer.pad_token_id or 0)
            input_ids = torch.full(
                (len(batch), max_length), pad_id, dtype=torch.long, device=self.device
            )
            attention_mask = torch.ones(
                (len(batch), prefix_length + max_length),
                dtype=torch.long,
                device=self.device,
            )
            for index, suffix in enumerate(suffixes):
                input_ids[index, : len(suffix)] = torch.tensor(
                    suffix, dtype=torch.long, device=self.device
                )
                if len(suffix) < max_length:
                    attention_mask[index, prefix_length + len(suffix) :] = 0

            output = self.backbone(
                input_ids=input_ids,
                attention_mask=attention_mask,
                past_key_values=replicate_cache(past_key_values, len(batch)),
                use_cache=False,
            )
            description_mask = torch.zeros(
                (len(batch), max_length), dtype=torch.bool, device=self.device
            )
            for index, (description_start, description_end) in enumerate(description_spans):
                description_mask[index, description_start:description_end] = True
            all_representations.append(
                mean_pool_description(output.last_hidden_state, description_mask)
            )

        representations = torch.cat(all_representations, dim=0)
        return self.head(representations), representations

    @torch.inference_mode()
    def predict_choice(
        self,
        state: Any,
        candidates: Sequence[CandidateLike],
        question: str = DEFAULT_QUESTION,
        temperature: float | None = None,
    ) -> ChoiceResponse:
        started = time.perf_counter()
        normalized = normalize_candidates(candidates)
        cache, prefix_length = self.encode_state(state)
        logits, _ = self.score_candidates_cached(
            cache, prefix_length, normalized, question=question
        )
        float_logits = logits.float()
        sorted_logits, _ = torch.sort(float_logits, descending=True)
        margin = float(
            (sorted_logits[0] - sorted_logits[1]).item()
            if len(sorted_logits) > 1
            else sorted_logits[0].item()
        )
        effective_temperature = self.temperature if temperature is None else float(temperature)
        probabilities = torch.softmax(float_logits / max(effective_temperature, 1e-4), dim=-1)
        selected_index = int(probabilities.argmax().item())
        probability_map = {
            candidate.id: float(probability.item())
            for candidate, probability in zip(normalized, probabilities, strict=True)
        }
        return ChoiceResponse(
            selected=normalized[selected_index].id,
            probabilities=probability_map,
            confidence=float(probabilities[selected_index].item()),
            decision_margin=margin,
            latency_ms=(time.perf_counter() - started) * 1000.0,
        )