DQN-深度Q網絡

深度Q網絡是用深度學習來解決強化中Q學習的問題，可以先了解一下Q學習的過程是一個怎樣的過程，實際上就是不斷的試錯，從試錯的經驗之中尋找最優解

關於Q學習，我看到一個非常好的例子，另外知乎上面也有相關的討論

其實早在13年的時候，deepmind出來了第一篇用深度學習來解決Q學習的問題的paper，那個時候deepmind還不夠火，和一般的Q學習不同的是，由於12年Alex率先用CNN解決圖像中的high level的語義的提取，deepmind也同時采用了CNN來直接對圖像進行特征提取，而非傳統的進行手工特征提取

我想從代碼的角度來看一下DQN是如何實現的

pytorcyh的代碼在官網上是有的，我也貼出了自己添加了注釋的代碼，以及寫一下自己的對於代碼的理解

# -*-coding:utf-8-*-
import gym
import math
import random
import numpy as np
import matplotlib
import matplotlib.pyplot as plt
from collections import namedtuple
from itertools import count
from PIL import Image

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import torchvision.transforms as T


env = gym.make('CartPole-v0').unwrapped

# set up matplotlib
is_ipython = 'inline' in matplotlib.get_backend()
if is_ipython:
    from IPython import display

plt.ion()

# if gpu is to be used
# device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

Transition = namedtuple('Transition',
                        ('state', 'action', 'next_state', 'reward'))  # 聲明一個name為Transition，里面的變量為以下的類似dict的


class ReplayMemory(object):

    def __init__(self, capacity):
        self.capacity = capacity
        self.memory = []
        self.position = 0

    def push(self, *args):
        """Saves a transition."""
        if len(self.memory) < self.capacity:
            self.memory.append(None)
        self.memory[self.position] = Transition(*args)
        self.position = (self.position + 1) % self.capacity

    def sample(self, batch_size):
        return random.sample(self.memory, batch_size)

    def __len__(self):  # 定義__len__以便於用len函數？
        return len(self.memory)


class DQN(nn.Module):

    def __init__(self):
        super(DQN, self).__init__()
        self.conv1 = nn.Conv2d(3, 16, kernel_size=5, stride=2)
        self.bn1 = nn.BatchNorm2d(16)
        self.conv2 = nn.Conv2d(16, 32, kernel_size=5, stride=2)
        self.bn2 = nn.BatchNorm2d(32)
        self.conv3 = nn.Conv2d(32, 32, kernel_size=5, stride=2)
        self.bn3 = nn.BatchNorm2d(32)
        self.head = nn.Linear(448, 2)

    def forward(self, x):
        x = F.relu(self.bn1(self.conv1(x)))
        x = F.relu(self.bn2(self.conv2(x)))
        x = F.relu(self.bn3(self.conv3(x)))
        return self.head(x.view(x.size(0), -1))


resize = T.Compose([T.ToPILImage(),
                    T.Resize(40, interpolation=Image.CUBIC),
                    T.ToTensor()])

# This is based on the code from gym.
screen_width = 600


def get_cart_location():
    world_width = env.x_threshold * 2
    scale = screen_width / world_width
    return int(env.state[0] * scale + screen_width / 2.0)  # MIDDLE OF CART


def get_screen():
    screen = env.render(mode='rgb_array').transpose(
        (2, 0, 1))  # transpose into torch order (CHW)
    # Strip off the top and bottom of the screen
    screen = screen[:, 160:320]
    view_width = 320
    cart_location = get_cart_location()
    if cart_location < view_width // 2:
        slice_range = slice(view_width)
    elif cart_location > (screen_width - view_width // 2):
        slice_range = slice(-view_width, None)
    else:
        slice_range = slice(cart_location - view_width // 2,
                            cart_location + view_width // 2)
    # Strip off the edges, so that we have a square image centered on a cart
    screen = screen[:, :, slice_range]
    # Convert to float, rescare, convert to torch tensor
    # (this doesn't require a copy)
    screen = np.ascontiguousarray(screen, dtype=np.float32) / 255
    screen = torch.from_numpy(screen)
    # Resize, and add a batch dimension (BCHW)
    return resize(screen).unsqueeze(0).cuda()


env.reset()
# plt.figure()
# plt.imshow(get_screen().cpu().squeeze(0).permute(1, 2, 0).numpy(),
#            interpolation='none')
# plt.title('Example extracted screen')
# plt.show()
BATCH_SIZE = 128
GAMMA = 0.999
EPS_START = 0.9
EPS_END = 0.05
EPS_DECAY = 200
TARGET_UPDATE = 10

policy_net = DQN().cuda()
target_net = DQN().cuda()
target_net.load_state_dict(policy_net.state_dict())
target_net.eval()

optimizer = optim.RMSprop(policy_net.parameters())
memory = ReplayMemory(10000)


steps_done = 0


def select_action(state):
    global steps_done
    sample = random.random()
    eps_threshold = EPS_END + (EPS_START - EPS_END) * \
        math.exp(-1. * steps_done / EPS_DECAY)
    steps_done += 1
    if sample > eps_threshold:
        with torch.no_grad():
            return policy_net(state).max(1)[1].view(1, 1)  # policy網絡的輸出
    else:
        return torch.tensor([[random.randrange(2)]], dtype=torch.long).cuda()  # 隨機的選擇一個網絡的輸出或者


episode_durations = []


def plot_durations():
    plt.figure(2)
    plt.clf()
    durations_t = torch.tensor(episode_durations, dtype=torch.float)
    plt.title('Training...')
    plt.xlabel('Episode')
    plt.ylabel('Duration')
    plt.plot(durations_t.numpy())
    # Take 100 episode averages and plot them too
    if len(durations_t) >= 100:
        means = durations_t.unfold(0, 100, 1).mean(1).view(-1)
        means = torch.cat((torch.zeros(99), means))
        plt.plot(means.numpy())

    plt.pause(0.001)  # pause a bit so that plots are updated
    if is_ipython:
        display.clear_output(wait=True)
        display.display(plt.gcf())


def optimize_model():
    if len(memory) < BATCH_SIZE:
        return
    transitions = memory.sample(BATCH_SIZE)  # 進行隨機的sample，序列問題是不存在的
    # print(transitions)
    # Transpose the batch (see http://stackoverflow.com/a/19343/3343043 for
    # detailed explanation).
    batch = Transition(*zip(*transitions))
    # print("current")
    # print(batch.state[0])
    # print("next")
    # print(batch.next_state[0])
    # print(torch.sum(batch.state[0]))
    # print(torch.sum(batch.next_state[0]))
    # print(torch.sum(batch.state[1]))
    # # print(type(batch))
    # print("@#$%^&*")

    # Compute a mask of non-final states and concatenate the batch elements
    non_final_mask = torch.tensor(tuple(map(lambda s: s is not None, batch.next_state)), dtype=torch.uint8).cuda()  # lambda表達式返回的是否為空的二值
    non_final_next_states = torch.cat([s for s in batch.next_state if s is not None])  # 空的不cat，所以長度不一定是batchsize
    # print("the non_final_mask is")
    # print(non_final_mask)
    # none_total = 0
    # total = 0
    # for s in batch.next_state:
    #     if s is None:
    #         none_total = none_total + 1
    #     else:
    #         total = total + 1
    # print(none_total, total)
    state_batch = torch.cat(batch.state)
    action_batch = torch.cat(batch.action)
    reward_batch = torch.cat(batch.reward)
    # print(action_batch)  # 非0即1
    # print(reward_batch)
    # print(len(non_final_mask))
    # Compute Q(s_t, a) - the model computes Q(s_t), then we select the
    # columns of actions taken
    state_action_values = policy_net(state_batch).gather(1, action_batch)  # gather將torch.tensor的中對應於action的index取出，dim為1
    # 從整體公式上而言，Q函數的值即為state_action_value的值
    # print((policy_net(state_batch)))
    # print(state_action_values)
    # Compute V(s_{t+1}) for all next states.
    next_state_values = torch.zeros(BATCH_SIZE).cuda()
    # print(next_state_values)
    # print("no final mask")
    # print(non_final_mask)
    # print("@#$%^&*")
    next_state_values[non_final_mask] = target_net(non_final_next_states).max(1)[0].detach()  # non_final_mask為1的地方進行賦值操作，其余仍為0
    # print(target_net(non_final_next_states).max(1)[0].detach())
    # print("12345")
    # print(next_state_values)
    # Compute the expected Q values
    expected_state_action_values = (next_state_values * GAMMA) + reward_batch

    # Compute Huber loss
    loss = F.smooth_l1_loss(state_action_values, expected_state_action_values.unsqueeze(1))

    # compare the parameters of 2 networks
    print(policy_net.state_dict()['head.bias'])
    print("!@#$%^&*")
    print(target_net.state_dict()['head.bias'])

    # Optimize the model
    optimizer.zero_grad()
    loss.backward()
    for param in policy_net.parameters():
        param.grad.data.clamp_(-1, 1)
    optimizer.step()


num_episodes = 50
for i_episode in range(num_episodes):
    # print("the episode is %f" % i_episode)
    # Initialize the environment and state
    env.reset()
    last_screen = get_screen()
    # print(last_screen)
    # print("#QW&*!$")
    current_screen = get_screen()  # 得到一張圖片，而非一個batch
    # print(current_screen)
    state = current_screen - last_screen  # 兩幀之間的差值，作為一個state，並且輸入網絡，類比於RNN對pose的估計
    for t in count():  # 創建一個無限循環迭代器，t的數值會一直增加
        # Select and perform an action
        action = select_action(state)
        _, reward, done, _ = env.step(action.item())  # done表示游戲是否結束, reward由gym內部決定；輸入action，gym展示下一個狀態
        reward = torch.tensor([reward]).cuda()

        # Observe new state
        last_screen = current_screen
        current_screen = get_screen()
        if not done:
            next_state = current_screen - last_screen
        else:
            next_state = None

        # Store the transition in memory
        memory.push(state, action, next_state, reward)  # memory存儲state，action，next_state,以及對應的reward
        # print("the length of the memory is %d" % len(memory))
        # Move to the next state
        state = next_state

        # Perform one step of the optimization (on the target network)
        optimize_model()
        if done:
            episode_durations.append(t + 1)
            plot_durations()
            break
    # Update the target network
    if i_episode % TARGET_UPDATE == 0:  # 只有在某個頻率下才會update target網絡結構
        target_net.load_state_dict(policy_net.state_dict())

print('Complete')
env.render()
env.close()
plt.ioff()
plt.show()
env.close()

作者調用了一個gym的庫，這個庫可以用作強化學習的訓練樣本，但是蛋疼的是，在用pycharm進行debug的時候，gym庫總會報錯，如果直接運行則不會，我想可能是因為gym庫並不可以進行調試

anyway，代碼的總體流程是，調用gym，聲明一個事件，在強化學習中被稱為agent，這個agent會展示當前的狀態，然后會接收一個action，輸出下一個的狀態以及這個action所得到的獎勵，ok，至於這個agent采取了action之后所得到的獎勵是如何計算的，

這個agent采取了這個action下一個狀態是啥，gym已經給你們寫好了

 

在定義網絡結構之前，作者實際上是把自己試錯的狀態存儲了起來，存儲的內容有，當前的state，采取action，以及nextstate，以及這個action相應的reward，而state並不是當前游戲的截屏，而是兩幀之間的差值，reward是gym自己返回的

至於為什么這樣做？有點兒類似與用RNN解決slam的問題，為什么輸入到網絡中的是視頻兩幀之間的差值，而不是視頻自己本身的內容，要給自己挖個坑

存儲了這些狀態之后就可以訓練網絡了，主體的網絡結構如下

class DQN(nn.Module):

    def __init__(self):
        super(DQN, self).__init__()
        self.conv1 = nn.Conv2d(3, 16, kernel_size=5, stride=2)
        self.bn1 = nn.BatchNorm2d(16)
        self.conv2 = nn.Conv2d(16, 32, kernel_size=5, stride=2)
        self.bn2 = nn.BatchNorm2d(32)
        self.conv3 = nn.Conv2d(32, 32, kernel_size=5, stride=2)
        self.bn3 = nn.BatchNorm2d(32)
        self.head = nn.Linear(448, 2)

    def forward(self, x):
        x = F.relu(self.bn1(self.conv1(x)))
        x = F.relu(self.bn2(self.conv2(x)))
        x = F.relu(self.bn3(self.conv3(x)))
        return self.head(x.view(x.size(0), -1))

網絡輸出的兩個值，分別是對應不同的action，其實也不難理解，訓練的網絡最終能夠產生的輸出當然是決策是怎樣的，不過這種自己不斷的試錯，並且把自己試錯的數據保存下來，嚴格意義上來說真的是無監督學習？

anyway，作者用這些試錯的數據進行訓練

不過，網絡的loss怎么設計？

loss如上，實際上就是求取兩個Q函數之間的差值，ok，前一個Q函數的自變量描述的是當前的狀態s以及對應的行為a，后一個r+Q描述的是當前的reward加上，在下一個state如何采取下一步行動能夠讓Q最大的項

而這兩項如何在代碼中體現，實際上作者定義了兩個網絡，一個成為policy，另外一個為target網絡

優化的目標是policy net，target網絡為定期對policy的copy，如下

    # Update the target network
    if i_episode % TARGET_UPDATE == 0:  # 只有在某個頻率下才會update target網絡結構
        target_net.load_state_dict(policy_net.state_dict())

policy net輸入state batch，並且將實際中的對應的action的那一列輸出，action非0即1,所以policy_net輸出的是batch_size的列向量

在這段代碼中，這個網絡的輸出就是Q函數的值，

target_net網絡輸入的是next_state，並且因為不知道其實際的action是多少，所以取最大的，輸出乘以一個gamma，並且加上當前狀態的reward即可

其實永遠是policy_net更新在前，更新的方向是讓兩個網絡的輸出盡可能的接近，其實也不僅僅是這樣，這中間還有一個reward變量，可是為什么target_net的更新要永遠滯后，一種更加極端的情況是，如果把next_state輸入到policy網絡中呢？