"""qev: System One decisions with qev-0.8b on plain PyTorch + transformers. from qev import QEV m = QEV.from_pretrained("labqoat/squad-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): "" in the state, or appended at the end m.decide("Customer screenshot: ", {"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)]) # specialists (fast CUDA backend): a router in front, LoRA slots for games and the e2e tester m = QEV.from_pretrained("labqoat/squad-0.8b", specialists=True) m.decide(state, questions); m.last_routes # ["games"] / ["e2e-tester"] / ["trunk"] python qev.py --model labqoat/squad-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). Specialists (specialists=True, fast backend; the specialists/ directory of the repo): before answering, a router reads the request with its images removed (the text prompt row, "" markers as plain text): the mean over its tokens of the unit-RMS-normalised hidden state after decoder layer 5, then a small head (LayerNorm -> Linear -> GELU -> Linear) picks one of {trunk only, slot 1 games, slot 2 e2e tester (id "e2e-tester": browser / computer-use decisions)}. The chosen slot is a rank-64 LoRA on the MLPs (gate / up / down) of layers 12..23, y = W x + 2 B A x, applied to every position of the row, image tokens included. For a text request the router's 6 layers are the first 6 layers of the answer pass (computed once). Rows routed to "trunk" get exactly the trunk's answers. last_routes: the route of each request of the last call (per question for per_question rows). 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 self.last_routes: list = [] # specialists: per request of the last call, its route ("trunk", "games", ...) 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) @classmethod def from_pretrained(cls, path: str = "labqoat/squad-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, specialists: bool | str = False, prefix_cache_mb: int = 0) -> "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). specialists (fast backend): True loads the router and LoRA slots of the repo's specialists/ directory (a path: that directory); False: the trunk alone, exactly as without them. prefix_cache_mb (fast backend): > 0 keeps the model state after each request's "Context:\n" + state on the GPU (LRU, this many MB), so a later request with the same state computes only its questions (answers equal up to bf16 round-off); 0: off.""" 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/*"] + (["specialists/*"] if specialists is True else [])) if backend not in ("auto", "fast", "transformers"): raise ValueError("backend: auto, fast or transformers") spec_dir = None if specialists: spec_dir = os.path.join(path, "specialists") if specialists is True else str(specialists) if not os.path.exists(os.path.join(spec_dir, "manifest.json")): raise FileNotFoundError(f"no specialists in {spec_dir} (manifest.json)") use_fast = backend == "fast" or (backend == "auto" and (device in (None, "cuda") or str(device).startswith("cuda")) and _fast_available()) if spec_dir and not use_fast: raise ValueError("specialists need the fast CUDA backend (pip install -r requirements-cuda.txt)") 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/squad-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, specialists=spec_dir, prefix_cache_mb=prefix_cache_mb) 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 = [] many = [(s, {k: v}) for k, v in q.items()] if len(q) > 1 else [(s, q)] if f.slots is not None and len(q) > 1: # specialists: the joint row too (the router's input) many.append((s, q)) for e in f.encoder.encode_many(many): if isinstance(e, Exception): raise e rows.append(e) joint = rows.pop() if f.slots is not None and len(q) > 1 else None probs = f.per_question_probs(rows, joint) 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) if f.prefix.cache_mb > 0 and len(encs) <= 4: f.mark_prefix(encs, requests) 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) if f.slots is not None and len(q) > 1: # specialists: every row takes the joint text prompt's route jt = f.encoder.encode_many([(s, q)])[0] if isinstance(jt, Exception): raise jt for e in rows: if "router_ids" in e: e["router_ids"] = jt["ids"] 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, max_pixels: int | None = None) -> list: f = self.fast if max_pixels and max_pixels != f.max_pixels: # this call's image budget (images resized to at most this) old = f.max_pixels f.max_pixels = int(max_pixels) try: res = self._fast_probs(requests, per_question) finally: f.max_pixels = old else: 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] sl = self.fast.slots if sl is not None: self.last_routes = [] for rs in res: names = [sl.name(r.get("route", 0)) for r, _ in rs] if len(rs) == 1 or len(set(names)) == 1: self.last_routes.append(names[0]) else: # per_question rows routed differently: per question id self.last_routes.append({k: nm for (r, _), nm in zip(rs, names, strict=True) for k in r["rqs"]}) return res def distributions_batch(self, requests: list, per_question: bool = False, max_pixels: int | None = None) -> 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, max_pixels): 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) self.last_routes = [] 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, max_pixels: int | None = None) -> 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. max_pixels: this call's image budget (default: the one given to from_pretrained, 448*448); documents and receipts with small print read better at 1024*1024.""" self._check_images(requests) if self.fast is not None: out = [] for rs in self._fast_results(requests, per_question, max_pixels): 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) self.last_routes = [] 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, max_pixels: int | None = None) -> dict: """images: {name: PIL image or path} (fast backend); "" in the state places one, the others go at the end of the context. max_pixels: this call's image budget.""" return self.decide_batch([(state, questions, images) if images else (state, questions)], per_question, max_pixels=max_pixels)[0] def main(): ap = argparse.ArgumentParser(description="qev-0.8b System One decisions") ap.add_argument("--model", default="labqoat/squad-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) ap.add_argument("--specialists", action="store_true", help="router + LoRA slots (fast backend)") a = ap.parse_args() m = QEV.from_pretrained(a.model, device=a.device, backend=a.backend, max_pixels=a.max_pixels, specialists=a.specialists) 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()