From 573cbedd9877dafadd513ff1b3420caeeeeaa653 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Mon, 9 Nov 2026 04:40:33 +0800 Subject: [PATCH] Implement bounded streaming CPU functional reuse pilot --- README.md | 4 +- docs/CPU_FIRST_VALIDATION.md | 18 ++- docs/SOP.md | 7 + functional/README.md | 68 ++++++++++ functional/__init__.py | 1 + functional/model.py | 242 ++++++++++++++++++++++++++++++++++ functional/preflight.py | 27 ++++ functional/reference.json | 5 + functional/reference_check.py | 142 ++++++++++++++++++++ functional/report.py | 49 +++++++ functional/run.py | 227 +++++++++++++++++++++++++++++++ functional/samples.json | 13 ++ functional/supervise.py | 73 ++++++++++ requirements-functional.txt | 5 + tests/test_functional.py | 62 +++++++++ 15 files changed, 940 insertions(+), 3 deletions(-) create mode 100644 functional/README.md create mode 100644 functional/__init__.py create mode 100644 functional/model.py create mode 100644 functional/preflight.py create mode 100644 functional/reference.json create mode 100644 functional/reference_check.py create mode 100644 functional/report.py create mode 100644 functional/run.py create mode 100644 functional/samples.json create mode 100644 functional/supervise.py create mode 100644 requirements-functional.txt create mode 100644 tests/test_functional.py diff --git a/README.md b/README.md index 5ef778a..0cc43f4 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,9 @@ Bounded, inference-free analysis of official unquantized model weights. The first experiment is **Qwen3.5-2B**. Model weights are downloaded and analyzed on the authorized remote host, -then deleted after archived results pass validation. No model weights are committed here. +then deleted after archived results pass validation for the completed static pilot. +The current CPU functional checkpoint is retained until the user explicitly requests deletion. +No model weights are committed here. - [CPU-first feasibility and validation](docs/CPU_FIRST_VALIDATION.md) - [Training-free functional reuse plan (proposed; not executed)](docs/TRAINING_FREE_FUNCTIONAL_PLAN.md) diff --git a/docs/CPU_FIRST_VALIDATION.md b/docs/CPU_FIRST_VALIDATION.md index 9623d02..0bbfe2d 100644 --- a/docs/CPU_FIRST_VALIDATION.md +++ b/docs/CPU_FIRST_VALIDATION.md @@ -28,7 +28,7 @@ NumPy 2.5.4，OPENBLAS_NUM_THREADS=1、OMP_NUM_THREADS=1，远程 timeout 60 秒 使用受信任的算子实现及小数组正确性检查；模型仓库代码不直接执行。 先确认首个短样本能以受控内存运行。不用不正确的“简化 Qwen”产生看似有价值的结果。 2. **真实分布上的候选**：batch=1、32 tokens，校准和留出各 4 段真实文本。 - 按顺序加载、计算、释放每层，保留很小的入口激活；探测选定的 3 个 full-attention 层对和 2 个 MLP 对。 + 按顺序加载、计算、释放每层，保留很小的入口激活；探测预先固定的 3 个 full-attention 层对、2 个 MLP 对，以及 6 个 full-attention 层内的 Q/K 归一化模块对，共 11 对。 每对在两个来源层的隐藏输入上分别测试，记录 JS、top-k、完整分支输出误差。 3. **最小干预证据**：最多对 1 个候选做单处替换；冻结参数，在留出文本上看输出变化， 条件允许时流式计算 token NLL。词表输出头分块归约 logsumexp，避免一次物化大 logits。 @@ -48,8 +48,22 @@ NumPy 2.5.4，OPENBLAS_NUM_THREADS=1、OMP_NUM_THREADS=1，远程 timeout 60 秒 - 先实现最小正确路径；线性状态递推、长上下文和生成速度不塞入首轮。 - 总计算 120 分钟，阶段预算见功能方案；无候选、超时或数值/实现核对失败均停止。 - 容量探测已完成，真实模型阶段尚未执行。后续需要恢复本实验自有的固定 revision 权重， - 分析后按既有规则校验归档并删除；不增加模型规模或自动租用 GPU。 + 权重持续保留，只有用户明确要求删除时才清理；不增加模型规模或自动租用 GPU。 CPU 上“找到输出接近的候选”能支持继续投入 GPU； “在指定小样本中单处复用保持质量且节省明确资源”才是初步正例。 两者都不能替代大规模能力验证。CPU 时延不外推为 GPU 加速率。 + +## 当前权重保留规则 + +用户于 2026-11-09 明确要求：本轮模型权重保留，只有用户明确要求删除时才删除。 +试验完成、超时、失败、归档或结果校验均不得自动触发清理。该要求覆盖先前逐模型分析后删除的约定。 +上一轮已删除的历史记录保留；本轮恢复下载的权重受新规则保护。 + +## 本轮执行补充 + +实际实验分支为 `codex/cpu-reuse-experiment`。在读取真实权重/功能指标前，新增固定的 +6 个层内 Q/K RMSNorm 增益共享候选，以检查较小粒度的功能复用；不是看过留出结果后扩充搜索。 +最多选择一个候选，必须分别验证两个替换方向；完整模块失败不会被小模块通过所掩盖。 +归一化共享最多节省少量字节，不预设能显著减少计算。 +采集真实隐藏输入必须正确执行前置线性注意力；它不进入首轮复用候选，但其递推计算纳入官方参考对照。 diff --git a/docs/SOP.md b/docs/SOP.md index 80689ab..025ef29 100644 --- a/docs/SOP.md +++ b/docs/SOP.md @@ -121,3 +121,10 @@ validation. See the [bounded plan](TRAINING_FREE_FUNCTIONAL_PLAN.md) and [primary-source literature](FUNCTIONAL_REUSE_LITERATURE.md). No forward pass, new download, or new model is started by adopting this document. The original static pilot remains inference-free; the proposed follow-up requires a separately confirmed execution environment. + +## Current functional-run retention override + +On 2026-11-09 the user explicitly changed retention: keep the CPU functional pilot +checkpoint until the user explicitly requests deletion. Completion, timeout, failure +and archive validation do not authorize cleanup. The earlier pilot cleanup remains +a historical action; do not apply its automatic cleanup step to the current run. diff --git a/functional/README.md b/functional/README.md new file mode 100644 index 0000000..31271df --- /dev/null +++ b/functional/README.md @@ -0,0 +1,68 @@ +# Bounded CPU functional pilot + +This is the execution implementation of the CPU-first plan, not the completed static +parameter pilot. Frozen Qwen3.5-2B weights are decoded to FP32 a projection block at a +time. CPU only, no autograd/optimization, no rotation fitting, no model-hub Python code. +The user explicitly requires retaining this checkpoint until they request deletion. + +## Fixed scope + +- Batch of eight distinct experiment-authored plain texts, first 32 tokens each: + four calibration and four heldout, two Chinese/two English in each split. +- Full attention pairs: (3,7), (11,15), (19,23). +- MLP pairs: (5,6), (20,21). +- Smaller-granularity Q/K RMSNorm pairs inside layers 3,7,11,15,19,23. + Added before observing real weights/functions for this pilot; these are small storage + opportunities, not a claim of substantial compute savings. +- At most one calibration-selected candidate, two separate substitution directions. + Selection prefers component size, then worst symmetric output error. No heldout-driven retry. +- Same-input relative L2 <=0.05 in both directions and token P95 <=0.1 in both directions; + attention additionally requires mean JS <=0.01 and top-5 overlap >=0.8. +- Heldout mean NLL increase <=0.01 nat/token and per-sequence P95 increase <=0.05. + These are pilot decision criteria, not mathematical capability guarantees. + +## Source and checks + +`reference.json` pins the official Transformers Qwen3.5 source commit, URL and hash. +`reference_check.py` extracts only named mathematical definitions from that exact file, +removes optional fused-kernel decorators, and compares the NumPy implementation against +Torch operators, full toy decoder layers and a two-layer text forward. Chunked delta and +recurrent delta implementations are both checked, including sequence padding boundaries. +The official source is Apache-2.0, copyright Qwen Team and HuggingFace; the reference file +is fetched for testing and is not bundled here. FP32 equivalence is checked, not equality +to every deployed BF16 kernel. + +`tests/test_functional.py` independently checks streamed projection/embedding/NLL, +causality, causal convolution, normalization conventions and basic metric identities. +These tests use tiny synthetic arrays. Real model evidence is produced only by `run.py`. + +## Remote execution + +Use the previously authorized CPU host and existing NumPy/Matplotlib environment. Install +only binary wheels: tokenizers==0.23.3 from HTTPS PyPI; torch==2.15.0+cpu from the official +CPU wheel index for independent reference checks. Use an experiment-owned on-disk TMPDIR: +the shared host's /tmp tmpfs was too small during the first wheel download. Do not change +system swap or shared temporary files. + +```sh +OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 venv/bin/python -B -m unittest tests.test_functional -v +OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 venv/bin/python -B -m functional.reference_check --source runs/cpu-qwen35-2b/reference.py --output runs/cpu-qwen35-2b/reference-check.json +OPENBLAS_NUM_THREADS=1 OMP_NUM_THREADS=1 venv/bin/python -B -m functional.supervise --root runs/cpu-qwen35-2b +``` + +The supervisor waits at most one hour for binary dependencies and four hours for the +already started pinned, verified checkpoint download. It does not install, retry downloads, +search extra candidates, mutate Git or delete weights. After reference checks, it runs one +8-token real text capacity check, then the bounded functional pilot. Reference, preflight +and functional compute together are capped at two hours with a reporting reserve; +download wait is recorded separately. Projection memory is checked, RSS soft stop is +512 MiB, available-system-memory floor 256 MiB; sustained host swap IO also stops the run. + +`preflight.py` is a capacity smoke check, not a semantic oracle. `run.py` records exact +sample/token IDs, fixed configuration, source hashes, activations, per-text/per-head metrics, +selection, per-token NLL, intervention results and truthful stop coverage. `report.py` reads +saved data to produce PNG/PDF/SVG figures without reopening weights. + +No generation throughput is measured in this short-text pilot. Any parameter-tying storage +benefit must be accounted for separately, including metadata. Neither same-input similarity +nor favorable tiny heldout NLL demonstrates broad mutual interchangeability or GPU speedup. diff --git a/functional/__init__.py b/functional/__init__.py new file mode 100644 index 0000000..faee6dc --- /dev/null +++ b/functional/__init__.py @@ -0,0 +1 @@ +"""CPU-only, training-free functional reuse pilot.""" diff --git a/functional/model.py b/functional/model.py new file mode 100644 index 0000000..dd64f49 --- /dev/null +++ b/functional/model.py @@ -0,0 +1,242 @@ +"""Streaming FP32 text-only Qwen3.5 reference, original BF16 weights. + +Mathematical semantics checked against Transformers Qwen3.5, commit +fd3cf0140de03f7619ac95b5cfed05b711ff9e22 (Apache-2.0, Qwen Team/HuggingFace). +No cached decode, padding, images, generation, fused kernels or autograd. +Full-sequence gated delta recurrence replaces the algebraically equivalent +chunk algorithm; both are independently compared in remote tests. +""" +import json +import time +from pathlib import Path +import numpy as np +from pilot.tensors import load_tensors, decode + +PREFIX = 'model.language_model.' + + +def sigmoid(x): + return np.exp(-np.logaddexp(np.float32(0), -x)) + + +def silu(x): + return x * sigmoid(x) + + +def norm(x, weight, eps=1e-6, centered=True): + return (x / np.sqrt(np.mean(x*x, axis=-1, keepdims=True) + eps) + * (1 + weight if centered else weight)).astype(np.float32) + + +def rope(x, theta=10000000., fraction=.25): + # Text-only: all three MRoPE position axes coincide. + dim = int(x.shape[-1] * fraction) + inv = np.float32(1) / np.float32(theta) ** (np.arange(0, dim, 2, dtype=np.float32) / dim) + freq = np.arange(x.shape[1], dtype=np.float32)[:, None] * inv[None, :] + angle = np.concatenate([freq, freq], axis=-1)[None, :, None, :] + a = x[..., :dim] + rotated = np.concatenate([-a[..., dim//2:], a[..., :dim//2]], axis=-1) + return np.concatenate([a*np.cos(angle) + rotated*np.sin(angle), x[..., dim:]], axis=-1) + + +def delta_rule(q, k, v, g, beta): + q = q / np.sqrt(np.sum(q*q, axis=-1, keepdims=True) + 1e-6) + k = k / np.sqrt(np.sum(k*k, axis=-1, keepdims=True) + 1e-6) + q = q / np.float32(np.sqrt(q.shape[-1])) + state = np.zeros((q.shape[0], q.shape[2], q.shape[3], v.shape[3]), dtype=np.float32) + out = np.empty_like(v) + for t in range(q.shape[1]): + state *= np.exp(g[:, t, :, None, None]) + predicted = np.sum(state * k[:, t, :, :, None], axis=-2) + delta = (v[:, t] - predicted) * beta[:, t, :, None] + state += k[:, t, :, :, None] * delta[:, :, None, :] + out[:, t] = np.sum(state * q[:, t, :, :, None], axis=-2) + return out, state + + +def causal_conv(x, kernel): + out = np.zeros_like(x) + width = kernel.shape[-1] + for j in range(width): + lag = width - 1 - j + if lag < x.shape[1]: + out[:, lag:] += x[:, :x.shape[1]-lag] * kernel[:, j] + return silu(out) + + +def attention(q, k, v): + repeats = q.shape[2] // k.shape[2] + k, v = np.repeat(k, repeats, axis=2), np.repeat(v, repeats, axis=2) + scores = np.einsum('bthd,bshd->bhts', q, k) / np.float32(np.sqrt(q.shape[-1])) + mask = np.triu(np.ones(scores.shape[-2:], dtype=bool), 1) + scores = np.where(mask, -np.inf, scores) + exp = np.exp(scores - scores.max(axis=-1, keepdims=True)) + p = exp / exp.sum(axis=-1, keepdims=True) + return np.einsum('bhts,bshd->bthd', p, v), p + + +class StreamingLM: + def __init__(self, weights, config, seconds=7200, enforce_memory=True): + self.tensors = load_tensors(Path(weights)) + self.config = config.get('text_config', config) + self.deadline = time.monotonic() + seconds + self.enforce_memory = enforce_memory + self.eps = self.config['rms_norm_eps'] + self.norm_replacement = None + self.capture_qk = False + self.qk_inputs = {} + self.swap_observation = None + self.swap_pressure_windows = 0 + + def guard(self): + if time.monotonic() > self.deadline: + raise TimeoutError('CPU computation deadline reached') + if not self.enforce_memory: + return + status = dict(line.split(':', 1) for line in Path('/proc/self/status').read_text().splitlines() if ':' in line) + rss_kib = int(status['VmRSS'].split()[0]) + available = int(next(line.split()[1] for line in Path('/proc/meminfo').read_text().splitlines() if line.startswith('MemAvailable:'))) + now = time.monotonic() + if self.swap_observation is None or now-self.swap_observation[0] >= 5: + counters = dict(line.split() for line in Path('/proc/vmstat').read_text().splitlines()) + count = int(counters['pswpin'])+int(counters['pswpout']) + if self.swap_observation is not None: + elapsed = now-self.swap_observation[0] + rate = (count-self.swap_observation[1])*4096/elapsed + self.swap_pressure_windows = self.swap_pressure_windows+1 if rate > 4*1024**2 else 0 + if self.swap_pressure_windows >= 3: + raise MemoryError('Sustained host swap IO above 4 MiB/s in three observation windows') + self.swap_observation = (now,count) + if rss_kib > 512*1024 or available < 256*1024: + raise MemoryError(f'Soft memory stop: RSS {rss_kib} KiB, available {available} KiB') + + def small(self, name): + t = self.tensors[PREFIX + name] + if t.numel > 1048576: + raise ValueError('Use blockwise linear for large tensors') + x = t.read().reshape(t.shape) + t.evict() + return x + + def linear(self, x, name, rows=256): + t = self.tensors[PREFIX + name + '.weight'] + outdim, indim = t.shape + flat = x.reshape(-1, indim) + result = np.empty((flat.shape[0], outdim), dtype=np.float32) + for start in range(0, outdim, rows): + self.guard() + end = min(start + rows, outdim) + w = t.read(slice(start*indim, end*indim)).reshape(end-start, indim) + result[:, start:end] = flat @ w.T + del w + t.evict() + return result.reshape(*x.shape[:-1], outdim) + + def embedding(self, ids): + t = self.tensors[PREFIX + 'embed_tokens.weight'] + raw = t.raw.reshape(t.shape) + result = decode(raw[ids], t.dtype) + t.evict() + return result + + def normalization(self, x, name): + if getattr(self, 'norm_replacement', None) and name == self.norm_replacement[0]: + name = self.norm_replacement[1] + return norm(x, self.small(name + '.weight'), self.eps) + + def mlp(self, h, layer): + p = f'layers.{layer}.mlp.' + gate = silu(self.linear(h, p+'gate_proj')) + up = self.linear(h, p+'up_proj') + return self.linear(gate*up, p+'down_proj') + + def full_attention(self, h, layer, shared_p=None): + p = f'layers.{layer}.self_attn.' + c = self.config + heads, kvheads, dim = c['num_attention_heads'], c['num_key_value_heads'], c['head_dim'] + # Q and output gate are interleaved per head, not globally split. + mixed = self.linear(h, p+'q_proj').reshape(*h.shape[:-1], heads, 2*dim) + q, gate = mixed[..., :dim], mixed[..., dim:] + k = self.linear(h, p+'k_proj').reshape(*h.shape[:-1], kvheads, dim) + v = self.linear(h, p+'v_proj').reshape(*h.shape[:-1], kvheads, dim) + if getattr(self, 'capture_qk', False): + self.qk_inputs[layer] = (q.copy(), k.copy()) + q = self.normalization(q, p+'q_norm') + k = self.normalization(k, p+'k_norm') + rp = c['rope_parameters'] + q, k = (rope(x, rp['rope_theta'], rp['partial_rotary_factor']) for x in (q,k)) + out, probs = attention(q,k,v) + if shared_p is not None: + out = np.einsum('bhts,bshd->bthd', shared_p, np.repeat(v, heads//kvheads, axis=2)) + out = (out*sigmoid(gate)).reshape(*h.shape[:-1], -1) + return self.linear(out, p+'o_proj'), probs + + def linear_attention(self, h, layer): + p = f'layers.{layer}.linear_attn.' + c = self.config + kh, vh = c['linear_num_key_heads'], c['linear_num_value_heads'] + kd, vd = c['linear_key_head_dim'], c['linear_value_head_dim'] + qkv = self.linear(h, p+'in_proj_qkv') + qkv = causal_conv(qkv, self.small(p+'conv1d.weight')[:,0,:]) + q, k, v = np.split(qkv, [kh*kd, 2*kh*kd], axis=-1) + q = q.reshape(*h.shape[:-1], kh, kd) + k = k.reshape(*h.shape[:-1], kh, kd) + v = v.reshape(*h.shape[:-1], vh, vd) + if vh != kh: + q, k = np.repeat(q, vh//kh, axis=2), np.repeat(k, vh//kh, axis=2) + z = self.linear(h, p+'in_proj_z').reshape(*h.shape[:-1], vh, vd) + beta = sigmoid(self.linear(h, p+'in_proj_b')) + a = self.linear(h, p+'in_proj_a') + g = -np.exp(self.small(p+'A_log')) * np.logaddexp(np.float32(0), a+self.small(p+'dt_bias')) + out, _ = delta_rule(q,k,v,g,beta) + out = norm(out, self.small(p+'norm.weight'), self.eps, centered=False)*silu(z) + return self.linear(out.reshape(*h.shape[:-1], -1), p+'out_proj') + + def forward(self, ids, capture=None, replacement=None, progress=None): + h = self.embedding(ids) + saved = {} + capture = set(capture or []) + for layer, kind in enumerate(self.config['layer_types']): + self.guard() + raw = h + n = self.normalization(h, f'layers.{layer}.input_layernorm') + if ('attn', layer) in capture: + saved[f'attn_{layer}'] = n.copy() + donor = replacement[2] if replacement and replacement[:2] == ('attn',layer) else layer + if kind == 'full_attention': + out, _ = self.full_attention(n,donor) + else: + out = self.linear_attention(n,donor) + h = raw + out + n = self.normalization(h, f'layers.{layer}.post_attention_layernorm') + if ('mlp',layer) in capture: + saved[f'mlp_{layer}'] = n.copy() + donor = replacement[2] if replacement and replacement[:2] == ('mlp',layer) else layer + h = h + self.mlp(n,donor) + if not np.isfinite(h).all(): + raise FloatingPointError(f'Nonfinite hidden states at layer {layer}') + if progress: + progress(layer) + return self.normalization(h,'norm'), saved + + def nll(self, h, ids, rows=512): + # Tied output embedding; online logsumexp avoids full vocabulary logits. + x = h[:,:-1].reshape(-1,h.shape[-1]) + targets = ids[:,1:].reshape(-1) + t = self.tensors[PREFIX+'embed_tokens.weight'] + mx = np.full(x.shape[0],-np.inf,dtype=np.float64) + total = np.zeros(x.shape[0],dtype=np.float64) + correct = np.empty(x.shape[0],dtype=np.float64) + for start in range(0,t.shape[0],rows): + self.guard() + end=min(start+rows,t.shape[0]) + w=t.read(slice(start*t.shape[1],end*t.shape[1])).reshape(end-start,t.shape[1]) + scores=x@w.T + m=np.maximum(mx,scores.max(axis=-1)) + total=total*np.exp(mx-m)+np.exp(scores-m[:,None]).sum(axis=-1) + mx=m + which=(targets>=start)&(targets0,p*np.log(np.maximum(p,1e-30)/np.maximum(m,1e-30)),0) + js=(term(a)+term(b)).sum(axis=-1)*.5 + overlap=[] + for row in range(a.shape[-2]): + k=min(5,row+1) + ai=np.argsort(a[:,:,row,:row+1],axis=-1)[...,-k:] + bi=np.argsort(b[:,:,row,:row+1],axis=-1)[...,-k:] + overlap.append(np.mean((ai[...,None]==bi[...,None,:]).any(axis=-1),axis=-1)) + # Attention-sink control: remove key zero, renormalize valid remaining mass. + aa,bb=a[...,1:],b[...,1:] + mass_a,mass_b=aa.sum(-1),bb.sum(-1) + valid=(mass_a>1e-12)&(mass_b>1e-12) + aa=aa/np.maximum(mass_a[...,None],1e-30) + bb=bb/np.maximum(mass_b[...,None],1e-30) + mm=(aa+bb)*.5 + conditional_js=.5*(np.where(aa>0,aa*np.log(np.maximum(aa,1e-30)/np.maximum(mm,1e-30)),0)+np.where(bb>0,bb*np.log(np.maximum(bb,1e-30)/np.maximum(mm,1e-30)),0)).sum(-1) + return {'first_key_mass_a':float(a[...,0].mean()),'first_key_mass_b':float(b[...,0].mean()), + 'js_without_first_key':float(conditional_js[valid].mean()) if valid.any() else None, + 'without_first_key_valid_rows':int(valid.sum()),'js_mean':float(js.mean()),'js_p95':float(np.quantile(js,.95)), + 'js_without_first_query':float(js[...,1:].mean()), + 'top5_overlap':float(np.stack(overlap,axis=-1).mean())} + + +def save_csv(path,rows): + if not rows:return + keys=list(dict.fromkeys(k for r in rows for k in r)) + with path.open('w',newline='') as f: + w=csv.DictWriter(f,fieldnames=keys);w.writeheader();w.writerows(rows) + + +def tokenize(root): + from tokenizers import Tokenizer + path=root/'tokenizer.json' + if not path.exists(): + url=f'https://huggingface.co/Qwen/Qwen3.5-2B/resolve/{REVISION}/tokenizer.json' + with urllib.request.urlopen(url,timeout=180) as r:path.write_bytes(r.read()) + tokenizer=Tokenizer.from_file(str(path)) + samples=json.loads(Path(__file__).with_name('samples.json').read_text())['samples'] + ids=[] + for s in samples: + tokens=tokenizer.encode(s['text'],add_special_tokens=False).ids + if len(tokens)<32:raise ValueError(f'Text too short: {s["id"]}') + s['text_sha256']=hashlib.sha256(s['text'].encode()).hexdigest() + s['full_token_count']=len(tokens);s['token_ids']=tokens[:32] + ids.append(tokens[:32]) + write_json(root/'samples-tokenized.json',{'revision':REVISION,'tokenizer_sha256':hashlib.sha256(path.read_bytes()).hexdigest(),'samples':samples}) + return samples,np.array(ids,dtype=np.int64) + + +def main(): + p=argparse.ArgumentParser();p.add_argument('--root',type=Path,required=True);p.add_argument('--seconds',type=int,default=7200) + args=p.parse_args();root=args.root;root.mkdir(exist_ok=True,parents=True) + write_json(root/'retention-policy.json',{'weights':'retain_until_explicit_user_delete_request','automatic_cleanup':False,'instruction_date':'2026-11-09'}) + samples,ids=tokenize(root) + cal=np.array([i for i,s in enumerate(samples) if s['split']=='calibration']) + held=np.array([i for i,s in enumerate(samples) if s['split']=='heldout']) + # Freeze all candidates/thresholds before opening weights or measuring functions. + experiment={'revision':REVISION,'attention_pairs':ATTN_PAIRS,'mlp_pairs':MLP_PAIRS,'norm_layers':NORM_LAYERS, + 'seconds':args.seconds,'output_relative_l2_limit':.05,'token_p95_limit':.1, + 'attention_js_limit':.01,'attention_top5_min':.8,'heldout_nll_mean_delta_limit':.01, + 'heldout_nll_sequence_p95_delta_limit':.05,'max_intervention_candidates':1, + 'directions_per_selected_candidate':2,'training':False,'search_R':False,'automatic_weight_cleanup':False, + 'dtype':'original weights decoded to FP32; not a BF16 backend benchmark', + 'blas_threads':os.environ.get('OPENBLAS_NUM_THREADS')} + write_json(root/'experiment.json',experiment) + started=time.time();status={'state':'running','started_unix':started,'phase':'baseline'} + write_json(root/'functional-status.json',status) + rows=[];headrows=[];interventions=[] + try: + config=json.loads((root/'model_config.json').read_text()) + lm=StreamingLM(root/'weights',config,seconds=args.seconds) + overall_deadline=lm.deadline + lm.deadline=min(overall_deadline,time.monotonic()+2400) + lm.capture_qk=True + capture=[('attn',i) for pair in ATTN_PAIRS for i in pair]+[('mlp',i) for pair in MLP_PAIRS for i in pair] + def progress(layer): + status.update(last_completed_layer=layer,elapsed_seconds=time.time()-started) + write_json(root/'functional-status.json',status) + print(json.dumps({'phase':status['phase'],'layer':layer,'elapsed':time.time()-started}),flush=True) + h,saved=lm.forward(ids,capture=capture,progress=progress) + np.savez(root/'baseline-activations.npz',**saved,**{f'qraw_{k}':v[0] for k,v in lm.qk_inputs.items()},**{f'kraw_{k}':v[1] for k,v in lm.qk_inputs.items()},final=h) + # Baseline heldout NLL uses no selection information. + nll=lm.nll(h[held],ids[held]);np.save(root/'baseline-heldout-token-nll.npy',nll) + baseline_nll=nll.mean(axis=1) + lm.capture_qk=False + lm.deadline=min(overall_deadline,time.monotonic()+1800) + status['phase']='calibration';write_json(root/'functional-status.json',status) + candidates=[] + for kind,pairs in [('attn',ATTN_PAIRS),('mlp',MLP_PAIRS)]: + for i,j in pairs: + pairid=f'{kind}_{i}_{j}';pairrows=[] + for source in (i,j): + x=saved[f'{kind}_{source}'][cal] + if kind=='attn': + a,pa=lm.full_attention(x,i);b,pb=lm.full_attention(x,j) + pm={} + for bi,si in enumerate(cal): + for hi in range(pa.shape[1]): + hpm=probability_metrics(pa[bi:bi+1,hi:hi+1],pb[bi:bi+1,hi:hi+1]) + headrows.append({'candidate':pairid,'source':source,'sample_id':samples[si]['id'],'head':hi,**hpm}) + save_csv(root/'attention-head-metrics.csv',headrows) + else: + a,b=lm.mlp(x,i),lm.mlp(x,j);pm={} + rec=error_rows(a,b,[samples[t] for t in cal],{'candidate':pairid,'kind':kind,'source':source,**pm}) + if kind=='attn': + for bi,r in enumerate(rec):r.update(probability_metrics(pa[bi:bi+1],pb[bi:bi+1])) + rows.extend(rec);pairrows.extend(rec) + save_csv(root/'calibration-metrics.csv',rows) + passing=all(r['relative_l2']<=.05 and r['reverse_relative_l2']<=.05 and r['token_relative_p95']<=.1 and r['reverse_token_relative_p95']<=.1 for r in pairrows) + if kind=='attn':passing=passing and all(r['js_mean']<=.01 and r['top5_overlap']>=.8 for r in pairrows) + prefix=f'model.language_model.layers.{i}.'+('self_attn.' if kind=='attn' else 'mlp.') + size=sum(t.bytes for name,t in lm.tensors.items() if name.startswith(prefix)) + candidates.append({'id':pairid,'kind':kind,'i':i,'j':j,'passed':passing,'maximum_symmetric_relative_error':max(max(r['relative_l2'],r['reverse_relative_l2']) for r in pairrows),'donor_component_bytes':size}) + for layer in NORM_LAYERS: + pairid=f'qk_norm_{layer}';pfx=f'layers.{layer}.self_attn.' + qw,kw=lm.small(pfx+'q_norm.weight'),lm.small(pfx+'k_norm.weight') + pairrows=[] + for source,x in zip(('q','k'),lm.qk_inputs[layer]): + a,b=norm(x[cal],qw,lm.eps),norm(x[cal],kw,lm.eps) + rec=error_rows(a,b,[samples[t] for t in cal],{'candidate':pairid,'kind':'qk_norm','source':source}) + pairrows.extend(rec);rows.extend(rec) + passing=all(r['relative_l2']<=.05 and r['reverse_relative_l2']<=.05 and r['token_relative_p95']<=.1 and r['reverse_token_relative_p95']<=.1 for r in pairrows) + candidates.append({'id':pairid,'kind':'qk_norm','i':layer,'j':layer,'passed':passing,'maximum_symmetric_relative_error':max(max(r['relative_l2'],r['reverse_relative_l2']) for r in pairrows),'donor_component_bytes':lm.tensors['model.language_model.'+pfx+'q_norm.weight'].bytes}) + save_csv(root/'calibration-metrics.csv',rows) + write_json(root/'candidate-summary.json',candidates) + passed=[c for c in candidates if c['passed']] + # Prefer larger potential savings, then smaller worst calibration error; never consult heldout. + selected=sorted(passed,key=lambda c:(-c['donor_component_bytes'],c['maximum_symmetric_relative_error'],c['id']))[:1] + write_json(root/'selection.json',{'selected':selected,'rule':'calibration pass; descending component bytes, ascending maximum symmetric error; maximum one candidate'}) + if selected: + c=selected[0] + heldrows=[] + if c['kind']=='qk_norm': + pfx=f'layers.{c["i"]}.self_attn.' + qw,kw=lm.small(pfx+'q_norm.weight'),lm.small(pfx+'k_norm.weight') + for source,x in zip(('q','k'),lm.qk_inputs[c['i']]): + heldrows.extend(error_rows(norm(x[held],qw,lm.eps),norm(x[held],kw,lm.eps),[samples[t] for t in held],{'candidate':c['id'],'kind':'qk_norm','source':source})) + else: + for source in (c['i'],c['j']): + x=saved[f'{c["kind"]}_{source}'][held] + if c['kind']=='attn': + a,_=lm.full_attention(x,c['i']);b,_=lm.full_attention(x,c['j']) + else:a,b=lm.mlp(x,c['i']),lm.mlp(x,c['j']) + heldrows.extend(error_rows(a,b,[samples[t] for t in held],{'candidate':c['id'],'kind':c['kind'],'source':source})) + save_csv(root/'heldout-function-metrics.csv',heldrows) + held_function_pass=all(r['relative_l2']<=.05 and r['reverse_relative_l2']<=.05 and r['token_relative_p95']<=.1 and r['reverse_token_relative_p95']<=.1 for r in heldrows) + status['heldout_function_pass']=held_function_pass + lm.deadline=min(overall_deadline,time.monotonic()+1800) + status['phase']='intervention';write_json(root/'functional-status.json',status) + for reverse in (False,True): + replacement=None;lm.norm_replacement=None + if c['kind']=='qk_norm': + pfx=f'layers.{c["i"]}.self_attn.' + target,donor=('q_norm','k_norm') if not reverse else ('k_norm','q_norm') + lm.norm_replacement=(pfx+target,pfx+donor) + label=target+'_uses_'+donor + else: + target,donor=(c['i'],c['j']) if not reverse else (c['j'],c['i']) + replacement=(c['kind'],target,donor);label=f'{target}_uses_{donor}' + t=time.monotonic();modified,_=lm.forward(ids[held],replacement=replacement,progress=progress) + changed=lm.nll(modified,ids[held]);np.save(root/f'intervention-{label}-token-nll.npy',changed) + delta=changed.mean(axis=1)-baseline_nll + rec={'candidate':c['id'],'direction':label,'elapsed_seconds':time.monotonic()-t, + 'nll_mean_delta':float(delta.mean()),'nll_sequence_p95_delta':float(np.quantile(delta,.95)), + 'sequence_nll_deltas':delta.tolist(),'passed_quality':bool(delta.mean()<=.01 and np.quantile(delta,.95)<=.05)} + interventions.append(rec);write_json(root/'interventions.json',interventions) + # No fallback candidate if heldout fails. + status.update(state='completed',phase='done',selected_candidate=selected, + positive_bidirectional_quality=bool(interventions and status.get('heldout_function_pass') and all(x['passed_quality'] for x in interventions)), + elapsed_seconds=time.time()-started,peak_rss_kib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, + claim_limit='Small text-only heldout pilot; no global interchangeability proof, no measured GPU speedup; component storage savings require explicit layout accounting.') + except Exception as exc: + status.update(state='stopped',error_type=type(exc).__name__,error=str(exc),elapsed_seconds=time.time()-started, + peak_rss_kib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) + (root/'functional-error.txt').write_text(traceback.format_exc()) + raise + finally: + write_json(root/'functional-status.json',status) + + +if __name__=='__main__':main() diff --git a/functional/samples.json b/functional/samples.json new file mode 100644 index 0000000..6230dba --- /dev/null +++ b/functional/samples.json @@ -0,0 +1,13 @@ +{ + "origin": "Experiment-authored plain text, no external corpus; fixed before functional metrics. Chinese/English balanced in each split. No chat template. First 32 tokenizer tokens retained.", + "samples": [ + {"id":"cal-zh-1","split":"calibration","language":"zh","text":"清晨，研究员打开实验记录本，检查昨天保存的数据。他先确认文件完整，再比较两组观测结果。为了避免偶然误差，他为每一步留下时间和条件，并将尚未验证的推测单独记录。"}, + {"id":"cal-zh-2","split":"calibration","language":"zh","text":"一辆列车从山间驶向海边，经过村庄、桥梁和开阔的田野。旅客望着窗外变化的景色，计划抵达后先步行到旅馆，再去市场购买晚餐需要的食材。天气预报说傍晚可能下雨。"}, + {"id":"cal-en-1","split":"calibration","language":"en","text":"A small library opens every morning beside the river. The librarian checks returned books, repairs damaged covers, and places new arrivals on a wooden table. Visitors often stop to read before continuing their journey across the old bridge."}, + {"id":"cal-en-2","split":"calibration","language":"en","text":"To compare two measurements, first record their units and the conditions under which they were collected. Compute the difference, inspect unusual observations, and repeat the procedure with independent examples. A short report should separate observations from explanations and unresolved questions."}, + {"id":"test-zh-1","split":"heldout","language":"zh","text":"午后，工程师检查温室中的传感器。温度记录出现一个异常值，她没有立刻删除，而是查看设备状态并补做一次测量。确认原因以后，她调整了采样计划，在报告中说明这次变化。"}, + {"id":"test-zh-2","split":"heldout","language":"zh","text":"周末，家人沿着湖边的小路散步。他们带了水和简单的午餐，打算在树林旁休息。途中遇到一处正在维修的桥，于是参考路标选择另一条道路，终于在日落前回到车站。"}, + {"id":"test-en-1","split":"heldout","language":"en","text":"The gardener planted several rows of beans behind the house. Each evening she checked the soil and wrote down how much water the plants received. After a week of warm weather, the first green shoots appeared near the fence beside the apple tree."}, + {"id":"test-en-2","split":"heldout","language":"en","text":"When a program reads a table, it should check column names and handle missing entries explicitly. The developer tested a small example, compared the output with a hand calculation, and saved the input alongside the result so another person could reproduce the check."} + ] +} diff --git a/functional/supervise.py b/functional/supervise.py new file mode 100644 index 0000000..45d1cab --- /dev/null +++ b/functional/supervise.py @@ -0,0 +1,73 @@ +"""One bounded CPU run; waits for verified download and binary dependencies. + +No installation, retry search, cleanup, training, GPU selection or Git mutation. +""" +import argparse +import hashlib +import importlib.util +import json +from pathlib import Path +import subprocess +import sys +import time +import traceback +from .run import write_json + + +def main(): + p=argparse.ArgumentParser();p.add_argument('--root',type=Path,required=True);p.add_argument('--download-wait-seconds',type=int,default=14400) + a=p.parse_args();root=a.root.resolve();start=time.monotonic() + state={'state':'waiting','phase':'dependencies','weight_retention':'until_explicit_user_delete_request'} + def status(**kwargs): + state.update(kwargs,supervisor_elapsed_seconds=time.monotonic()-start) + write_json(root/'supervisor-status.json',state) + def execute(module,args,timeout,log): + with (root/log).open('w') as out: + r=subprocess.run([sys.executable,'-B','-m',module,*args],stdout=out,stderr=subprocess.STDOUT,timeout=timeout) + if r.returncode:raise RuntimeError(f'{module} failed: exit {r.returncode}, see {log}') + status() + try: + while not all(importlib.util.find_spec(name) for name in ('torch','tokenizers')): + if time.monotonic()-start>3600:raise TimeoutError('Binary reference dependencies not ready within one hour') + time.sleep(15) + compute_start=time.monotonic() + status(state='running',phase='reference_check') + provenance={str(p):hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(Path('functional').glob('*')) if p.is_file()} + write_json(root/'functional-code-hashes.json',provenance) + execute('functional.reference_check',['--source',str(root/'reference.py'),'--output',str(root/'reference-check.json')],300,'reference-check.log') + reference_seconds=time.monotonic()-compute_start + status(state='waiting',phase='verified_weights',reference_seconds=reference_seconds) + while True: + path=root/'manifest.json' + try: + manifest=json.loads(path.read_text()) + weights=[f for f in manifest.get('files',[]) if f['name'].endswith('.safetensors')] + ready=(manifest.get('completed_unix') is not None and weights and + all((root/'weights'/f['name']).is_file() and (root/'weights'/f['name']).stat().st_size==f['bytes'] and f['sha256']==f['upstream_sha256'] for f in weights)) + except (OSError,ValueError,KeyError):ready=False + if ready:break + if time.monotonic()-start>a.download_wait_seconds:raise TimeoutError('Verified weight download not ready before wait deadline') + time.sleep(15) + status(state='running',phase='real_weight_preflight') + compute_start=time.monotonic() + execute('functional.preflight',['--root',str(root)],1200,'preflight.log') + preflight_seconds=time.monotonic()-compute_start + remaining=int(7200-reference_seconds-preflight_seconds-60) + if remaining<60:raise TimeoutError('No remaining bounded functional computation budget') + status(phase='functional_analysis',preflight_seconds=preflight_seconds,remaining_seconds=remaining) + execute('functional.run',['--root',str(root),'--seconds',str(remaining)],remaining+15,'functional-run.log') + status(phase='report') + execute('functional.report',['--root',str(root)],120,'functional-report.log') + status(state='completed',phase='done') + except Exception as exc: + status(state='stopped',error_type=type(exc).__name__,error=str(exc)) + (root/'supervisor-error.txt').write_text(traceback.format_exc()) + # Retain all weights/results. No automatic retries or candidate expansion. + if not (root/'functional-status.json').exists(): + write_json(root/'functional-status.json',{'state':'stopped','phase':state['phase'],'error_type':type(exc).__name__,'error':str(exc)}) + try:execute('functional.report',['--root',str(root)],120,'functional-report.log') + except Exception:pass + raise + + +if __name__=='__main__':main() diff --git a/requirements-functional.txt b/requirements-functional.txt new file mode 100644 index 0000000..7787628 --- /dev/null +++ b/requirements-functional.txt @@ -0,0 +1,5 @@ +# Existing requirements.txt supplies NumPy and plotting dependencies. +# Install binary wheels only. Torch is a reference-check dependency, not the streaming runtime. +tokenizers==0.23.3 +# Install separately with --index-url https://download.pytorch.org/whl/cpu: +# torch==2.15.0+cpu diff --git a/tests/test_functional.py b/tests/test_functional.py new file mode 100644 index 0000000..08f14d2 --- /dev/null +++ b/tests/test_functional.py @@ -0,0 +1,62 @@ +"""CPU functional implementation tests; run on authorized remote server only.""" +import json +from pathlib import Path +import struct +import tempfile +import unittest +import numpy as np +from functional.model import StreamingLM,causal_conv,silu,attention,norm +from functional.run import probability_metrics,error_rows + + +class FunctionalTests(unittest.TestCase): + def test_streamed_projection_embedding_and_output_head(self): + rng=np.random.default_rng(7) + arrays={'model.language_model.embed_tokens.weight':rng.normal(size=(19,4)).astype(np.float32), + 'model.language_model.test.weight':rng.normal(size=(9,4)).astype(np.float32)} + header={};offset=0;body=b'' + for name,arr in arrays.items(): + raw=arr.tobytes();header[name]={'shape':list(arr.shape),'dtype':'F32','data_offsets':[offset,offset+len(raw)]} + body+=raw;offset+=len(raw) + encoded=json.dumps(header).encode() + with tempfile.TemporaryDirectory() as d: + Path(d,'model.safetensors').write_bytes(struct.pack('