"""qev_image: System One decisions with images, on qev-0.8b (or any Qwen3.5 image-text checkpoint). from qev_image import QEVImage m = QEVImage.from_pretrained("labqoat/squad-0.8b") # Qwen3_5ForConditionalGeneration, bf16 m.decide({"ticket": "A customer sent this screenshot: "}, # or a plain string state {"urgent": {"type": "noul", "instructions": "Does the screenshot show an error dialog?"}}, images={"shot": "error.png"}) # PIL image or path m.decide_batch([(state, questions, images), ...]) # images optional per request: (state, questions) It is qev.py (the text runner, same prompt layout, letter-logit readout and per-type temperatures) with images in the "Context:" part. Where the state mentions "" the image goes there; an image the state does not mention is appended at the end of the context. An image becomes <|vision_start|> + <|image_pad|> x N + <|vision_end|>, N = (resized height/32) * (resized width/32) for the Qwen3.5 patch size 16 and 2x2 merge. The vision tower and merger are the base model's (qev's training touched only the text layers), the text stack is qev's. A text-only request takes the same path with no pixels and answers as qev.py does (within bf16 round-off). Image size is bounded: the image is resized (aspect kept) to at most max_pixels (default 448*448 = 196 tokens) and at least min_pixels, each side a multiple of 32. decide_batch(..., usage=True) returns (answers, input tokens, image tokens) per request; image_tokens(image) gives the count for one image. Needs: torch, transformers with Qwen3.5 (5.x), pillow, torchvision, decider-ai==1.9.0 (requirements.txt). Run it on a GPU (CPU works, slowly). The image processor is the base model's (preprocessor_config.json of Qwen3.5-0.8B-Base). """ from __future__ import annotations import argparse import json import math import os import re 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 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"]} # ---- images --------------------------------------------------------------------------------------------------------- IMAGE_MARK = re.compile(r"\s]+)>") FACTOR = 32 # patch 16 x merge 2 def _device(device): import torch if device: return torch.device(device) if torch.cuda.is_available(): return torch.device("cuda") return torch.device("cpu") def _open(im): from PIL import Image if isinstance(im, (str, os.PathLike)): im = Image.open(im) return im.convert("RGB") def fit_size(w: int, h: int, min_pixels: int, max_pixels: int, factor: int = FACTOR) -> tuple[int, int]: """(width, height) a multiple of `factor` each, aspect kept, with min_pixels <= w*h <= max_pixels (Qwen smart_resize).""" hh = max(factor, round(h / factor) * factor) ww = max(factor, round(w / factor) * factor) if hh * ww > max_pixels: beta = math.sqrt(h * w / max_pixels) hh, ww = max(factor, math.floor(h / beta / factor) * factor), max(factor, math.floor(w / beta / factor) * factor) elif hh * ww < min_pixels: beta = math.sqrt(min_pixels / (h * w)) hh, ww = math.ceil(h * beta / factor) * factor, math.ceil(w * beta / factor) * factor return ww, hh class QEVImage: def __init__(self, model, tok, ip, config: dict, device, dtype, max_pixels: int, min_pixels: int): import torch from decider.prompt import letter_ids self.model, self.tok, self.ip, self.config, self.device, self.dtype = model, tok, ip, config, device, dtype self.max_pixels, self.min_pixels = max_pixels, min_pixels self.temps = dict(config["temperatures"]) mc = model.config self.start_id, self.end_id, self.pad_id_img = mc.vision_start_token_id, mc.vision_end_token_id, mc.image_token_id self.start_tok, self.end_tok = tok.convert_ids_to_tokens(self.start_id), tok.convert_ids_to_tokens(self.end_id) self.pad_tok = tok.convert_ids_to_tokens(self.pad_id_img) 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() self.pad = tok.pad_token_id if tok.pad_token_id is not None else (tok.eos_token_id or 0) self.last_image_tokens: list = [] @classmethod def from_pretrained(cls, path: str, device: str | None = None, dtype=None, config: str | dict | None = None, max_pixels: int = 448 * 448, min_pixels: int = 64 * 64, revision: str | None = None) -> "QEVImage": """path: a local directory (model.safetensors with the vision tower, config.json with vision_config, preprocessor_config.json, tokenizer) or a Hugging Face repo id.""" import torch from transformers import AutoImageProcessor, AutoModelForImageTextToText, AutoTokenizer if not os.path.isdir(path): from huggingface_hub import snapshot_download path = snapshot_download(path, revision=revision) dev = _device(device) dtype = dtype or (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))) tok = use_training_pretokenizer(AutoTokenizer.from_pretrained(path)) ip = AutoImageProcessor.from_pretrained(path) model = AutoModelForImageTextToText.from_pretrained(path, dtype=dtype).to(dev).eval() return cls(model, tok, ip, cfg, dev, dtype, max_pixels, min_pixels) # ---- images -> pixel values + placeholder text ------------------------------------------------------------------ def _prep(self, im): """PIL/path -> (pixel_values [patches, D], grid_thw [1, 3], n image tokens).""" im = _open(im) w, h = fit_size(im.size[0], im.size[1], self.min_pixels, self.max_pixels) im = im.resize((w, h), 3) out = self.ip(images=[im], do_resize=False, return_tensors="pt") thw = out["image_grid_thw"] n = int(thw.prod(-1).sum()) // 4 return out["pixel_values"], thw, n def image_tokens(self, im) -> int: return self._prep(im)[2] def _row(self, state, questions: dict, images: dict | None): """One prompt row (qev.encode with the images expanded in the context) and its pixel inputs.""" 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") images = dict(images or {}) ctx = render_state(state) named = [] for m in IMAGE_MARK.finditer(ctx): if m.group(1) not in images: raise ValueError(f"the state mentions but no such image was given") named.append(m.group(1)) for k in images: # an image the state does not mention goes at the end if k not in named: ctx += f"\n" named.append(k) prepped = {k: self._prep(images[k]) for k in dict.fromkeys(named)} ctx = IMAGE_MARK.sub(lambda m: self.start_tok + self.pad_tok * prepped[m.group(1)][2] + self.end_tok, ctx) mo, ms = self.config["max_options"], self.config["max_state_tokens"] rqs = {k: render_question(v) for k, v in questions.items()} flat_q, index = plan_rows(rqs, False) item = build(_Example(ctx, [_Q(r["question"], neutralize_options(r["options"]), 0) for r in flat_q]), self.tok, _NoShuffle(), max_options=mo, max_ctx_tokens=ms, layout="state_first", chat=None) enc = {"ids": item["ids"], "slots": item["slots"], "nopts": item["nopts"], "rqs": rqs, "index": index, "types": row_types(rqs, index)} order = named # occurrence order in the context = order of the placeholder runs want = sum(prepped[k][2] for k in order) got = sum(1 for t in enc["ids"] if t == self.pad_id_img) if want != got: raise RuntimeError(f"image placeholder tokens: expected {want}, the tokenizer produced {got}") enc["pix"] = [prepped[k][0] for k in order] enc["thw"] = [prepped[k][1] for k in order] enc["image_tokens"] = want return enc # ---- rows -> letter logits ---------------------------------------------------------------------------------------- def _logits(self, rows: list, max_batch_tokens: int = 16384) -> list: 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) mask = torch.zeros((len(grp), L), dtype=torch.long) types = torch.zeros((len(grp), L), dtype=torch.int) for b, j in enumerate(grp): r = torch.as_tensor(rows[j]["ids"]) ids[b, : len(r)] = r mask[b, : len(r)] = 1 types[b, : len(r)] = (r == self.pad_id_img).int() kw = {} pix = [p for j in grp for p in rows[j]["pix"]] if pix: kw["pixel_values"] = torch.cat(pix).to(self.device, self.dtype) kw["image_grid_thw"] = torch.cat([t for j in grp for t in rows[j]["thw"]]).to(self.device) if _takes_mm_types(self.model): kw["mm_token_type_ids"] = types.to(self.device) with torch.no_grad(): h = self.model.model(input_ids=ids.to(self.device), attention_mask=mask.to(self.device), use_cache=False, **kw).last_hidden_state 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 _norm(self, req): s, q, *rest = req return s, q, (rest[0] if rest else None) def distributions_batch(self, requests: list, max_batch_tokens: int = 16384) -> list: """[(state, questions[, images])] -> per request {question id: {option name: probability}} (tempered).""" rows = [self._row(*self._norm(r)) for r in requests] self.last_image_tokens = [r["image_tokens"] for r in rows] return [flat(r, probabilities(r, lg, self.temps)) for r, lg in zip(rows, self._logits(rows, max_batch_tokens), strict=True)] def decide_batch(self, requests: list, usage: bool = False, max_batch_tokens: int = 16384) -> list: """[(state, questions[, images])] -> per request {question id: answer} (decider's /v1/systemone answer format). usage=True: per request (answers, input tokens, image tokens).""" rows = [self._row(*self._norm(r)) for r in requests] self.last_image_tokens = [r["image_tokens"] for r in rows] res = [answers(r, probabilities(r, lg, self.temps)) for r, lg in zip(rows, self._logits(rows, max_batch_tokens), strict=True)] return [(a, len(r["ids"]), r["image_tokens"]) for a, r in zip(res, rows, strict=True)] if usage else res def decide(self, state, questions: dict, images: dict | None = None) -> dict: return self.decide_batch([(state, questions, images)])[0] def _takes_mm_types(model) -> bool: import inspect try: return "mm_token_type_ids" in inspect.signature(model.model.forward).parameters except (TypeError, ValueError): return False def main(): ap = argparse.ArgumentParser(description="qev-0.8b System One decisions with images") ap.add_argument("--model", required=True, help="local directory (with the vision tower) or Hugging Face repo id") ap.add_argument("--state", required=True, help="the state: a string (use ), or JSON") ap.add_argument("--questions", required=True, help="JSON {id: {type, instructions, ...}}") ap.add_argument("--image", action="append", default=[], help="NAME=PATH (repeatable)") ap.add_argument("--max-pixels", type=int, default=448 * 448) ap.add_argument("--device", default=None) a = ap.parse_args() m = QEVImage.from_pretrained(a.model, device=a.device, max_pixels=a.max_pixels) try: state = json.loads(a.state) except json.JSONDecodeError: state = a.state imgs = dict(x.split("=", 1) for x in a.image) print(json.dumps(m.decide(state, json.loads(a.questions), imgs), indent=1, ensure_ascii=False)) if __name__ == "__main__": main()