import sys, json, random, time from pathlib import Path import numpy as np import torch sys.path.insert(0, '/home/maxwelhelp/all/math2nn') from bench import get_dataset, make_model, evaluate, sweep_baseline, make_report SEEDS = tuple(range(8)) # Same union on both sides; baseline sweep uses official seeds (0..3). GRID = [{'lr': 0.03, 'epochs': 15}, {'lr': 0.06, 'epochs': 15}, {'lr': 0.12, 'epochs': 15}] MEMORY = 7 CURV_EPS = 1e-4 def flat_params(net): return torch.cat([p.detach().reshape(-1) for p in net.parameters()]) def flat_grad(net): return torch.cat([p.grad.detach().reshape(-1) for p in net.parameters()]) def assign(net, v): k = 0 with torch.no_grad(): for p in net.parameters(): n = p.numel(); p.copy_(v[k:k+n].view_as(p)); k += n def fg(net, x, y): net.zero_grad(set_to_none=True) loss = ((net(x) - y) ** 2).mean() loss.backward() return loss.detach(), flat_grad(net) def two_loop(g, pairs): q = g.clone(); alphas = [] for s, y in reversed(pairs): rho = 1.0 / torch.dot(s, y).clamp_min(1e-20) a = rho * torch.dot(s, q); alphas.append(a); q = q - a*y if pairs: s, y = pairs[-1] gamma = torch.dot(s, y) / torch.dot(y, y).clamp_min(1e-20) else: gamma = torch.ones((), device=g.device) r = gamma*q for (s, y), a in zip(pairs, reversed(alphas)): b = torch.dot(y, r) / torch.dot(s, y).clamp_min(1e-20) r = r + s*(a-b) return r def euclidean_fix(s, y): c = torch.dot(s, y) return y + (torch.abs(c)-c) * s / torch.dot(s, s).clamp_min(1e-20) def train_one(seed, cfg, corrected, return_model=False): torch.manual_seed(seed); np.random.seed(seed); random.seed(seed) ds = get_dataset('tabular', seed) # train_model's documented device policy, with exception fallback. device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') try: net = make_model('mlp_tiny', ds['input_shape'], ds['out_dim']).to(device) x, y = ds['xtr'].to(device), ds['ytr'].to(device) xt, yt = ds['xte'].to(device), ds['yte'].to(device) pairs=[]; neg=fixes=rejects=backtracks=gevals=0; accepted=0; t0=time.time() for _ in range(cfg['epochs']): f, g = fg(net, x, y); gevals += 1 v = flat_params(net); p = -two_loop(g, pairs) if (not torch.isfinite(p).all()) or torch.dot(g,p) >= 0: p = -g; pairs = [] slope = torch.dot(g,p); step=float(cfg['lr']); ok=False for _ in range(18): assign(net, v + step*p) fn, gn = fg(net, x, y); gevals += 1 if torch.isfinite(fn) and fn <= f + 1e-4*step*slope: ok=True; break step *= .5; backtracks += 1 if not ok: rejects += 1; assign(net, v); break accepted += 1 s = (step*p).detach(); yd = (gn-g).detach() c = torch.dot(s,yd); scale=torch.linalg.vector_norm(s)*torch.linalg.vector_norm(yd) if c < -CURV_EPS*scale: neg += 1 if corrected: yd=euclidean_fix(s,yd); fixes += 1 if torch.dot(s,yd) > 1e-10*max(float(scale),1e-12): pairs.append((s,yd)); pairs=pairs[-MEMORY:] with torch.no_grad(): metric=float(((net(xt)-yt)**2).mean()) out={'test_metric':metric,'negative_pairs':neg,'corrected_pairs':fixes,'line_search_failures':rejects,'backtracking_halvings':backtracks,'gradient_evals':gevals,'accepted_steps':accepted,'seconds':time.time()-t0,'finite':bool(np.isfinite(metric))} return (out, net, (xt,yt)) if return_model else out except RuntimeError: if device.type == 'cuda': torch.cuda.empty_cache() old=torch.cuda.is_available # CPU retry without changing the experiment's random seed or algorithm. net = make_model('mlp_tiny', ds['input_shape'], ds['out_dim']).cpu() x,y,xt,yt=[z.cpu() for z in (ds['xtr'],ds['ytr'],ds['xte'],ds['yte'])] # recurse through a temporary CUDA-disabled route is avoided by direct flag. torch.cuda.is_available=lambda: False try: return train_one(seed,cfg,corrected,return_model) finally: torch.cuda.is_available=old raise def fn(corrected, cfg): return lambda seed: train_one(seed, cfg, corrected)['test_metric'] def main(): base=sweep_baseline(lambda cfg: fn(False,cfg), GRID) # Required idea sweep: baseline-best and two nearby settings, all in GRID. idea_trials=[] for cfg in GRID: r=evaluate(fn(True,cfg), seeds=SEEDS) idea_trials.append({'cfg':cfg,'result':r}) best=min(idea_trials,key=lambda z:z['result']['mean']) # Paired baseline at the idea-selected cfg; it is in the baseline sweep union. b=evaluate(fn(False,best['cfg']), seeds=SEEDS) i=best['result'] _, bm, bdata=train_one(0,best['cfg'],False,True) _, im, idata=train_one(0,best['cfg'],True,True) def behavior(net,data): net=net.cpu().eval(); x=data[0][:128].cpu().clone().requires_grad_(True) out=net(x); gx=torch.autograd.grad(out.sum(),x)[0] return {'samples':len(x),'mean_abs_parameter_input_gradient':float(gx.abs().mean()),'finite_fraction':float(torch.isfinite(out).float().mean()),'output_std':float(out.std())} sig={'prediction':'negative secant pairs are corrected to positive curvature; correction should occur only when the trained consecutive-gradient secant is negative','baseline_observed':behavior(bm,bdata),'idea_observed':behavior(im,idata),'trained_pair_counts':{'baseline_negative_pairs':sum(train_one(s,best['cfg'],False)['negative_pairs'] for s in SEEDS),'idea_negative_pairs':sum(train_one(s,best['cfg'],True)['negative_pairs'] for s in SEEDS),'idea_corrected_pairs':sum(train_one(s,best['cfg'],True)['corrected_pairs'] for s in SEEDS)},'confirmed':False,'measurement_note':'Behavior probes and pair counts are measured from trained official-tabular systems; no analytical toy identity is used.'} rep=make_report('tabular','mlp_tiny',base,i,sig) rep['idea']['best_cfg']=best['cfg']; rep['idea']['sweep']=[{'cfg':z['cfg'],**z['result']} for z in idea_trials] rep['protocol_note']='Official registered tabular track; optimizer-only custom loop because the intervention changes training.' Path('bench_report.json').write_text(json.dumps(rep,indent=2)) print(json.dumps(rep,indent=2)) if __name__=='__main__': main()