"""The answer contract engine: contract v3 on the reference engine's float32 cached path. Adds to solomon/engine_reference.py (which is not modified): * a switchable LoRA layer, so an adapter can act on every token ('full', as the reference engine) or only on the question branch ('question'): off while the document is prefilled, on for the branch, and in full execution applied from the prefix boundary onward; * documents made of parts: text and page images, in any order. The vision tower runs once per document, the library's positions are used for the prefix, and the state carries the rope delta; * task blocks for every answer type, and ten answer letters; * state format v3. """ import copy import hashlib import json from pathlib import Path import numpy as np from solomon.engine_reference import (BOOLEAN_TASK, CHOICE_TASK, CHUNK, SYSTEM, TOKEN_CAP, Engine, Ledger, boolean_block, # noqa: F401 choice_block, dump, read_rows, sha, softmax) ROOT = Path(__file__).resolve().parents[1] MODEL = ROOT / 'runtime/qwen38-8bit' ART = ROOT / 'artifacts/contract' CONTRACT = 'solomon-answer-contract-v3' STATE_FORMAT = 'solomon-answer-state-v3' LETTERS = 'ABCDEFGHIJ' PAGE = '<|vision_start|><|image_pad|><|vision_end|>' IMAGE_PAD, VISION_END = '<|image_pad|>', '<|vision_end|>' RESERVED = ['The document does not state this', 'The document gives conflicting answers'] NOT_STATED, CONFLICTING = 'not_stated', 'conflicting' SINGLE_R = ('Task: choose the single option that the document establishes as the answer. If the document does not establish any of ' 'the other listed answers, choose the option that says it does not state this. If the document establishes two different ' 'listed answers and does not say which prevails, choose the option that says it gives conflicting answers. A replaced or ' 'withdrawn statement establishes nothing. Respond with exactly one letter. Do not explain.') ORDERED_R = ('Task: the options form an ordered scale, lowest first, followed by two special options. Choose the single level that the ' 'document establishes. If the document does not establish any level, choose the option that says it does not state this. If ' 'it establishes two different levels and does not say which prevails, choose the option that says it gives conflicting ' 'answers. A replaced or withdrawn statement establishes nothing. Respond with exactly one letter. Do not explain.') SINGLE_S = CHOICE_TASK ORDERED_S = ('Task: the options form an ordered scale, lowest first. Choose the single level that the document best supports. ' 'Respond with exactly one letter. Do not explain.') SUFFICIENCY = ('Task: classify what the document establishes about the answer to the question, using only the document. ' 'A = it establishes exactly one of the listed answers. B = it does not establish any of the listed answers. ' 'C = it establishes two or more different listed answers and does not say which prevails. ' 'A missing fact is not a negative fact. A replaced or withdrawn statement establishes nothing. ' 'Respond with exactly one letter: A, B, or C. Do not explain.') LABEL = ('Task: decide whether the label applies, using only the document. A = the document establishes that it applies, and nothing in ' 'it establishes that it does not. B = the document establishes that it does not apply, and nothing in it establishes that it ' 'does. C = the document establishes neither. D = the document establishes both. A missing fact is not a negative fact. ' 'Evidence about another person or subject does not count. A replaced or withdrawn statement establishes nothing. ' 'Respond with exactly one letter: A, B, C, or D. Do not explain.') OPTION = ('Task: decide whether the proposed answer is correct, using only the document. A = the document establishes this answer, and ' 'nothing in it establishes a different one. B = the document establishes a different answer, or establishes that this one is ' 'wrong, and nothing in it establishes this one. C = the document establishes neither. D = the document establishes both this ' 'answer and a different one. A missing fact is not a negative fact. A replaced or withdrawn statement establishes nothing. ' 'Respond with exactly one letter: A, B, C, or D. Do not explain.') def _lettered(options): return '\n'.join(f'{LETTERS[i]}. {text}' for i, text in enumerate(options)) def listwise_block(question, options, ordered=False, reserved=True): """Caller options in caller order; with `reserved` the service appends the two reserved outcomes.""" if not 2 <= len(options) <= 8 or len(set(options)) != len(options): raise ValueError('needs 2 to 8 distinct options') shown = list(options) + (RESERVED if reserved else []) task = (ORDERED_R if ordered else SINGLE_R) if reserved else (ORDERED_S if ordered else SINGLE_S) return task + '\nQuestion: ' + question + '\nOptions:\n' + _lettered(shown) + '\nAnswer (one letter):', len(shown) def sufficiency_block(question, options): return SUFFICIENCY + '\nQuestion: ' + question + '\nListed answers:\n' + '\n'.join('- ' + o for o in options) + '\nAnswer (one letter):', 3 def label_block(question, label): return LABEL + '\nQuestion: ' + question + '\nLabel: ' + label + '\nAnswer (one letter):', 4 def option_block(question, option): return OPTION + '\nQuestion: ' + question + '\nProposed answer: ' + option + '\nAnswer (one letter):', 4 def four_state_block(question): return boolean_block(question), 4 CTL = {'start': None} # None: adapter off. int s: adapter applied to positions >= s of the current call. def _switch_lora_class(): import mlx.core as mx import mlx.nn as nn class SwitchLoRA(nn.Module): def __init__(self, linear, lora_a, lora_b, scale): super().__init__() self.linear = linear self.lora_a = lora_a self.lora_b = lora_b self.scale = scale def __call__(self, x): y = self.linear(x) s = CTL['start'] if s is None or s >= x.shape[1]: return y xs = x if s <= 0 else x[:, s:] z = (self.scale * ((xs @ self.lora_a) @ self.lora_b)).astype(y.dtype) if s <= 0: return y + z return mx.concatenate([y[:, :s], y[:, s:] + z], axis=1) return SwitchLoRA def attach_adapter(lm, path, scale=2.0): """Wrap the projections named in the adapter file. Returns (wrapped count, rank).""" import mlx.core as mx weights = mx.load(str(path)) cls = _switch_lora_class() names = sorted({k.rsplit('.', 1)[0] for k in weights}) rank = None for name in names: parts = name.split('.') # model.layers.N.owner.key owner = lm.model.layers[int(parts[2])] for p in parts[3:-1]: owner = getattr(owner, p) a, b = weights[name + '.lora_a'].astype(mx.float32), weights[name + '.lora_b'].astype(mx.float32) rank = a.shape[1] setattr(owner, parts[-1], cls(getattr(owner, parts[-1]), a, b, scale)) lm.freeze() mx.eval(lm.parameters()) return len(names), rank def as_parts(document): """A document is a string (text) or a list of parts {'text': str} / {'image': path}.""" if isinstance(document, str): return [{'text': document}] if not document or any(set(p) not in ({'text'}, {'image'}) for p in document): raise ValueError('parts must be {"text": ...} or {"image": ...}') return document def document_key(document): h = hashlib.sha256() for p in as_parts(document): h.update(b'T' + p['text'].encode() if 'text' in p else b'I' + bytes.fromhex(sha(p['image']))) return h.hexdigest() class ContractEngine(Engine): def __init__(self, arithmetic='float32', ledger=None, adapter=None, placement='question'): if placement not in ('question', 'full'): raise ValueError('placement must be question or full') super().__init__(arithmetic, ledger, adapter=None) self.placement = placement self.adapter = None self.rank = None if adapter is not None: self.wrapped, self.rank = attach_adapter(self.lm, adapter) self.adapter = str(adapter) self._pad = self.t.convert_tokens_to_ids(IMAGE_PAD) # ---- rendering ------------------------------------------------------------------- def _content(self, parts): out = '' for i, p in enumerate(parts): if 'text' in p: out += ('\n' if i and 'image' in parts[i - 1] else '') + p['text'] else: out += ('\n' if i and 'text' in parts[i - 1] else '') + PAGE return out def _text(self, parts, block): msgs = [{'role': 'system', 'content': SYSTEM}, {'role': 'user', 'content': 'Document:\n' + self._content(parts) + '\n\n' + block}] return self.t.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True, enable_thinking=False) def _expand(self, ids, counts): out, k = [], 0 for tok in ids: if tok == self._pad: out += [tok] * counts[k] k += 1 else: out.append(tok) if k != len(counts): raise ValueError('image placeholder count differs from the number of images') return out def _pixels(self, parts): from PIL import Image images = [Image.open(p['image']).convert('RGB') for p in parts if 'image' in p] if not images: return None, None, [] out = self.processor.image_processor(images=images) grid = np.array(out['image_grid_thw']) merge = self.model.config.vision_config.spatial_merge_size ** 2 counts = [int(g[0] * g[1] * g[2]) // merge for g in grid] return self.mx.array(np.array(out['pixel_values'])), self.mx.array(grid), counts def _vision(self, pixels, grid): """Vision tower one image at a time (memory stays flat for long scans).""" mx = self.mx dtype = self.model.vision_tower.patch_embed.proj.weight.dtype feats, offset = [], 0 for i in range(grid.shape[0]): n = int(np.prod(np.array(grid[i]))) h, _ = self.model.vision_tower(pixels[offset:offset + n].astype(dtype), grid[i:i + 1]) mx.eval(h) feats.append(h) offset += n return mx.concatenate(feats, axis=0) # ---- execution ------------------------------------------------------------------- def _forward(self, ids, pos, cache, embeds=None, chunk=CHUNK, adapter_from=None): """Run ids through the language model; logits for the final position. adapter_from indexes into ids.""" mx = self.mx n = len(ids) if self.ledger is not None: self.ledger.add(forwards=1, input_tokens=n) step = n if chunk is None else chunk if cache is None and step < n: raise ValueError('Chunked execution needs a cache') logits = None for a in range(0, n, step): b = min(a + step, n) CTL['start'] = None if adapter_from is None else max(0, adapter_from - a) kwargs = dict(cache=cache, position_ids=pos[:, :, a:b], skip_logits=True) if embeds is not None: kwargs['inputs_embeds'] = embeds[:, a:b] if b == n: out = self.lm(mx.array([ids[a:b]]), return_hidden=True, **kwargs) lg = self.lm.lm_head(out.hidden_states[-1][:, -1:, :])[0, -1].astype(mx.float32) mx.eval(lg) logits = np.array(lg) del out else: self.lm(mx.array([ids[a:b]]), **kwargs) mx.eval([c.state for c in cache]) CTL['start'] = None if not np.isfinite(logits).all(): raise ValueError('Non-finite model output') if mx.get_peak_memory() > 100 * 2**30: raise MemoryError('100 GiB MLX allocation cap exceeded') return logits def _positions(self, start, n): mx = self.mx return mx.broadcast_to(mx.arange(start, start + n)[None, None, :], (3, 1, n)) def prefill(self, document, chunk=CHUNK): """Encode the task-neutral document prefix once. Returns state format v3.""" mx = self.mx parts = as_parts(document) text = self._text(parts, 'X') end = text.rfind('\n\nX') if end < 0: raise ValueError('Document boundary missing') raw = self.t.encode(text[:end], add_special_tokens=False) if 'text' in parts[-1]: raw = raw[:-1] # leave the boundary token uncached: its tokenisation can depend on what follows pixels, grid, counts = self._pixels(parts) ids = self._expand(raw, counts) if not ids or len(ids) > TOKEN_CAP: raise ValueError(f'prefix of {len(ids)} tokens is empty or exceeds the {TOKEN_CAP}-token cap') embeds, delta, feats = None, 0, None if counts: feats = self._vision(pixels, grid) f = self.model.get_input_embeddings(mx.array([ids]), pixels, image_grid_thw=grid, cached_image_features=feats) embeds, pos = f.inputs_embeds, f.position_ids delta = int(np.array(f.rope_deltas).reshape(-1)[0]) if delta != int(np.array(pos).max()) + 1 - len(ids): raise ValueError('rope delta disagrees with the prefix positions') else: pos = self._positions(0, len(ids)) cache = self.lm.make_cache() self._forward(ids, pos, cache, embeds, chunk, adapter_from=0 if (self.adapter and self.placement == 'full') else None) mx.clear_cache() return {'format': STATE_FORMAT, 'prefix_ids': ids, 'cache': cache, 'rope_delta': delta, 'parts': parts, 'image_tokens': int(sum(counts)), 'image_sha256': [sha(p['image']) for p in parts if 'image' in p], 'layout': ''.join('T' if 'text' in p else 'I' for p in parts), 'prefix_sha256': hashlib.sha256(json.dumps(ids).encode()).hexdigest(), '_pixels': pixels, '_grid': grid, '_counts': counts, '_feats': feats} def letters_for(self, text, n): ids = self.t.encode(text, add_special_tokens=False) out = [] for letter in LETTERS[:n]: if letter not in self._letters: ext = self.t.encode(text + letter, add_special_tokens=False) if len(ext) != len(ids) + 1 or ext[:-1] != ids: raise ValueError('Unstable answer-letter continuation') self._letters[letter] = ext[-1] out.append(self._letters[letter]) return out, ids def ask(self, state, block, n_letters, execution='cached', full_chunk=None): """Answer-letter distribution for one task block against a prefilled state.""" mx = self.mx text = self._text(state['parts'], block) letters, raw = self.letters_for(text, n_letters) ids = self._expand(raw, state['_counts']) prefix = state['prefix_ids'] P = len(prefix) fallback = None if ids[:P] != prefix: execution, fallback = 'full', 'prefix token mismatch' on = self.adapter is not None if execution == 'cached': branch = copy.deepcopy(state['cache']) logits = self._forward(ids[P:], self._positions(P + state['rope_delta'], len(ids) - P), branch, chunk=None, adapter_from=0 if on else None) del branch reused = P else: embeds = None if state['_counts']: f = self.model.get_input_embeddings(mx.array([ids]), state['_pixels'], image_grid_thw=state['_grid'], cached_image_features=None if len(state['_counts']) <= 6 else state['_feats']) embeds, pos = f.inputs_embeds, f.position_ids else: pos = self._positions(0, len(ids)) start = None if not on else (0 if self.placement == 'full' else P) logits = self._forward(ids, pos, self.lm.make_cache() if full_chunk else None, embeds, full_chunk, adapter_from=start) reused = 0 full = softmax(logits) mx.clear_cache() return {'letter_logits': logits[letters].astype(np.float64), 'probabilities': softmax(logits[letters]), 'mass': float(full[letters].sum()), 'top_is_letter': int(logits.argmax()) in letters, 'execution': execution, 'fallback': fallback, 'prompt_tokens': len(ids), 'reused_prefix_tokens': reused, 'branch_tokens': len(ids) - reused} def identity(engine): """Runtime fingerprint for state format v3. With question placement the adapter is not part of the state.""" from importlib.metadata import version ip = engine.processor.image_processor payload = {'contract': CONTRACT, 'state_format': STATE_FORMAT, 'system_sha256': hashlib.sha256(SYSTEM.encode()).hexdigest(), 'arithmetic': engine.arithmetic, 'engine_sha256': sha(Path(__file__)), 'base_engine_sha256': sha(ROOT / 'solomon/engine_reference.py'), 'model_config_sha256': sha(MODEL / 'config.json'), 'image_processor': {'class': type(ip).__name__, 'config_sha256': sha(MODEL / 'preprocessor_config.json') if (MODEL / 'preprocessor_config.json').exists() else None}, 'state_adapter': (engine.adapter and sha(engine.adapter)) if engine.placement == 'full' else None, 'versions': {p: version(p) for p in ('mlx', 'mlx-vlm', 'transformers')}} payload['fingerprint'] = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest() payload['branch_adapter'] = engine.adapter and {'sha256': sha(engine.adapter), 'placement': engine.placement, 'rank': engine.rank} return payload