Text Classification
Transformers
Safetensors
English
qwen3_5
image-text-to-text
decision-model
classification
calibration
routing
guardrails
baseten
Instructions to use baselabs/elephant-1 with libraries, inference providers, notebooks, and local apps. Follow these links to get started.
- Libraries
- Transformers
How to use baselabs/elephant-1 with Transformers:
# Use a pipeline as a high-level helper from transformers import pipeline pipe = pipeline("text-classification", model="baselabs/elephant-1")# pip install -U transformers accelerate # Load model directly from transformers import AutoProcessor, AutoModelForMultimodalLM processor = AutoProcessor.from_pretrained("baselabs/elephant-1") model = AutoModelForMultimodalLM.from_pretrained("baselabs/elephant-1", device_map="auto") - Notebooks
- Google Colab
- Kaggle
Download serve.patch from baselabs/elephant-1: direct link, hf CLI and curl.
- Browser
- Download file 41 kB
-
https://huggingface.co/baselabs/elephant-1/resolve/main/serve.patch
- Command line
-
hf download hf://baselabs/elephant-1/serve.patch
-
curl -L -o serve.patch https://huggingface.co/baselabs/elephant-1/resolve/main/serve.patch
41 kB
| diff --git a/qev/engine.py b/qev/engine.py | |
| index 2e4c7b0..cb79e4f 100644 | |
| --- a/qev/engine.py | |
| +++ b/qev/engine.py | |
| tolerance of 0.05 and only guards against structural failures (an unknown cache | |
| """ | |
| from __future__ import annotations | |
| +import contextlib | |
| import copy | |
| import hashlib | |
| import logging | |
| from pathlib import Path | |
| from typing import Any | |
| from .answers import make_choice, make_noul, make_score | |
| +from . import readout as ro | |
| from .calibrate import Calibration, softmax | |
| from .prompt import (DEFAULT_STYLE, LETTERS, MAX_SLOTS, PROMPT_VERSION, SLOT_LABELS, PromptStyle, RenderedQuestion, | |
| build_messages, render_question) | |
| class DecisionEngine: | |
| self.model_type = config.model_type | |
| self.template_kwargs = {"enable_thinking": False} if "qwen" in self.model_type else {} | |
| self.slot_ids = self._verify_slots() | |
| + rcfg = ro.ReadoutConfig.load(model_path) if local else ro.ReadoutConfig() | |
| + self.readout = ro.Readout.attach(self.model, rcfg, slot_ids=self.slot_ids, | |
| + head_path=Path(model_path) / ro.HEAD_FILE if rcfg.head else None, | |
| + device=self.device) | |
| + if rcfg.head: | |
| + self.max_slots = min(self.max_slots, rcfg.n_slots) | |
| + if calibration is None and local and (Path(model_path) / "calibration.json").exists(): | |
| + self.calibration = Calibration.load(Path(model_path) / "calibration.json") | |
| self.pad_id = self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else self.tokenizer.eos_token_id | |
| self.info = EngineInfo(model_path=model_path, model_type=self.model_type, dtype=dtype, device=str(self.device), | |
| fork_mode="sequential", fork_max_abs_diff=None, | |
| transformers_version=transformers.__version__, torch_version=torch.__version__, | |
| - extra={"head_dtype": head_dtype}) | |
| + extra={"head_dtype": head_dtype, "attention": rcfg.attention, "readout": rcfg.readout, | |
| + "calibration": self.calibration.source}) | |
| if fork_mode == "auto": | |
| self._select_fork_mode() | |
| else: | |
| class DecisionEngine: | |
| with torch.no_grad(): | |
| torch.cuda.empty_cache() if self.device.type == "cuda" else None | |
| + def _split(self, state: Any, rendered: list[RenderedQuestion]): | |
| + """Prefix ids, suffix ids, and per branch the option token spans relative to its suffix (None for slot_lm). | |
| + The layout is the one train.common.batch_scores computes, so serving matches training: under noncausal | |
| + attention the cache boundary must be the prefix block boundary.""" | |
| + prefix_ids, suffixes = self._prefix_and_suffix_ids(state, rendered) | |
| + cfg = self.readout.config | |
| + if not cfg.needs_layout: | |
| + return prefix_ids, suffixes, [None] * len(rendered) | |
| + P = len(prefix_ids) | |
| + spans = [] | |
| + for r, suf in zip(rendered, suffixes): | |
| + prompt = self._render_prompt(state, r.suffix) | |
| + start = prompt.rindex(r.suffix) | |
| + ids, prefix, sp = ro.token_layout(self.tokenizer, prompt, start, | |
| + [(start + a, start + b) for a, b in r.option_spans or []]) | |
| + if ids != prefix_ids + suf: | |
| + raise ValueError("layout tokenization differs from the branch tokenization") | |
| + if cfg.noncausal and prefix != P: | |
| + raise ValueError(f"prefix block ends at token {prefix} but the shared cache at {P}") | |
| + if any(a < P for a, _ in sp): | |
| + raise ValueError("an option span starts inside the shared prefix") | |
| + spans.append([(a - P, b - P) for a, b in sp]) | |
| + return prefix_ids, suffixes, spans | |
| + | |
| + def _blocks(self, prefixes: list[int], lengths: list[int], width: int): | |
| + if not self.readout.config.noncausal: | |
| + return contextlib.nullcontext() | |
| + return ro.blocks(ro.block_ids(prefixes, lengths, width, self.device)) | |
| + | |
| + def _scores(self, o, hidden, row: int, logit_pos: int, end: int, n_slots: int, spans) -> tuple[list[float], float]: | |
| + """Scores of one branch from a forward's output: the slot letters' LM logits, or the readout head over the | |
| + captured final hidden states (`end` and `spans` index the suffix tokens of this row).""" | |
| + if self.readout.head is None: | |
| + return self._readout(o.logits[row, logit_pos], n_slots) | |
| + h = hidden[row: row + 1] | |
| + scores = self.readout.head(h[:, end], ro.pool_options(h, [spans], n_slots))[0] | |
| + return scores.tolist(), 1.0 | |
| + | |
| def _prefix_and_suffix_ids(self, state: Any, rendered: list[RenderedQuestion]) -> tuple[list[int], list[list[int]]]: | |
| """Tokenize every branch fully, then find the longest common token prefix that ends at or | |
| before the end of the state block, so the shared prefix is exact by construction.""" | |
| class DecisionEngine: | |
| def _run_sequential(self, state: Any, rendered: list[RenderedQuestion]) -> list[BranchResult]: | |
| import torch | |
| - prefix_ids, suffixes = self._prefix_and_suffix_ids(state, rendered) | |
| + prefix_ids, suffixes, spans = self._split(state, rendered) | |
| dev = self.device | |
| + P = len(prefix_ids) | |
| with torch.inference_mode(): | |
| - out = self.model(input_ids=torch.tensor([prefix_ids], device=dev), | |
| - attention_mask=torch.ones((1, len(prefix_ids)), dtype=torch.long, device=dev), | |
| - use_cache=True, return_dict=True, logits_to_keep=1) | |
| + with self._blocks([P], [P], P): | |
| + out = self.model(input_ids=torch.tensor([prefix_ids], device=dev), | |
| + attention_mask=torch.ones((1, P), dtype=torch.long, device=dev), | |
| + use_cache=True, return_dict=True, logits_to_keep=1) | |
| base = out.past_key_values | |
| del out | |
| results = [] | |
| - for r, suf in zip(rendered, suffixes): | |
| + for r, suf, sp in zip(rendered, suffixes, spans): | |
| cache = copy.deepcopy(base) | |
| - o = self.model(input_ids=torch.tensor([suf], device=dev), | |
| - attention_mask=torch.ones((1, len(prefix_ids) + len(suf)), dtype=torch.long, device=dev), | |
| - past_key_values=cache, use_cache=True, return_dict=True, logits_to_keep=1) | |
| - logits, mass = self._readout(o.logits[0, -1], r.n_slots) | |
| + with self._blocks([P], [P + len(suf)], P + len(suf)): | |
| + o = self.model(input_ids=torch.tensor([suf], device=dev), | |
| + attention_mask=torch.ones((1, P + len(suf)), dtype=torch.long, device=dev), | |
| + past_key_values=cache, use_cache=True, return_dict=True, logits_to_keep=1) | |
| + hidden = self.readout.capture.pop() if self.readout.capture else None | |
| + logits, mass = self._scores(o, hidden, 0, -1, len(suf) - 1, r.n_slots, sp) | |
| results.append(BranchResult(r.qid, r.kind, r.labels, r.descriptions, logits, mass, len(suf))) | |
| del o, cache | |
| del base | |
| class DecisionEngine: | |
| caller falls back to sequential branches.""" | |
| import torch | |
| - prefix_ids, suffixes = self._prefix_and_suffix_ids(state, rendered) | |
| + prefix_ids, suffixes, spans = self._split(state, rendered) | |
| dev = self.device | |
| P = len(prefix_ids) | |
| chunk = max(1, min(len(rendered), self.fork_token_budget // max(P, 1))) | |
| results: list[BranchResult | None] = [None] * len(rendered) | |
| with torch.inference_mode(): | |
| - out = self.model(input_ids=torch.tensor([prefix_ids], device=dev), | |
| - attention_mask=torch.ones((1, P), dtype=torch.long, device=dev), | |
| - use_cache=True, return_dict=True, logits_to_keep=1) | |
| + with self._blocks([P], [P], P): | |
| + out = self.model(input_ids=torch.tensor([prefix_ids], device=dev), | |
| + attention_mask=torch.ones((1, P), dtype=torch.long, device=dev), | |
| + use_cache=True, return_dict=True, logits_to_keep=1) | |
| base = out.past_key_values | |
| del out | |
| start = 0 | |
| while start < len(rendered): | |
| idx = list(range(start, min(start + chunk, len(rendered)))) | |
| try: | |
| - self._run_chunk(base, P, idx, rendered, suffixes, results) | |
| + self._run_chunk(base, P, idx, rendered, suffixes, results, spans) | |
| except torch.OutOfMemoryError: | |
| torch.cuda.empty_cache() | |
| if chunk == 1: | |
| class DecisionEngine: | |
| del base | |
| return results | |
| - def _run_chunk(self, base, P: int, idx: list[int], rendered, suffixes, results) -> None: | |
| + def _run_chunk(self, base, P: int, idx: list[int], rendered, suffixes, results, spans=None) -> None: | |
| import torch | |
| dev = self.device | |
| class DecisionEngine: | |
| ends.append(len(s) - 1) | |
| keep = sorted(set(ends)) | |
| cache = self._fork(base, len(idx)) | |
| + spans = spans or [None] * len(rendered) | |
| try: | |
| - o = self.model(input_ids=ids, attention_mask=mask, position_ids=pos, past_key_values=cache, | |
| - use_cache=True, return_dict=True, | |
| - logits_to_keep=torch.tensor(keep, dtype=torch.long, device=dev)) | |
| + with self._blocks([P] * len(idx), [P + len(s) for s in sufs], P + width): | |
| + o = self.model(input_ids=ids, attention_mask=mask, position_ids=pos, past_key_values=cache, | |
| + use_cache=True, return_dict=True, | |
| + logits_to_keep=torch.tensor(keep, dtype=torch.long, device=dev)) | |
| + hidden = self.readout.capture.pop() if self.readout.capture else None | |
| for row, i in enumerate(idx): | |
| r = rendered[i] | |
| - logits, mass = self._readout(o.logits[row, keep.index(ends[row])], r.n_slots) | |
| + logits, mass = self._scores(o, hidden, row, keep.index(ends[row]), ends[row], r.n_slots, spans[i]) | |
| results[i] = BranchResult(r.qid, r.kind, r.labels, r.descriptions, logits, mass, len(sufs[row])) | |
| del o | |
| finally: | |
| diff --git a/qev/prompt.py b/qev/prompt.py | |
| index 8eaadd3..de49cca 100644 | |
| --- a/qev/prompt.py | |
| +++ b/qev/prompt.py | |
| class RenderedQuestion: | |
| labels: list[str] | |
| descriptions: list[Any] | |
| suffix: str | |
| + option_spans: list[tuple[int, int]] | None = None | |
| n_slots: int = field(init=False) | |
| def __post_init__(self) -> None: | |
| self.n_slots = len(self.labels) | |
| -def _options_block(names: Sequence[str], descriptions: Sequence[Any], show_names: bool = True) -> str: | |
| +def _options_lines(names: Sequence[str], descriptions: Sequence[Any], show_names: bool = True) -> list[str]: | |
| lines = [] | |
| for letter, name, desc in zip(SLOT_LABELS, names, descriptions): | |
| text = render_entry(desc) | |
| def _options_block(names: Sequence[str], descriptions: Sequence[Any], show_names | |
| lines.append(f"{letter}. {name}: {text}") | |
| else: | |
| lines.append(f"{letter}. {name}") | |
| - return "\n".join(lines) | |
| + return lines | |
| + | |
| + | |
| +def _options_block(names: Sequence[str], descriptions: Sequence[Any], show_names: bool = True) -> str: | |
| + return "\n".join(_options_lines(names, descriptions, show_names)) | |
| def render_question(qid: str, q: Question, order: Sequence[int] | None = None, style: PromptStyle = DEFAULT_STYLE) -> RenderedQuestion: | |
| def render_question(qid: str, q: Question, order: Sequence[int] | None = None, s | |
| descs = [descs[i] for i in order] | |
| letters = list(SLOT_LABELS[: len(labels)]) | |
| + spans = None | |
| if style.layout == "json": | |
| payload = {"criterion": instructions, | |
| "options": [{"letter": l, "description": (f"{n}: {render_entry(d)}" if (style.show_names and n is not None and render_entry(d)) else (render_entry(d) or n))} | |
| def render_question(qid: str, q: Question, order: Sequence[int] | None = None, s | |
| suffix = json.dumps(payload, ensure_ascii=False)[1:] | |
| else: | |
| header = style.question_header if kind != "score" else style.score_header | |
| + head = f"{header} {instructions}\n\nOptions:\n" | |
| + lines = _options_lines(names, descs, style.show_names) | |
| + spans, pos = [], len(head) | |
| + for line in lines: | |
| + spans.append((pos, pos + len(line))) | |
| + pos += len(line) + 1 | |
| suffix = ( | |
| - f"{header} {instructions}\n\n" | |
| - f"Options:\n{_options_block(names, descs, style.show_names)}\n\n" | |
| + head + "\n".join(lines) + "\n\n" | |
| + (style.answer_line if len(letters) <= len(LETTERS) else style.answer_line_many).format(letters=", ".join(letters)) | |
| ) | |
| - return RenderedQuestion(qid=qid, kind=kind, labels=labels, descriptions=descs, suffix=suffix) | |
| + return RenderedQuestion(qid=qid, kind=kind, labels=labels, descriptions=descs, suffix=suffix, option_spans=spans) | |
| def state_block(state: Any, style: PromptStyle = DEFAULT_STYLE) -> str: | |
| diff --git a/qev/readout.py b/qev/readout.py | |
| new file mode 100644 | |
| index 0000000..02707ca | |
| +"""Readout variants, shared by training (train.common / train.sft) and serving (qev.engine). | |
| + | |
| +attention: | |
| + causal the base model's mask. | |
| + noncausal in the full-attention layers every token attends bidirectionally within its block and causally to | |
| + earlier blocks. A rendered prompt has two blocks: the shared prefix (system turn and state block) and | |
| + the question suffix (question, options, answer line, generation prompt). The prefix never sees a | |
| + suffix, so its cache is still shared exactly by all questions of a request, and one forward of a | |
| + suffix scores all of its options. The GatedDeltaNet (linear-attention) layers are recurrent and stay | |
| + causal. | |
| + | |
| +readout: | |
| + slot_lm softmax over the LM-head logits of the slot letters at the answer position. | |
| + head a learned head over the final (normed) hidden states: | |
| + score_i = <w_i, h_ans> + <W_q h_ans, W_k mean(h over option i's line)> / sqrt(d) | |
| + w has one row per slot and starts as the LM-head rows of the slot letters, W_k starts at zero, so the | |
| + untrained head reproduces slot_lm exactly; the pointer term scores each option by its content. | |
| + | |
| +Both settings are recorded next to the weights, so a checkpoint is always served the way it was trained. The | |
| +attention mask is declared per layer type under `modeling_overrides` in `config.json`, where serving stacks | |
| +read it; `readout.json` holds the rest. Checkpoints without the overrides block carry `attention` in `readout.json`. | |
| +""" | |
| +from __future__ import annotations | |
| + | |
| +import bisect | |
| +import contextlib | |
| +import contextvars | |
| +import importlib | |
| +import json | |
| +import math | |
| +from dataclasses import asdict, dataclass | |
| +from pathlib import Path | |
| +from typing import Sequence | |
| + | |
| +import torch | |
| + | |
| +from .prompt import MAX_SLOTS | |
| + | |
| +CONFIG_FILE = "readout.json" | |
| +MODEL_CONFIG_FILE = "config.json" | |
| +OVERRIDES_KEY = "modeling_overrides" | |
| +FULL_ATTENTION_MASK = {"causal": "causal", "noncausal": "block_bidirectional"} | |
| +HEAD_FILE = "readout_head.pt" | |
| +ATTENTION = ("causal", "noncausal") | |
| +READOUT = ("slot_lm", "head") | |
| + | |
| + | |
| +@dataclass | |
| +class ReadoutConfig: | |
| + attention: str = "causal" | |
| + readout: str = "slot_lm" | |
| + head_dim: int = 256 | |
| + n_slots: int = MAX_SLOTS | |
| + | |
| + def __post_init__(self) -> None: | |
| + if self.attention not in ATTENTION: | |
| + raise ValueError(f"attention must be one of {ATTENTION}, got {self.attention!r}") | |
| + if self.readout not in READOUT: | |
| + raise ValueError(f"readout must be one of {READOUT}, got {self.readout!r}") | |
| + | |
| + @property | |
| + def noncausal(self) -> bool: | |
| + return self.attention == "noncausal" | |
| + | |
| + @property | |
| + def head(self) -> bool: | |
| + return self.readout == "head" | |
| + | |
| + @property | |
| + def needs_layout(self) -> bool: | |
| + return self.noncausal or self.head | |
| + | |
| + @classmethod | |
| + def load(cls, model_dir: str | Path) -> "ReadoutConfig": | |
| + path, model_path = Path(model_dir) / CONFIG_FILE, Path(model_dir) / MODEL_CONFIG_FILE | |
| + fields = json.loads(path.read_text()) if path.exists() else {} | |
| + overrides = json.loads(model_path.read_text()).get(OVERRIDES_KEY) if model_path.exists() else None | |
| + if overrides: | |
| + mask = overrides["attention"]["full_attention"] | |
| + attention = {v: k for k, v in FULL_ATTENTION_MASK.items()}.get(mask) | |
| + if attention is None: | |
| + raise ValueError(f"unsupported full_attention mask {mask!r} in {model_path}") | |
| + if fields.get("attention", attention) != attention: | |
| + raise ValueError(f"{path} says attention {fields['attention']!r} but {model_path} says {mask!r}") | |
| + fields["attention"] = attention | |
| + return cls(**fields) | |
| + | |
| + def save(self, model_dir: str | Path) -> None: | |
| + fields = asdict(self) | |
| + model_path = Path(model_dir) / MODEL_CONFIG_FILE | |
| + if model_path.exists(): # LoRA adapters have no config.json; their readout.json keeps the attention | |
| + attention = fields.pop("attention") | |
| + config = json.loads(model_path.read_text()) | |
| + config[OVERRIDES_KEY] = {"attention": {"full_attention": FULL_ATTENTION_MASK[attention], | |
| + "linear_attention": "causal"}, **fields} | |
| + model_path.unlink() # may be a symlink to the source checkpoint (model_v5/mkvariant.py) | |
| + model_path.write_text(json.dumps(config, indent=2) + "\n") | |
| + (Path(model_dir) / CONFIG_FILE).write_text(json.dumps(fields, indent=2)) | |
| + | |
| + | |
| +_BLOCKS: contextvars.ContextVar = contextvars.ContextVar("qev_readout_blocks", default=None) | |
| + | |
| + | |
| +def install_noncausal(model) -> None: | |
| + """Route block ids (set with `blocks`) into the full-attention mask of the model's modeling module. Outside a | |
| + `blocks` context the model is unchanged. Needs a mask-based attention implementation (sdpa, eager, flex).""" | |
| + impl = getattr(model.config.get_text_config(), "_attn_implementation", None) or model.config._attn_implementation | |
| + if impl not in ("sdpa", "eager", "flex_attention"): | |
| + raise ValueError(f"noncausal attention needs sdpa, eager or flex_attention, not {impl}") | |
| + mod = importlib.import_module(type(model).__module__) | |
| + if getattr(mod, "_qev_noncausal", False): | |
| + return | |
| + if not hasattr(mod, "create_causal_mask"): | |
| + raise ValueError(f"{mod.__name__} builds no causal mask to extend") | |
| + original = mod.create_causal_mask | |
| + | |
| + def create_causal_mask(*args, **kwargs): | |
| + ids = _BLOCKS.get() | |
| + if ids is not None and kwargs.get("block_sequence_ids") is None: | |
| + kwargs["block_sequence_ids"] = ids | |
| + return original(*args, **kwargs) | |
| + | |
| + mod.create_causal_mask = create_causal_mask | |
| + mod._qev_noncausal = True | |
| + | |
| + | |
| +@contextlib.contextmanager | |
| +def blocks(ids): | |
| + """Block ids [B, kv_length] over absolute positions (cached prefix included); None leaves the mask causal.""" | |
| + token = _BLOCKS.set(ids) | |
| + try: | |
| + yield | |
| + finally: | |
| + _BLOCKS.reset(token) | |
| + | |
| + | |
| +def block_ids(prefix_lens: Sequence[int], lengths: Sequence[int], width: int, device) -> torch.Tensor: | |
| + """0 on the prefix, 1 on the suffix, -1 on right padding (causal, and masked as a key anyway).""" | |
| + pos = torch.arange(width, device=device)[None] | |
| + p = torch.tensor(list(prefix_lens), device=device)[:, None] | |
| + n = torch.tensor(list(lengths), device=device)[:, None] | |
| + return torch.where(pos < p, 0, torch.where(pos < n, 1, -1)) | |
| + | |
| + | |
| +def token_layout(tokenizer, prompt: str, suffix_start: int, option_spans: Sequence[tuple[int, int]] | None): | |
| + """Token ids of `prompt`, the prefix length (tokens that end at or before the suffix) and each option line's | |
| + token span [start, end), from the tokenizer's character offsets. Option spans are characters of the prompt.""" | |
| + enc = tokenizer(prompt, add_special_tokens=False, return_offsets_mapping=True) | |
| + ids, offs = enc["input_ids"], enc["offset_mapping"] | |
| + ends = [e for _, e in offs] | |
| + starts = [s for s, _ in offs] | |
| + prefix = bisect.bisect_right(ends, suffix_start) | |
| + spans = [] | |
| + for cs, ce in option_spans or []: | |
| + a = bisect.bisect_right(ends, cs) | |
| + b = bisect.bisect_left(starts, ce) | |
| + if b <= a: | |
| + raise ValueError(f"option span {cs}:{ce} covers no token") | |
| + spans.append((a, b)) | |
| + return ids, prefix, spans | |
| + | |
| + | |
| +class ReadoutHead(torch.nn.Module): | |
| + def __init__(self, hidden: int, n_slots: int = MAX_SLOTS, head_dim: int = 256): | |
| + super().__init__() | |
| + self.slot = torch.nn.Parameter(torch.zeros(n_slots, hidden)) | |
| + self.q = torch.nn.Linear(hidden, head_dim, bias=False) | |
| + self.k = torch.nn.Linear(hidden, head_dim, bias=False) | |
| + torch.nn.init.normal_(self.q.weight, std=hidden ** -0.5) | |
| + torch.nn.init.zeros_(self.k.weight) | |
| + | |
| + @classmethod | |
| + def from_lm_head(cls, lm_head_weight, slot_ids: Sequence[int], head_dim: int) -> "ReadoutHead": | |
| + rows = lm_head_weight.detach()[list(slot_ids)].float() | |
| + head = cls(rows.shape[1], rows.shape[0], head_dim) | |
| + with torch.no_grad(): | |
| + head.slot.copy_(rows.to(head.slot.device)) | |
| + return head | |
| + | |
| + def forward(self, h_ans, h_opt): | |
| + """h_ans [B, H], h_opt [B, K, H] -> scores [B, K] (fp32).""" | |
| + h_ans, h_opt = h_ans.float(), h_opt.float() | |
| + k = h_opt.shape[1] | |
| + slot = h_ans @ self.slot[:k].T | |
| + ptr = torch.einsum("bd,bkd->bk", self.q(h_ans), self.k(h_opt)) / math.sqrt(self.q.out_features) | |
| + return slot + ptr | |
| + | |
| + | |
| +def pool_options(hidden, spans: Sequence[Sequence[tuple[int, int]]], kmax: int): | |
| + """Mean of hidden [B, T, H] over each option's token span -> [B, kmax, H] fp32 (zeros past a row's options).""" | |
| + bsz, _, dim = hidden.shape | |
| + dev = hidden.device | |
| + rows, toks, segs = [], [], [] | |
| + for b, sp in enumerate(spans): | |
| + for k, (s, e) in enumerate(sp): | |
| + n = e - s | |
| + rows.append(torch.full((n,), b, dtype=torch.long)) | |
| + toks.append(torch.arange(s, e, dtype=torch.long)) | |
| + segs.append(torch.full((n,), b * kmax + k, dtype=torch.long)) | |
| + rows, toks, segs = (torch.cat(x).to(dev) for x in (rows, toks, segs)) | |
| + flat = hidden[rows, toks].float() | |
| + total = torch.zeros(bsz * kmax, dim, dtype=torch.float32, device=dev).index_add(0, segs, flat) | |
| + count = torch.zeros(bsz * kmax, dtype=torch.float32, device=dev).index_add( | |
| + 0, segs, torch.ones_like(segs, dtype=torch.float32)) | |
| + return (total / count.clamp(min=1)[:, None]).view(bsz, kmax, dim) | |
| + | |
| + | |
| +def final_norm(model): | |
| + """The final norm of a (possibly multimodal) Qwen-style decoder: its output is the readout hidden state.""" | |
| + inner = getattr(model, "model", model) | |
| + inner = getattr(inner, "language_model", inner) | |
| + norm = getattr(inner, "norm", None) | |
| + if norm is None: | |
| + raise ValueError(f"cannot find the final norm of {type(model).__name__}") | |
| + return norm | |
| + | |
| + | |
| +class HiddenCapture: | |
| + """Keeps the final-norm output of the latest forward (with its graph, so the head trains the backbone).""" | |
| + | |
| + def __init__(self, model): | |
| + self.value = None | |
| + self.handle = final_norm(model).register_forward_hook(self._hook) | |
| + | |
| + def _hook(self, module, args, output): | |
| + self.value = output | |
| + | |
| + def pop(self): | |
| + out, self.value = self.value, None | |
| + if out is None: | |
| + raise RuntimeError("no hidden state captured; the forward did not reach the final norm") | |
| + return out | |
| + | |
| + | |
| +@dataclass | |
| +class Readout: | |
| + """A model's readout at runtime: its config, and for `head` the head module and the hidden-state hook.""" | |
| + | |
| + config: ReadoutConfig | |
| + head: ReadoutHead | None = None | |
| + capture: HiddenCapture | None = None | |
| + | |
| + @classmethod | |
| + def attach(cls, model, config: ReadoutConfig, slot_ids: Sequence[int] | None = None, | |
| + head_path: str | Path | None = None, device=None) -> "Readout": | |
| + """Install the variant on `model`. A head is loaded from `head_path` when given, else initialised from the | |
| + model's LM-head rows of `slot_ids` (call before the model is sharded).""" | |
| + if config.noncausal: | |
| + install_noncausal(model) | |
| + if not config.head: | |
| + return cls(config) | |
| + if head_path is not None: | |
| + state = torch.load(head_path, map_location="cpu", weights_only=True) | |
| + head = ReadoutHead(state["slot"].shape[1], state["slot"].shape[0], state["q.weight"].shape[0]) | |
| + head.load_state_dict(state) | |
| + else: | |
| + head = ReadoutHead.from_lm_head(model.get_output_embeddings().weight, slot_ids[: config.n_slots], | |
| + config.head_dim) | |
| + head = head.to(device or next(model.parameters()).device) | |
| + return cls(config, head, HiddenCapture(model)) | |
| + | |
| + def save(self, out_dir: str | Path) -> None: | |
| + self.config.save(out_dir) | |
| + if self.head is not None: | |
| + torch.save({k: v.detach().cpu() for k, v in self.head.state_dict().items()}, Path(out_dir) / HEAD_FILE) | |
| + | |
| + @property | |
| + def max_slots(self) -> int: | |
| + return self.config.n_slots | |
| diff --git a/train/common.py b/train/common.py | |
| index 331206f..896e6d3 100644 | |
| --- a/train/common.py | |
| +++ b/train/common.py | |
| class Example: | |
| target: list[float] | |
| source: str | |
| task: str | |
| + suffix_start: int = -1 | |
| + option_spans: list[tuple[int, int]] | None = None | |
| PAD_FAMILIES = ("next_action",) | |
| def render_example(row: dict, tmpl, idx: int, seed: int | None = None, augment: | |
| compact = augment and rng.random() < 0.5 | |
| state = serialize_state(row.get("state"), compact) | |
| prompt = tmpl(state, rendered.suffix) | |
| + suffix_start = prompt.rindex(rendered.suffix) | |
| + spans = [(suffix_start + a, suffix_start + b) for a, b in rendered.option_spans or []] | |
| return Example( | |
| idx=idx, | |
| prompt=prompt, | |
| def render_example(row: dict, tmpl, idx: int, seed: int | None = None, augment: | |
| target=target_vector(row["target"], rendered.labels), | |
| source=str(row.get("source", "unknown")), | |
| task=str(row.get("task", "unknown")), | |
| + suffix_start=suffix_start, | |
| + option_spans=spans or None, | |
| ) | |
| class Bundle: | |
| template_kwargs: dict | |
| processor: Any = None | |
| counter: Any = None | |
| + readout: Any = None | |
| @property | |
| def multimodal(self) -> bool: | |
| def readout_logits(model, input_ids, attention_mask, ends): | |
| return out.logits[rows, pos] | |
| +def batch_scores(bundle: Bundle, model, examples: Sequence[Example], rows: Sequence[dict], device): | |
| + """Option scores [B, Kmax] (-inf past each row's options) and their mask for a right-padded batch, under the | |
| + bundle's readout (qev.readout): causal or noncausal attention, slot_lm or head scoring. The same function scores | |
| + training steps, in-training evals and train.eval_mm, and qev.engine reproduces it for serving.""" | |
| + import torch | |
| + | |
| + from qev import readout as ro | |
| + | |
| + rd = bundle.readout | |
| + cfg = rd.config if rd is not None else ro.ReadoutConfig() | |
| + ks = [len(ex.labels) for ex in examples] | |
| + if bundle.multimodal: | |
| + from train import mm | |
| + | |
| + inputs, ends = mm.mm_inputs(bundle.processor, [ex.prompt for ex in examples], | |
| + [r.get("media") or [] for r in rows], device) | |
| + shifts = [bundle.counter.media_extra(r.get("media") or []) for r in rows] | |
| + else: | |
| + input_ids, attention_mask, ends = pad_batch([bundle.encode(ex.prompt) for ex in examples], bundle.tokenizer.pad_token_id, device) | |
| + inputs = {"input_ids": input_ids, "attention_mask": attention_mask} | |
| + shifts = [0] * len(examples) | |
| + ctx = ro.blocks(None) | |
| + spans = None | |
| + if cfg.needs_layout: | |
| + prefixes, spans = [], [] | |
| + for ex, shift, end in zip(examples, shifts, ends): | |
| + ids, prefix, sp = ro.token_layout(bundle.tokenizer, ex.prompt, ex.suffix_start, ex.option_spans) | |
| + if len(ids) + shift != end + 1: | |
| + raise ValueError(f"layout of row {ex.idx} has {len(ids) + shift} tokens, the batch has {end + 1}") | |
| + if cfg.head and len(sp) != len(ex.labels): | |
| + raise ValueError(f"row {ex.idx}: {len(sp)} option spans for {len(ex.labels)} options") | |
| + prefixes.append(prefix + shift) | |
| + spans.append([(a + shift, b + shift) for a, b in sp]) | |
| + if cfg.noncausal: | |
| + width = inputs["input_ids"].shape[1] | |
| + ctx = ro.blocks(ro.block_ids(prefixes, [e + 1 for e in ends], width, device)) | |
| + with ctx: | |
| + if cfg.head: | |
| + out = model(**inputs, use_cache=False, return_dict=True, logits_to_keep=1) | |
| + hidden = rd.capture.pop() | |
| + kmax = max(ks) | |
| + rows_ix = torch.arange(len(examples), device=hidden.device) | |
| + h_ans = hidden[rows_ix, torch.tensor(ends, device=hidden.device)] | |
| + scores = rd.head(h_ans, ro.pool_options(hidden, spans, kmax)) | |
| + # every FSDP unit (the LM head too) must take part in the backward | |
| + scores = scores + 0.0 * out.logits.float().sum() | |
| + mask = torch.zeros((len(ks), kmax), dtype=torch.bool, device=scores.device) | |
| + for i, k in enumerate(ks): | |
| + mask[i, :k] = True | |
| + return scores.masked_fill(~mask, float("-inf")), mask | |
| + if bundle.multimodal: | |
| + from train import mm | |
| + | |
| + vocab_logits = mm.mm_readout_logits(model, inputs, ends) | |
| + else: | |
| + vocab_logits = readout_logits(model, inputs["input_ids"], inputs["attention_mask"], ends) | |
| + return slot_view(vocab_logits, bundle.slot_ids, ks) | |
| + | |
| + | |
| def slot_view(vocab_logits, slot_ids, ks): | |
| """Gather the first K slot logits per row into a padded [B, Kmax] tensor plus a mask.""" | |
| import torch | |
| def collect_slot_logits(bundle: Bundle, rows: Sequence[dict], max_len: int = 409 | |
| readout_logits(bundle.model, ids, torch.ones_like(ids), [0]) | |
| for bi, batch in enumerate(batches): | |
| exs = [examples[i] for i in batch] | |
| - id_lists = [bundle.encode(ex.prompt) for ex in exs] | |
| - input_ids, attention_mask, ends = pad_batch(id_lists, pad_id, dev) | |
| - ks = [len(ex.labels) for ex in exs] | |
| - vocab_logits = readout_logits(bundle.model, input_ids, attention_mask, ends) | |
| - sel, _ = slot_view(vocab_logits, bundle.slot_ids, ks) | |
| + sel, _ = batch_scores(bundle, bundle.model, exs, [rows[ex.idx] for ex in exs], dev) | |
| sel = sel.float().cpu().tolist() | |
| for r, ex in enumerate(exs): | |
| out.append(Scored(kind=ex.kind, k=len(ex.labels), logits=sel[r][: len(ex.labels)], | |
| diff --git a/train/eval_mm.py b/train/eval_mm.py | |
| index e87e72d..03a6022 100644 | |
| --- a/train/eval_mm.py | |
| +++ b/train/eval_mm.py | |
| def main() -> int: | |
| mm.set_root(args.media_root) | |
| bundle = common.load_bundle(args.model, device=args.device, dtype="bfloat16", multimodal=args.multimodal, | |
| chat_template=args.chat_template) | |
| + from qev.readout import HEAD_FILE, Readout, ReadoutConfig | |
| + | |
| + rcfg = ReadoutConfig.load(args.model) | |
| + head_path = Path(args.model) / HEAD_FILE | |
| + bundle.readout = Readout.attach(bundle.model, rcfg, slot_ids=bundle.slot_ids, | |
| + head_path=head_path if rcfg.head else None) | |
| + print(f"readout: attention {rcfg.attention}, scores {rcfg.readout}", flush=True) | |
| results = {} | |
| for spec in args.evals: | |
| name, _, path = spec.partition("=") | |
| diff --git a/train/sft.py b/train/sft.py | |
| index 4f6f7bd..ae428d9 100644 | |
| --- a/train/sft.py | |
| +++ b/train/sft.py | |
| def build_parser() -> argparse.ArgumentParser: | |
| "train one step on each listed micro-batch (one GPU), peak memory reset per step, no final eval") | |
| p.add_argument("--dump-first-grad", default=None, | |
| help="test hook: after the first step's all-reduce, rank 0 saves the adapter gradients here") | |
| + p.add_argument("--attention", choices=["causal", "noncausal"], default="causal", | |
| + help="noncausal: bidirectional within the prefix and within the question suffix in the full-attention " | |
| + "layers (qev.readout)") | |
| + p.add_argument("--readout", choices=["slot_lm", "head"], default="slot_lm", | |
| + help="head: learned slot + pointer head over the final hidden states instead of LM-head letter logits") | |
| + p.add_argument("--head-dim", type=int, default=256, help="pointer projection width of the readout head") | |
| + p.add_argument("--head-lr", type=float, default=1e-4, help="peak learning rate of the pointer projections " | |
| + "(the head's slot rows use --lr)") | |
| p.add_argument("--disable-fla-kernels", action="store_true", | |
| help="force the reference PyTorch gated delta rule instead of the fla kernel " | |
| "(needed on H100/H200 with Triton in [3.4.0, 3.7.1); correct but much slower)") | |
| def collect_mm(bundle, rows, max_len: int, batch_tokens: int, max_rows: int, dev | |
| common.readout_logits(bundle.model, ids, torch.ones_like(ids), [0]) | |
| for b in batches: | |
| exs = [examples[usable[j]] for j in b] | |
| - inputs, ends = mm.mm_inputs(bundle.processor, [ex.prompt for ex in exs], | |
| - [rows[ex.idx].get("media") or [] for ex in exs], device) | |
| - vocab_logits = mm.mm_readout_logits(bundle.model, inputs, ends) | |
| - sel, _ = common.slot_view(vocab_logits, bundle.slot_ids, [len(ex.labels) for ex in exs]) | |
| + sel, _ = common.batch_scores(bundle, bundle.model, exs, [rows[ex.idx] for ex in exs], device) | |
| sel = sel.float().cpu().tolist() | |
| for r, ex in enumerate(exs): | |
| out.append(common.Scored(kind=ex.kind, k=len(ex.labels), logits=sel[r][: len(ex.labels)], | |
| def collect_mm(bundle, rows, max_len: int, batch_tokens: int, max_rows: int, dev | |
| def save_weights(model, bundle, out_dir: Path, full: bool, sharded: bool = False, is_main: bool = True) -> None: | |
| + _save_weights(model, bundle, out_dir, full, sharded, is_main) | |
| + if is_main and bundle.readout is not None: | |
| + out_dir.mkdir(parents=True, exist_ok=True) | |
| + bundle.readout.save(out_dir) | |
| + | |
| + | |
| +def _save_weights(model, bundle, out_dir: Path, full: bool, sharded: bool = False, is_main: bool = True) -> None: | |
| """LoRA: the adapter. Full fine-tuning: the whole model in bf16 (from the fp32 master weights, copied to CPU), | |
| with tokenizer and processor, loadable like the base checkpoint. Sharded (--fsdp): every rank must call it, | |
| the weights are gathered to rank 0's CPU and written there as HF safetensors shards.""" | |
| def main(argv=None) -> int: | |
| raise SystemExit("--fsdp is implemented without --share-prefix") | |
| if args.disable_fla_kernels: | |
| common.block_fla_kernels() | |
| + if (args.attention != "causal" or args.readout != "slot_lm") and args.share_prefix: | |
| + raise SystemExit("--share-prefix supports only causal slot_lm") | |
| import torch | |
| from peft import LoraConfig, get_peft_model | |
| def main(argv=None) -> int: | |
| if args.plan_only: | |
| return plan_only(bundle, rows, args, world) | |
| log.info("model_type=%s template_kwargs=%s", bundle.model_type, bundle.template_kwargs) | |
| + from qev.readout import Readout, ReadoutConfig | |
| + | |
| + bundle.readout = Readout.attach(bundle.model, ReadoutConfig(attention=args.attention, readout=args.readout, | |
| + head_dim=args.head_dim), | |
| + slot_ids=bundle.slot_ids, device=torch.device(args.device)) | |
| + head = bundle.readout.head | |
| + if head is not None and world > 1: | |
| + for prm in head.parameters(): | |
| + dist.broadcast(prm.data, 0) | |
| + log.info("readout: attention %s, scores %s", args.attention, args.readout) | |
| if args.full_finetune: | |
| model = bundle.model | |
| def main(argv=None) -> int: | |
| opt = torch.optim.AdamW(params, lr=args.lr, weight_decay=args.weight_decay, betas=(0.9, 0.95), eps=1e-8, | |
| fused=bool(args.full_finetune)) | |
| sched = get_cosine_schedule_with_warmup(opt, num_warmup_steps=warmup, num_training_steps=max(total_steps, 1)) | |
| + head_opt = head_sched = None | |
| + if head is not None: | |
| + head_opt = torch.optim.AdamW([{"params": [head.slot], "lr": args.lr}, | |
| + {"params": [head.q.weight, head.k.weight], "lr": args.head_lr}], | |
| + weight_decay=0.0, betas=(0.9, 0.95), eps=1e-8) | |
| + head_sched = get_cosine_schedule_with_warmup(head_opt, num_warmup_steps=warmup, | |
| + num_training_steps=max(total_steps, 1)) | |
| logf = open(log_path, "a", encoding="utf-8") if is_main else open(os.devnull, "w") | |
| step = 0 | |
| def main(argv=None) -> int: | |
| examples.append(common.render_example(rows[i], bundle.template_for(rows[i]), i, seed=seed, | |
| augment=args.augment, option_pool=OPTION_POOL, | |
| pad_prob=args.pad_options)) | |
| - if bundle.multimodal: | |
| - from train import mm | |
| - | |
| - inputs, ends = mm.mm_inputs(bundle.processor, [ex.prompt for ex in examples], | |
| - [rows[i].get("media") or [] for i in idx], device) | |
| - with torch.autocast("cuda", dtype=torch.bfloat16, enabled=use_autocast(args)): | |
| - vocab_logits = mm.mm_readout_logits(model, inputs, ends) | |
| - else: | |
| - id_lists = [bundle.encode(ex.prompt) for ex in examples] | |
| - input_ids, attention_mask, ends = common.pad_batch(id_lists, pad_id, device) | |
| - with torch.autocast("cuda", dtype=torch.bfloat16, enabled=use_autocast(args)): | |
| - vocab_logits = common.readout_logits(model, input_ids, attention_mask, ends) | |
| + with torch.autocast("cuda", dtype=torch.bfloat16, enabled=use_autocast(args)): | |
| + sel, mask = common.batch_scores(bundle, model, examples, [rows[i] for i in idx], device) | |
| ks = [len(ex.labels) for ex in examples] | |
| targets = [ex.target for ex in examples] | |
| kmax = max(ks) | |
| tgt = torch.zeros((len(ks), kmax), dtype=torch.float32, device=device) | |
| for r, t in enumerate(targets): | |
| tgt[r, : len(t)] = torch.tensor(t, dtype=torch.float32, device=device) | |
| - sel, mask = common.slot_view(vocab_logits, bundle.slot_ids, ks) | |
| + if kind == "group": | |
| + sel, mask = common.slot_view(vocab_logits, bundle.slot_ids, ks) | |
| return sel, mask, tgt | |
| if eval_sets and args.eval_at_start: | |
| def main(argv=None) -> int: | |
| gnorm = torch.nn.utils.clip_grad_norm_(params, args.max_grad_norm) | |
| if hasattr(gnorm, "full_tensor"): | |
| gnorm = gnorm.full_tensor() | |
| + if head is not None: | |
| + for prm in head.parameters(): | |
| + if prm.grad is None: | |
| + prm.grad = torch.zeros_like(prm) | |
| + if world > 1: | |
| + dist.all_reduce(prm.grad) | |
| + prm.grad /= grad_scale | |
| + torch.nn.utils.clip_grad_norm_(list(head.parameters()), args.max_grad_norm) | |
| + head_opt.step() | |
| + head_sched.step() | |
| + head_opt.zero_grad(set_to_none=True) | |
| try: | |
| opt.step() | |
| except torch.OutOfMemoryError: | |