qev_fast: image input on the fast CUDA engine; qev.py / qev_server.py images; card speed numbers
fd083b9 verified | """qev: System One decisions with qev-0.8b on plain PyTorch + transformers. | |
| from qev import QEV | |
| m = QEV.from_pretrained("labqoat/qev-0.8b") # CUDA, then Apple MPS, then CPU | |
| m.decide({"ticket": "My card was charged twice, please refund one."}, | |
| {"dept": {"type": "choice", "instructions": "Which team should handle this?", | |
| "criteria": {"billing": "payments and refunds", "tech": "bugs", "sales": None}}, | |
| "urgent": {"type": "noul", "instructions": "Is this urgent?"}}) | |
| m.decide_batch([(state, questions), ...]) | |
| # images (fast CUDA backend; pillow + torchvision): "<image:NAME>" in the state, or appended at the end | |
| m.decide("Customer screenshot: <image:shot>", {"err": {"type": "noul", "instructions": "Is an error shown?"}}, | |
| images={"shot": "screenshot.png"}) # PIL image or path | |
| m.decide_batch([(state, questions, {"shot": img}), (state2, questions2)]) | |
| python qev.py --model labqoat/qev-0.8b --state '{"ticket": "..."}' --questions '{"urgent": {"type": "noul", "instructions": "Is this urgent?"}}' | |
| How a request is answered (no text is generated): | |
| 1. The state and the questions become one prompt row (decider-ai's state-first System One layout: the state once, | |
| then every question with its lettered options and an "Answer: (" slot). per_question=True instead builds one row | |
| per question, each with its own copy of the state (higher consistency across questions, see the model card). | |
| 2. One forward pass of the model gives the hidden state at each answer slot; the logits of the option letter tokens | |
| are read from the tied embedding rows of those letters only. | |
| 3. Each question's answer is the softmax over its own options, renormalised over the named options, then sharpened | |
| per question type with the temperatures of qev_config.json (p -> p^(1/t), renormalised). | |
| 4. The answers are in decider's /v1/systemone format (decider.systemone.assemble / format_answer). | |
| Backends: "fast" (the default on CUDA when flash-linear-attention is installed: qev_fast/, a packed varlen engine | |
| with CUDA graphs and a shared-prefix per-question path; pip install -r requirements-cuda.txt) or "transformers" | |
| (plain transformers forward; CPU, Apple MPS or CUDA). backend="auto" picks fast when it can. | |
| Images: the fast backend reads images as qev_image.QEVImage does (same resize to at most max_pixels, default 448*448 | |
| = 196 image tokens, same placeholders and prompt layout, same readout and temperatures): the checkpoint's vision tower | |
| once per distinct image, its embeddings at the image placeholders, 3-D M-RoPE positions. The transformers backend of | |
| this file is text-only (use qev_image.py there). | |
| Requires: torch, transformers (Qwen3.5 support), decider-ai==1.9.0 (Apache-2.0; prompt layout and answer format); | |
| for the fast backend also flash-linear-attention (requirements-cuda.txt); for images pillow and torchvision. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import math | |
| import os | |
| DEFAULT_CONFIG = { | |
| "format": "qev-systemone/1", | |
| "temperatures": {"choice": 1.7, "noul": 0.9, "score": 1.25}, | |
| "max_options": 255, | |
| "max_state_tokens": 1000000, | |
| } | |
| NEUTRAL_NONE = "not listed here" | |
| # The pre-tokenizer the training data was encoded with (the Qwen2 split pattern). transformers' Qwen tokenizer | |
| # classes impose their own pattern whatever tokenizer.json says (5.17: this one; 5.20+: the Qwen3.5 pattern with | |
| # combining marks), so it is set explicitly: otherwise words in scripts with combining marks (Indic and others) | |
| # split differently from training. | |
| TRAINING_PRE_TOKENIZER = json.loads('{"type": "Sequence", "pretokenizers": [{"type": "Split", "pattern": {"Regex": "(?i:\'s|\'t|\'re|\'ve|\'m|\'ll|\'d)|[^\\\\r\\\\n\\\\p{L}\\\\p{N}]?\\\\p{L}+|\\\\p{N}| ?[^\\\\s\\\\p{L}\\\\p{N}]+[\\\\r\\\\n]*|\\\\s*[\\\\r\\\\n]+|\\\\s+(?!\\\\S)|\\\\s+"}, "behavior": "Isolated", "invert": false}, {"type": "ByteLevel", "add_prefix_space": false, "trim_offsets": true, "use_regex": false}]}') | |
| def use_training_pretokenizer(tok): | |
| """Set the training-time pre-tokenizer on a transformers fast tokenizer (in place) -> tok.""" | |
| from tokenizers import Tokenizer | |
| t = json.loads(tok.backend_tokenizer.to_str()) | |
| t["pre_tokenizer"] = TRAINING_PRE_TOKENIZER | |
| tok.backend_tokenizer.pre_tokenizer = Tokenizer.from_str(json.dumps(t)).pre_tokenizer | |
| return tok | |
| # ---- prompt rows (decider-ai's System One layout) ---------------------------------------------------------------- | |
| def neutralize_options(options): | |
| """Options reading like 'none of the above' -> a neutral phrasing (decider.infer.neutralize_options).""" | |
| out = [] | |
| for o in options: | |
| key = o.strip().lower() | |
| out.append(NEUTRAL_NONE if (key.startswith("none of the above") or key in ("none of the above", "none", "n/a", | |
| "none of these")) else o) | |
| return out | |
| class _Q: | |
| __slots__ = ("gold", "options", "text") | |
| def __init__(self, text, options, gold=0): | |
| self.text, self.options, self.gold = text, options, gold | |
| class _Example: | |
| __slots__ = ("context", "image", "qs", "task") | |
| def __init__(self, context, qs): | |
| self.context, self.qs, self.task, self.image = context, qs, "infer", None | |
| class _NoShuffle: | |
| def shuffle(self, x): | |
| pass | |
| def sample(self, xs, k): | |
| return xs[:k] | |
| def encode(tok, state, questions: dict, max_options: int = 255, max_state_tokens: int = 10**6) -> dict: | |
| """One prompt row: {"ids", "slots", "nopts", "rqs", "index", "types"} (every question behind one copy of the | |
| state). A string state is used as it is; an object or array is rendered by decider (JSON with array indices | |
| annotated), as decider's own server does.""" | |
| from decider.prompt import build | |
| from decider.systemone import plan_rows, render_question, render_state, row_types | |
| if not isinstance(questions, dict) or not questions: | |
| raise ValueError("questions: a non-empty {id: question} object") | |
| ctx = render_state(state) | |
| rqs = {k: render_question(v) for k, v in questions.items()} | |
| flat, index = plan_rows(rqs, False) | |
| item = build(_Example(ctx, [_Q(r["question"], neutralize_options(r["options"]), 0) for r in flat]), tok, | |
| _NoShuffle(), max_options=max_options, max_ctx_tokens=max_state_tokens, layout="state_first", | |
| chat=None) | |
| return {"ids": item["ids"], "slots": item["slots"], "nopts": item["nopts"], "rqs": rqs, "index": index, | |
| "types": row_types(rqs, index)} | |
| def _softmax(xs): | |
| m = max(xs) | |
| e = [math.exp(x - m) for x in xs] | |
| s = sum(e) | |
| return [x / s for x in e] | |
| def probabilities(enc: dict, letter_logits, temps: dict) -> list: | |
| """Per question row: softmax over its option letters, renormalised over the named options, tempered by type.""" | |
| out = [] | |
| for row, n, t, (k, _kind, _s, _c) in zip(letter_logits, enc["nopts"], enc["types"], enc["index"], strict=True): | |
| names = enc["rqs"][k]["names"] | |
| p = _softmax([float(x) for x in row[:n]])[: len(names)] | |
| z = sum(p) or 1e-9 | |
| p = [x / z for x in p] | |
| temp = temps.get(t, 1.0) | |
| if temp != 1.0: | |
| w = [max(x, 1e-9) ** (1 / temp) for x in p] | |
| z = sum(w) | |
| p = [x / z for x in w] | |
| out.append(p) | |
| return out | |
| def answers(enc: dict, probs: list) -> dict: | |
| from decider.systemone import assemble | |
| return assemble(enc["rqs"], enc["index"], probs) | |
| def tempered_from_slot_probs(enc: dict, rows: list, temps: dict) -> list: | |
| """Each slot's option probabilities (softmax over its options) -> per question renormalised over its named | |
| options, tempered by type (the same numbers probabilities() gives from logits).""" | |
| out = [] | |
| for (k, _kind, s, _c), t in zip(enc["index"], enc["types"], strict=True): | |
| p = rows[s][: len(enc["rqs"][k]["names"])] | |
| z = sum(p) or 1e-9 | |
| p = [x / z for x in p] | |
| temp = temps.get(t, 1.0) | |
| if temp != 1.0: | |
| w = [max(x, 1e-9) ** (1 / temp) for x in p] | |
| z = sum(w) | |
| p = [x / z for x in w] | |
| out.append(p) | |
| return out | |
| def _fast_available() -> bool: | |
| import importlib.util | |
| import torch | |
| return torch.cuda.is_available() and importlib.util.find_spec("fla") is not None | |
| def flat(enc: dict, probs: list) -> dict: | |
| """{question id: {option name: probability}} (unrounded; the served distribution).""" | |
| return {k: {str(nm): float(x) for nm, x in zip(enc["rqs"][k]["names"], probs[s], strict=False)} | |
| for (k, _kind, s, _c) in enc["index"]} | |
| # ---- the model --------------------------------------------------------------------------------------------------- | |
| def _device(device: str | None): | |
| import torch | |
| if device: | |
| return torch.device(device) | |
| if torch.cuda.is_available(): | |
| return torch.device("cuda") | |
| if getattr(torch.backends, "mps", None) is not None and torch.backends.mps.is_available(): | |
| return torch.device("mps") | |
| return torch.device("cpu") | |
| class QEV: | |
| def __init__(self, model, tok, config: dict, device, dtype, fast=None): | |
| import torch | |
| from decider.prompt import letter_ids | |
| self.model, self.tok, self.config, self.device, self.dtype = model, tok, config, device, dtype | |
| self.fast = fast # a qev_fast.FastQEV (backend "fast"), else the transformers forward below | |
| self.backend = "fast" if fast is not None else "transformers" | |
| self.temps = dict(config["temperatures"]) | |
| self.last_image_tokens: list = [] # per request of the last decide_batch / distributions_batch call | |
| if fast is not None: | |
| return | |
| letters = config.get("letter_token_ids") or letter_ids(tok) | |
| emb = model.get_output_embeddings().weight # tied with the input embedding | |
| self.W = emb[torch.as_tensor(letters, device=emb.device)].detach() # [255, H]: the letter rows only | |
| self.pad = tok.pad_token_id if tok.pad_token_id is not None else (tok.eos_token_id or 0) | |
| def from_pretrained(cls, path: str = "labqoat/qev-0.8b", device: str | None = None, dtype=None, | |
| config: str | dict | None = None, revision: str | None = None, backend: str = "auto", | |
| precapture: bool = True, images: bool = False, max_pixels: int = 448 * 448, | |
| min_pixels: int = 64 * 64) -> "QEV": | |
| """path: a local directory or a Hugging Face repo id. backend: "auto" (fast on CUDA with | |
| flash-linear-attention, else transformers), "fast" or "transformers". dtype (transformers backend): bf16 on | |
| CUDA / MPS, fp32 on CPU by default. precapture (fast): capture the common CUDA graphs now, the rest in the | |
| background. images (fast): load the vision tower now (else on the first request with images) and capture | |
| the image graphs with the others; max_pixels / min_pixels bound the resized image (as qev_image.py).""" | |
| import sys | |
| import torch | |
| from transformers import AutoModelForCausalLM, AutoTokenizer | |
| if not os.path.isdir(path): | |
| from huggingface_hub import snapshot_download | |
| path = snapshot_download(path, revision=revision, | |
| allow_patterns=["*.json", "*.safetensors", "*.txt", "*.py", "*.jinja", | |
| "qev_fast/*"]) | |
| if backend not in ("auto", "fast", "transformers"): | |
| raise ValueError("backend: auto, fast or transformers") | |
| use_fast = backend == "fast" or (backend == "auto" and (device in (None, "cuda") or str(device).startswith("cuda")) | |
| and _fast_available()) | |
| if backend == "auto" and not use_fast: | |
| if torch.cuda.is_available(): | |
| print("qev: using the plain transformers backend; for the fast CUDA engine: " | |
| "pip install -r requirements-cuda.txt (flash-linear-attention, causal-conv1d)", file=sys.stderr) | |
| elif getattr(torch.backends, "mps", None) is not None and torch.backends.mps.is_available(): | |
| print("qev: on Apple silicon, labqoat/qev-0.8b-MLX (qev_mlx.py) is several times faster than this " | |
| "PyTorch path", file=sys.stderr) | |
| dev = _device(device) | |
| if dtype is None: | |
| dtype = torch.float32 if dev.type == "cpu" else torch.bfloat16 | |
| cfg = dict(DEFAULT_CONFIG) | |
| if isinstance(config, dict): | |
| cfg.update(config) | |
| else: | |
| cp = config or os.path.join(path, "qev_config.json") | |
| if os.path.exists(cp): | |
| cfg.update(json.load(open(cp))) | |
| if use_fast: | |
| here = os.path.dirname(os.path.abspath(__file__)) | |
| for d in (here, path): # qev_fast/ next to qev.py, or in the model snapshot | |
| if os.path.isdir(os.path.join(d, "qev_fast")) and d not in sys.path: | |
| sys.path.insert(0, d) | |
| from qev_fast import FastQEV | |
| fast = FastQEV(path) | |
| fast.max_pixels, fast.min_pixels = max_pixels, min_pixels | |
| if images: | |
| fast.enable_images(max_pixels, min_pixels) | |
| if precapture: | |
| fast.precapture(background=True, images=images) | |
| return cls(None, fast.tok, cfg, torch.device("cuda"), torch.bfloat16, fast) | |
| tok = use_training_pretokenizer(AutoTokenizer.from_pretrained(path)) | |
| model = AutoModelForCausalLM.from_pretrained(path, dtype=dtype).to(dev).eval() | |
| return cls(model, tok, cfg, dev, dtype) | |
| # ---- rows -> letter logits ---------------------------------------------------------------------------------- | |
| def _logits(self, rows: list, max_batch_tokens: int = 16384) -> list: | |
| """Prompt rows (encode()) -> per row its letter logits [slots, 255] as lists. Rows run in length-sorted, | |
| right-padded batches (causal model: padding after a row never reaches its tokens).""" | |
| import torch | |
| order = sorted(range(len(rows)), key=lambda i: len(rows[i]["ids"])) | |
| out = [None] * len(rows) | |
| i = 0 | |
| while i < len(order): | |
| grp = [order[i]] | |
| i += 1 | |
| while i < len(order) and len(rows[order[i]]["ids"]) * (len(grp) + 1) <= max_batch_tokens: | |
| grp.append(order[i]) | |
| i += 1 | |
| L = max(len(rows[j]["ids"]) for j in grp) | |
| ids = torch.full((len(grp), L), self.pad, dtype=torch.long) | |
| for b, j in enumerate(grp): | |
| ids[b, : len(rows[j]["ids"])] = torch.as_tensor(rows[j]["ids"]) | |
| with torch.no_grad(): | |
| h = self.model.model(input_ids=ids.to(self.device), use_cache=False).last_hidden_state # final-normed | |
| for b, j in enumerate(grp): | |
| hs = h[b, torch.as_tensor(rows[j]["slots"], device=h.device)] | |
| out[j] = (hs.float() @ self.W.float().T).cpu().tolist() | |
| return out | |
| # ---- requests ------------------------------------------------------------------------------------------------- | |
| def _rows(self, state, questions: dict, per_question: bool) -> list: | |
| mo, ms = self.config["max_options"], self.config["max_state_tokens"] | |
| if per_question and len(questions) > 1: | |
| return [encode(self.tok, state, {k: q}, mo, ms) for k, q in questions.items()] | |
| return [encode(self.tok, state, questions, mo, ms)] | |
| def _fast_probs(self, requests: list, per_question: bool) -> list: | |
| """Fast backend: [(state, questions[, images])] -> per request [(row encoding, tempered probabilities)].""" | |
| f = self.fast | |
| if any(len(r) > 2 and r[2] for r in requests): | |
| return self._fast_image_probs(requests, per_question) | |
| requests = [tuple(r[:2]) for r in requests] | |
| if per_question: | |
| res = [] | |
| for s, q in requests: | |
| if not isinstance(q, dict) or not q: | |
| raise ValueError("questions: a non-empty {id: question} object") | |
| rows = [] | |
| for e in f.encoder.encode_many([(s, {k: v}) for k, v in q.items()] if len(q) > 1 else [(s, q)]): | |
| if isinstance(e, Exception): | |
| raise e | |
| rows.append(e) | |
| probs = f.per_question_probs(rows) if len(rows) > 1 else f.slot_probs(rows) | |
| res.append([(r, tempered_from_slot_probs(r, p, self.temps)) for r, p in zip(rows, probs, strict=True)]) | |
| return res | |
| encs = [] | |
| for (s, q), e in zip(requests, f.encoder.encode_many(requests), strict=True): | |
| if isinstance(e, Exception): | |
| raise e | |
| if not isinstance(q, dict) or not q: | |
| raise ValueError("questions: a non-empty {id: question} object") | |
| encs.append(e) | |
| return [[(e, tempered_from_slot_probs(e, p, self.temps))] for e, p in zip(encs, f.slot_probs(encs), strict=True)] | |
| def _fast_image_probs(self, requests: list, per_question: bool) -> list: | |
| """Fast backend, a call with images: rows from FastQEV.encode_requests (image placeholders expanded).""" | |
| f = self.fast | |
| for r in requests: | |
| if not isinstance(r[1], dict) or not r[1]: | |
| raise ValueError("questions: a non-empty {id: question} object") | |
| if per_question: | |
| res = [] | |
| for r in requests: | |
| s, q, imgs = r[0], r[1], (r[2] if len(r) > 2 else None) | |
| rows = [] | |
| for e in f.encode_requests([(s, {k: v}, imgs) for k, v in q.items()] if len(q) > 1 else [(s, q, imgs)]): | |
| if isinstance(e, Exception): | |
| raise e | |
| rows.append(e) | |
| probs = f.per_question_probs(rows) if len(rows) > 1 else f.slot_probs(rows) | |
| res.append([(r_, tempered_from_slot_probs(r_, p, self.temps)) for r_, p in zip(rows, probs, strict=True)]) | |
| return res | |
| encs = f.encode_requests(requests) | |
| for e in encs: | |
| if isinstance(e, Exception): | |
| raise e | |
| return [[(e, tempered_from_slot_probs(e, p, self.temps))] for e, p in zip(encs, f.slot_probs(encs), strict=True)] | |
| def _check_images(self, requests: list): | |
| if self.fast is None and any(len(r) > 2 and r[2] for r in requests): | |
| raise ValueError("images need the fast CUDA backend of qev.py; on other devices use qev_image.py") | |
| def _fast_results(self, requests: list, per_question: bool) -> list: | |
| res = self._fast_probs(requests, per_question) | |
| self.last_image_tokens = [sum(r.get("image_tokens", 0) for r, _ in rs) for rs in res] | |
| return res | |
| def distributions_batch(self, requests: list, per_question: bool = False) -> list: | |
| """[(state, questions[, images])] -> per request {question id: {option name: probability}} (tempered, | |
| unrounded). images: {name: PIL image or path} (fast backend).""" | |
| self._check_images(requests) | |
| if self.fast is not None: | |
| out = [] | |
| for rs in self._fast_results(requests, per_question): | |
| d = {} | |
| for r, p in rs: | |
| d.update(flat(r, p)) | |
| out.append(d) | |
| return out | |
| rows, owner = [], [] | |
| for i, (s, q, *_) in enumerate(requests): | |
| for r in self._rows(s, q, per_question): | |
| rows.append(r) | |
| owner.append(i) | |
| self.last_image_tokens = [0] * len(requests) | |
| res = [{} for _ in requests] | |
| for i, r, lg in zip(owner, rows, self._logits(rows), strict=True): | |
| res[i].update(flat(r, probabilities(r, lg, self.temps))) | |
| return res | |
| def decide_batch(self, requests: list, per_question: bool = False, usage: bool = False) -> list: | |
| """[(state, questions[, images])] -> per request {question id: answer} (decider's /v1/systemone answer | |
| format). images: {name: PIL image or path} (fast backend). usage=True: per request (answers, input tokens | |
| read, image tokens included); the image tokens per request are in last_image_tokens.""" | |
| self._check_images(requests) | |
| if self.fast is not None: | |
| out = [] | |
| for rs in self._fast_results(requests, per_question): | |
| d = {} | |
| for r, p in rs: | |
| d.update(answers(r, p)) | |
| out.append((d, sum(len(r["ids"]) for r, _ in rs)) if usage else d) | |
| return out | |
| rows, owner = [], [] | |
| for i, (s, q, *_) in enumerate(requests): | |
| for r in self._rows(s, q, per_question): | |
| rows.append(r) | |
| owner.append(i) | |
| self.last_image_tokens = [0] * len(requests) | |
| res, toks = [{} for _ in requests], [0] * len(requests) | |
| for i, r, lg in zip(owner, rows, self._logits(rows), strict=True): | |
| res[i].update(answers(r, probabilities(r, lg, self.temps))) | |
| toks[i] += len(r["ids"]) | |
| return list(zip(res, toks, strict=True)) if usage else res | |
| def decide(self, state, questions: dict, per_question: bool = False, images: dict | None = None) -> dict: | |
| """images: {name: PIL image or path} (fast backend); "<image:NAME>" in the state places one, the others go at | |
| the end of the context.""" | |
| return self.decide_batch([(state, questions, images) if images else (state, questions)], per_question)[0] | |
| def main(): | |
| ap = argparse.ArgumentParser(description="qev-0.8b System One decisions") | |
| ap.add_argument("--model", default="labqoat/qev-0.8b", help="local directory or Hugging Face repo id") | |
| ap.add_argument("--state", help="the state: a string, or JSON") | |
| ap.add_argument("--questions", help='JSON {id: {"type": "choice"|"noul"|"score", "instructions": ..., ...}}') | |
| ap.add_argument("--requests", help='a JSONL file of {"state": ..., "questions": {...}} (one answer line each)') | |
| ap.add_argument("--per-question", action="store_true", help="one prompt row per question") | |
| ap.add_argument("--device", default=None) | |
| ap.add_argument("--backend", default="auto", choices=["auto", "fast", "transformers"]) | |
| ap.add_argument("--image", action="append", default=[], help="NAME=PATH (repeatable; fast backend)") | |
| ap.add_argument("--max-pixels", type=int, default=448 * 448) | |
| a = ap.parse_args() | |
| m = QEV.from_pretrained(a.model, device=a.device, backend=a.backend, max_pixels=a.max_pixels) | |
| imgs = dict(x.split("=", 1) for x in a.image) or None | |
| if a.requests: | |
| reqs = [json.loads(line) for line in open(a.requests) if line.strip()] | |
| for r in m.decide_batch([(r["state"], r["questions"]) for r in reqs], a.per_question): | |
| print(json.dumps(r, ensure_ascii=False)) | |
| return | |
| if a.state is None or a.questions is None: | |
| ap.error("--state and --questions, or --requests") | |
| try: | |
| state = json.loads(a.state) | |
| except json.JSONDecodeError: | |
| state = a.state | |
| print(json.dumps(m.decide(state, json.loads(a.questions), a.per_question, images=imgs), indent=1, | |
| ensure_ascii=False)) | |
| if __name__ == "__main__": | |
| main() | |