Download switch.py from opticalfibre/PIN: direct link, hf CLI and curl.
- Browser
- Download file 24.5 kB
-
https://huggingface.co/opticalfibre/PIN/resolve/main/switch.py
- Command line
-
hf download hf://opticalfibre/PIN/switch.py
-
curl -L -o switch.py https://huggingface.co/opticalfibre/PIN/resolve/main/switch.py
24.5 kB
| """ | |
| SHAPE SHIFTING: HOW MANY EPOCHS TO RECOVER? | |
| A fold is W[i,j] = v[idx(i,j)], so the architecture lives entirely in the | |
| index and the knowledge lives in the values. Changing shape is therefore | |
| swapping the index and keeping the values, and the question is not whether | |
| that is possible but what it COSTS and how quickly the cost is repaid. | |
| Four reshapes are already known to be free or nearly so. Splitting a value | |
| group leaves the function unchanged at exactly zero. Widening by | |
| concatenation is prediction averaging, verified to one part in ten | |
| trillion. Adding a layer as a delta filter moved the logits by 0.09 with | |
| every prediction preserved. A residual path added at zero weight is exact | |
| by construction. For those, an adaptation phase would be idle. | |
| The interesting case is the LOSSY direction, and it is the one deployment | |
| actually wants: fitting a trained model onto a smaller device. §6 of the | |
| paper shows a model COARSENED DURING TRAINING recovers, eleven or twelve | |
| seeds of twelve, but it had thirty epochs between rungs. That says recovery | |
| is possible; it does not say a single epoch buys it. | |
| One epoch is 2.5% of a training run. If that is enough, reshaping stops | |
| being a research operation and becomes a deployment step: ship one model, | |
| adapt it to whatever the target hardware prefers, on arrival. If it takes | |
| twenty, this is retraining with a warm start, which is a much weaker claim | |
| and should be said plainly. | |
| THREE RESHAPES, all of which keep the convolution structure so the fast | |
| path still applies and an adaptation epoch costs seconds: | |
| SHRINK halve every layer's channels, merging pairs by averaging. The | |
| lossy one, and the one a smaller device wants. | |
| WIDEN double them, duplicating each channel and halving what it sends | |
| on, which preserves the function exactly. | |
| DEEPEN insert a layer initialised as a delta filter, whose scale and | |
| shift are set to undo the normalisation that follows. | |
| Each is measured at 0, 1, 2 and 3 epochs of adaptation, against training | |
| that shape from scratch. The SHAPE of that curve is the finding: steep at | |
| one epoch means shape shifting is practical, flat until twenty means it is | |
| retraining wearing a different name. | |
| """ | |
| import numpy as np | |
| import time | |
| import json | |
| import os | |
| try: | |
| import cupy as _cp | |
| _GPU = _cp.cuda.runtime.getDeviceCount() > 0 | |
| except Exception: | |
| _GPU = False | |
| xp = _cp if _GPU else np | |
| DT = np.float32 | |
| def to_dev(a, dtype=DT): | |
| a = np.asarray(a, dtype=dtype) | |
| return xp.asarray(a) if _GPU else a | |
| def to_host(a): | |
| return _cp.asnumpy(a) if _GPU and isinstance(a, _cp.ndarray) else np.asarray(a) | |
| class FastConv: | |
| def __init__(self, g_in, c_in, k, c_out, stride): | |
| self.g_in, self.c_in, self.k = g_in, c_in, k | |
| self.c_out, self.stride = c_out, stride | |
| self.g_out = g_in // stride | |
| self.ins = c_in*g_in*g_in | |
| self.out = c_out*self.g_out*self.g_out | |
| self.K = c_in*c_out*k*k + 1 | |
| self.taps = c_in*k*k | |
| go, gi = self.g_out, g_in | |
| oy, ox = np.divmod(np.arange(go*go), go) | |
| ci = np.arange(c_in)[:, None, None] | |
| dy = np.arange(k)[None, :, None] | |
| dx = np.arange(k)[None, None, :] | |
| iy = oy[:, None, None, None]*stride - k//2 + dy | |
| ix = ox[:, None, None, None]*stride - k//2 + dx | |
| ok = (iy >= 0) & (iy < gi) & (ix >= 0) & (ix < gi) | |
| flat = (ci*gi*gi + np.clip(iy, 0, gi-1)*gi + np.clip(ix, 0, gi-1)) | |
| flat = np.where(ok, flat, self.ins).reshape(go*go, self.taps) | |
| self.fwd_idx = to_dev(flat, np.int64) if _GPU else flat.astype(np.int64) | |
| readers = {} | |
| for p in range(go*go): | |
| for tc in range(self.taps): | |
| s = int(flat[p, tc]) | |
| if s < self.ins: | |
| readers.setdefault(s, []).append(p*self.taps + tc) | |
| w = max((len(v) for v in readers.values()), default=1) | |
| tbl = np.full((self.ins, w), go*go*self.taps, np.int64) | |
| for s, v in readers.items(): | |
| tbl[s, :len(v)] = v | |
| self.bwd_idx = to_dev(tbl, np.int64) if _GPU else tbl | |
| def weights(self, v): | |
| return v[:-1].reshape(self.c_in, self.c_out, self.k*self.k) \ | |
| .transpose(0, 2, 1).reshape(self.taps, self.c_out) | |
| def forward(self, v, x): | |
| n = x.shape[0] | |
| xz = xp.concatenate([x, xp.zeros((n, 1), DT)], 1) | |
| col = xz[:, self.fwd_idx].reshape(n*self.g_out**2, self.taps) | |
| z = (col @ self.weights(v)).reshape(n, self.g_out**2, self.c_out) | |
| return z.transpose(0, 2, 1).reshape(n, self.out), col | |
| def backward(self, v, col, dz, need_input=True): | |
| n = dz.shape[0] | |
| d = dz.reshape(n, self.c_out, self.g_out**2).transpose(0, 2, 1) \ | |
| .reshape(n*self.g_out**2, self.c_out) | |
| gv = xp.zeros(self.K, DT) | |
| gv[:-1] = (col.T @ d).reshape(self.c_in, self.k*self.k, self.c_out) \ | |
| .transpose(0, 2, 1).reshape(-1) | |
| if not need_input: | |
| return gv, None | |
| dcol = (d @ self.weights(v).T).reshape(n, -1) | |
| pad = xp.concatenate([dcol, xp.zeros((n, 1), DT)], 1) | |
| return gv, pad[:, self.bwd_idx].sum(2) | |
| # ------------------------------------------------------------ the model | |
| class Stack: | |
| """The testbed's architecture, rebuildable at any channel widths.""" | |
| def __init__(self, grid, c_in, chans, strides, n_out, seed=0, P=None): | |
| self.chans, self.strides = list(chans), list(strides) | |
| rg = np.random.default_rng(seed) | |
| self.fcs, self.P = [], [] | |
| g, cin = grid, c_in | |
| for ch, st in zip(chans, strides): | |
| fc = FastConv(g, cin, 3, ch, st) | |
| self.fcs.append(fc) | |
| v = rg.normal(0, np.sqrt(2.0/(cin*9)), fc.K).astype(np.float32) | |
| v[-1] = 0.0 | |
| self.P += [to_dev(v), xp.ones(fc.out, DT), xp.zeros(fc.out, DT)] | |
| g, cin = fc.g_out, ch | |
| self.feat, self.g_final = cin, g | |
| self.P += [to_dev(rg.normal(0, np.sqrt(2.0/cin), (cin, n_out))), | |
| xp.zeros(n_out, DT)] | |
| self.head = len(self.P)-2 | |
| if P is not None: | |
| self.P = [p.copy() for p in P] | |
| def values(self): | |
| return sum(f.K for f in self.fcs) | |
| def forward(self, x, keep=False): | |
| cache, h = [], x | |
| for li, fc in enumerate(self.fcs): | |
| p = 3*li | |
| z, col = fc.forward(self.P[p], h) | |
| var = z.var(1, keepdims=True) + 1e-5 | |
| zn = (z - z.mean(1, keepdims=True))/xp.sqrt(var) | |
| zs = zn*self.P[p+1] + self.P[p+2] | |
| h = xp.maximum(zs, 0) | |
| if keep: | |
| cache.append((col, zn, zs, var)) | |
| feat = h.reshape(h.shape[0], self.feat, self.g_final**2).mean(2) | |
| lg = feat @ self.P[self.head] + self.P[self.head+1] | |
| return (feat, lg, cache) if keep else (feat, lg) | |
| def accuracy(self, X, y): | |
| out = [] | |
| for s in range(0, X.shape[0], 512): | |
| _, lg = self.forward(X[s:s+512]) | |
| out.append(to_host(lg)) | |
| return float((np.concatenate(out).argmax(1) == y).mean()) | |
| def train(net, X, Y, Xte, yte, epochs, lr, batch, seed, M=None, V=None, | |
| t0=0, quiet=True): | |
| P = net.P | |
| M = M or [xp.zeros_like(p) for p in P] | |
| V = V or [xp.zeros_like(p) for p in P] | |
| rg = np.random.default_rng(seed) | |
| n = X.shape[0]; t = t0 | |
| L = len(net.fcs) | |
| for ep in range(epochs): | |
| for b in np.array_split(rg.permutation(n), max(1, n//batch)): | |
| bd = to_dev(b, np.int64) if _GPU else b | |
| x, y = X[bd], Y[bd] | |
| feat, lg, cache = net.forward(x, keep=True) | |
| e = xp.exp(lg - lg.max(1, keepdims=True)) | |
| d = (e/e.sum(1, keepdims=True) - y)/len(b) | |
| G = [xp.zeros_like(p) for p in P] | |
| G[net.head] = feat.T @ d; G[net.head+1] = d.sum(0) | |
| npos = net.g_final**2 | |
| dh = xp.broadcast_to((d @ P[net.head].T)[:, :, None]/npos, | |
| (len(b), net.feat, npos) | |
| ).reshape(len(b), net.feat*npos) | |
| for li in range(L-1, -1, -1): | |
| col, zn, zs, var = cache[li] | |
| p = 3*li | |
| dzs = dh*(zs > 0) | |
| G[p+1] = (dzs*zn).sum(0); G[p+2] = dzs.sum(0) | |
| dzn = dzs*P[p+1] | |
| dz = (dzn - dzn.mean(1, keepdims=True) | |
| - zn*(dzn*zn).mean(1, keepdims=True))/xp.sqrt(var) | |
| gv, dx = net.fcs[li].backward(P[p], col, dz, li > 0) | |
| G[p] = gv | |
| if li > 0: | |
| dh = dx | |
| t += 1 | |
| for i in range(len(P)): | |
| M[i] = 0.9*M[i] + 0.1*G[i] | |
| V[i] = 0.999*V[i] + 0.001*G[i]*G[i] | |
| P[i] = P[i] - lr*(M[i]/(1-0.9**t))/(xp.sqrt(V[i]/(1-0.999**t))+1e-8) | |
| for i in range(L): | |
| P[3*i][-1] = 0.0 | |
| return net, M, V, t | |
| # --------------------------------------------------------- the reshapes | |
| def shrink(net, n_out): | |
| """Halve every layer's channels, merging pairs by averaging. | |
| The LOSSY direction, and the one a smaller device wants. Averaging a | |
| pair of channels is the least-squares merge of two value groups into | |
| one, which is what §6's annealing does between rungs.""" | |
| new_ch = [max(1, c//2) for c in net.chans] | |
| out = Stack(net.fcs[0].g_in, net.fcs[0].c_in, new_ch, net.strides, | |
| n_out, seed=0) | |
| cin_old, cin_new = net.fcs[0].c_in, net.fcs[0].c_in | |
| for li, (fc_o, fc_n) in enumerate(zip(net.fcs, out.fcs)): | |
| v = to_host(net.P[3*li])[:-1].reshape(fc_o.c_in, fc_o.c_out, 9) | |
| # merge output channels in pairs, then input channels to match | |
| v = v[:, :fc_n.c_out*2].reshape(fc_o.c_in, fc_n.c_out, 2, 9).mean(2) | |
| if fc_n.c_in != fc_o.c_in: | |
| v = v[:fc_n.c_in*2].reshape(fc_n.c_in, 2, fc_n.c_out, 9).mean(1) | |
| nv = np.zeros(fc_n.K, np.float32); nv[:-1] = v.reshape(-1) | |
| out.P[3*li] = to_dev(nv) | |
| for off, src in ((1, net.P[3*li+1]), (2, net.P[3*li+2])): | |
| s = to_host(src).reshape(fc_o.c_out, -1)[:fc_n.c_out*2] | |
| s = s.reshape(fc_n.c_out, 2, -1).mean(1) | |
| out.P[3*li+off] = to_dev(s.reshape(-1)) | |
| hd = to_host(net.P[net.head]).reshape(net.feat, -1) | |
| hd = hd[:out.feat*2].reshape(out.feat, 2, -1).mean(1) | |
| out.P[out.head] = to_dev(hd) | |
| out.P[out.head+1] = net.P[net.head+1].copy() | |
| return out | |
| def widen(net, n_out): | |
| """Double every layer's channels by DUPLICATING each and halving what | |
| it sends on, which preserves the function exactly.""" | |
| new_ch = [c*2 for c in net.chans] | |
| out = Stack(net.fcs[0].g_in, net.fcs[0].c_in, new_ch, net.strides, | |
| n_out, seed=0) | |
| for li, (fc_o, fc_n) in enumerate(zip(net.fcs, out.fcs)): | |
| v = to_host(net.P[3*li])[:-1].reshape(fc_o.c_in, fc_o.c_out, 9) | |
| v = np.repeat(v, 2, axis=1) # duplicate outputs | |
| if fc_n.c_in != fc_o.c_in: | |
| v = np.repeat(v, 2, axis=0)/2.0 # halve the inputs | |
| nv = np.zeros(fc_n.K, np.float32); nv[:-1] = v.reshape(-1) | |
| out.P[3*li] = to_dev(nv) | |
| for off in (1, 2): | |
| s = to_host(net.P[3*li+off]).reshape(fc_o.c_out, -1) | |
| out.P[3*li+off] = to_dev(np.repeat(s, 2, axis=0).reshape(-1)) | |
| hd = to_host(net.P[net.head]) | |
| out.P[out.head] = to_dev(np.repeat(hd, 2, axis=0)/2.0) | |
| out.P[out.head+1] = net.P[net.head+1].copy() | |
| return out | |
| def deepen(net, n_out, sample): | |
| """Insert a layer initialised as a DELTA FILTER, with its scale and | |
| shift set to undo the normalisation that follows. Measured elsewhere: | |
| that calibration takes the transition from moving the logits by 1.28 | |
| and losing a sixth of predictions, to 0.09 and losing none.""" | |
| ch = net.chans + [net.chans[-1]] | |
| st = net.strides + [1] | |
| out = Stack(net.fcs[0].g_in, net.fcs[0].c_in, ch, st, n_out, seed=0) | |
| L = len(net.fcs) | |
| for li in range(L): | |
| for off in range(3): | |
| out.P[3*li+off] = net.P[3*li+off].copy() | |
| fc = out.fcs[L] | |
| v = np.zeros(fc.K, np.float32) | |
| for c in range(min(fc.c_in, fc.c_out)): | |
| v[(c*fc.c_out + c)*9 + 4] = 1.0 | |
| out.P[3*L] = to_dev(v) | |
| h, _ = net.forward(sample) | |
| hh = sample | |
| for li, f in enumerate(net.fcs): | |
| z, _ = f.forward(net.P[3*li], hh) | |
| var = z.var(1, keepdims=True) + 1e-5 | |
| zn = (z - z.mean(1, keepdims=True))/xp.sqrt(var) | |
| hh = xp.maximum(zn*net.P[3*li+1] + net.P[3*li+2], 0) | |
| mu = float(to_host(hh.mean(1).mean())); sd = float(to_host(hh.std(1).mean())) | |
| out.P[3*L+1] = xp.full(fc.out, DT(max(sd, 1e-6))) | |
| out.P[3*L+2] = xp.full(fc.out, DT(mu)) | |
| out.P[out.head] = net.P[net.head].copy() | |
| out.P[out.head+1] = net.P[net.head+1].copy() | |
| return out | |
| # --------------------------------------------- does it only LOOK the same? | |
| def chi_of(net, X, n_probe=12, seed=0): | |
| """Mean squared singular value of each layer's Jacobian, by probing. | |
| Two models at the same accuracy can have quite different sensitivity, | |
| and that is what mimicry would look like from the inside.""" | |
| rg = np.random.default_rng(seed) | |
| hs, h = [X], X | |
| for li, fc in enumerate(net.fcs): | |
| z, _ = fc.forward(net.P[3*li], h) | |
| var = z.var(1, keepdims=True) + 1e-5 | |
| zn = (z - z.mean(1, keepdims=True))/xp.sqrt(var) | |
| h = xp.maximum(zn*net.P[3*li+1] + net.P[3*li+2], 0) | |
| hs.append(h) | |
| out = [] | |
| for li, fc in enumerate(net.fcs): | |
| hin = hs[li] | |
| def f(u): | |
| z, _ = fc.forward(net.P[3*li], u) | |
| var = z.var(1, keepdims=True) + 1e-5 | |
| zn = (z - z.mean(1, keepdims=True))/xp.sqrt(var) | |
| return xp.maximum(zn*net.P[3*li+1] + net.P[3*li+2], 0) | |
| base = f(hin) | |
| hn = float(to_host(xp.linalg.norm(hin, axis=1).mean())) | |
| tot = 0.0 | |
| for _ in range(n_probe): | |
| v = to_dev(rg.normal(size=hin.shape)) | |
| v = v/xp.linalg.norm(v, axis=1, keepdims=True) | |
| e = 1e-3*hn | |
| tot += float(to_host( | |
| (xp.linalg.norm((f(hin + e*v) - base)/e, axis=1)**2).mean())) | |
| out.append(tot/n_probe) | |
| return out | |
| def probe(net, ref, Xte, yte, Xh, yh, n_member=400, lam=1.0, seed=0): | |
| """Everything accuracy cannot see. | |
| AGREEMENT two models at the same score can be right on different | |
| examples. Chance overlap with equal accuracy a means a^2 + | |
| (1-a)^2/(classes-1) roughly, so anything near that is a | |
| different function wearing the same number. | |
| A MEMBER fit a head for classes NEITHER model was trained on. A | |
| model that has memorised a boundary keeps its own accuracy | |
| and gives a WORSE BASE, because the representation decayed | |
| where the readout did not. This is the test this framework | |
| is built to run. | |
| CHI the sensitivity of each layer, against the original's.""" | |
| pr = lambda m: np.concatenate([ | |
| to_host(m.forward(Xte[s:s+512])[1]) for s in range(0, len(yte), 512) | |
| ]).argmax(1) | |
| a, b = pr(net), pr(ref) | |
| ok_a, ok_b = a == yte, b == yte | |
| feats = lambda m, Z: np.concatenate([ | |
| to_host(m.forward(Z[s:s+512])[0]) for s in range(0, Z.shape[0], 512)]) | |
| F = feats(net, Xh) | |
| rg = np.random.default_rng(seed) | |
| cls = np.unique(yh) | |
| sub = rg.choice(len(yh), min(n_member, len(yh)), replace=False) | |
| A = np.concatenate([F[sub], np.ones((len(sub), 1), np.float32)], 1) | |
| T = (yh[sub][:, None] == cls[None, :]).astype(np.float32) | |
| W = np.linalg.solve(A.T @ A + lam*np.eye(A.shape[1], dtype=np.float32), | |
| A.T @ T) | |
| rest = np.setdiff1d(np.arange(len(yh)), sub) | |
| P2 = np.concatenate([F[rest], np.ones((len(rest), 1), np.float32)], 1) @ W | |
| member = float((cls[P2.argmax(1)] == yh[rest]).mean()) | |
| return dict( | |
| agreement=float((a == b).mean()), | |
| both_right=float((ok_a & ok_b).mean()), | |
| same_errors=float((~ok_a & ~ok_b & (a == b)).mean() | |
| / max((~ok_b).mean(), 1e-9)), | |
| member=member, | |
| chi=chi_of(net, Xte[:256])) | |
| # ------------------------------------------------------------- the run | |
| CFG = dict(grid=32, c_in=3, chans=(48, 96, 192), strides=(2, 2, 2), | |
| n_classes=100, n_held=20, batch=128, lr=1e-3, | |
| pretrain=60, adapt=(0, 1, 2, 3), scratch=60, | |
| seed=0, out="/content/drive/MyDrive/pin_switch") | |
| def load(cfg): | |
| from tensorflow import keras | |
| (a, b), (c, d) = keras.datasets.cifar100.load_data() | |
| X = np.concatenate([a, c]).astype(np.float32)/255.0 | |
| y = np.concatenate([b, d]).ravel().astype(np.int64) | |
| rg = np.random.default_rng(0) | |
| order = rg.permutation(cfg["n_classes"]) | |
| base = np.sort(order[cfg["n_held"]:]) | |
| remap = -np.ones(cfg["n_classes"], np.int64) | |
| remap[base] = np.arange(len(base)) | |
| m = np.isin(y, base) | |
| mu, sd = X[m].mean((0, 1, 2)), X[m].std((0, 1, 2)) + 1e-8 | |
| X = np.ascontiguousarray(((X-mu)/sd).astype(np.float32) | |
| .transpose(0, 3, 1, 2)).reshape(len(X), -1) | |
| i = np.where(m)[0]; rg.shuffle(i) | |
| cut = int(0.85*len(i)) | |
| held = np.where(~m)[0][:4000] | |
| return X, remap[y], i[:cut], i[cut:], len(base), held, y[held] | |
| def ensure_output(path): | |
| if path.startswith("/content/drive") and not os.path.isdir( | |
| "/content/drive/MyDrive"): | |
| try: | |
| from google.colab import drive | |
| drive.mount("/content/drive") | |
| except Exception as e: | |
| alt = os.path.basename(path.rstrip("/")) or "out" | |
| print(f"\n ** DRIVE UNAVAILABLE ({e}); writing to ./{alt} **\n") | |
| os.makedirs(alt, exist_ok=True); return alt | |
| os.makedirs(path, exist_ok=True) | |
| return path | |
| def main(**over): | |
| CFG.update(over) | |
| t0 = time.time() | |
| print("=" * 78) | |
| print("SHAPE SHIFTING: HOW MANY EPOCHS TO RECOVER?") | |
| print(" build 2026-08-15a") | |
| print("=" * 78) | |
| print(f" backend: {'cupy (GPU)' if _GPU else 'numpy (CPU)'}") | |
| for k, v in CFG.items(): | |
| print(f" {k:10s} = {v}") | |
| print("=" * 78, flush=True) | |
| X, y, tr, te, n_out, hi, yh = load(CFG) | |
| Xh = to_dev(X[hi]) | |
| Xd, Xte = to_dev(X[tr]), to_dev(X[te]) | |
| Y = to_dev(np.eye(n_out, dtype=np.float32)[y[tr]]); yte = y[te] | |
| net = Stack(CFG["grid"], CFG["c_in"], CFG["chans"], CFG["strides"], | |
| n_out, CFG["seed"]) | |
| print(f"\n pretraining {CFG['pretrain']} epochs, " | |
| f"{net.values():,} values", flush=True) | |
| net, _, _, _ = train(net, Xd, Y, Xte, yte, CFG["pretrain"], CFG["lr"], | |
| CFG["batch"], CFG["seed"]+1) | |
| base_acc = net.accuracy(Xte, yte) | |
| print(f" the model to be reshaped: {base_acc:.4f} " | |
| f"[{time.time()-t0:.0f}s]", flush=True) | |
| ops = [("shrink", lambda: shrink(net, n_out), "halve every channel"), | |
| ("widen", lambda: widen(net, n_out), "double every channel"), | |
| ("deepen", lambda: deepen(net, n_out, Xte[:512]), "add a layer")] | |
| res = {} | |
| for name, make, what in ops: | |
| print(f"\n {name.upper()}: {what}", flush=True) | |
| row = {} | |
| for ne in CFG["adapt"]: | |
| m = make() | |
| if ne: | |
| m, _, _, _ = train(m, Xd, Y, Xte, yte, ne, CFG["lr"], | |
| CFG["batch"], CFG["seed"]+50+ne) | |
| row[ne] = m.accuracy(Xte, yte) | |
| p = probe(m, net, Xte, yte, Xh, yh) | |
| row[f"probe{ne}"] = p | |
| print(f" {ne} epochs: {row[ne]:.4f} agrees with the " | |
| f"original {p['agreement']:.0%} a member on the unseen " | |
| f"twenty {p['member']:.4f} [{time.time()-t0:.0f}s]", | |
| flush=True) | |
| shape = make() | |
| sc = Stack(CFG["grid"], CFG["c_in"], shape.chans, shape.strides, | |
| n_out, CFG["seed"]+9) | |
| sc, _, _, _ = train(sc, Xd, Y, Xte, yte, CFG["scratch"], CFG["lr"], | |
| CFG["batch"], CFG["seed"]+11) | |
| row["scratch"] = sc.accuracy(Xte, yte) | |
| row["probe_scratch"] = probe(sc, net, Xte, yte, Xh, yh) | |
| row["values"] = shape.values() | |
| print(f" that shape trained from scratch ({CFG['scratch']} " | |
| f"epochs): {row['scratch']:.4f} [{time.time()-t0:.0f}s]", | |
| flush=True) | |
| res[name] = row | |
| out = ensure_output(CFG["out"]) | |
| json.dump(dict(base=base_acc, values=net.values(), ops=res), | |
| open(f"{out}/switch.json", "w"), indent=2) | |
| print("\n" + "=" * 78) | |
| print(" RECOVERY") | |
| print("=" * 78) | |
| print(f" the model before reshaping: {base_acc:.4f}, " | |
| f"{net.values():,} values\n") | |
| print(f" {'reshape':>8s} {'values':>9s} " + | |
| " ".join(f"{str(e)+' ep':>8s}" for e in CFG["adapt"]) + | |
| f" {'scratch':>9s}") | |
| for name, _, _ in ops: | |
| r = res[name] | |
| print(f" {name:>8s} {r['values']:9,} " + | |
| " ".join(f"{r[e]:8.4f}" for e in CFG["adapt"]) + | |
| f" {r['scratch']:9.4f}") | |
| print(f"\n as a share of what that shape reaches trained from scratch:") | |
| print(f" {'reshape':>8s} " + | |
| " ".join(f"{str(e)+' ep':>8s}" for e in CFG["adapt"])) | |
| for name, _, _ in ops: | |
| r = res[name] | |
| print(f" {name:>8s} " + | |
| " ".join(f"{r[e]/max(r['scratch'],1e-9):7.0%} " | |
| for e in CFG["adapt"])) | |
| ref_p = probe(net, net, Xte, yte, Xh, yh) | |
| print(f"\n WHAT ACCURACY CANNOT SEE. The original agrees with itself") | |
| print(f" 100% and supports a member on the twenty unseen classes at") | |
| print(f" {ref_p['member']:.4f}; its chi runs " | |
| + ", ".join(f"{c:.2f}" for c in ref_p["chi"]) + ".\n") | |
| print(f" {'reshape':>8s} {'adapt':>6s} {'accuracy':>9s} {'agrees':>8s} " | |
| f"{'member':>8s} {'chi':>22s}") | |
| for name, _, _ in ops: | |
| for e in list(CFG["adapt"]) + ["scratch"]: | |
| k = f"probe{e}" if e != "scratch" else "probe_scratch" | |
| p = res[name][k] | |
| print(f" {name:>8s} {str(e):>6s} " | |
| f"{res[name][e if e != 'scratch' else 'scratch']:9.4f} " | |
| f"{p['agreement']:8.0%} {p['member']:8.4f} " | |
| + " ".join(f"{c:6.2f}" for c in p["chi"])) | |
| print(f"\n A RESHAPED MODEL THAT MERELY MIMICS would keep its accuracy") | |
| print(f" and lose the other three: it would agree with the original") | |
| print(f" only as often as two models of that accuracy agree by chance,") | |
| print(f" support a WORSE member because the representation decayed") | |
| print(f" where the readout did not, and drift in chi. A reshaped model") | |
| print(f" that is the same FUNCTION keeps all four.") | |
| print("\n" + "=" * 78) | |
| print(" READOUT") | |
| print("=" * 78) | |
| s = res["shrink"] | |
| gain1 = s[1] - s[0] if 1 in s else 0.0 | |
| total = s[max(CFG["adapt"])] - s[0] | |
| print(f" SHRINKING is the lossy direction and the one a smaller device") | |
| print(f" wants. Halving every channel costs " | |
| f"{base_acc - s[0]:+.4f} immediately,") | |
| print(f" and one epoch of adaptation returns {gain1:+.4f} of it.") | |
| if total > 1e-9 and gain1/total > 0.6: | |
| print(f"\n MOST OF THE RECOVERY IS IN THE FIRST EPOCH " | |
| f"({gain1/total:.0%} of what") | |
| print(f" three epochs achieve). Reshaping is a DEPLOYMENT STEP:") | |
| print(f" ship one model, adapt it to the target's preferred shape") | |
| print(f" on arrival, and pay 2.5% of a training run for it.") | |
| elif total > 1e-9: | |
| print(f"\n RECOVERY IS GRADUAL: the first epoch is " | |
| f"{gain1/total:.0%} of what three") | |
| print(f" achieve, so this is retraining with a warm start rather") | |
| print(f" than a switch, and the honest claim is the smaller one.") | |
| else: | |
| print(f"\n ADAPTATION DOES NOTHING at this scale. The reshape is") | |
| print(f" either already at its ceiling or beyond repair by three") | |
| print(f" epochs, and the curve says which.") | |
| for nm in ("widen", "deepen"): | |
| r = res[nm] | |
| print(f"\n {nm.upper()} starts at {r[0]:.4f} against the original's") | |
| print(f" {base_acc:.4f} ({r[0]-base_acc:+.4f}), which is the") | |
| print(f" exactness these operations are supposed to have, and") | |
| print(f" reaches {r[max(CFG['adapt'])]:.4f} after " | |
| f"{max(CFG['adapt'])} epochs " | |
| f"({r[max(CFG['adapt'])]-r[0]:+.4f}).") | |
| print(f"\n saved to {out}; total {time.time()-t0:.0f}s") | |
| main() | |