SPB Git

spb/localvm-research Public License

Running LLMs larger than memory on a consumer Mac — falsification-driven research: margin-gated deferred refinement, out-of-core verification on Apple Silicon. TR-01 published.

Python 63.2% JavaScript 23.5% CSS 11.8% Shell 0.9% Makefile 0.5%

candidate_01 scale: StreamingVerifier (per-layer materialize->compute->re-lazify) + 32B benchmark

q4-32B resident (17.5 GB) + q8-32B (34.8 GB) layer-streamed from SSD per
sweep; peak residency = base + a few layers; SSD sees one sequential
checkpoint pass per sweep. Judge: independent 8B bf16.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Simon-Pierre Boucher committed 4 h ago (Aug 12, 2026) parent 12c4857

Showing 3 changed files with +267 and −2

added experiments/candidate_01/benchmark_scale.py +174 −0
@@ -0,0 +1,174 @@
1 +#!/usr/bin/env python3
2 +# =============================================================================
3 +# Project : localvm-research
4 +# File : experiments/candidate_01/benchmark_scale.py
5 +# Purpose : Scale run — 32B model whose q8 does NOT fit beside the resident
6 +# base: q4 resident + layer-streamed q8 verification sweeps
7 +# Author : Simon-Pierre Boucher
8 +# Contact : contact@spboucher.ai
9 +# Created : 2026-08-12
10 +# Modified : 2026-08-12
11 +# Platform : macOS / Apple Silicon (arm64) — MLX / Metal
12 +# License : All rights reserved (research code)
13 +# =============================================================================
14 +"""Candidate-01 scale benchmark (the regime the architecture exists for).
15 +
16 +Qwen3-32B on a 48 GB Mac: q4 (17.5 GB) resident; q8 (34.8 GB) cannot be
17 +co-resident — sweeps stream it layer-by-layer from SSD (StreamingVerifier).
18 +Baseline: pure q4 (the only real alternative on this machine). Quality judged
19 +by Qwen3-8B-bf16 (independent judge; the 32B bf16 obviously cannot run).
20 +
21 +Usage:
22 + .venv/bin/python benchmark_scale.py [--per-domain 2] [--max-tokens 96]
23 +"""
24 +
25 +from __future__ import annotations
26 +
27 +import argparse
28 +import gc
29 +import json
30 +import sys
31 +import time
32 +from datetime import datetime, timezone
33 +from pathlib import Path
34 +
35 +import mlx.core as mx
36 +from huggingface_hub import snapshot_download
37 +from mlx_lm import load
38 +
39 +REPO_ROOT = Path(__file__).resolve().parents[2]
40 +sys.path.insert(0, str(REPO_ROOT / "benchmarks"))
41 +sys.path.insert(0, str(Path(__file__).parent / "implementation"))
42 +from hardware_manifest import collect_manifest # noqa: E402
43 +from runtime import generate_deferred # noqa: E402
44 +from streaming_verifier import StreamingVerifier # noqa: E402
45 +
46 +Q4_REPO = "mlx-community/Qwen3-32B-4bit"
47 +Q8_REPO = "mlx-community/Qwen3-32B-8bit"
48 +JUDGE_REPO = "mlx-community/Qwen3-8B-bf16"
49 +
50 +
51 +def greedy_baseline(model, tokenizer, prompt_ids, max_tokens):
52 + from mlx_lm.models.cache import make_prompt_cache
53 +
54 + cache = make_prompt_cache(model)
55 + tokens = []
56 + inp = mx.array(list(prompt_ids))[None]
57 + t0 = time.perf_counter()
58 + for _ in range(max_tokens):
59 + nxt = int(mx.argmax(model(inp, cache=cache)[0, -1]).item())
60 + if nxt == tokenizer.eos_token_id:
61 + break
62 + tokens.append(nxt)
63 + inp = mx.array([[nxt]])
64 + return tokens, time.perf_counter() - t0
65 +
66 +
67 +def main() -> None:
68 + ap = argparse.ArgumentParser()
69 + ap.add_argument("--per-domain", type=int, default=2)
70 + ap.add_argument("--max-tokens", type=int, default=96)
71 + ap.add_argument("--window", type=int, default=32)
72 + ap.add_argument("--taus", default="2.0")
73 + ap.add_argument("--modes", default="margin,verify-all")
74 + args = ap.parse_args()
75 +
76 + q4_path = snapshot_download(Q4_REPO)
77 + q8_path = snapshot_download(Q8_REPO)
78 + domains = json.loads((REPO_ROOT / "benchmarks/datasets/eval_prompts.json").read_text())["domains"]
79 +
80 + print("loading q4 resident …", flush=True)
81 + base_model, tokenizer = load(q4_path)
82 + verifier = StreamingVerifier(q8_path)
83 + q8_bytes = verifier.weight_bytes
84 + print(f"q8 checkpoint (streamed): {q8_bytes/1e9:.1f} GB", flush=True)
85 +
86 + prompts = []
87 + for domain, plist in domains.items():
88 + for prompt in plist[: args.per_domain]:
89 + ids = tokenizer.apply_chat_template(
90 + [{"role": "user", "content": prompt}], add_generation_prompt=True)
91 + prompts.append({"domain": domain, "ids": list(ids)})
92 +
93 + print("baseline: pure q4 …", flush=True)
94 + q4_out, q4_times = [], []
95 + for k, p in enumerate(prompts):
96 + toks, dt = greedy_baseline(base_model, tokenizer, p["ids"], args.max_tokens)
97 + q4_out.append(toks); q4_times.append((len(toks), dt))
98 + print(f" {k+1}/{len(prompts)} ({len(toks)} tok, {len(toks)/dt:.1f} tok/s)", flush=True)
99 +
100 + outputs = {"pure_q4": q4_out}
101 + runs = []
102 + for mode in args.modes.split(","):
103 + for tau in ([float(x) for x in args.taus.split(",")] if mode == "margin" else [2.0]):
104 + print(f"runtime: mode={mode} tau={tau} W={args.window} …", flush=True)
105 + outs, agg = [], {"tokens": 0, "deferred": 0, "sweeps": 0, "rollbacks": 0,
106 + "sweep_s": 0.0, "gen_s": 0.0, "logical_bytes": 0, "io_s": []}
107 + for k, p in enumerate(prompts):
108 + toks, st = generate_deferred(
109 + base_model, verifier, tokenizer, p["ids"],
110 + args.max_tokens, tau, args.window, mode, q8_bytes)
111 + outs.append(toks)
112 + agg["tokens"] += st.tokens_out; agg["deferred"] += st.deferred
113 + agg["sweeps"] += st.sweeps; agg["rollbacks"] += st.rollbacks
114 + agg["sweep_s"] += st.sweep_time_s; agg["gen_s"] += st.gen_time_s
115 + agg["logical_bytes"] += st.sweep_logical_bytes
116 + print(f" {k+1}/{len(prompts)} ({st.tokens_out} tok, {st.sweeps} sweeps, "
117 + f"{st.rollbacks} rollbacks, last sweep io {verifier.last_sweep_io_s:.1f}s)",
118 + flush=True)
119 + n = max(agg["tokens"], 1)
120 + runs.append({
121 + "mode": mode, "tau": tau, "window": args.window,
122 + "tokens_per_s": n / (agg["gen_s"] + agg["sweep_s"]),
123 + "deferral_rate": agg["deferred"] / n,
124 + "rollback_rate": agg["rollbacks"] / n,
125 + "sweep_latency_s_mean": agg["sweep_s"] / max(agg["sweeps"], 1),
126 + "logical_verify_bytes_per_token": agg["logical_bytes"] / n,
127 + "raw": {k: v for k, v in agg.items() if k != "io_s"},
128 + })
129 + outputs[f"{mode}_tau{tau}"] = outs
130 + r = runs[-1]
131 + print(f" tok/s={r['tokens_per_s']:.2f} sweepLat={r['sweep_latency_s_mean']:.1f}s "
132 + f"GB/token(logical)={r['logical_verify_bytes_per_token']/1e9:.2f}", flush=True)
133 +
134 + print("freeing 32B models; loading 8B bf16 judge …", flush=True)
135 + del base_model, verifier
136 + gc.collect(); mx.clear_cache()
137 + judge, _ = load(JUDGE_REPO)
138 + quality = {}
139 + for name, outs in outputs.items():
140 + vals = []
141 + for p, toks in zip(prompts, outs):
142 + if len(toks) < 2:
143 + continue
144 + full = p["ids"] + list(toks)
145 + logits = judge(mx.array(full)[None])[0]
146 + sel = logits[len(p["ids"]) - 1 : len(full) - 1].astype(mx.float32)
147 + lp = sel - mx.logsumexp(sel, axis=-1, keepdims=True)
148 + tok_lp = mx.take_along_axis(lp, mx.array(toks)[:, None], axis=-1)
149 + mx.eval(tok_lp)
150 + vals.append(float(mx.mean(tok_lp).item()))
151 + quality[name] = {"mean_logprob_8b_judge": sum(vals) / len(vals), "n": len(vals)}
152 + print(f" {name:>18}: {quality[name]['mean_logprob_8b_judge']:.4f}", flush=True)
153 +
154 + ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
155 + out_dir = REPO_ROOT / "results" / "candidate_01_scale32b" / ts
156 + out_dir.mkdir(parents=True)
157 + (out_dir / "results.json").write_text(json.dumps({
158 + "experiment": "candidate_01_scale32b",
159 + "author": "Simon-Pierre Boucher",
160 + "contact": "contact@spboucher.ai",
161 + "manifest": collect_manifest(),
162 + "config": vars(args),
163 + "models": {"base": Q4_REPO, "verify": Q8_REPO, "judge": JUDGE_REPO},
164 + "q8_streamed_bytes": q8_bytes,
165 + "baseline_pure_q4_tokens_per_s":
166 + sum(t for t, _ in q4_times) / max(sum(d for _, d in q4_times), 1e-9),
167 + "runs": runs,
168 + "quality_8b_judge": quality,
169 + }, indent=2))
170 + print(f"\nwrote {out_dir / 'results.json'}")
171 +
172 +
173 +if __name__ == "__main__":
174 + main()
modified experiments/candidate_01/implementation/runtime.py +8 −2
@@ -65,7 +65,10 @@ def generate_deferred(base_model, verify_model, tokenizer, prompt_ids: list[int]
65 65 # Incremental verify cache: sweeps feed only the unseen suffix, cutting
66 66 # sweep compute from O(context) to O(window). v_pos = tokens of the
67 67 # current sequence already ingested; invariant at sweep: v_pos <= base-1.
68 − vcache = make_prompt_cache(verify_model)
68 + # verify_model may be a StreamingVerifier (layer-streamed weights for
69 + # models that do not fit beside the resident base).
70 + is_streaming = hasattr(verify_model, "forward_chunk")
71 + vcache = make_prompt_cache(verify_model.model if is_streaming else verify_model)
69 72 v_pos = 0
70 73 t0 = time.perf_counter()
71 74 inp = mx.array(prompt_ids)[None]
@@ -85,7 +88,10 @@ def generate_deferred(base_model, verify_model, tokenizer, prompt_ids: list[int]
85 88 # feed full[v_pos : end-1]; logits rows are absolute v_pos..end-2,
86 89 # predictions for tokens v_pos+1..end-1 ⊇ the pending window
87 90 chunk = full[v_pos : len(full) - 1]
88 − logits = verify_model(mx.array(chunk)[None], cache=vcache)[0]
91 + if is_streaming:
92 + logits = verify_model.forward_chunk(chunk, vcache)
93 + else:
94 + logits = verify_model(mx.array(chunk)[None], cache=vcache)[0]
89 95 sel = logits[base_of_window - 1 - v_pos :].astype(mx.float32)
90 96 v_pos = len(full) - 1
91 97 q8_argmax = mx.argmax(sel, axis=-1)
added experiments/candidate_01/implementation/streaming_verifier.py +85 −0
@@ -0,0 +1,85 @@
1 +# =============================================================================
2 +# Project : localvm-research
3 +# File : experiments/candidate_01/implementation/streaming_verifier.py
4 +# Purpose : Layer-streamed q8 verification for models larger than free RAM —
5 +# per-layer materialize → compute → re-lazify on unified memory
6 +# Author : Simon-Pierre Boucher
7 +# Contact : contact@spboucher.ai
8 +# Created : 2026-08-12
9 +# Modified : 2026-08-12
10 +# Platform : macOS / Apple Silicon (arm64) — MLX / Metal
11 +# License : All rights reserved (research code)
12 +# =============================================================================
13 +"""StreamingVerifier — runs a verification forward pass through a quantized
14 +model whose weights do NOT fit in free memory alongside the resident base.
15 +
16 +Mechanism: the model is built with lazy (mmap-backed) weights. During a
17 +forward pass we walk the layers manually; each layer's weights materialize
18 +on first use, and immediately after the layer's output is evaluated we
19 +re-assign that layer's parameters to FRESH lazy arrays (a new mx.load view),
20 +dropping the concrete buffers. Peak residency ≈ resident base + a few
21 +layers, while the SSD sees one sequential pass over the checkpoint per
22 +sweep — exactly the expH-friendly access pattern.
23 +"""
24 +
25 +from __future__ import annotations
26 +
27 +import glob
28 +import time
29 +from pathlib import Path
30 +
31 +import mlx.core as mx
32 +from mlx_lm import load as mlx_load
33 +from mlx_lm.models.base import create_attention_mask
34 +
35 +
36 +class StreamingVerifier:
37 + def __init__(self, model_path: str):
38 + self.path = Path(model_path)
39 + # lazy=True: parameters are mmap-backed lazy arrays, nothing evaluated
40 + self.model, self.tokenizer = mlx_load(str(model_path), lazy=True)
41 + self.shards = sorted(glob.glob(str(self.path / "*.safetensors")))
42 + self.weight_bytes = sum(Path(s).stat().st_size for s in self.shards)
43 + self.last_sweep_io_s = 0.0
44 +
45 + def _fresh_lazy_weights(self) -> dict:
46 + w = {}
47 + for s in self.shards:
48 + w.update(mx.load(s)) # lazy by default: no eval performed
49 + return w
50 +
51 + def _relazify(self, weights: dict, prefix: str) -> None:
52 + subset = [(k, v) for k, v in weights.items() if k.startswith(prefix)]
53 + if subset:
54 + self.model.load_weights(subset, strict=False)
55 +
56 + def forward_chunk(self, chunk_ids: list[int], cache) -> mx.array:
57 + """Teacher-force `chunk_ids` through the model with per-layer weight
58 + streaming. `cache` is a make_prompt_cache(self.model) list; it is
59 + advanced by len(chunk_ids). Returns logits (T, vocab)."""
60 + t0 = time.perf_counter()
61 + fresh = self._fresh_lazy_weights()
62 + inner = self.model.model
63 + x = mx.array(chunk_ids)[None]
64 + h = inner.embed_tokens(x)
65 + mx.eval(h)
66 + self._relazify(fresh, "model.embed_tokens")
67 + mask = create_attention_mask(h, cache)
68 + for i, layer in enumerate(inner.layers):
69 + h = layer(h, mask, cache=cache[i] if cache else None)
70 + mx.eval(h)
71 + self._relazify(fresh, f"model.layers.{i}.")
72 + if (i + 1) % 8 == 0:
73 + mx.clear_cache() # release Metal allocator pools
74 + h = inner.norm(h)
75 + if hasattr(self.model, "lm_head"):
76 + logits = self.model.lm_head(h)
77 + else: # tied embeddings
78 + logits = inner.embed_tokens.as_linear(h)
79 + logits = logits[0]
80 + mx.eval(logits)
81 + self._relazify(fresh, "model.norm")
82 + self._relazify(fresh, "lm_head")
83 + mx.clear_cache()
84 + self.last_sweep_io_s = time.perf_counter() - t0
85 + return logits
86