File size: 16,275 Bytes
96a4100 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 | from __future__ import annotations
import json
import hashlib
import os
from pathlib import Path
from typing import Any, Protocol, Sequence
from .assets import load_codebook, load_runtime_manifest, sha256_file
from .compiler import NumericField, compile_numeric, compile_vision_numeric, decode_numeric
from .native_onepass import compile_joint, score_joint
class Backend(Protocol):
model_id: str
def numeric(self, context: str, fields: Sequence[NumericField], image: Any | None = None) -> dict[str, Any]: ...
def atomic(self, context: str, field_key: str, choices: Sequence[Any]) -> dict[str, Any]: ...
def finite_joint(self, context: str, schema: dict[str, Any], candidates: Sequence[dict[str, Any]]) -> dict[str, Any]: ...
class TransformersBackend:
"""Pinned Transformers/PEFT backend; no custom CUDA kernels or generation."""
@staticmethod
def _context_limits(manifest: dict[str, Any], model_config: dict[str, Any],
environ: dict[str, str]) -> tuple[int, int]:
capacity = int(model_config["text_config"]["max_position_embeddings"])
default = int(manifest["admission"]["default_max_input_tokens"])
limit = int(environ.get("GPC1_MAX_INPUT_TOKENS", str(default)))
if not 0 < limit <= capacity:
raise ValueError(f"GPC1_MAX_INPUT_TOKENS must be between 1 and {capacity}")
padded = int(environ.get("GPC1_MAX_PADDED_TOKENS", str(max(131072, 2 * limit))))
if padded <= 0:
raise ValueError("GPC1_MAX_PADDED_TOKENS must be positive")
return limit, padded
@staticmethod
def _verify_model_file_closure(model_path: Path, expected_files: set[str]) -> None:
top_files={p.name for p in model_path.iterdir() if p.is_file()}
suspicious={name for name in top_files if (
name.endswith((".safetensors",".json",".jinja",".txt"))
or "config" in name or "tokenizer" in name or "processor" in name or "index" in name)}
if suspicious != expected_files:
raise RuntimeError(f"loader-visible model file closure mismatch: {sorted(suspicious ^ expected_files)}")
@staticmethod
def _verify_adapter_file_closure(adapter_path: Path) -> None:
adapter_files={p.name for p in adapter_path.iterdir() if p.is_file()}
if adapter_files != {"adapter_config.json","adapter_model.safetensors"}:
raise RuntimeError("adapter directory must contain exactly the two bound release files")
@staticmethod
def _adapter_path(manifest: dict[str, Any], package_root: Path,
environ: dict[str, str]) -> Path:
if manifest.get("weight_format") != "adapter" or not isinstance(manifest.get("adapter"), dict) \
or manifest.get("merged_full_weights") is not None:
raise RuntimeError("release runtime requires adapter weight metadata only")
return Path(environ.get("GPC1_ADAPTER_PATH", package_root / "adapter"))
def __init__(self) -> None:
import torch
from transformers import AutoProcessor
from transformers import Qwen3_5MoeForConditionalGeneration
self.torch = torch
torch.backends.cuda.matmul.allow_tf32 = False
torch.backends.cudnn.allow_tf32 = False
torch.set_float32_matmul_precision("highest")
self.manifest = load_runtime_manifest()
package_root = Path(__file__).resolve().parents[1]
model_path = Path(os.environ["GPC1_MODEL_PATH"])
self.max_input_tokens, self.max_padded_tokens = self._context_limits(
self.manifest, json.loads((model_path / "config.json").read_text()), os.environ)
adapter_path = self._adapter_path(self.manifest, package_root, os.environ)
self._verify_artifacts(model_path, adapter_path)
self.processor = AutoProcessor.from_pretrained(model_path, local_files_only=True,
trust_remote_code=False)
base = Qwen3_5MoeForConditionalGeneration.from_pretrained(
model_path, local_files_only=True, trust_remote_code=False,
dtype=torch.bfloat16, attn_implementation="sdpa",
low_cpu_mem_usage=True, device_map={"": 0})
from peft import PeftModel
self.model = PeftModel.from_pretrained(base, adapter_path, is_trainable=False,
autocast_adapter_dtype=True)
self.model.eval()
for parameter in self.model.parameters():
parameter.requires_grad_(False)
if base.__class__.__name__ != "Qwen3_5MoeForConditionalGeneration" \
or getattr(base.config, "model_type", None) != "qwen3_5_moe":
raise RuntimeError("loaded base is not the pinned Qwen3.5 MoE architecture")
adapter_parameters = {name: p for name, p in self.model.named_parameters() if "lora_" in name}
base_dtypes = {p.dtype for name, p in self.model.named_parameters()
if p.is_floating_point() and "lora_" not in name}
if base_dtypes != {torch.bfloat16} or not adapter_parameters \
or {p.dtype for p in adapter_parameters.values()} != {torch.float32}:
raise RuntimeError("loaded base/adapter dtype inventory differs from BF16/FP32 contract")
from peft.utils import get_peft_model_state_dict
digest = hashlib.sha256()
state = get_peft_model_state_dict(self.model)
for name in sorted(state):
tensor = state[name].detach().cpu().contiguous()
metadata = json.dumps({"dtype":str(tensor.dtype),"name":name,"shape":list(tensor.shape)},
sort_keys=True,separators=(",", ":")).encode()
raw = tensor.view(torch.uint8).numpy().tobytes()
digest.update(len(metadata).to_bytes(8,"big")); digest.update(metadata)
digest.update(len(raw).to_bytes(8,"big")); digest.update(raw)
if digest.hexdigest() != self.manifest["adapter"]["adapter_tensor_digest"]:
raise RuntimeError("loaded adapter tensor digest mismatch")
self.model_id = self.manifest["served_model_id"]
self.codebook = load_codebook()
self.max_joint_rows = int(os.environ.get("GPC1_MAX_JOINT_ROWS", "32"))
self.max_logit_vectors = int(os.environ.get("GPC1_MAX_LOGIT_VECTORS", "512"))
self.max_logits_bytes = int(os.environ.get("GPC1_MAX_LOGITS_BYTES", str(512 * 1024 * 1024)))
accepted_modes = set(self.manifest["admission"]["accepted_modes"])
self.admit_vision = "numeric101_image" in accepted_modes \
and os.environ.get("GPC1_ADMIT_VISION_NUMERIC", "accepted") == "accepted"
self.admit_joint = "finite_joint" in accepted_modes \
and os.environ.get("GPC1_ADMIT_FINITE_JOINT", "accepted") == "accepted"
cache_mb = int(os.environ.get("GPC1_IMAGE_CACHE_MB", "64"))
if not 0 <= cache_mb <= 64:
raise ValueError("GPC1_IMAGE_CACHE_MB must be an integer between 0 and 64")
self._vision_cache = None
if cache_mb:
from .vision_cache import VisionFeatureCache
self._vision_cache = VisionFeatureCache(
self.model.base_model.model.model.visual, torch, cache_mb * 1024 * 1024)
def _verify_artifacts(self, model_path: Path, adapter_path: Path) -> None:
expected = self.manifest
package_root = Path(__file__).resolve().parents[1]
inventory_path = package_root / "assets" / "base_files.sha256.json"
inventory = json.loads(inventory_path.read_text())
if inventory["model_id"] != expected["base"]["model_id"] or inventory["revision"] != expected["base"]["revision"]:
raise RuntimeError("model-file inventory identity mismatch")
self._verify_model_file_closure(model_path, set(inventory["files"]))
for relative, digest in inventory["files"].items():
path = model_path / relative
if not path.is_file() or sha256_file(path) != digest:
raise RuntimeError(f"pinned model file identity mismatch: {relative}")
self._verify_adapter_file_closure(adapter_path)
if sha256_file(adapter_path / "adapter_model.safetensors") != expected["adapter"]["adapter_model_sha256"]:
raise RuntimeError("bundled adapter identity mismatch")
if sha256_file(adapter_path / "adapter_config.json") != expected["adapter"]["adapter_config_sha256"]:
raise RuntimeError("bundled adapter config identity mismatch")
def _forward(self, inputs: dict[str, Any], positions: Sequence[int]) -> Any:
torch = self.torch
device = next(self.model.parameters()).device
moved = {k: v.to(device) if hasattr(v, "to") else v for k, v in inputs.items()}
with torch.inference_mode():
# Project logits only at requested positions.
output = self.model(**moved, logits_to_keep=[int(p) for p in positions],
use_cache=False, return_dict=True)
if output.logits.ndim != 3 or tuple(output.logits.shape[:2]) != (1, len(positions)):
raise RuntimeError("native sparse logits do not align to query positions")
return output.logits[0].float()
def numeric(self, context: str, fields: Sequence[NumericField], image: Any | None = None) -> dict[str, Any]:
vocab = int(self.model.config.text_config.vocab_size)
if len(fields) * vocab * 4 > self.max_logits_bytes:
raise ValueError("numeric sparse logits exceed byte admission")
if image is None:
compiled = compile_numeric(self.processor.tokenizer, context, fields, self.codebook,
self.max_input_tokens)
ids = self.torch.tensor([compiled.input_ids], dtype=self.torch.long)
mask = self.torch.ones_like(ids)
inputs = {"input_ids": ids, "attention_mask": mask,
"position_ids": mask.cumsum(dim=-1) - 1}
positions = compiled.positions
else:
if not self.admit_vision:
raise ValueError("vision numeric requires an accepted same-hardware parity gate")
vision = compile_vision_numeric(self.processor, image, context, fields, self.codebook,
self.max_input_tokens, self.model.config)
compiled = vision.numeric; inputs = vision.processor_inputs; positions = vision.positions
logits = self._forward(inputs, positions)
token_ids = self.torch.tensor(compiled.candidate_token_ids, device=logits.device)
results = {}
for index, field in enumerate(fields):
raw = logits[index].index_select(0, token_ids).float()
probs = self.torch.softmax(raw, dim=-1).cpu().tolist()
results[field.key] = decode_numeric(probs, field)
return {"fields": results, "accounting": {"backbone_calls": 1,
"physical_rows": 1, "field_slots": len(fields), "truncated": False},
"semantics": "per-field marginals; not a learned full joint distribution"}
def atomic(self, context: str, field_key: str, choices: Sequence[Any]) -> dict[str, Any]:
if not 2 <= len(choices) <= 255 or len({json.dumps(v,sort_keys=True) for v in choices}) != len(choices):
raise ValueError("atomic choices must contain 2..255 unique JSON values")
schema={"field_order":[field_key],"fields":{field_key:{"type":"enum","choices":list(choices)}}}
support="\n".join(f'{self.codebook[i]["code"]} = '+json.dumps(v,ensure_ascii=False,sort_keys=True,separators=(",",":")) for i,v in enumerate(choices))
user=("Classify the active field from the complete input.\n\nContext:\n"+context+"\n\n"
"Registered schema gpc1-public-atomic@1:\n"+json.dumps(schema,ensure_ascii=False,sort_keys=True,separators=(",",":"))
+"\n\nAllowed complete outputs in registered semantic order:\n"+support+"\n\nReturn exactly one allowed output.")
prompt=self.processor.tokenizer.apply_chat_template([
{"role":"system","content":"You are a fixed-schema lookup model. Return exactly one listed opaque code and no other text."},
{"role":"user","content":user}],tokenize=False,add_generation_prompt=True,enable_thinking=False)
ids=tuple(self.processor.tokenizer.encode(prompt,add_special_tokens=False))
if len(ids)>self.max_input_tokens: raise ValueError("atomic request exceeds token admission; no truncation performed")
for row in self.codebook[:len(choices)]:
combined=tuple(self.processor.tokenizer.encode(prompt+str(row["code"]),add_special_tokens=False))
if combined != ids + (int(row["token_id"]),):
raise ValueError("atomic response-boundary tokenization changed")
tensor=self.torch.tensor([ids]); logits=self._forward({"input_ids":tensor,"attention_mask":self.torch.ones_like(tensor)},[len(ids)-1])[0]
indices=self.torch.tensor([int(r["token_id"]) for r in self.codebook[:len(choices)]],device=logits.device)
probs=self.torch.softmax(logits.index_select(0,indices).float(),dim=-1).cpu().tolist(); selected=max(range(len(choices)),key=probs.__getitem__)
return {"value":choices[selected],"probabilities":probs,"selected_index":selected,"calibrated":False,
"accounting":{"backbone_calls":1,"physical_rows":1,"truncated":False},
"semantics":"conditional distribution over the complete caller-declared choice set"}
def finite_joint(self, context: str, schema: dict[str, Any], candidates: Sequence[dict[str, Any]]) -> dict[str, Any]:
if not self.admit_joint:
raise ValueError("finite_joint requires an accepted same-hardware parity gate")
tokenizer = self.processor.tokenizer
canonical = [json.dumps(v, ensure_ascii=False, sort_keys=True,
separators=(",", ":"), allow_nan=False) for v in candidates]
support = "\n".join(f"{value} = {value}" for value in canonical)
user = ("Select one complete allowed JSON record from the complete input.\n\nContext:\n" + context
+ "\n\nRegistered schema gpc1-public-finite-joint@1:\n"
+ json.dumps(schema, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)
+ "\n\nAllowed complete outputs in registered semantic order:\n" + support
+ "\n\nReturn exactly one allowed output.")
prompt = tokenizer.apply_chat_template([
{"role":"system","content":"You are a fixed-schema lookup model. Return exactly one listed complete JSON record and no other text."},
{"role":"user","content":user}], tokenize=False, add_generation_prompt=True,
enable_thinking=False)
prefix = tuple(tokenizer.encode(prompt, add_special_tokens=False))
terminal_text = tokenizer.eos_token or "<|im_end|>"
terminal = tuple(tokenizer.encode(terminal_text, add_special_tokens=False))
suffixes = []
for value in canonical:
complete = tuple(tokenizer.encode(prompt + value + terminal_text, add_special_tokens=False))
if complete[:len(prefix)] != prefix: raise ValueError("candidate tokenization changed prompt prefix")
suffixes.append(complete[len(prefix):])
compiled = compile_joint(prefix, tuple(str(i) for i in range(len(candidates))), suffixes,
terminal, pad_token_id=int(tokenizer.pad_token_id), max_rows=self.max_joint_rows,
max_padded_tokens=self.max_padded_tokens, max_logit_vectors=self.max_logit_vectors,
max_input_tokens=self.max_input_tokens)
vocab = int(self.model.config.text_config.vocab_size)
if compiled.logit_vectors * vocab * 4 > self.max_logits_bytes:
raise ValueError("joint sparse logits exceed byte admission")
device = next(self.model.parameters()).device
result = score_joint(self.model, compiled, device=device)
result["value"] = candidates[result["selected_index"]]
result["semantics"] = "conditional distribution over the complete caller-declared joint support"
return result
|