强化学习——DQN算法_强化学习dqn

1、DQN算法介绍

        DQN算与sarsa算法和Q-learning算法类似，对于sarsa和Q-learning，我们使用一个Q矩阵，记录所有的state（状态）和action（动作）的价值，不断学习更新，最后使得机器选择在某种状态下，价值最高的action进行行动。但是当state和action的数量特别大的时候，甚至有限情况下不可数时，这时候再用Q矩阵去记录对应价值就会有很大的局限性，而DQN就是在这一点上进行了改进。在DQN中，对于每种state下采取的action所对应价值，我们将会使用神经网络来进行计算。

2、平衡车游戏的实例

        平衡车游戏在gym库中，这里需要下载gym库，这个游戏中，状态由四个数字来进行表示（我也不知道这四个数字代表什么，但是无伤大雅），接着只会有两种行动，并且reward并不需要我们进行设置，这个游戏进行过程中会自己返回reward。现在先搭建环境

env = gym.make('CartPole-v1')
env.reset()
#打印游戏
def show():
    plt.imshow(env.render(mode='rgb_array'))
    plt.axis('off')
    plt.show()
#show()

 搭建环境后再创建两个神经网络，并且要让两个神经网络的参数一致，在后续的过程中，一个神经网络会延迟更新。这两个神经网络会以四个状态参数作为输入，然后以两个动作的评分作为输出

#创建神经网络
 
#计算动作模型，也就是真正需要使用的模型
model = torch.nn.Sequential(
    torch.nn.Linear(4,128),
    torch.nn.ReLU(),
    torch.nn.Linear(128,2),
)
 
#经验网络，用于评估状态分数
next_model = torch.nn.Sequential(
    torch.nn.Linear(4,128),
    torch.nn.ReLU(),
    torch.nn.Linear(128,2)
)
 
#把两个神经网络的参数统一一下
next_model.load_state_dict(model.state_dict())
 
#print(model,next_model)

接下来我们需要创建一个样本池，神经网络会在这个样本池中进行学习，随后需要不断更新我们的样本池，当我们有一个新的行动时，我们应该添加新的样本，删除旧的样本，保持样本池最大数量不变

#想样本池中添加一些数据，删除一些古老的数据
def update_date():
    old_count = len(datas)
 
    while len(datas) - old_count<200:
        #初始化
        state = env.reset()
 
        over = False
        while not over:
            #获取当前状态得到一个动作
            action = get_action(state)
 
            #执行动作，得到反馈
            next_state,reward,over,_ = env.step(action)
 
            #记录样本
            datas.append((state,action,reward,next_state,over))
 
            #更新状态
            state = next_state
 
    update_count = len(datas) - old_count
    drop_count = max(len(datas)-10000,0)
 
    while len(datas)>10000:
        datas.pop(0)
    return update_count,drop_count

        接下来需要进行采样，并将样本格式转换为所需要的格式

#获取一批数据样本
def get_sample():
    # 从样本池中采样
    samples = random.sample(datas, 64)
 
    state = np.array([i[0] for i in samples])
    action = np.array([i[1] for i in samples])
    reward = np.array([i[2] for i in samples])
    next_state = np.array([i[3] for i in samples])
    over = np.array([i[4] for i in samples])
 
    state = torch.FloatTensor(state).reshape(-1, 4)
    action = torch.LongTensor(action).reshape(-1, 1)
    reward = torch.FloatTensor(reward).reshape(-1, 1)
    next_state = torch.FloatTensor(next_state).reshape(-1, 4)
    over = torch.LongTensor(over).reshape(-1, 1)
 
    return state, action, reward, next_state, over

        接着是价值函数（直接交给神经网络就可以了）

def get_value(state,action):
    value = model(state)
 
    value = value.gather(dim=1,index=action)
    return value

        接下来是获取target函数，这个函数的意义在于，我们是不知道游戏的全貌的，这样的话在一个状态下所采取的行动，不仅仅至于它本身有关，更和接下来所到达的状态和接下来应该采取的行动有关，价值value应该要想target靠近

def get_target(reward,next_state,over):
    with torch.no_grad():
        target = next_model(next_state)
 
    target = target.max(dim = 1)[0]
    target = target.reshape(-1,1)
 
    target *= 0.98
 
    target *= (1-over)#游戏结束了就不用玩了
 
    target += reward
 
    return target

然后就是测试函数并开始训练了

def train():
    model.train()
    optimizer = torch.optim.Adam(model.parameters(),lr=2e-3)
    loss_fn = torch.nn.MSELoss()
 
 
    for epoch in range(500):
        update_count,drop_count = update_date()
 
        for i in range(200):
            state,action,reward,next_state,over = get_sample()
 
            value = get_value(state,action)
            target = get_target(reward,next_state,over)
 
            loss = loss_fn(value,target)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
 
            if (i+1)%10==0:
                next_model.load_state_dict(model.state_dict())
 
            if epoch%50==0:
                test_result = sum([tes(play=False) for _ in range(20)])/20
                print(f"Epoch: {epoch}, Data Size: {len(datas)}, Update: {update_count}, Drop: {drop_count}, Test Reward: {test_result}")
 

接下来是完整代码

#这里会使用神经网络
import gym
import torch
import random
import numpy as np
import matplotlib.pyplot as plt
from IPython import display
from matplotlib.animation import FuncAnimation
 
env = gym.make('CartPole-v1')
env.reset()
#打印游戏
def show():
    plt.imshow(env.render(mode='rgb_array'))
    plt.axis('off')
    plt.show()
#show()
 
#创建神经网络
 
#计算动作模型，也就是真正需要使用的模型
model = torch.nn.Sequential(
    torch.nn.Linear(4,128),
    torch.nn.ReLU(),
    torch.nn.Linear(128,2),
)
 
#经验网络，用于评估状态分数
next_model = torch.nn.Sequential(
    torch.nn.Linear(4,128),
    torch.nn.ReLU(),
    torch.nn.Linear(128,2)
)
 
#把两个神经网络的参数统一一下
next_model.load_state_dict(model.state_dict())
 
#print(model,next_model)
 
 
#定义动作函数
def get_action(state):
    if random.random()<0.01:
        return random.choice([0,1])
 
    state = torch.FloatTensor(state).reshape(1,4)
 
    return model(state).argmax().item()
 
#样本池
datas = []
 
#想样本池中添加一些数据，删除一些古老的数据
def update_date():
    old_count = len(datas)
 
    while len(datas) - old_count<200:
        #初始化
        state = env.reset()
 
        over = False
        while not over:
            #获取当前状态得到一个动作
            action = get_action(state)
 
            #执行动作，得到反馈
            next_state,reward,over,_ = env.step(action)
 
            #记录样本
            datas.append((state,action,reward,next_state,over))
 
            #更新状态
            state = next_state
 
    update_count = len(datas) - old_count
    drop_count = max(len(datas)-10000,0)
 
    while len(datas)>10000:
        datas.pop(0)
    return update_count,drop_count
 
#获取一批数据样本
def get_sample():
    # 从样本池中采样
    samples = random.sample(datas, 64)
 
    state = np.array([i[0] for i in samples])
    action = np.array([i[1] for i in samples])
    reward = np.array([i[2] for i in samples])
    next_state = np.array([i[3] for i in samples])
    over = np.array([i[4] for i in samples])
 
    state = torch.FloatTensor(state).reshape(-1, 4)
    action = torch.LongTensor(action).reshape(-1, 1)
    reward = torch.FloatTensor(reward).reshape(-1, 1)
    next_state = torch.FloatTensor(next_state).reshape(-1, 4)
    over = torch.LongTensor(over).reshape(-1, 1)
 
    return state, action, reward, next_state, over
 
 
def get_value(state,action):
    value = model(state)
 
    value = value.gather(dim=1,index=action)
    return value
 
def get_target(reward,next_state,over):
    with torch.no_grad():
        target = next_model(next_state)
 
    target = target.max(dim = 1)[0]
    target = target.reshape(-1,1)
 
    target *= 0.98
 
    target *= (1-over)
 
    target += reward
 
    return target
 
#测试函数
def tes(play):
    #初始化
    state = env.reset()
 
    #记录reward之和，越大越好
    reward_sum = 0
 
    over = False
    while not over:
        #获取动作
        action = get_action(state)
 
        #执行动作
        state,reward,over,_ = env.step(action)
        reward_sum +=reward
 
        #打印动画
        if play and random.random()<0.2:#跳帧
            display.clear_output(wait=True)
            show()
    return reward_sum
 
def train():
    model.train()
    optimizer = torch.optim.Adam(model.parameters(),lr=2e-3)
    loss_fn = torch.nn.MSELoss()
 
 
    for epoch in range(500):
        update_count,drop_count = update_date()
 
        for i in range(200):
            state,action,reward,next_state,over = get_sample()
 
            value = get_value(state,action)
            target = get_target(reward,next_state,over)
 
            loss = loss_fn(value,target)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
 
            if (i+1)%10==0:
                next_model.load_state_dict(model.state_dict())
 
            if epoch%50==0:
                test_result = sum([tes(play=False) for _ in range(20)])/20
                print(f"Epoch: {epoch}, Data Size: {len(datas)}, Update: {update_count}, Drop: {drop_count}, Test Reward: {test_result}")
 
 
train()

为什么需要两个神经网络？

这个两个神经网络其实是一致的，只是其中一个会延迟更新参数。

这个原因在于假如在一个游戏中，我们的目标状态并不是固定，可能是一直变换的，就如这个游戏中，平衡的状态是多种多样的，那么我们一直跟踪这个目标就会变得困难，这时我们不妨固定住某一个曾经是目标的状态，让机器先尝试去达到这种状态，再去跟踪下一个固定目标状态，这样的方式会使得机器更容易找到目标状态。这也是为什么需要一个一样的，但是延迟更新的神经网络。