| """Offline CPU inference with independently pinned model catalog and artifacts. |
| |
| No downloads, training, benchmark scoring or workspace-dependent imports occur. |
| The catalog digest must be obtained from a trusted release, not from a changed |
| local catalog. Checkpoints are the original complete weights-only PyTorch files. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| from pathlib import Path, PurePosixPath |
| import re |
|
|
| import numpy as np |
| import torch |
| from tokenizers import Tokenizer |
|
|
| import portable_core as core |
|
|
|
|
| def require(ok, message): |
| if not ok: |
| raise ValueError(message) |
|
|
|
|
| def canonical_sha(value): |
| return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()).hexdigest() |
|
|
|
|
| def file_sha(path): |
| with Path(path).open("rb") as stream: |
| return hashlib.file_digest(stream, "sha256").hexdigest() |
|
|
|
|
| def safe_path(root, relative): |
| require(isinstance(relative, str) and relative and "\\" not in relative and "\0" not in relative, |
| "Invalid artifact path") |
| item = PurePosixPath(relative) |
| require(not item.is_absolute() and str(item) == relative and not any(x in ("", ".", "..") for x in item.parts), |
| "Artifact path must be canonical and relative") |
| root = Path(root).resolve() |
| target = (root / relative).resolve() |
| require(target.is_relative_to(root), "Artifact escapes artifact root") |
| return target |
|
|
|
|
| def verify_file(path, descriptor): |
| require(Path(path).is_file() and Path(path).stat().st_size == descriptor["bytes"], "Artifact size differs") |
| require(file_sha(path) == descriptor["sha256"], "Artifact SHA-256 differs") |
|
|
|
|
| def read_catalog(path, expected_sha256): |
| require(isinstance(expected_sha256, str) and re.fullmatch(r"[0-9a-f]{64}", expected_sha256), |
| "Trusted catalog SHA-256 is required") |
| raw = Path(path).read_bytes() |
| require(hashlib.sha256(raw).hexdigest() == expected_sha256, "Catalog SHA-256 differs") |
| catalog = json.loads(raw) |
| require(catalog.get("schema_version") == 1 and catalog.get("release_id") == "bio2nl-2026-10-02-models-v1", |
| "Unknown model catalog") |
| entries = catalog["models"] |
| expected = {f"{c}-pt{p}" for c in ("EP", "ES", "EE") for p in range(3)} |
| expected |= {f"{c}-pt{p}-ft{f}" for c in ("EP", "ES", "EE") for p in range(3) for f in range(3)} |
| require(len(entries) == 36 and len({x["id"] for x in entries}) == 36 and {x["id"] for x in entries} == expected, |
| "Catalog model grid differs") |
| for entry in entries: |
| require(entry["config_sha256"] == canonical_sha(catalog["config"]), "Config identity differs") |
| require(entry["tokenizer_sha256"] == catalog["tokenizer"]["sha256"], "Tokenizer identity differs") |
| job = entry["job"] |
| classifier = entry["kind"] == "source_classifier" |
| require(entry["kind"] in ("causal_lm", "source_classifier") and job.get("smoke") is False, |
| "Unexpected checkpoint kind/job") |
| suffix = f"-ft{job['ft_seed']}" if classifier else "" |
| require(entry["id"] == f"{job['condition']}-pt{job['pt_seed']}" + suffix, "Model/job identity differs") |
| if classifier: |
| parents = [p for p in entries if p["id"] == entry["pretraining_parent"]["id"]] |
| require(len(parents) == 1, "Missing classifier parent") |
| parent = parents[0] |
| require(parent["kind"] == "causal_lm" and parent["job"]["condition"] == job["condition"] and |
| parent["job"]["pt_seed"] == job["pt_seed"] and |
| entry["pretraining_parent"]["sha256"] == parent["checkpoint"]["sha256"] and |
| entry["pretraining_parent"]["state_sha256"] == parent["state_sha256"], |
| "Classifier pretraining-parent provenance differs") |
| require(type(entry["selected_epoch"]) is int and 1 <= entry["selected_epoch"] <= 5, |
| "Invalid source-selected epoch") |
| return catalog |
|
|
|
|
| def select(catalog, model_id): |
| rows = [entry for entry in catalog["models"] if entry["id"] == model_id] |
| require(len(rows) == 1, "Unknown model identifier") |
| return rows[0] |
|
|
|
|
| def load_tokenizer(catalog, artifact_root): |
| descriptor = catalog["tokenizer"] |
| path = safe_path(artifact_root, descriptor["path"]) |
| verify_file(path, descriptor) |
| tokenizer = Tokenizer.from_file(str(path)) |
| require(tokenizer.get_vocab_size() == 32000 and |
| [tokenizer.token_to_id(s) for s in descriptor["special_tokens"]] == [0, 1, 2, 3], |
| "Tokenizer vocabulary or special-token identities differ") |
| require(tokenizer.padding is None and tokenizer.truncation is None, "Unexpected tokenizer padding/truncation") |
| return tokenizer |
|
|
|
|
| def state_parts(state, kind): |
| prefix = "transformer." if kind == "causal_lm" else "backbone." |
| backbone = {key[len(prefix):]: value for key, value in state.items() if key.startswith(prefix)} |
| head_name = "lm_head.weight" if kind == "causal_lm" else "score.weight" |
| require(backbone and head_name in state, "Incomplete backbone/head state") |
| return core.state_digest(backbone), core.state_digest({head_name: state[head_name]}) |
|
|
|
|
| def config_for_core(public_config): |
| """Restore integer-keyed config maps without mutating the public catalog.""" |
| config = json.loads(json.dumps(public_config)) |
| for field in ("id2label", "pruned_heads"): |
| config[field] = {int(key): value for key, value in config[field].items()} |
| return config |
|
|
|
|
| def original_config_sha_after_load(config): |
| """Transformers 4.45.2 from_dict adds only this runtime key in-place.""" |
| original = dict(config) |
| require(original.pop("attn_implementation", None) is None, "Unexpected injected attention implementation") |
| return canonical_sha(original) |
|
|
|
|
| def load_model(catalog, model_id, artifact_root, checkpoint_override=None): |
| entry = select(catalog, model_id) |
| path = Path(checkpoint_override) if checkpoint_override is not None else safe_path(artifact_root, entry["checkpoint"]["path"]) |
| require(path.stat().st_size == entry["checkpoint"]["bytes"], "Checkpoint size differs") |
| model, payload = core.load_checkpoint(path, |
| expected_sha256=entry["checkpoint"]["sha256"], expected_kind=entry["kind"], expected_job=entry["job"], |
| expected_state_sha256=entry["state_sha256"], expected_protocol_sha256=catalog["training_protocol_sha256"], |
| expected_manifest_sha256=catalog["prepared_manifest_sha256"], expected_epoch=entry.get("selected_epoch"), |
| expected_config=config_for_core(catalog["config"])) |
| require(original_config_sha_after_load(payload["config"]) == entry["config_sha256"], "Full checkpoint config differs") |
| backbone_sha, head_sha = state_parts(payload["state_dict"], entry["kind"]) |
| require(backbone_sha == entry["backbone_state_sha256"] and head_sha == entry["head_state_sha256"], |
| "Backbone/head identity differs") |
| require(len(payload["state_dict"]) == entry["state_tensor_count"], "Tensor inventory differs") |
| if entry["kind"] == "source_classifier": |
| require(model.score.bias is None and tuple(model.score.weight.shape) == (2, 768), "Classifier head shape differs") |
| del payload |
| return model |
|
|
|
|
| def encode_pairs(tokenizer, pairs): |
| require(isinstance(pairs, list) and pairs, "Supply at least one sentence pair") |
| ids = np.zeros((len(pairs), 512), dtype=np.int64) |
| masks = np.zeros_like(ids) |
| details = [] |
| for index, pair in enumerate(pairs): |
| require(isinstance(pair, dict) and set(pair) == {"sentence1", "sentence2"}, |
| "Each pair requires only sentence1 and sentence2; labels are not accepted") |
| original = [] |
| for field in ("sentence1", "sentence2"): |
| text = pair[field] |
| require(isinstance(text, str) and bool(text), "Empty/nontext endpoint") |
| content = tokenizer.encode(text, add_special_tokens=False).ids |
| require(content and all(4 <= token < 32000 for token in content), "Unknown/special content token") |
| require(tokenizer.decode(content, skip_special_tokens=False) == text, "Text roundtrip failed") |
| original.append(content) |
| joined = original[0][:255] + [2] + original[1][:255] + [1] |
| ids[index, :len(joined)] = joined |
| masks[index, :len(joined)] = 1 |
| details.append({"original_content_tokens": [len(x) for x in original], |
| "truncated_endpoints": [len(x) > 255 for x in original], "encoded_length": len(joined)}) |
| return {"input_ids": ids, "attention_mask": masks, "labels": np.zeros(len(pairs), dtype=np.int64)}, details |
|
|
|
|
| def predict_pairs(model, tokenizer, pairs, batch_size=1): |
| arrays, details = encode_pairs(tokenizer, pairs) |
| predicted = core.predict_source(model, arrays, batch_size=batch_size, device="cpu", precision="float32") |
| values = [] |
| for logp, pred, detail in zip(predicted["log_probabilities"], predicted["predictions"], details): |
| values.append({**detail, "log_probabilities": logp.tolist(), "probability_label_1": float(np.exp(logp[1])), |
| "logit_margin_label_1_minus_0": float(logp[1] - logp[0]), "predicted_label": int(pred)}) |
| return values |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("command", choices=("verify", "predict")) |
| parser.add_argument("--catalog", type=Path, required=True) |
| parser.add_argument("--catalog-sha256", required=True) |
| parser.add_argument("--artifact-root", type=Path, required=True) |
| parser.add_argument("--model-id", required=True, help="For example EP-pt0-ft0 or EP-pt0") |
| parser.add_argument("--checkpoint", type=Path, help="Explicit relocated checkpoint; all hashes still apply") |
| parser.add_argument("--pairs", type=Path, help="JSON array with sentence1/sentence2, without labels") |
| parser.add_argument("--batch-size", type=int, default=1) |
| parser.add_argument("--cpu-threads", type=int, default=4) |
| args = parser.parse_args() |
| require(args.cpu_threads > 0, "Invalid CPU thread count") |
| torch.set_num_threads(args.cpu_threads) |
| torch.set_float32_matmul_precision("highest") |
| catalog = read_catalog(args.catalog, args.catalog_sha256) |
| tokenizer = load_tokenizer(catalog, args.artifact_root) |
| model = load_model(catalog, args.model_id, args.artifact_root, args.checkpoint) |
| result = {"model_id": args.model_id, "checkpoint_sha256": select(catalog, args.model_id)["checkpoint"]["sha256"], |
| "catalog_sha256": args.catalog_sha256, "device": "cpu", "precision": "float32", "verified": True} |
| if args.command == "predict": |
| require(select(catalog, args.model_id)["kind"] == "source_classifier", "Pair inference requires a complete classifier") |
| require(args.pairs is not None, "--pairs is required for prediction") |
| result.update(fixed_direction="label_1", decision_rule="argmax_ties_label_0", |
| predictions=predict_pairs(model, tokenizer, json.loads(args.pairs.read_text()), args.batch_size)) |
| print(json.dumps(result, indent=2, allow_nan=False)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
