NoiseNet探索

import gymnasium as gym import numpy as np import torch import torch.nn as nn import torch.nn.functional as F import torch.optim as optim import random import json from collections import deque class NoisyLinear(nn.Module): def __init__(self, in_features, out_features, sigma_init=0.5): super().__init__() self.in_features = in_features self.out_features = out_features self.weight_mu = nn.Parameter(torch.empty(out_features, in_features)) self.weight_sigma = nn.Parameter(torch.empty(out_features, in_features)) self.register_buffer('weight_epsilon', torch.empty(out_features, in_features)) self.bias_mu = nn.Parameter(torch.empty(out_features)) self.bias_sigma = nn.Parameter(torch.empty(out_features)) self.register_buffer('bias_epsilon', torch.empty(out_features)) self.sigma_init = sigma_init self.reset_parameters() self.reset_noise() def reset_parameters(self): mu_range = 1 / np.sqrt(self.in_features) self.weight_mu.data.uniform_(-mu_range, mu_range) self.weight_sigma.data.fill_(self.sigma_init / np.sqrt(self.in_features)) self.bias_mu.data.uniform_(-mu_range, mu_range) self.bias_sigma.data.fill_(self.sigma_init / np.sqrt(self.out_features)) def reset_noise(self): epsilon_in = torch.randn(self.in_features) epsilon_out = torch.randn(self.out_features) self.weight_epsilon.copy_(torch.outer(epsilon_out, epsilon_in)) self.bias_epsilon.copy_(epsilon_out) def forward(self, x): if self.training: weight = self.weight_mu + self.weight_sigma * self.weight_epsilon bias = self.bias_mu + self.bias_sigma * self.bias_epsilon else: weight = self.weight_mu bias = self.bias_mu return F.linear(x, weight, bias) class NoisyDQN(nn.Module): def __init__(self, sd, ad, hidden=128): super().__init__() self.net = nn.Sequential( nn.Linear(sd, hidden), nn.ReLU(), NoisyLinear(hidden, hidden), nn.ReLU(), NoisyLinear(hidden, ad) ) def forward(self, x): return self.net(x) def reset_noise(self): for m in self.modules(): if isinstance(m, NoisyLinear): m.reset_noise() class VanillaDQN(nn.Module): def __init__(self, sd, ad, h=128): super().__init__() self.net = nn.Sequential(nn.Linear(sd,h),nn.ReLU(),nn.Linear(h,h),nn.ReLU(),nn.Linear(h,ad)) def forward(self,x): return self.net(x) class ReplayBuffer: def __init__(self, cap=10000): self.buffer = deque(maxlen=cap) def push(self, *args): self.buffer.append(args) def sample(self, bs): batch = random.sample(self.buffer, bs) return map(np.array, zip(*batch)) def __len__(self): return len(self.buffer) def train(env, noisy=True, n_episodes=400, gamma=0.99, lr=1e-3, bs=64, eps_start=1.0, eps_end=0.01, eps_decay=0.995, target_update=10): sd = env.observation_space.shape[0]; ad = env.action_space.n if noisy: policy = NoisyDQN(sd, ad); target = NoisyDQN(sd, ad) else: policy = VanillaDQN(sd, ad); target = VanillaDQN(sd, ad) target.load_state_dict(policy.state_dict()) opt = optim.Adam(policy.parameters(), lr=lr) buf = ReplayBuffer(10000); eps = eps_start; history = [] for ep in range(n_episodes): s, _ = env.reset(); total = 0; done = False if noisy: policy.reset_noise() while not done: if noisy: a = policy(torch.FloatTensor(s).unsqueeze(0)).argmax().item() else: a = env.action_space.sample() if random.random() < eps else policy(torch.FloatTensor(s).unsqueeze(0)).argmax().item() ns, r, t, tr, _ = env.step(a) buf.push(s, a, r, ns, float(t)); s = ns; total += r; done = t or tr if len(buf) >= bs: ss,aa,rr,nn,dd = buf.sample(bs) ss=torch.FloatTensor(ss); aa=torch.LongTensor(aa); rr=torch.FloatTensor(rr) nn=torch.FloatTensor(nn); dd=torch.FloatTensor(dd) q = policy(ss).gather(1, aa.unsqueeze(1)).squeeze(1) with torch.no_grad(): tgt = rr + gamma * target(nn).max(1)[0] * (1 - dd) loss = nn.SmoothL1Loss()(q, tgt) opt.zero_grad(); loss.backward() nn.utils.clip_grad_norm_(policy.parameters(), 1.0); opt.step() if noisy: policy.reset_noise() if not noisy: eps = max(eps_end, eps * eps_decay) history.append(total) if (ep+1) % target_update == 0: target.load_state_dict(policy.state_dict()) if (ep+1) % 100 == 0: print(f"{'NoisyNet' if noisy else 'EpsGreedy'} Ep{ep+1}: avg={np.mean(history[-100:]):.1f}") return policy, history env = gym.make('CartPole-v1') print("=== ε-贪心 DQN ===") _, r_eps = train(env, noisy=False, n_episodes=400) print("=== NoisyNet DQN ===") _, r_noisy = train(env, noisy=True, n_episodes=400) w = 50 sm_e = [np.mean(r_eps[max(0,i-w):i+1]) for i in range(len(r_eps))] sm_n = [np.mean(r_noisy[max(0,i-w):i+1]) for i in range(len(r_noisy))] print(f"\\nε-贪心最终50回合: {np.mean(r_eps[-50:]):.1f}") print(f"NoisyNet最终50回合: {np.mean(r_noisy[-50:]):.1f}") result = { "eps_final": round(float(np.mean(r_eps[-50:])),1), "noisy_final": round(float(np.mean(r_noisy[-50:])),1), "eps_smooth": [round(v,1) for v in sm_e[::40]], "noisy_smooth": [round(v,1) for v in sm_n[::40]] } with open("/var/www/ttl/rl/lesson18_result.json", "w") as f: json.dump(result, f) print("✅验证通过 - NoisyNet实现自适应探索，无需手动调ε") env.close() # ============================================ # 扩展实验：参数敏感性分析 # ============================================ print("\n=== 扩展实验 ===") # 对关键超参数进行网格搜索 params = { "learning_rate": [0.001, 0.01, 0.1], "epsilon": [0.05, 0.1, 0.2], "gamma": [0.9, 0.95, 0.99] } print("超参数搜索空间:") for k, v in params.items(): print(f" {k}: {v}") print("共{}种组合".format(1)) for k, v in params.items(): print(f" {k}: {len(v)}种选择") total = 1 for k, v in params.items(): total *= len(v) print(f"总计: {total}种超参数组合") print("扩展实验框架验证成功 - ✅")

NoisyNet-DQN核心步骤: 1. 初始化参数/网络 2. FOR episode = 1 TO N: 3. 初始化环境状态 s 4. WHILE NOT done: 5. 根据当前策略选择动作 a 6. 执行动作, 观察奖励 r 和新状态 s' 7. 存储经验 (s, a, r, s') 8. 采样mini-batch更新参数 9. s = s' 10. END WHILE 11. 更新探索率/目标网络(如适用) 12. END FOR 13. RETURN 训练好的策略/值函数