| """ |
| A MEMBER THAT CHANGES ONLY THE CONFIDENCE. |
| |
| Two results boxed the correction loop in. A member that changes the ANSWER |
| has a damage channel, and the arithmetic is brutal: right answers outnumber |
| wrong ones 6.6 to 1, so a point of damage costs 6.6 times what a point of |
| correction earns, and no gate we built came out ahead. And the base cannot |
| say where it needs help — its confidence separates what it can do from what |
| it has never seen at AUC 0.4245, BELOW chance. |
| |
| Lee's move: manipulate the confidence only. The base is confident that its |
| ignorance is correct, so change that rather than the answer. |
| |
| A member applied as a positive SCALE on the logits cannot alter the argmax |
| — accuracy is invariant by construction, not by measurement — so it has no |
| damage channel at all. What it can alter is how peaked the distribution is, |
| which is exactly what abstention reads. |
| |
| logits -> logits / exp(s(a)) |
| |
| with s a linear read of the frozen base's features, fitted by the same |
| closed-form solve as any other member. Large s flattens the distribution |
| and the model reports doubt; small s sharpens it. |
| |
| THE QUESTION UNDERNEATH is whether the information exists at all. The |
| output layer demonstrably cannot express unfamiliarity — it was trained to |
| name one of seven classes and it always does. But the FEATURES might still |
| carry it, and a probe would find it if so. If a ridge read of the features |
| separates seen from unseen where the softmax cannot, then ignorance is |
| visible one layer down and the failsafe has something to run on. If it |
| cannot, the representation genuinely does not encode unfamiliarity and no |
| member can supply it. |
| |
| Measured three ways: |
| |
| DOES IT SEPARATE? AUC of the new confidence against the base's 0.4245 |
| IS ACCURACY SAFE? it must be EXACTLY unchanged; anything else is a bug |
| IS IT USEFUL? risk-coverage — accuracy on the most confident X%, |
| before and after, which is what abstention buys |
| """ |
|
|
| import numpy as np |
| import time |
| import json |
|
|
| 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) |
|
|
|
|
| def windowed(g, c_in, k, c_out): |
| ni, no = c_in*g*g, c_out*g*g |
| ii, jj = np.meshgrid(np.arange(ni), np.arange(no), indexing='ij') |
| ci, pi = ii // (g*g), ii % (g*g) |
| co, po = jj // (g*g), jj % (g*g) |
| dr = pi // g - (po // g - k//2) |
| dc = pi % g - (po % g - k//2) |
| inside = (dr >= 0) & (dr < k) & (dc >= 0) & (dc < k) |
| K = c_in*c_out*k*k + 1 |
| idx = np.where(inside, (ci*c_out + co)*k*k + dr*k + dc, K-1) |
| return idx.ravel().astype(np.int32), K, no |
|
|
|
|
| _FIXED = {} |
|
|
|
|
| class FixedScatter: |
| def __init__(self, idx, K, cap=8192): |
| h = to_host(idx).astype(np.int64).reshape(-1) |
| order = np.argsort(h, kind="stable") |
| counts = np.bincount(h, minlength=K) |
| starts = np.cumsum(counts) - counts |
| big = np.where(counts > cap)[0] |
| small = np.where(counts <= cap)[0] |
| self.K = K |
| self.order = to_dev(order, np.int64) if _GPU else order |
| self.big = [(int(b), int(starts[b]), int(starts[b]+counts[b])) |
| for b in big] |
| self.small = to_dev(small, np.int64) if _GPU else small |
| self.width = int(counts[small].max()) if len(small) else 0 |
| if self.width: |
| pos = np.concatenate([np.arange(counts[s]) for s in small]) |
| src = np.concatenate([np.arange(starts[s], starts[s]+counts[s]) |
| for s in small]) |
| row = np.repeat(np.arange(len(small)), counts[small]) |
| self.src = to_dev(src, np.int64) if _GPU else src |
| sl = row*self.width + pos |
| self.slot = to_dev(sl, np.int64) if _GPU else sl |
| self.buf = xp.zeros(len(small)*self.width, DT) |
| self._keep = idx |
|
|
| def __call__(self, g): |
| gs = g.reshape(-1)[self.order] |
| out = xp.zeros(self.K, DT) |
| if self.width: |
| self.buf[:] = 0 |
| self.buf[self.slot] = gs[self.src] |
| out[self.small] = self.buf.reshape(-1, self.width).sum(1) |
| for b, a, z in self.big: |
| out[b] = gs[a:z].sum() |
| return out |
|
|
|
|
| def scatter(dW, idx, K): |
| key = (id(idx), K) |
| if key not in _FIXED: |
| _FIXED[key] = FixedScatter(idx, K) |
| return _FIXED[key](dW) |
|
|
|
|
| def train_base(Xtr, Ytr, cfg, seed): |
| D, g, ch = Xtr.shape[1], cfg["grid"], cfg["chan"] |
| rg = np.random.default_rng(seed) |
| layers, cin = [], cfg["c_in"] |
| for l in range(cfg["depth"]): |
| idx, K, no = windowed(g, cin, 3, ch) |
| layers.append(dict(idx=to_dev(idx, np.int32) if _GPU else idx, |
| K=K, out=no, taps=cin*9, |
| ins=D if l == 0 else layers[-1]["out"])) |
| cin = ch |
| L = cfg["depth"] |
| P = [] |
| for l in layers: |
| v = rg.normal(0, np.sqrt(2.0/l["taps"]), l["K"]).astype(np.float32) |
| v[-1] = 0.0 |
| P.append(to_dev(v)) |
| P += [xp.ones(l["out"], DT) for l in layers] |
| P += [xp.zeros(l["out"], DT) for l in layers] |
| P += [to_dev(rg.normal(0, np.sqrt(2.0/layers[-1]["out"]), |
| (layers[-1]["out"], 10))), xp.zeros(10, DT)] |
| HEAD, OB = 3*L, 3*L+1 |
| M = [xp.zeros_like(p) for p in P]; V = [xp.zeros_like(p) for p in P] |
| n = Xtr.shape[0]; t = 0 |
| ag = np.random.default_rng(seed + 991) |
|
|
| def fwd(x): |
| cache = []; h = x |
| for li, l in enumerate(layers): |
| W = P[li][l["idx"]].reshape(l["ins"], l["out"]) |
| z = h @ W |
| var = z.var(1, keepdims=True) + 1e-5 |
| zn = (z - z.mean(1, keepdims=True))/xp.sqrt(var) |
| zs = zn*P[L+li] + P[2*L+li] |
| a = xp.maximum(zs, 0) |
| cache.append((h, W, var, zn, zs)) |
| h = a |
| return h, cache |
|
|
| for ep in range(cfg["epochs"]): |
| perm = ag.permutation(n) |
| for st in range(0, n, cfg["batch"]): |
| b = perm[st:st+cfg["batch"]] |
| x = Xtr[b]; y = Ytr[b] |
| h, cache = fwd(x) |
| lg = h @ P[HEAD] + P[OB] |
| 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[HEAD] = h.T @ d; G[OB] = d.sum(0) |
| dh = d @ P[HEAD].T |
| for li in range(L-1, -1, -1): |
| hin, W, var, zn, zs = cache[li] |
| dzs = dh*(zs > 0) |
| G[L+li] = (dzs*zn).sum(0); G[2*L+li] = dzs.sum(0) |
| dzn = dzs*P[L+li] |
| dz = (dzn - dzn.mean(1, keepdims=True) |
| - zn*(dzn*zn).mean(1, keepdims=True))/xp.sqrt(var) |
| G[li] = scatter(hin.T @ dz, layers[li]["idx"], layers[li]["K"]) |
| if li > 0: |
| dh = dz @ W.T |
| t += 1 |
| for i, (p_, gr) in enumerate(zip(P, G)): |
| M[i] = 0.9*M[i] + 0.1*gr |
| V[i] = 0.999*V[i] + 0.001*gr*gr |
| P[i] = p_ - cfg["lr"]*(M[i]/(1-0.9**t)) \ |
| / (xp.sqrt(V[i]/(1-0.999**t))+1e-8) |
| return P, fwd, HEAD, OB, layers[-1]["out"] |
|
|
|
|
| def ridge(A, R, lam): |
| """The closed-form member: one solve, no steps. |
| |
| Fitted on the RESIDUAL between the target and what the base already |
| says, so a member that explains nothing contributes nothing — and as |
| lambda dominates the solution shrinks toward zero and the base answers. |
| That is the failsafe as arithmetic rather than as a rule.""" |
| n, d = A.shape |
| A1 = xp.concatenate([A, xp.ones((n, 1), DT)], 1) |
| if d + 1 <= n: |
| G = A1.T @ A1 + lam*xp.eye(d+1, dtype=DT) |
| W = xp.linalg.solve(G, A1.T @ R) |
| else: |
| |
| |
| |
| G = A1 @ A1.T + lam*xp.eye(n, dtype=DT) |
| W = A1.T @ xp.linalg.solve(G, R) |
| return W[:-1], W[-1] |
|
|
|
|
| def descent(A, R, lam, steps, lr): |
| """The same fit by gradient descent, for comparison. The closed form is |
| only worth having if it matches.""" |
| n, d = A.shape |
| W = xp.zeros((d, R.shape[1]), DT); b = xp.zeros(R.shape[1], DT) |
| M = [xp.zeros_like(W), xp.zeros_like(b)] |
| V = [xp.zeros_like(W), xp.zeros_like(b)] |
| for t in range(1, steps+1): |
| E = A @ W + b - R |
| G = [A.T @ E/n + lam*W/n, E.mean(0)] |
| for i, (p_, gr) in enumerate(zip([W, b], G)): |
| M[i] = 0.9*M[i] + 0.1*gr |
| V[i] = 0.999*V[i] + 0.001*gr*gr |
| upd = p_ - lr*(M[i]/(1-0.9**t))/(xp.sqrt(V[i]/(1-0.999**t))+1e-8) |
| if i == 0: |
| W = upd |
| else: |
| b = upd |
| return W, b |
|
|
|
|
| def load(cfg): |
| from tensorflow import keras |
| (a, b), (c, d) = keras.datasets.fashion_mnist.load_data() |
| X = np.concatenate([a, c]).astype(np.float32)/255.0 |
| y = np.concatenate([b, d]).ravel().astype(np.int64) |
| if cfg["grid"] != 28: |
| s = 28//cfg["grid"] |
| X = X.reshape(-1, cfg["grid"], s, cfg["grid"], s).mean(axis=(2, 4)) |
| rg = np.random.default_rng(0); p = rg.permutation(len(X)) |
| tr, te = p[:cfg["n_train"]], p[cfg["n_train"]:cfg["n_train"]+10000] |
| mu, sd = X[tr].mean(), X[tr].std()+1e-8 |
| f = lambda Z: ((Z-mu)/sd).reshape(len(Z), -1) |
| Y = np.zeros((len(tr), 10), np.float32); Y[np.arange(len(tr)), y[tr]] = 1 |
| return f(X[tr]), Y, y[tr], f(X[te]), y[te] |
|
|
|
|
|
|
|
|
| |
| |
| |
| |
| |
|
|
|
|
|
|
|
|
| CFG = dict(grid=14, c_in=1, chan=16, depth=3, n_train=20000, batch=128, |
| lr=1e-3, epochs=30, seed=0, |
| base_classes=(0, 1, 2, 3, 4, 5, 6), |
| new_classes=(7, 8, 9), |
| n_holdout=4000, N=200, lam=1.0, draws=5, |
| |
| |
| |
| base_seeds=(0, 1, 2), |
| coverage=(0.2, 0.4, 0.6, 0.8, 1.0)) |
|
|
|
|
| def auc_of(score, pos): |
| """Probability a positive ranks above a negative.""" |
| r = np.argsort(np.argsort(score)) |
| n1, n0 = pos.sum(), (~pos).sum() |
| if n1 == 0 or n0 == 0: |
| return float("nan") |
| return float((r[pos].sum() - n1*(n1-1)/2)/(n1*n0)) |
|
|
|
|
| def conf_of(lg): |
| e = np.exp(lg - lg.max(1, keepdims=True)) |
| return (e/e.sum(1, keepdims=True)).max(1) |
|
|
|
|
| def risk_coverage(conf, correct, cover): |
| """Accuracy on the most confident fraction — what abstention buys.""" |
| order = np.argsort(-conf) |
| out = [] |
| for c in cover: |
| k = max(1, int(c*len(conf))) |
| out.append(float(correct[order[:k]].mean())) |
| return out |
|
|
|
|
| def main(**over): |
| CFG.update(over) |
| t0 = time.time() |
| print("=" * 78) |
| print("A MEMBER THAT CHANGES ONLY THE CONFIDENCE") |
| print("=" * 78) |
| print(f" backend: {'cupy (GPU)' if _GPU else 'numpy (CPU)'}") |
| for k, v in CFG.items(): |
| print(f" {k:14s} = {v}") |
| print(f"\n the base is trained only on {CFG['base_classes']}; a positive") |
| print(f" scale on its logits cannot change the argmax, so accuracy is") |
| print(f" invariant BY CONSTRUCTION and only confidence can move") |
| print("=" * 78, flush=True) |
|
|
| X, Y, y, Xte, yte = load(CFG) |
| old_m = np.isin(y, CFG["base_classes"]) |
| tr_i = np.where(old_m)[0] |
| |
| |
| |
| ho_new = np.where(~old_m)[0][:CFG["n_holdout"]//2] |
| ho_old = tr_i[-(CFG["n_holdout"]//2):] |
| tr_i = tr_i[:-(CFG["n_holdout"]//2)] |
| ho_i = np.concatenate([ho_old, ho_new]) |
| Xtr, Ytr = to_dev(X[tr_i]), to_dev(Y[tr_i]) |
| Xho = to_dev(X[ho_i]); yho = y[ho_i] |
| unseen_ho = np.isin(yho, CFG["new_classes"]) |
| Xte_d = to_dev(Xte) |
|
|
| A_auc, A_err, A_acc, A_rc = [], [], [], [] |
| B_auc_fam, B_rc_fam, B_auc_all = [], [], [] |
| base_seen_unseen, base_fam_rw, rc0_all, rc0_fam_all = [], [], [], [] |
| for bs in CFG["base_seeds"]: |
| P, fwd, HEAD, OB, width = train_base(Xtr, Ytr, CFG, bs) |
| fho, _ = fwd(Xho); fte, _ = fwd(Xte_d) |
| base_te = to_host(fte @ P[HEAD] + P[OB]) |
| base_ho = to_host(fho @ P[HEAD] + P[OB]) |
| unseen = np.isin(yte, CFG["new_classes"]) |
| pred = base_te.argmax(1); correct = pred == yte |
| c0 = conf_of(base_te) |
| fam_ho = ~unseen_ho |
| wrong_ho = base_ho.argmax(1) != yho |
| fam_te = ~unseen |
| base_seen_unseen.append(auc_of(-c0, unseen)) |
| base_fam_rw.append(auc_of(-c0[fam_te], ~correct[fam_te])) |
| rc0_all.append(risk_coverage(c0, correct, CFG["coverage"])) |
| rc0_fam_all.append(risk_coverage(c0[fam_te], correct[fam_te], |
| CFG["coverage"])) |
| print(f" base seed {bs}: {float(correct[~unseen].mean()):.4f} on its " |
| f"seven, confidence separates seen from unseen at AUC " |
| f"{base_seen_unseen[-1]:.4f} [{time.time()-t0:.0f}s]", |
| flush=True) |
| for dr in range(CFG["draws"]): |
| rg = np.random.default_rng(4000 + 83*bs + dr) |
| sub = rg.choice(len(ho_i), min(CFG["N"], len(ho_i)), replace=False) |
| sd_ = to_dev(sub, np.int64) if _GPU else sub |
| tgt = to_dev(np.where(unseen_ho[sub], 1.0, -1.0) |
| .astype(np.float32)[:, None]) |
| W, b = ridge(fho[sd_], tgt, CFG["lam"]) |
| sc = to_host(fte @ W + b).ravel() |
| scaled = base_te/np.exp(np.clip(sc, -4, 4))[:, None] |
| c1 = conf_of(scaled) |
| A_auc.append(auc_of(-c1, unseen)) |
| A_err.append(auc_of(-c1, ~correct)) |
| A_acc.append(float((scaled.argmax(1) == yte).mean()) |
| - float(correct.mean())) |
| A_rc.append(risk_coverage(c1, correct, CFG["coverage"])) |
| fi = np.where(fam_ho)[0] |
| rgb = np.random.default_rng(6000 + 83*bs + dr) |
| subb = rgb.choice(fi, min(CFG["N"], len(fi)), replace=False) |
| sb = to_dev(subb, np.int64) if _GPU else subb |
| tb = to_dev(np.where(wrong_ho[subb], 1.0, -1.0) |
| .astype(np.float32)[:, None]) |
| Wb, bb = ridge(fho[sb], tb, CFG["lam"]) |
| s2 = to_host(fte @ Wb + bb).ravel() |
| c2 = conf_of(base_te/np.exp(np.clip(s2, -4, 4))[:, None]) |
| B_auc_fam.append(auc_of(-c2[fam_te], ~correct[fam_te])) |
| B_auc_all.append(auc_of(-c2, unseen)) |
| B_rc_fam.append(risk_coverage(c2[fam_te], correct[fam_te], |
| CFG["coverage"])) |
| a0 = float(np.mean(base_seen_unseen)) |
| fb = float(np.mean(base_fam_rw)) |
| aucs, aucs_err, accs, rc_new = A_auc, A_err, A_acc, A_rc |
| print(f"\n {len(CFG['base_seeds'])} bases; their confidence separates") |
| print(f" seen from unseen at {min(base_seen_unseen):.4f} to " |
| f"{max(base_seen_unseen):.4f}, and right from wrong on FAMILIAR") |
| print(f" input at {min(base_fam_rw):.4f} to {max(base_fam_rw):.4f}") |
|
|
| rc0 = np.mean(rc0_all, 0) |
| rc1 = np.mean(A_rc, 0) |
| print("\n" + "=" * 78) |
| print(" DOES A MEMBER SUPPLY WHAT THE OUTPUT LAYER CANNOT?") |
| print("=" * 78) |
| print(f" {'':>28s} {'AUC':>8s} {'sd':>7s}") |
| print(f" {'base confidence, seen/unseen':>28s} {a0:8.4f} " |
| f"{np.std(base_seen_unseen):7.4f}") |
| print(f" {'scaled confidence, seen/unseen':>28s} {np.mean(aucs):8.4f} " |
| f"{np.std(aucs):7.4f}") |
| print(f" {'scaled confidence, right/wrong':>28s} " |
| f"{np.mean(aucs_err):8.4f} {np.std(aucs_err):7.4f}") |
| print(f"\n BUT THAT right/wrong FIGURE IS INFLATED: 40% of the test set") |
| print(f" is unseen and therefore wrong by construction. The honest") |
| print(f" question is errors on FAMILIAR input, where there is no") |
| print(f" category boundary to give them away:\n") |
| fb = auc_of(-c0[fam_te], ~correct[fam_te]) |
| print(f" {'':>34s} {'AUC':>8s} {'sd':>7s}") |
| print(f" {'base confidence, familiar right/wrong':>34s} {fb:8.4f}") |
| print(f" {'probe B on familiar right/wrong':>34s} " |
| f"{np.mean(B_auc_fam):8.4f} {np.std(B_auc_fam):7.4f}") |
| print(f" {'probe B also spotting the unseen':>34s} " |
| f"{np.mean(B_auc_all):8.4f} {np.std(B_auc_all):7.4f}") |
| print(f"\n accuracy change from the scale, across every base and " |
| f"draw: {np.abs(accs).max():.3e}") |
| print(f" (a positive scale cannot move an argmax; this is a bug check)") |
|
|
| print("\n" + "=" * 78) |
| print(" WHAT ABSTENTION BUYS") |
| print("=" * 78) |
| print(f" {'coverage':>9s} {'base':>9s} {'with member':>12s} " |
| f"{'gain':>8s}") |
| for i, c in enumerate(CFG["coverage"]): |
| print(f" {c:8.0%} {rc0[i]:9.4f} {rc1[i]:12.4f} " |
| f"{rc1[i]-rc0[i]:+8.4f}") |
|
|
| rcb = np.mean(B_rc_fam, 0) |
| rc0f = np.mean(rc0_fam_all, 0) |
| print(f"\n and abstention on FAMILIAR input alone, which is what a") |
| print(f" correction loop would actually run on:") |
| print(f" {'coverage':>9s} {'base':>9s} {'probe B':>12s} {'gain':>8s}") |
| for i, c in enumerate(CFG["coverage"]): |
| print(f" {c:8.0%} {rc0f[i]:9.4f} {rcb[i]:12.4f} " |
| f"{rcb[i]-rc0f[i]:+8.4f}") |
|
|
| print("\n" + "=" * 78) |
| print(" READOUT") |
| print("=" * 78) |
| fam_gain = np.mean(B_auc_fam) - fb |
| print(f" THERE ARE TWO BLIND SPOTS AND ONLY ONE IS BLIND.\n") |
| print(f" On FAMILIAR input the base already knows when it is unsure —") |
| print(f" AUC {fb:.4f} — and a {CFG['N']}-example probe on the same") |
| print(f" features reaches {np.mean(B_auc_fam):.4f} ({fam_gain:+.4f}). A") |
| print(f" member adds nothing there.") |
| print(f"\n On UNFAMILIAR input it has no idea: {a0:.4f}, below chance,") |
| print(f" and a member supplies {np.mean(aucs):.4f}.") |
| print(f"\n AND THE TWO DETECTORS DO NOT TRANSFER: probe B, fitted on") |
| print(f" familiar errors, spots the unseen at {np.mean(B_auc_all):.4f}.") |
| print(f" Knowing you are unsure and knowing you are out of your depth") |
| print(f" are different quantities living in different places.") |
| print(f"\n WHICH MEANS THE GATE NEVER FAILED FOR WANT OF A SIGNAL. The") |
| print(f" 0.5481 that made the base's confidence look useless was an") |
| print(f" aggregate over a test set 40% of which it had never seen. What") |
| print(f" killed the correction loop was the 6.6-to-1 arithmetic and a") |
| print(f" member that could only reach 17% of irreducible errors.") |
| print() |
| gain = np.mean(aucs) - a0 |
| sd = max(np.std(aucs), 1e-9) |
| if np.mean(aucs) > 0.7: |
| print(f" IGNORANCE IS VISIBLE ONE LAYER DOWN. The output layer") |
| print(f" separates seen from unseen at {a0:.4f} — below chance —") |
| print(f" and a {CFG['N']}-example linear read of the SAME frozen") |
| print(f" features reaches {np.mean(aucs):.4f}. The representation") |
| print(f" encodes unfamiliarity; the head was trained to name one of") |
| print(f" seven classes and never to admit it could not.") |
| print(f"\n And it costs nothing to try: a positive scale cannot") |
| print(f" change the argmax, so the failure mode of a bad confidence") |
| print(f" member is a worse confidence, never a worse answer.") |
| elif np.mean(aucs) > 0.6: |
| print(f" PARTLY VISIBLE: {np.mean(aucs):.4f} against the output") |
| print(f" layer's {a0:.4f}. Real signal, not enough to abstain on") |
| print(f" confidently.") |
| else: |
| print(f" NOT VISIBLE EITHER. A probe on the features reaches") |
| print(f" {np.mean(aucs):.4f}, so the representation does not encode") |
| print(f" unfamiliarity and no member can supply it. Ignorance is") |
| print(f" invisible from the inside at every level, which is a") |
| print(f" stronger and more useful claim than the output-layer") |
| print(f" result alone.") |
| print(f"\n total {time.time()-t0:.0f}s; wrote confidence_member.json") |
| json.dump(dict(base_auc=a0, member_auc=float(np.mean(aucs)), |
| member_auc_sd=float(np.std(aucs)), |
| err_auc=float(np.mean(aucs_err)), |
| rc_base=rc0, rc_member=list(map(float, rc1))), |
| open("confidence_member.json", "w"), indent=2) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
