### 使用PyTorch实现双重深度Q网络(DDQN) #### 创建环境和初始化 为了构建一个基于PyTorch的DDQN模型，首先需要定义好使用的环境以及一些必要的超参数设置。这通常涉及到选择合适的学习率、折扣因子γ以及其他可能影响训练过程的因素。 ```python import gymnasium as gym import torch import torch.nn.functional as F from collections import deque import random import numpy as np env = gym.make('CartPole-v1') state_size = env.observation_space.shape[0] action_size = env.action_space.n learning_rate = 0.001 gamma = 0.99 epsilon_decay = 0.995 min_epsilon = 0.01 batch_size = 64 memory_capacity = 20000 update_target_frequency = 1000 ``` #### 构建神经网络结构 接下来是设计两个几乎相同的评估网络`eval_net`用于预测当前状态下的动作价值函数；目标网络`target_net`则用来提供稳定的目标值，在每次更新权重之前保持不变一段时间以减少数据间的关联性[^2]。 ```python class Net(torch.nn.Module): def __init__(self, state_dim, action_dim): super().__init__() self.fc1 = torch.nn.Linear(state_dim, 50) self.out = torch.nn.Linear(50, action_dim) def forward(self, x): x = F.relu(self.fc1(x)) actions_value = self.out(x) return actions_value eval_net, target_net = Net(state_size, action_size), Net(state_size, action_size) optimizer = torch.optim.Adam(eval_net.parameters(), lr=learning_rate) loss_func = torch.nn.MSELoss() ``` #### 经验回放缓冲区 引入经验重放技术来打破样本之间的顺序依赖关系，并提高学习效率。通过随机采样过去的经验批次来进行梯度下降优化。 ```python class ReplayBuffer(object): def __init__(self, capacity): self.buffer = deque(maxlen=capacity) def push(self, state, action, reward, next_state, done): state = np.expand_dims(state, 0) next_state = np.expand_dims(next_state, 0) self.buffer.append((state, action, reward, next_state, done)) def sample(self, batch_size): state, action, reward, next_state, done = zip(*random.sample(self.buffer, batch_size)) return np.concatenate(state), action, reward, np.concatenate(next_state), done def __len__(self): return len(self.buffer) replay_buffer = ReplayBuffer(memory_capacity) ``` #### 训练循环与策略执行 最后一步是在环境中不断尝试新的行为并记录下这些经历以便后续分析调整。每当累积了一定量的新鲜记忆后就从中抽取一批次作为输入给定至上述提到过的两个网络中去计算损失进而反向传播求解最优解路径直至满足终止条件为止。 ```python def select_action(state, epsilon): if np.random.rand() <= epsilon: return random.randrange(action_size) else: with torch.no_grad(): q_values = eval_net.forward(torch.FloatTensor([state])) return int(q_values.argmax().item()) total_steps = 0 for episode in range(1000): state = env.reset()[0] eps_reward = 0 while True: total_steps += 1 # 动作选取 action = select_action(state, max(min_epsilon, (1 - epsilon_decay ** total_steps))) # 执行动作获取反馈 next_state, reward, terminated, truncated, info = env.step(action) done = terminated or truncated replay_buffer.push(state, action, reward, next_state, done) state = next_state eps_reward += reward if len(replay_buffer) >= batch_size * 10 and total_steps % update_target_frequency == 0: # 抽取mini-batch states, actions, rewards, next_states, dones = replay_buffer.sample(batch_size) # 转换成tensor形式 states_tensor = torch.tensor(states).float() actions_tensor = torch.tensor(actions).long() rewards_tensor = torch.tensor(rewards).float() next_states_tensor = torch.tensor(next_states).float() non_final_mask = ~torch.tensor(dones).bool() current_q_values = eval_net(states_tensor).gather(dim=-1, index=actions_tensor.unsqueeze(-1)).squeeze(-1) next_max_actions = eval_net(next_states_tensor).argmax(axis=1)[non_final_mask].unsqueeze(-1) next_q_values = torch.zeros_like(current_q_values) next_q_values[non_final_mask] = target_net(next_states_tensor).gather( dim=-1, index=next_max_actions ).squeeze(-1) expected_q_values = rewards_tensor + gamma * next_q_values loss = loss_func(current_q_values, expected_q_values.detach()) optimizer.zero_grad() loss.backward() optimizer.step() # 更新目标网络 if total_steps % update_target_frequency == 0: target_net.load_state_dict(eval_net.state_dict()) if done: break print(f"Episode {episode}: Total Reward={eps_reward}") ```