Weighted Resolvent-Equivariant Attention / experiment.py

Unverified

Raw ⬇ ZIP
  1import json, math, random
  2from pathlib import Path
  3import numpy as np
  4
  5SEED = 2327
  6np.random.seed(SEED)
  7random.seed(SEED)
  8
  9
 10def frob(x):
 11    return float(np.linalg.norm(x, 'fro'))
 12
 13
 14def comm(a, b):
 15    return a @ b - b @ a
 16
 17
 18def run_math():
 19    n = 6
 20    sigma = np.arange(n)[::-1]
 21    kappa = np.array([1.0, 1.7, 0.8, 1.3, 2.0, 0.6])
 22    G = np.zeros((n, n))
 23    G[np.arange(n), sigma] = kappa
 24
 25    # A matrix polynomial in G commutes exactly with weighted reflection.
 26    P0 = np.eye(n) + 0.055 * G + 0.009 * (G @ G)
 27    rng = np.random.default_rng(SEED + 1)
 28    E = rng.normal(size=(n, n))
 29    E /= frob(E)
 30    C0 = frob(comm(G, E))
 31
 32    # Prediction 1: ||[G,P0+eps E]|| is exactly linear in eps.
 33    epses = np.array([0.0, .001, .003, .01, .03, .1, .3])
 34    cvals = np.array([frob(comm(G, P0 + e * E)) for e in epses])
 35    slope = float(np.dot(epses, cvals) / max(np.dot(epses, epses), 1e-30))
 36    linear_rel_err = float(np.max(np.abs(cvals - epses * C0)) / max(cvals[-1], 1e-30))
 37
 38    # Prediction 2: exact commutation propagates to powers and resolvents.
 39    power_errors = []
 40    Ppow = np.eye(n)
 41    for t in range(0, 9):
 42        power_errors.append(frob(comm(G, Ppow)))
 43        Ppow = Ppow @ P0
 44    gamma0 = 0.37
 45    R0 = np.linalg.inv(np.eye(n) - (1 - gamma0) * P0)
 46    resolvent_zero = frob(comm(G, R0))
 47
 48    # Prediction 3: [G,R] = (1-gamma) R [G,P] R; it vanishes at gamma=1
 49    # and is approximately linear in alpha=1-gamma near alpha=0.
 50    P1 = P0 + 0.08 * E
 51    gammas = np.array([0.0, .1, .25, .5, .7, .85, .95, 1.0])
 52    rvals, identity_errors = [], []
 53    for gamma in gammas:
 54        alpha = 1.0 - gamma
 55        R = np.linalg.inv(np.eye(n) - alpha * P1)
 56        lhs = comm(G, R)
 57        rhs = alpha * R @ comm(G, P1) @ R
 58        rvals.append(frob(lhs))
 59        identity_errors.append(frob(lhs-rhs) / max(frob(lhs), 1e-30))
 60    rvals = np.array(rvals)
 61    alpha = 1-gammas
 62    near = alpha <= .3
 63    through_origin = float(np.dot(alpha[near], rvals[near]) / max(np.dot(alpha[near], alpha[near]), 1e-30))
 64    # report normalized values so the damping trend is easy to inspect
 65    normalized = (rvals / max(rvals[0], 1e-30)).tolist()
 66
 67    return {
 68        'weighted_G': G.tolist(),
 69        'prediction_1_linear_perturbation': {
 70            'eps': epses.tolist(), 'observed_commutators': cvals.tolist(),
 71            'predicted_slope_norm_[G,E]': C0, 'fit_slope': slope,
 72            'max_relative_deviation': linear_rel_err
 73        },
 74        'prediction_2_exact_multi_step': {
 75            'power_commutator_norms_t0_to_8': power_errors,
 76            'resolvent_commutator_norm_gamma_0.37': resolvent_zero
 77        },
 78        'prediction_3_damping_resolvent': {
 79            'gamma': gammas.tolist(), 'observed_norms': rvals.tolist(),
 80            'normalized_to_gamma_0': normalized,
 81            'resolvent_identity_relative_errors': identity_errors,
 82            'near_gamma_1_fit_slope_vs_(1-gamma)': through_origin,
 83            'gamma_1_norm': float(rvals[-1])
 84        }
 85    }
 86
 87
 88def attention_trial(comm_lambda, seed, steps=180):
 89    import torch
 90    torch.manual_seed(seed); np.random.seed(seed)
 91    device = 'cuda' if torch.cuda.is_available() else 'cpu'
 92    try:
 93        dev = torch.device(device)
 94        N, L, D = 768, 8, 8
 95        x = torch.randn(N, L, D, device=dev)
 96        # Reversal-invariant signal, with deliberately unequal noise scales by side.
 97        latent = x[:, :, 0].sum(1) + .35 * x[:, :, 1].sum(1)
 98        x[:, :L//2] += .45 * torch.randn(N, L//2, D, device=dev)
 99        x[:, L//2:] += 1.25 * torch.randn(N, L//2, D, device=dev)
100        y = (latent > 0).long()
101        tr, va = torch.arange(0, 600, device=dev), torch.arange(600, N, device=dev)
102        q = torch.nn.Linear(D, D, bias=False).to(dev)
103        k = torch.nn.Linear(D, D, bias=False).to(dev)
104        v = torch.nn.Linear(D, D, bias=False).to(dev)
105        out = torch.nn.Linear(D, 2).to(dev)
106        opt = torch.optim.Adam(list(q.parameters())+list(k.parameters())+list(v.parameters())+list(out.parameters()), lr=.012)
107        rev = torch.arange(L-1, -1, -1, device=dev)
108        kap = torch.tensor([1., 1.35, .85, 1.15, 1.15, .85, 1.35, 1.], device=dev)
109        G = torch.zeros(L,L,device=dev); G[torch.arange(L),rev] = kap
110        last_loss = None
111        for _ in range(steps):
112            xb, yb = x[tr], y[tr]
113            P = torch.softmax(q(xb) @ k(xb).transpose(1,2) / math.sqrt(D), dim=-1)
114            z = (P @ v(xb)).mean(1)
115            loss = torch.nn.functional.cross_entropy(out(z), yb)
116            GP = G.unsqueeze(0) @ P; PG = P @ G.unsqueeze(0)
117            penalty = ((GP-PG)**2).sum() / (G.square().sum()*P.square().sum().mean()+1e-8)
118            total = loss + comm_lambda * penalty
119            opt.zero_grad(); total.backward(); opt.step(); last_loss=float(loss.detach())
120        with torch.no_grad():
121            P = torch.softmax(q(x[va]) @ k(x[va]).transpose(1,2) / math.sqrt(D), dim=-1)
122            pred = out((P @ v(x[va])).mean(1)).argmax(1)
123            acc = float((pred == y[va]).float().mean())
124            ce = float(torch.nn.functional.cross_entropy(out((P @ v(x[va])).mean(1)), y[va]))
125            cm = torch.linalg.vector_norm(G@P-P@G).item() / math.sqrt(len(va))
126        return {'accuracy':acc, 'val_loss':ce, 'commutator_rms':cm, 'device':str(dev), 'train_loss':last_loss}
127    except Exception as exc:
128        # CUDA can be shared; retrying on CPU is part of the experiment contract.
129        if device == 'cuda':
130            torch.cuda.empty_cache()
131            old = torch.cuda.is_available
132            # A clean CPU subprocess is simpler than mutating torch's device state.
133            return attention_trial_cpu(comm_lambda, seed, steps)
134        raise
135
136
137def attention_trial_cpu(comm_lambda, seed, steps):
138    import torch
139    # Force CPU by temporarily using a CPU-only equivalent implementation.
140    # The compact fallback duplicates the computation with tensors on CPU.
141    torch.manual_seed(seed); np.random.seed(seed)
142    N,L,D=768,8,8; dev=torch.device('cpu')
143    x=torch.randn(N,L,D); latent=x[:,:,0].sum(1)+.35*x[:,:,1].sum(1)
144    x[:,:4]+=.45*torch.randn(N,4,D); x[:,4:]+=1.25*torch.randn(N,4,D); y=(latent>0).long()
145    q=torch.nn.Linear(D,D,bias=False); k=torch.nn.Linear(D,D,bias=False); v=torch.nn.Linear(D,D,bias=False); out=torch.nn.Linear(D,2)
146    opt=torch.optim.Adam(list(q.parameters())+list(k.parameters())+list(v.parameters())+list(out.parameters()),lr=.012)
147    rev=torch.arange(7,-1,-1); kap=torch.tensor([1.,1.35,.85,1.15,1.15,.85,1.35,1.]); G=torch.zeros(L,L); G[torch.arange(L),rev]=kap
148    for _ in range(steps):
149        P=torch.softmax(q(x[:600])@k(x[:600]).transpose(1,2)/math.sqrt(D),-1); z=(P@v(x[:600])).mean(1); loss=torch.nn.functional.cross_entropy(out(z),y[:600]); pen=((G@P-P@G)**2).sum()/(G.square().sum()*P.square().sum().mean()+1e-8); opt.zero_grad(); (loss+comm_lambda*pen).backward(); opt.step()
150    with torch.no_grad():
151        P=torch.softmax(q(x[600:])@k(x[600:]).transpose(1,2)/math.sqrt(D),-1); logits=out((P@v(x[600:])).mean(1)); return {'accuracy':float((logits.argmax(1)==y[600:]).float().mean()),'val_loss':float(torch.nn.functional.cross_entropy(logits,y[600:])),'commutator_rms':float(torch.linalg.vector_norm(G@P-P@G).item()/math.sqrt(168)),'device':'cpu','train_loss':float(loss)}
152
153
154def main():
155    math_result=run_math()
156    attention={}
157    for lam in (0.0, 1.0):
158        vals=[attention_trial(lam,s) for s in (11,22)]
159        attention[str(lam)]={'seeds':vals,'mean_accuracy':float(np.mean([v['accuracy'] for v in vals])),'mean_val_loss':float(np.mean([v['val_loss'] for v in vals])),'mean_commutator_rms':float(np.mean([v['commutator_rms'] for v in vals]))}
160    result={'seed':SEED,'math':math_result,'attention_mini_experiment':attention}
161    Path('results.json').write_text(json.dumps(result,indent=2))
162    print(json.dumps(result,indent=2))
163
164if __name__=='__main__': main()