深度强化学习中Double DQN算法（Q-Learning+CNN）的讲解及在Asterix游戏上的实战（超详细附源码）

梦里梦外; 2023-09-30 17:11 3阅读 0赞

> **需要源码和环境搭建请点赞关注收藏后评论区留下QQ~~~**

## 一、核心思想 ##

针对DQN中出现的高估问题，有人提出深度双Q网络算法（DDQN），该算法是将强化学习中的双Q学习应用于DQN中。在强化学习中，双Q学习的提出能在一定程度上缓解Q学习带来的过高估计问题。

DDQN的主要思想是在目标值计算时将动作的选择和评估分离，在更新过程中，利用两个网络来学习两组权重，分别是预测网络的权重W和目标网络的权重W'，在DQN中，动作选择和评估都是通过目标网络来实现的，而在DDQN中，计算目标Q值时，采取目标网络获取最优动作，再通过预测网络估计该最优动作的目标Q值，这样就可以将最优动作选额和动作值函数估计分离，采用不用的样本保证独立性

## 二、允许结果与分析 ##

本节实验在Asterix游戏上，通过控制参数变量对DQN,DDQN算法进行性能对比从而验证了在一定程度上，DDQN算法可以缓解DQN算法的高估问题，DDQN需要两个不同的参数网络，每1000步后预测预测网络的参数同步更新给目标网络，实验设有最大能容纳1000000记录的缓冲池，每个Atari游戏，DDQN算法训练1000000时间步

实战结果如下图所示，图中的DDQN算法最后收敛回报明显大于DQN，并且在实验过程中，可以发现DQN算法容易陷入局部的情况，其问题主要在于Q-Learning中的最大化操作，Agent在选择动作时每次都取最大Q值得动作，对于真实的策略来说，在给定的状态下并不是每次都选择Q值最大的动作，因为一般真实的策略都是随机性策略，所以在这里目标值直接选择动作最大的Q值往往会导致目标值高于真实值

![c0393aa71d9d4052aee579f31680c295.png][]

为了解决值函数高估计的问题，DDQN算法将动作的选择和动作的评估分别用不同的值函数来实现，结果表明DDQN能够估计出更准确的Q值，在一些Atari2600游戏中可获得更稳定有效的策略

## 三、代码 ##

部分源码如下

import gym, random, pickle, os.path, math, glob
    
    import numpy as np
    import pandas as pd
    import matplotlib.pyplot as plt
    
    import torch
    import torch.optim as optim
    import torch.nn as nn
    import torch.nn.functional as F
    import torch.autograd as autograd
    import pdb
    
    from atari_wrappers import make_atari, wrap_deepmind,LazyFrames
    
        def __init__(self, in_channels=4, num_actions=5):
           nnels: number of channel of input.
                    i.e The number of most recent frames stacked together as describe in the paper
                num_actions: number of action-value to output, one-to-one correspondence to action in game.
            """
            super(DQN, self).__init__()
            self.conv1 = nn.Conv2d(in_channels, 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.fc4 = nn.Linear(7 * 7 * 64, 512)
            self.fc5 = nn.Linear(512, num_actions)
    
        def forward(self, x):
            x = F.relu(self.conv1(x))
            x = F.relu(self.conv2(x))
            x = F.relu(self.conv3(x))
            x = F.relu(self.fc4(x.view(x.size(0), -1)))
            return self.fc5(x)
    
    class Memory_Buffer(object):
        def __init__(self, memory_size=1000):
            self.buffer = []
            self.memory_size = memory_size
            self.next_idx = 0
    
        def push(self, state, action, reward, next_state, done):
            data = (state, action, reward, next_state, done)
            if len(self.buffer) <= self.memory_size: # buffer not full
                self.buffer.append(data)
            else: # buffer is full
                self.buffer[self.next_idx] = data
            self.next_idx = (self.next_idx + 1) % self.memory_size
    
        def sample(self, batch_size):
            states, actions, rewards, next_states, dones = [], [], [], [], []
            for i in range(batch_size):
                idx = random.randint(0, self.size() - 1)
                data = self.buffer[idx]
                state, action, reward, next_state, done= data
                states.append(state)
                actions.append(action)
                rewards.append(reward)
                next_states.append(next_state)
                dones.append(done)
    
    
            return np.concatenate(states), actions, rewards, np.concatenate(next_states), dones
    
        def size(self):
            return len(self.buffer)
    
    class DDQNAgent:
        def __init__(self, in_channels = 1, action_space = [], USE_CUDA = False, memory_size = 10000, epsilon  = 1, lr = 1e-4):
            self.epsilon = epsilon
            self.action_space = action_space
            self.memory_buffer = Memory_Buffer(memory_size)
            self.DQN = DQN(in_channels = in_channels, num_actions = action_space.n)
            self.DQN_target = DQN(in_channels = in_channels, num_actions = action_space.n)
            self.DQN_target.load_state_dict(self.DQN.state_dict())
    
    
            self.USE_CUDA = USE_CUDA
            if USE_CUDA:
                self.DQN = self.DQN.to(device)
                self.DQN_target = self.DQN_target.to(device)
            self.optimizer = optim.RMSprop(self.DQN.parameters(),lr=lr, eps=0.001, alpha=0.95)
    
        def observe(self, lazyframe):
            # from Lazy frame to tensor
            state =  torch.from_numpy(lazyframe._force().transpose(2,0,1)[None]/255).float()
            if self.USE_CUDA:
                state = state.to(device)
            return state
    
        def value(self, state):
            q_values = self.DQN(state)
            return q_values
    
        def act(self, state, epsilon = None):
            """
            sample actions with epsilon-greedy policy
            recap: with p = epsilon pick random action, else pick action with highest Q(s,a)
            """
            if epsilon is None: epsilon = self.epsilon
    
            q_values = self.value(state).cpu().detach().numpy()
            if random.random()<epsilon:
                aciton = random.randrange(self.action_space.n)
            else:
                aciton = q_values.argmax(1)[0]
            return aciton
    
        def compute_td_loss(self, states, actions, rewards, next_states, is_done, gamma=0.99):
            """ Compute td loss using torch operations only. Use the formula above. """
            actions = torch.tensor(actions).long()    # shape: [batch_size]
            rewards = torch.tensor(rewards, dtype =torch.float)  # shape: [batch_size]
            is_done = torch.tensor(is_done, dtype = torch.uint8)  # shape: [batch_size]
    
            if self.USE_CUDA:
                actions = actions.to(device)
                rewards = rewards.to(device)
                is_done = is_done.to(device)
    
            # get q-values for all actions in current states
            predicted_qvalues = self.DQN(states)
    
            # select q-values for chosen actions
            predicted_qvalues_for_actions = predicted_qvalues[
              range(states.shape[0]), actions
            ]
    
            # compute q-values for all actions in next states
            ## Where DDQN is different from DQN
            predicted_next_qvalues_current = self.DQN(next_states)
            predicted_next_qvalues_target = self.DQN_target(next_states)
            # compute V*(next_states) using predicted next q-values
            next_state_values =  predicted_next_qvalues_target.gather(1, torch.max(predicted_next_qvalues_current, 1)[1].unsqueeze(1)).squeeze(1)
    
            # compute "target q-values" for loss - it's what's inside square parentheses in the above formula.
            target_qvalues_for_actions = rewards + gamma *next_state_values
    
            # at the last state we shall use simplified formula: Q(s,a) = r(s,a) since s' doesn't exist
            target_qvalues_for_actions = torch.where(
                is_done, rewards, target_qvalues_for_actions)
    
            # mean squared error loss to minimize
            #loss = torch.mean((predicted_qvalues_for_actions -
            #                   target_qvalues_for_actions.detach()) ** 2)
            loss = F.smooth_l1_loss(predicted_qvalues_for_actions, target_qvalues_for_actions.detach())
    
            return loss
    
        def sample_from_buffer(self, batch_size):
            states, actions, rewards, next_states, dones = [], [], [], [], []
            for i in range(batch_size):
                idx = random.randint(0, self.memory_buffer.size() - 1)
                data = self.memory_buffer.buffer[idx]
                frame, action, reward, next_frame, done= data
                states.append(self.observe(frame))
                actions.append(action)
                rewards.append(reward)
                next_states.append(self.observe(next_frame))
                dones.append(done)
            return torch.cat(states), actions, rewards, torch.cat(next_states), dones
    
        def learn_from_experience(self, batch_size):
            if self.memory_buffer.size() > batch_size:
                states, actions, rewards, next_states, dones = self.sample_from_buffer(batch_size)
                td_loss = self.compute_td_loss(states, actions, rewards, next_states, dones)
                self.optimizer.zero_grad()
                td_loss.backward()
                for param in self.DQN.parameters():
                    param.grad.data.clamp_(-1, 1)
    
                self.optimizer.step()
                return(td_loss.item())
            else:
                return(0)
    
    def moving_average(a, n=3) :
        ret = np.cumsum(a, dtype=float)
        ret[n:] = ret[n:] - ret[:-n]
        return ret[n - 1:] / n
    
    def plot_training(frame_idx, rewards, losses):
        clear_output(True)
        plt.figure(figsize=(20,5))
        plt.subplot(131)
        plt.title('frame %s. reward: %s' % (frame_idx, np.mean(rewards[-100:])))
        plt.plot(moving_average(rewards,20))
        plt.subplot(132)
        plt.title('loss, average on 100 stpes')
        plt.plot(moving_average(losses, 100),linewidth=0.2)
        plt.show()
    
    # if __name__ == '__main__':
    
    # Training DQN in PongNoFrameskip-v4
    env = make_atari('PongNoFrameskip-v4')
    env = wrap_deepmind(env, scale = False, frame_stack=True)
    
    gamma = 0.99
    epsilon_max = 1
    epsilon_min = 0.01
    eps_decay = 30000
    frames = 1000000
    USE_CUDA = True
    learning_rate = 2e-4
    max_buff = 100000
    update_tar_interval = 1000
    batch_size = 32
    print_interval = 1000
    log_interval = 1000
    learning_start = 10000
    win_reward = 18     # Pong-v4
    win_break = True
    
    action_space = env.action_space
    action_dim = env.action_space.n
    state_dim = env.observation_space.shape[0]
    state_channel = env.observation_space.shape[2]
    agent = DDQNAgent(in_channels = state_channel, action_space= action_space, USE_CUDA = USE_CUDA, lr = learning_rate)
    
    #frame = env.reset()
    
    episode_reward = 0
    all_rewards = []
    losses = []
    episode_num = 0
    is_win = False
    # tensorboard
    summary_writer = SummaryWriter(log_dir = "DDQN", comment= "good_makeatari")
    
    # e-greedy decay
    epsilon_by_frame = lambda frame_idx: epsilon_min + (epsilon_max - epsilon_min) * math.exp(
                -1. * frame_idx / eps_decay)
    # plt.plot([epsilon_by_frame(i) for i in range(10000)])
    
    for i in range(frames):
        epsilon = epsilon_by_frame(i)
        #state_tensor = agent.observe(frame)
        #action = agent.act(state_tensor, epsilon)
    
        #next_frame, reward, done, _ = env.step(action)
    
        #episode_reward += reward
        #agent.memory_buffer.push(frame, action, reward, next_frame, done)
        #frame = next_frame
    
        loss = 0
        if agent.memory_buffer.size() >= learning_start:
            loss = agent.learn_from_experience(batch_size)
            losses.append(loss)
    
        if i % print_interval == 0:
            print("frames: %5d, reward: %5f, loss: %4f, epsilon: %5f, episode: %4d" % (i, np.mean(all_rewards[-10:]), loss, epsilon, episode_num))
            summary_writer.add_scalar("Temporal Difference Loss", loss, i)
            summary_writer.add_scalar("Mean Reward", np.mean(all_rewards[-10:]), i)
            summary_writer.add_scalar("Epsilon", epsilon, i)
    
        if iQN_dict.pth.tar")
    plot_training(i, all_rewards, losses)

**创作不易 觉得有帮助请点赞关注收藏~~~**

[c0393aa71d9d4052aee579f31680c295.png]: https://img-blog.csdnimg.cn/c0393aa71d9d4052aee579f31680c295.png