import json, math, random, time from pathlib import Path import numpy as np import torch from torch import nn from sklearn.datasets import make_moons from sklearn.model_selection import train_test_split SEED = 636 random.seed(SEED); np.random.seed(SEED); torch.manual_seed(SEED) try: if torch.cuda.is_available(): device = torch.device('cuda') else: device = torch.device('cpu') except Exception: device = torch.device('cpu') def cosine(a, b): den = np.linalg.norm(a)*np.linalg.norm(b) return float(np.dot(a,b)/den) if den > 1e-12 else 0.0 def scores(updates, projection): """Updates is a list of (n_chunks, parameter_dim) arrays, one tensor per layer. Projection puts layer/chunk directions in one common low-dimensional space.""" local = [] directions = [] for U in updates: V = np.asarray([projection @ x for x in U]) V = V / np.maximum(np.linalg.norm(V, axis=1, keepdims=True), 1e-12) local.extend(cosine(V[i], V[i+1]) for i in range(len(V)-1)) directions.append(np.mean(V, axis=0)) u = np.asarray(directions) u = u / np.maximum(np.linalg.norm(u, axis=1, keepdims=True), 1e-12) mean = u.mean(axis=0) G = ((u-mean).T @ (u-mean))/len(u) eig = np.linalg.eigvalsh(G) tr = float(np.trace(G)) C = float(eig[-1]/tr) if tr > 1e-12 else 0.0 return float(np.mean(local)), C def controller_eta(eta, Q, C, kappa=1.0, C0=.45, q0=.3): # Formula from the proposal, with the stated high-Q gate. if Q > q0 and C < C0: return eta * math.exp(-kappa*(1-Q)*max(C0-C, 0.0)) return eta def math_check(): # Verify the controller on the smallest clear construction: each block has # internally aligned neighboring updates (Q high), while four blocks point # into distinct orthogonal global sectors (C low). rng = np.random.default_rng(4) d = 4 sectors = np.eye(d) competing = [np.tile(sectors[i], (4, 1)) + .001*rng.normal(size=(4, d)) for i in range(4)] Q, C = scores(competing, np.eye(d)) eta0 = 1.; eta1 = controller_eta(eta0, Q, C, kappa=2., C0=.8) # A single global sector has negligible dispersion; the covariance ratio is # conventionally set to zero when its trace vanishes. aligned = [np.tile(np.array([1., 0, 0, 0]), (4, 1)) + .001*rng.normal(size=(4, d)) for _ in range(4)] Qa, Ca = scores(aligned, np.eye(d)) return {'Q_competing': Q, 'C_competing': C, 'eta_ratio_competing': eta1/eta0, 'Q_aligned': Qa, 'C_aligned': Ca, 'eta_ratio_aligned': controller_eta(1, Qa, Ca, C0=.8), 'controller_triggered_on_competing': bool(eta1 < eta0)} class Net(nn.Module): def __init__(self): super().__init__() self.layers = nn.ModuleList([nn.Linear(2,32), nn.Linear(32,32), nn.Linear(32,2)]) def forward(self,x): x=torch.tanh(self.layers[0](x)); x=torch.tanh(self.layers[1](x)); return self.layers[2](x) def run(controlled, steps=900): X,y=make_moons(n_samples=1800, noise=.22, random_state=SEED) X=(X-X.mean(0))/X.std(0) xt,xv,yt,yv=train_test_split(X,y,test_size=.3,random_state=SEED,stratify=y) x=torch.tensor(xt,dtype=torch.float32,device=device); y=torch.tensor(yt,dtype=torch.long,device=device) xv=torch.tensor(xv,dtype=torch.float32,device=device); yv=torch.tensor(yv,dtype=torch.long,device=device) torch.manual_seed(SEED) model=Net().to(device) opt=torch.optim.SGD(model.parameters(),lr=.12,momentum=.9) params=list(model.parameters()) # A shared random sketch makes block directions comparable despite different tensor sizes. rng=np.random.default_rng(99); d=18 projections=[rng.normal(0,1/math.sqrt(d),(d,p.numel()//p.shape[0] if p.ndim>1 else p.numel())) for p in params] # For each layer, combine weight and bias rows into equal-sized row chunks by flattening tensor. history=[]; eta_trace=[]; q_trace=[]; c_trace=[]; loss_trace=[] batch=96 for step in range(steps): ix=torch.randint(0,len(x),(batch,),device=device) before=[p.detach().clone() for p in params] opt.zero_grad(); loss=nn.functional.cross_entropy(model(x[ix]),y[ix]); loss.backward(); opt.step() if step % 30 == 29: updates=[] for li,(p,old) in enumerate(zip(params,before)): delta=(p.detach()-old).cpu().numpy().reshape(-1) # four neighboring temporal-like shards within each tensor; local consistency is adjacent shards. chunks=np.array_split(delta,4) # project with a fresh deterministic resize matrix for each chunk to common coordinates U=[] for ci,ch in enumerate(chunks): rr=np.random.default_rng(1000+li*10+ci) R=rr.normal(0,1/math.sqrt(d),(d,len(ch))) U.append(R@ch) updates.append(np.asarray(U)) Q,C=scores(updates,np.eye(d)) eta=opt.param_groups[0]['lr'] # Baseline cosine schedule is applied every diagnostic interval; controller adds coarsening slowdown. base=.12*(.015 + .985*0.5*(1+math.cos(math.pi*(step+1)/steps))) neweta=base if controlled and Q>.3 and C<.45: neweta=controller_eta(base,Q,C,kappa=2.5,C0=.45,q0=.3) for g in opt.param_groups:g['lr']=neweta q_trace.append(Q); c_trace.append(C); eta_trace.append(neweta) with torch.no_grad(): vl=nn.functional.cross_entropy(model(xv),yv).item(); acc=(model(xv).argmax(1)==yv).float().mean().item() history.append({'step':step+1,'loss':vl,'acc':acc,'Q':Q,'C':C,'lr':neweta}) return {'final_acc':history[-1]['acc'],'final_loss':history[-1]['loss'], 'mean_Q':float(np.mean(q_trace)),'mean_C':float(np.mean(c_trace)), 'highQ_lowC_fraction':float(np.mean([(q>.3 and c<.45) for q,c in zip(q_trace,c_trace)])), 'min_lr':float(min(eta_trace)), 'records':history} def main(): math_result=math_check() # Separate process-level failure fallback is handled by rerunning on CPU if CUDA errors. try: base=run(False); idea=run(True) except Exception as e: global device device=torch.device('cpu') base=run(False); idea=run(True) math_result['cuda_error']=repr(e) out={'device':str(device),'math_check':math_result,'baseline':base,'idea':idea} Path('results.json').write_text(json.dumps(out,indent=2)) print(json.dumps({'device':str(device),'math_check':math_result, 'baseline':{k:v for k,v in base.items() if k!='records'}, 'idea':{k:v for k,v in idea.items() if k!='records'}},indent=2)) if __name__=='__main__': main()