""" TELEPATHIC MNIST — SIMULATION ============================= A frozen random projection M_T sits IN FRONT of the activation function. Only the private (trainable) weights receive gradient updates. What's being tested: A tiny, frozen "telepathic" basis can carry enough structure that the MLP reaches high accuracy. If the basis is reseeded between training and inference (broken coupling), accuracy should collapse. Usage: python telepathic_mnist_sim.py --epochs 5 \ --tele-widths 2 4 8 16 32 64 128 \ --seeds 1729 4242 9001 Outputs: telepathy_sim_results.csv — per-(T, seed) accuracy telepathy_sim_plot.png — 3-panel figure telepathy_summary.txt — text table """ import argparse, math, time, random, csv from dataclasses import dataclass from typing import List import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from torch.utils.data import DataLoader from torchvision import datasets, transforms try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt HAS_MPL = True except Exception: HAS_MPL = False # ── Reproducibility helpers ────────────────────────────────────── def seed_all(s: int): random.seed(s) np.random.seed(s) torch.manual_seed(s) if torch.cuda.is_available(): torch.cuda.manual_seed_all(s) # ── The telepathic layer itself ────────────────────────────────── class TelepathicLayer(nn.Module): """ A linear projection that NEVER updates. Whatever activation comes after it will operate in this fixed basis — it is the layer 'in front of your activation layer'. """ def __init__(self, d_in: int, d_out: int, seed: int): super().__init__() g = torch.Generator().manual_seed(int(seed) & 0xFFFFFFFF) W = torch.randn(d_in, d_out, generator=g) / math.sqrt(d_in) self.weight = nn.Parameter(W, requires_grad=False) def forward(self, x): return x @ self.weight # no bias — pure projection @torch.no_grad() def reseed(self, new_seed: int): g = torch.Generator().manual_seed(int(new_seed) & 0xFFFFFFFF) shp = self.weight.shape W = torch.randn(shp[0], shp[1], generator=g) / math.sqrt(shp[0]) self.weight.data.copy_(W) # ── The MLP ────────────────────────────────────────────────────── class TelepathicMLP(nn.Module): """ 784 → W_in → [M_T FROZEN] → GELU → W_after → GELU → W_out → 10 """ def __init__(self, d_in: int = 784, d_hid: int = 256, d_tele: int = 8, d_out: int = 10, mt_seed: int = 1729, num_after: int = 2): super().__init__() self.W_in = nn.Linear(d_in, d_hid) # private self.tel = TelepathicLayer(d_hid, d_tele, mt_seed) # ← FROZEN self.after = nn.ModuleList( [nn.Linear(d_tele if i == 0 else d_hid, d_hid) for i in range(num_after)] ) self.W_out = nn.Linear(d_hid, d_out) # private n_all = sum(p.numel() for p in self.parameters()) n_tel = self.tel.weight.numel() self.n_total = n_all self.n_tele = n_tel self.n_private = n_all - n_tel self.tele_pct = 100.0 * n_tel / n_all def forward(self, x): h = self.W_in(x) h = self.tel(h) # ← IN FRONT OF ACTIVATION h = F.gelu(h) for layer in self.after: h = F.gelu(layer(h)) return self.W_out(h) def private_params(self): return [p for n, p in self.named_parameters() if "tel" not in n] # ── Data ───────────────────────────────────────────────────────── def make_loaders(root: str, batch: int = 256): tf = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)) ]) train = datasets.MNIST(root, train=True, download=True, transform=tf) test = datasets.MNIST(root, train=False, transform=tf) kw = dict(batch_size=batch, num_workers=2, pin_memory=True) return ( DataLoader(train, shuffle=True, drop_last=True, **kw), DataLoader(test, shuffle=False, **kw), ) # ── Eval ───────────────────────────────────────────────────────── @torch.no_grad() def evaluate(model: nn.Module, loader: DataLoader, device: str) -> float: model.eval() correct = total = 0 for x, y in loader: x = x.to(device).view(-1, 784) y = y.to(device) correct += (model(x).argmax(1) == y).sum().item() total += x.size(0) return correct / total # ── Train one run ──────────────────────────────────────────────── @dataclass class Run: d_tele: int mt_seed: int tele_pct: float n_total: int n_tele: int n_private: int train_acc: float test_acc: float test_acc_broken: float epochs: int elapsed_s: float def train_one(d_tele: int, mt_seed: int, epochs: int, lr: float, device: str, data_root: str, batch: int, broken_seed, verbose: bool = True) -> Run: """Independent random init for private weights (different seed).""" seed_all(mt_seed * 1009 + 7) train_loader, test_loader = make_loaders(data_root, batch=batch) model = TelepathicMLP(d_tele=d_tele, mt_seed=mt_seed).to(device) opt = torch.optim.AdamW(model.private_params(), lr=lr, weight_decay=1e-4) sched = torch.optim.lr_scheduler.CosineAnnealingLR( opt, T_max=epochs * len(train_loader)) t0 = time.time() for ep in range(1, epochs + 1): model.train() c = t = 0 loss_sum = 0.0 for x, y in train_loader: x = x.to(device).view(-1, 784) y = y.to(device) opt.zero_grad(set_to_none=True) logits = model(x) loss = F.cross_entropy(logits, y) loss.backward() opt.step() sched.step() c += (logits.argmax(1) == y).sum().item() t += x.size(0) loss_sum += loss.item() * x.size(0) tr = c / t te = evaluate(model, test_loader, device) if verbose: print(f" ep{ep}/{epochs} loss={loss_sum/t:.4f} " f"train={tr:.4f} test={te:.4f}") # Broken-coupling control if broken_seed is not None: backup = model.tel.weight.data.clone() model.tel.reseed(broken_seed) te_broken = evaluate(model, test_loader, device) model.tel.weight.data.copy_(backup) else: te_broken = float("nan") return Run( d_tele=d_tele, mt_seed=mt_seed, tele_pct=model.tele_pct, n_total=model.n_total, n_tele=model.n_tele, n_private=model.n_private, train_acc=tr, test_acc=te, test_acc_broken=te_broken, epochs=epochs, elapsed_s=time.time() - t0, ) # ── Sweep ──────────────────────────────────────────────────────── def sweep(widths: List[int], seeds: List[int], epochs: int, lr: float, device: str, data_root: str, batch: int, broken_seed, out_csv: str) -> List[Run]: rows: List[Run] = [] for T in widths: for s in seeds: print(f"\n[run] T={T:<4} mt_seed={s:<6} epochs={epochs} " f"device={device}") r = train_one(T, s, epochs, lr, device, data_root, batch, broken_seed) extra = (f" broken={r.test_acc_broken:.4f}" if not np.isnan(r.test_acc_broken) else "") print(f" ↳ test={r.test_acc:.4f}{extra}") rows.append(r) with open(out_csv, "w", newline="") as f: w = csv.writer(f) w.writerow(["d_tele","mt_seed","tele_pct","n_total","n_tele", "n_private","train_acc","test_acc", "test_acc_broken","epochs","elapsed_s"]) for r in rows: w.writerow([r.d_tele, r.mt_seed, f"{r.tele_pct:.4f}", r.n_total, r.n_tele, r.n_private, f"{r.train_acc:.4f}", f"{r.test_acc:.4f}", (f"{r.test_acc_broken:.4f}" if not np.isnan(r.test_acc_broken) else "nan"), r.epochs, f"{r.elapsed_s:.1f}"]) print(f"\n[csv] → {out_csv}") return rows # ── Report ─────────────────────────────────────────────────────── def summarise(rows: List[Run]) -> str: by_T = {} for r in rows: by_T.setdefault(r.d_tele, []).append(r) out = [ "=" * 72, "TELEPATHIC LAYER — IN FRONT OF ACTIVATION — sweep results", "=" * 72, f"{'T':>4} {'%':>6} {'test':>10} {'broken':>10} {'Δ(intact-broken)':>16}", "-" * 72, ] for T in sorted(by_T): rs = by_T[T] ta = np.array([r.test_acc for r in rs]) ba = np.array([r.test_acc_broken for r in rs]) out.append( f"{T:>4} {rs[0].tele_pct:>5.2f}% " f"{ta.mean():>10.4f} {ba.mean():>10.4f} " f"{(ta-ba).mean():>+16.4f}" ) out.append("=" * 72) drops = [(T, np.mean([r.test_acc - r.test_acc_broken for r in by_T[T]])) for T in sorted(by_T)] worst = max(drops, key=lambda x: x[1]) out.append(f"\n★ Peak semantic-gauge failure: T={worst[0]} Δ={worst[1]:+.4f}") out.append(" (i.e. the T whose frozen basis the private weights depend on most)") txt = "\n".join(out) print("\n" + txt) with open("telepathy_summary.txt", "w") as f: f.write(txt) return txt # ── Plot ───────────────────────────────────────────────────────── def plot(rows: List[Run], path: str): if not HAS_MPL: print("[plot] matplotlib missing – skipping"); return by_T = {} for r in rows: by_T.setdefault(r.d_tele, []).append(r) Ts = sorted(by_T) tele = [by_T[T][0].tele_pct for T in Ts] acc = np.array([[r.test_acc for r in by_T[T]] for T in Ts]) brk = np.array([[r.test_acc_broken for r in by_T[T]] for T in Ts]) fig, axes = plt.subplots(1, 3, figsize=(15, 4.5)) fig.suptitle("Telepathic layer in front of activation — MNIST", fontsize=13, y=1.02) # 1) acc vs T ax = axes[0] ax.errorbar(Ts, acc.mean(1), yerr=acc.std(1), marker='o', color='#ff6fa1', lw=2, capsize=3, label="Intact M_T") ax.errorbar(Ts, brk.mean(1), yerr=brk.std(1), marker='x', color='#6fd0ff', lw=2, ls='--', capsize=3, label="Reseeded M_T") ax.axhline(0.99, color='#6fffb1', ls=':', lw=1, label="99%") ax.set_xscale('log', base=2) ax.set_xticks(Ts) ax.set_xticklabels([str(t) for t in Ts]) ax.set_xlabel("T — telepathic width") ax.set_ylabel("test accuracy") ax.set_title("Accuracy vs tele-layer width") ax.legend(fontsize=9, loc='lower right') ax.grid(alpha=.3) # 2) acc vs % ax = axes[1] ax.plot(tele, acc.mean(1), 'o-', color='#ff6fa1', lw=2, label="intact") ax.plot(tele, brk.mean(1), 'x--', color='#6fd0ff', lw=2, label="broken") ax.axhline(0.99, color='#6fffb1', ls=':', lw=1) ax.set_xlabel("telepathic % of total params") ax.set_ylabel("test accuracy") ax.set_title("The 3% line") ax.legend(fontsize=9) ax.grid(alpha=.3) # 3) semantic-gauge failure ax = axes[2] delta = acc.mean(1) - brk.mean(1) colors = ['#ff6fa1' if d > 0.5 else '#6fd0ff' for d in delta] ax.bar([str(T) for T in Ts], delta, color=colors) ax.set_xlabel("telepathic width T") ax.set_ylabel("Δ acc (intact − broken)") ax.set_title("Semantic-gauge failure") ax.grid(alpha=.3, axis='y') fig.tight_layout() fig.savefig(path, dpi=120, bbox_inches='tight') print(f"[plot] → {path}") # ── Entrypoint ─────────────────────────────────────────────────── def main(): ap = argparse.ArgumentParser() ap.add_argument("--data-root", default="./data") ap.add_argument("--out-csv", default="./telepathy_sim_results.csv") ap.add_argument("--out-fig", default="./telepathy_sim_plot.png") ap.add_argument("--epochs", type=int, default=5) ap.add_argument("--lr", type=float, default=2e-3) ap.add_argument("--batch", type=int, default=256) ap.add_argument("--tele-widths", type=int, nargs="+", default=[2, 4, 8, 16, 32, 64, 128]) ap.add_argument("--seeds", type=int, nargs="+", default=[1729, 4242, 9001]) ap.add_argument("--broken-seed", type=int, default=31415) ap.add_argument("--cpu", action="store_true") args = ap.parse_args() device = "cuda" if torch.cuda.is_available() and not args.cpu else "cpu" print(f"[boot] device={device} torch={torch.__version__}") print(f"[boot] widths={args.tele_widths} seeds={args.seeds} " f"epochs={args.epochs} lr={args.lr}") rows = sweep(args.tele_widths, args.seeds, args.epochs, args.lr, device, args.data_root, args.batch, args.broken_seed, args.out_csv) summarise(rows) plot(rows, args.out_fig) if __name__ == "__main__": main()