281
个编辑
更改
跳到导航
跳到搜索
第253行：
第253行： + + + + + + + + + + + + + + + + + + + + + + + + + + + + 第264行：
第292行： + + + + + + + + + + + + + + + + 第273行：
第317行： − 工程中，使用随机梯度下降 SGD 来完成优化，另外当令 <math> θ_{i}^{−} = θ_{i−1} </math>时，退化为原始的 Q-Learning 更新形式+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
→DQN 的核心思想与结构
在论文中DQN所用的网络结构是三个卷积层加两个全连接层，网络的输入是经过预处理后的数据，预处理指的是将210*160的游戏图像转化为84*84的图像等操作，得到算法结构中的具体代码实现和网络结构见图和以下代码：
在论文中DQN所用的网络结构是三个卷积层加两个全连接层，网络的输入是经过预处理后的数据，预处理指的是将210*160的游戏图像转化为84*84的图像等操作，得到算法结构中的具体代码实现和网络结构见图和以下代码：
[[文件:网络结构.png|缩略图|网络具体结构]]
[[文件:网络结构.png|缩略图|网络具体结构]]
<syntaxhighlight lang="python">
# 预处理图像
def preprocess(frame):
# 转灰度图
frame = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)
# 调整大小
frame = cv2.resize(frame, (84, 84))
# 归一化
frame = frame.astype(np.float32) / 255.0
return frame
# Q-Network (CNN)
class DQN(nn.Module):
def __init__(self, action_dim):
super(DQN, self).__init__()
self.conv1 = nn.Conv2d(4, 32, kernel_size=8, stride=4)
self.conv2 = nn.Conv2d(32, 64, kernel_size=4, stride=2)
self.conv3 = nn.Conv2d(64, 64, kernel_size=3, stride=1)
self.fc1 = nn.Linear(64 * 7 * 7, 512)
self.fc2 = nn.Linear(512, action_dim)
def forward(self, x):
x = F.relu(self.conv1(x))
x = F.relu(self.conv2(x))
x = F.relu(self.conv3(x))
x = torch.flatten(x, start_dim=1)
x = F.relu(self.fc1(x))
</syntaxhighlight>
'''核心意义'''：
'''核心意义'''：
这种经验回放的技巧可以打破数据之间的关联性。
这种经验回放的技巧可以打破数据之间的关联性。
[[文件:经验回放.jpg|缩略图|经验回收]]
[[文件:经验回放.jpg|缩略图|经验回收]]
代码实现可参考：
<syntaxhighlight lang="python">
# 经验回放
class ReplayMemory:
def __init__(self, capacity):
self.memory = deque(maxlen=capacity)
def push(self, state, action, reward, next_state, done):
self.memory.append((state, action, reward, next_state, done))
def sample(self, batch_size):
return random.sample(self.memory, batch_size)
def __len__(self):
return len(self.memory)
</syntaxhighlight>
===目标网络（Target Network）的引入与参数更新策略===
===目标网络（Target Network）的引入与参数更新策略===
<math> ▽_{θ_i} L_i ( θ_i ) = E_{s , a , r , s ′ }[ ( r + γ max _{a ′ }Q ( s ′ , a ′ , θ_{i}^{−} ) − Q ( s , a , θ_{i }) ) ▽ θ_i Q ( s , a , θ_i ) ] </math>
<math> ▽_{θ_i} L_i ( θ_i ) = E_{s , a , r , s ′ }[ ( r + γ max _{a ′ }Q ( s ′ , a ′ , θ_{i}^{−} ) − Q ( s , a , θ_{i }) ) ▽ θ_i Q ( s , a , θ_i ) ] </math>
工程中，使用随机梯度下降来完成优化，另外当令 <math> θ_{i}^{−} = θ_{i−1} </math>时，退化为原始的 Q-Learning 更新形式。
DQN的训练与参数更新策略参考以下代码实现：
<syntaxhighlight lang="python">
# DQN Agent
class DQNAgent:
def __init__(self, action_dim):
self.action_dim = action_dim
self.memory = ReplayMemory(10000)
self.policy_net = DQN(action_dim).to(device)
self.target_net = DQN(action_dim).to(device)
self.target_net.load_state_dict(self.policy_net.state_dict())
self.optimizer = optim.Adam(self.policy_net.parameters(), lr=0.0001)
self.exploration_rate = 1.0
self.steps_done = 0
def select_action(self, state):
# ϵ-贪心策略
if random.random() < self.exploration_rate:
return random.randrange(self.action_dim)
else:
with torch.no_grad():
return self.policy_net(state).argmax().item()
def train(self):
# 样本不足时直接返回
if len(self.memory) < 64:
return
# 从经验回放中采样
batch = self.memory.sample(64)
states, actions, rewards, next_states, dones = zip(*batch)
# 转为张量，并去除多余维度
states = torch.tensor(np.array(states), dtype=torch.float32).to(device) # shape: (64, 1, 4, 84, 84)
states = states.squeeze(1) # shape: (64, 4, 84, 84)
actions = torch.tensor(actions, dtype=torch.int64).unsqueeze(1).to(device) # shape: (64, 1)
rewards = torch.tensor(rewards, dtype=torch.float32).to(device) # shape: (64,)
next_states = torch.tensor(np.array(next_states), dtype=torch.float32).to(device) # shape: (64, 1, 4, 84, 84)
next_states = next_states.squeeze(1) # shape: (64, 4, 84, 84)
dones = torch.tensor(dones, dtype=torch.bool).to(device) # shape: (64,)
# 计算当前 Q 值
q_values = self.policy_net(states).gather(1, actions).squeeze()
# 计算下一时刻最大 Q 值
next_q_values = self.target_net(next_states).max(1)[0].detach()
# 计算目标 Q 值
target_q_values = rewards + (0.99 * next_q_values * ~dones)
# 优化
loss = F.mse_loss(q_values, target_q_values)
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
# 更新探索率
if self.exploration_rate > 0.01:
self.exploration_rate *= 0.995
def update_target_net(self):
self.target_net.load_state_dict(self.policy_net.state_dict())
# 训练循环
env = gym.make('Pong-v4', frameskip=4)
action_dim = env.action_space.n
agent = DQNAgent(action_dim)
episodes = 500000
total_rewards = []
for episode in range(episodes):
state, info = env.reset()
state = preprocess(state)
# 初始化 4 帧堆叠
state_stack = deque([state] * 4, maxlen=4)
# 构造当前状态张量：(1,4,84,84)
state_tensor = torch.tensor(np.array(state_stack), dtype=torch.float32).unsqueeze(0).to(device)
total_reward = 0
done = False
while not done:
# 根据策略选择动作
action = agent.select_action(state_tensor)
next_state, reward, done, truncated, info = env.step(action)
done = done or truncated
# 预处理后压入堆叠
next_state = preprocess(next_state)
state_stack.append(next_state)
next_state_tensor = torch.tensor(np.array(state_stack), dtype=torch.float32).unsqueeze(0).to(device)
# 存储到经验回放
agent.memory.push(state_tensor.cpu().numpy(), action, reward, next_state_tensor.cpu().numpy(), done)
# 训练一步
agent.train()
# 更新状态
state_tensor = next_state_tensor
total_reward += reward
# 定期更新目标网络
if episode % 10 == 0:
agent.update_target_net()
total_rewards.append(total_reward)
print(f"Episode {episode}: Total Reward = {total_reward}")
env.close()
</syntaxhighlight>
==DQN在pytorch中的简单完整实现==
==DQN在pytorch中的简单完整实现==