Noise-Whitened Trajectory-KL Policy Regularization / experiment.py
Failed on benchmark
1import json, math, os
2import numpy as np
3import torch
4
5SEED = 1267
6np.random.seed(SEED)
7torch.manual_seed(SEED)
8try:
9 device = 'cuda' if torch.cuda.is_available() else 'cpu'
10 if device == 'cuda':
11 torch.cuda.set_device(0)
12 _ = torch.zeros(1, device='cuda')
13except Exception:
14 device = 'cpu'
15
16
17def solve_local(R, B, a, g, lam, u0=None, bdiff=None):
18 """Exact minimizer of 1/2 u'Ru + g'u + lam/2 (b+Bu-b0-Bu0)'a^-1(...)."""
19 R, B, a, g = map(np.asarray, (R, B, a, g))
20 n = R.shape[0]
21 u0 = np.zeros(n) if u0 is None else np.asarray(u0)
22 bdiff = np.zeros(B.shape[0]) if bdiff is None else np.asarray(bdiff)
23 ai = np.linalg.pinv(a)
24 H = R + lam * B.T @ ai @ B
25 rhs = -g - lam * B.T @ ai @ (bdiff - B @ u0)
26 return np.linalg.solve(H, rhs), H
27
28
29def toy_verification():
30 # Two drift coordinates with equal actuation but very different diffusion.
31 R = np.eye(2)
32 B = np.eye(2)
33 a = np.diag([0.25, 4.0])
34 g = np.array([-2.0, -2.0])
35 u0 = np.zeros(2)
36 lambdas = np.array([0., .1, .3, 1., 3., 10., 30., 100.])
37 us = np.array([solve_local(R, B, a, g, x, u0)[0] for x in lambdas])
38 weights = np.diag(np.linalg.pinv(a))
39 drift_kl = .5 * np.sum((us @ B.T) ** 2 * weights, axis=1)
40 # Prediction 1: KL decreases monotonically as lambda increases.
41 monotone = bool(np.all(np.diff(drift_kl) <= 1e-10))
42 # Prediction 2: each coordinate's deviation has asymptotic lambda*u -> -a*g.
43 tail = lambdas >= 10
44 asymp_expected = -np.diag(a) * g
45 asymp_observed = np.mean(us[tail] * lambdas[tail, None], axis=0)
46 rel_asymp_err = float(np.max(np.abs(asymp_observed - asymp_expected) /
47 (np.abs(asymp_expected) + 1e-12)))
48 # Prediction 3: regularization curvature ratio equals inverse-noise ratio.
49 curv0 = np.diag(R)
50 curv_lam = np.diag(R + 1.0 * B.T @ np.linalg.pinv(a) @ B)
51 observed_ratio = curv_lam[0] / curv_lam[1]
52 predicted_ratio = (1 + 1/a[0, 0]) / (1 + 1/a[1, 1])
53 return {
54 'lambdas': lambdas.tolist(), 'solutions': us.tolist(), 'trajectory_kl': drift_kl.tolist(),
55 'prediction_1_kl_monotone': monotone,
56 'prediction_2_expected_lambda_u_tail': asymp_expected.tolist(),
57 'prediction_2_observed_lambda_u_tail': asymp_observed.tolist(),
58 'prediction_2_relative_error': rel_asymp_err,
59 'prediction_3_predicted_curvature_ratio': float(predicted_ratio),
60 'prediction_3_observed_curvature_ratio': float(observed_ratio),
61 'prediction_3_relative_error': float(abs(observed_ratio-predicted_ratio)/predicted_ratio),
62 }
63
64
65def finite_difference_check():
66 R = np.diag([1.3, .7]); B = np.array([[1., .2], [.1, 1.]])
67 a = np.diag([.4, 2.5]); g = np.array([-.8, .3]); lam = 2.7
68 u, H = solve_local(R, B, a, g, lam)
69 def f(x):
70 d = B @ x
71 return .5*x@R@x + g@x + .5*lam*d@np.linalg.pinv(a)@d
72 eps = 1e-5
73 grad = np.array([(f(u+eps*np.eye(2)[i])-f(u-eps*np.eye(2)[i]))/(2*eps) for i in range(2)])
74 hess = np.empty((2,2))
75 for i in range(2):
76 for j in range(2):
77 ei, ej = np.eye(2)[i]*eps, np.eye(2)[j]*eps
78 hess[i,j] = (f(u+ei+ej)-f(u+ei-ej)-f(u-ei+ej)+f(u-ei-ej))/(4*eps*eps)
79 return {'stationarity_gradient_norm': float(np.linalg.norm(grad)),
80 'hessian_error_norm': float(np.linalg.norm(hess-H)), 'analytic_u': u.tolist()}
81
82
83def train_policy():
84 # Same synthetic states and task for action-norm baseline versus noise-whitened drift KL.
85 torch.manual_seed(SEED)
86 n, d, act = 4096, 3, 2
87 x = torch.randn(n, d, device=device)
88 target = x @ torch.tensor([[1.0, -.4], [.2, .8], [-.7, .3]], device=device)
89 # State-dependent but known diagonal diffusion; one direction is highly noisy.
90 var = torch.tensor([.2, 4.0], device=device)
91 B = torch.eye(act, device=device)
92 lam = 2.0
93 def fit(kind):
94 torch.manual_seed(SEED + (0 if kind == 'action' else 1))
95 W = torch.nn.Parameter(torch.zeros(d, act, device=device))
96 opt = torch.optim.Adam([W], lr=.08)
97 for _ in range(700):
98 u = x @ W
99 task = .5 * ((u-target)**2).mean()
100 if kind == 'action':
101 reg = .5 * (u**2).sum(1).mean()
102 else:
103 reg = .5 * ((u**2)/var).sum(1).mean()
104 loss = task + lam*reg
105 opt.zero_grad(); loss.backward(); opt.step()
106 with torch.no_grad():
107 u = x @ W
108 task = .5*((u-target)**2).mean().item()
109 action_reg = .5*(u**2).sum(1).mean().item()
110 traj_kl = .5*((u**2)/var).sum(1).mean().item()
111 return {'task_mse_half': task, 'action_penalty': action_reg,
112 'trajectory_kl': traj_kl, 'W': W.detach().cpu().numpy().tolist()}
113 return {'device': device, 'lambda': lam, 'action_regularization': fit('action'),
114 'noise_whitened_drift_regularization': fit('drift')}
115
116
117def main():
118 out = {'seed': SEED, 'math_check': finite_difference_check(),
119 'toy_predictions': toy_verification(), 'policy_experiment': train_policy()}
120 with open('results.json', 'w') as f: json.dump(out, f, indent=2)
121 print(json.dumps(out, indent=2))
122
123if __name__ == '__main__':
124 main()