算法DDQN

DQN是强化学习里最经典的算法之一，网上也有很多文章讲解DQN。但是有个问题是DQN会高估Q值（因为DQN在计算Q时会用max Q来计算）。
Double DQN其实就是Double Q learning在DQN上的拓展，它有两套Q值，分别对应DQN的policy network（更新的快）和target network（每隔一段时间与policy network同步）。
DQN总是选择Target Q网络的最大输出值。而DDQN不同，DDQN首先从Q网络中找到最大输出值的那个动作，然后再找到这个动作对应的Target Q网络的输出值。

实践

环境准备

GYM及PARL+paddle

我使用的是百度的paddle作为模型及parl框架来做强化学习，parl很方便，可以省去不少agent和算法的交互，并且提供了不少主流算法实现可以直接食用，还支持并发计算。
游戏环境为gym的 LunarLander-v2

月球登入，环境输出的观察空间为长度为8的元组（数值是-1，1之间），行动范围为4（0，1，2，3）

env = gym.make('LunarLander-v2')
action_dim = env.action_space  # 查看行动空间
obs_space = env.observation_space.shape[0]  # 查看观察空间

每个step会返回observation(环境数据),reward(当前分数),done(是否结束游戏),message(一些其他信息，目前没啥用)。
一个step可以理解为一帧。

parl的框架结构

我项目的目录结构如下：

各个py文件内容

agent.py: 负责与环境和模型交互
model.py: 神经网络的搭建
algorithm.py: 负责算法部分，DDQN的更新模型的地方
replay_memory.py: DDQN需要从历史经验中抽取行动及奖励数据进行学习，这里就是存经验回放池的地方
train.py: 主程序，设置参数及训练episode的地方

agent构建

agent是继承parl.Agent类，主要有sample、predict、learn三个方法组成
DDQN更新需要的数据为5个， 分别是：当前step的环境数据obs, 当前step的行动act, 当前step的分数reward, 下一个step的环境数据next_obs, 当前step是否继续游戏done

# -*- coding: utf-8 -*-
import parl
import paddle
import numpy as np
from parl.utils import logger


class Agent(parl.Agent):
    def __init__(self, algorithm, act_dim, e_greed=0.1, e_greed_decrement=0):
        super(Agent, self).__init__(algorithm)
        assert isinstance(act_dim, int)
        self.act_dim = act_dim

        self.global_step = 0
        self.update_target_steps = 200  # 每隔200个training steps再把model的参数复制到target_model中

        self.e_greed = e_greed  # 有一定概率随机选取动作，探索
        self.e_greed_decrement = e_greed_decrement  # 随着训练逐步收敛，探索的程度慢慢降低

    def sample(self, obs):
        """ 
        根据观测值 obs 采样（带探索）一个动作
        """
        sample = np.random.random()  # 产生0~1之间的小数
        if sample < self.e_greed:
            act = np.random.randint(self.act_dim)  # 探索：每个动作都有概率被选择
        else:
            act = self.predict(obs)  # 选择最优动作
        self.e_greed = max(
            0.01, self.e_greed - self.e_greed_decrement)  # 随着训练逐步收敛，探索的程度慢慢降低，最低探索度是1%，可自己调整
        return act

    def predict(self, obs):
        """ 
        根据观测值 obs 选择最优动作
        """
        obs = paddle.to_tensor(obs, dtype='float32')
        pred_q = self.alg.predict(obs)
        act = pred_q.argmax().numpy()[0]  # 选择Q最大的下标，即对应的动作
        return act

    def learn(self, obs, act, reward, next_obs, terminal):
        """ 
        根据训练数据更新一次模型参数
        """
        if self.global_step % self.update_target_steps == 0:
            self.alg.sync_target()
        self.global_step += 1
        act = np.expand_dims(act, axis=-1)
        reward = np.expand_dims(reward, axis=-1)
        terminal = np.expand_dims(terminal, axis=-1)

        obs = paddle.to_tensor(obs, dtype='float32')
        act = paddle.to_tensor(act, dtype='int32')
        reward = paddle.to_tensor(reward, dtype='float32')
        next_obs = paddle.to_tensor(next_obs, dtype='float32')
        terminal = paddle.to_tensor(terminal, dtype='float32')
        loss = self.alg.learn(obs, act, reward, next_obs, terminal)  # 训练一次网络
        return loss.numpy()[0]

搭建神经网络

月球登入的环境数据obs并不复杂，只是一个长度为8的元祖，只需要比较简单的神经网络就可以达到预期目标
我用的3层全连接层，激活函数用的relu
第三层不用激活函数，输出的是4个动作概率从大到小排序

import parl
import paddle.nn as nn
import paddle
import paddle.nn.functional as F


class Model(parl.Model):

    def __init__(self, obs, act_dim):
        super(Model, self).__init__()
        hid1_size = obs * 32
        hid2_size = obs * 32
        # 3层全连接网络
        self.fc1 = nn.Linear(obs, hid1_size)
        self.fc2 = nn.Linear(hid1_size, hid2_size)
        self.fc3 = nn.Linear(hid2_size, act_dim)

    def forward(self, x, *args, **kwargs):
        x = paddle.to_tensor(x, dtype='float32')
        fc1 = F.relu(self.fc1(x))
        fc2 = F.relu(self.fc2(fc1))
        return self.fc3(fc2)

replay_memory经验回放池

collections.deque是python的标准库，类似于list的容器，可以快速的在队列头部和尾部添加、删除元素

import random
import collections
import numpy as np


class ReplayMemory(object):
    def __init__(self, max_size):
    	"""设定容器的最列队数量，超过范围的会删除最早插入的数据"""
        self.buffer = collections.deque(maxlen=max_size)

    def append(self, exp):
    	"""插入数据"""
        self.buffer.append(exp)

    def sample(self, batch_size):
    	"""对数据进行随机采样"""
        mini_batch = random.sample(self.buffer, batch_size)
        obs_batch, action_batch, reward_batch, next_obs_batch, done_batch = [], [], [], [], []

        for experience in mini_batch:
            s, a, r, s_p, done = experience
            obs_batch.append(s)
            action_batch.append(a)
            reward_batch.append(r)
            next_obs_batch.append(s_p)
            done_batch.append(done)

        return np.array(obs_batch).astype('float32'), 
            np.array(action_batch).astype('float32'), np.array(reward_batch).astype('float32'),
            np.array(next_obs_batch).astype('float32'), np.array(done_batch).astype('float32')

    def __len__(self):
        return len(self.buffer)

algorithm算法

parl其实已经预存了DDQN的算法模型，自己写的话还可能出错，我就默认使用了官方的算法，只是进行调参。有兴趣的可以去看看源码，大佬可以自行对算法进行复写改良！

train训练主程序

这里包括训练、评估及DDQN的超参设置

训练任务

先写一个训练任务

# -*- coding: utf-8 -*-
import gym
import numpy as np
import parl
from parl.utils import logger  # 日志打印工具
from model import Model
from agent import Agent
from replay_memory import ReplayMemory

rpm = ReplayMemory(MEMORY_SIZE)  # DQN的经验回放池

def run_episode(env, agent, rpm):
	"""
	训练一个episode，一个episode=一个游戏循环，游戏结束done会返回true
    :param env: 游戏环境类
    :param agent: 
    :param rpm: DQN的经验回放池
    :return: 
	"""
    total_reward = 0
    obs = env.reset()  # 重制游戏
    step = 0
    while True:
        step += 1
        action = agent.sample(obs)  # 采样动作，所有动作都有概率被尝试到
        next_obs, reward, done, _ = env.step(action)  # 运行一个step后返回的信息
        rpm.append((obs, action, reward, next_obs, done))

        # 这里判断经验池是否达到预存的量和训练频率
        # 在模型学习之前先要给经验池一定的数据才能训练；频率的话不必每一步都训练，可以根据需要设置频率；
        if (len(rpm) > MEMORY_WARMUP_SIZE) and (step % LEARN_FREQ == 0):
            (batch_obs, batch_action, batch_reward, batch_next_obs,
             batch_done) = rpm.sample(BATCH_SIZE)
            train_loss = agent.learn(batch_obs, batch_action, batch_reward,
                                     batch_next_obs,
                                     batch_done)  # s,a,r,s',done

        total_reward += reward
        obs = next_obs
        if done or step >= 500:  # 这里设置了step超过500就停止本局游戏，正常一局大概在300-400个step就能结束，超过的肯定是在摸鱼，游戏好像是设定1200才自动结束，没必要拿摸鱼的数据存在经验池里
            break
    return total_reward

评估模型

我这里是跑5个episode，看平均分数和每回合最终奖励(游戏如果平稳降落地在旗子之间，最终reward给100分，坠机-100，不在旗子之间降落给-1到1之间的分数)

def evaluate(env, agent, render=False):
    """
    
    :param env: 
    :param agent: 
    :param render: 是否展示游戏画面，True为展示，默认关闭
    :return: 
    """
    eval_reward = []
    end_rewards = []
    step = 0
    for i in range(5):
        step += 1
        obs = env.reset()
        episode_reward = 0
        while True:
            action = agent.predict(obs)  # 预测动作，只选最优动作
            obs, reward, done, _ = env.step(action)
            episode_reward += reward
            if render:
                env.render()
            if done or step >= 500:
                break
        eval_reward.append(episode_reward)
        end_rewards.append(reward)
    r = 0
    for i in end_rewards:  # 判断是否降落在旗子之间
        if i >= 100:
            r += 1
    return np.mean(eval_reward), r 

超参数及训练主程序

参数设定

DDQN的超参数就是奖励衰减因子和学习率:
gamma (float): discounted factor for reward computation.
lr (float): learning rate.
设置的所有参数如下：

LEARN_FREQ = 8  # 训练频率，不需要每一个step都learn，攒一些新增经验后再learn，提高效率
MEMORY_SIZE = 20000  # replay memory的大小，越大越占用内存，也不是越大越好
MEMORY_WARMUP_SIZE = 2000  # replay_memory 里需要预存一些经验数据，再从里面sample一个batch的经验让agent去learn
BATCH_SIZE = 32  # 每次给agent learn的数据数量，从replay memory随机里sample一批数据出来，同样不是越大越好
LEARNING_RATE = 0.005  # 学习率 一般设置0.01-0.0001之间，设置太大收敛不稳定，设置太小收敛太慢了，如果是比较复杂的项目可以设置动态学习率
GAMMA = 0.99  # reward 的衰减因子，一般取0.9到0.999不等，这里设置0.99是为了让它学习到降落旗子之间的到的奖励将影响前面的决策，可以试试0.98-0.95，不要>=1

主程序

def main():
    env = gym.make('LunarLander-v2')  
    action_dim = env.action_space.n  # 游戏行动范围
	obs_space = env.observation_space.shape[0] # 游戏反馈环境范围
	
    rpm = ReplayMemory(MEMORY_SIZE)  # DDQN的经验回放池

    model = Model(obs_space, act_dim=action_dim)
    algorithm = parl.algorithms.DDQN(model, gamma=GAMMA, lr=LEARNING_RATE)
    agent = Agent(
        algorithm,
        act_dim=action_dim,
        e_greed=0.1,  # 有一定概率随机选取动作，探索，设置的是10%
        e_greed_decrement=2e-8)  # 随着训练逐步收敛，探索的程度慢慢降低
    
    # 先往经验池里存一些数据，避免最开始训练的时候样本丰富度不够
    logger.info('开始收集训练数据')
    while len(rpm) < MEMORY_WARMUP_SIZE:
        run_episode(env, agent, rpm)
    logger.info('训练数据{}条收集完毕'.format(MEMORY_WARMUP_SIZE))

    max_episode = 20000  # 虽然设置了2万次循环，但实际不需要那么多
    # start train
    episode = 0
    while episode < max_episode:  # 训练max_episode个回合，test部分不计算入episode数量
        # 训练部分
        for i in range(0, 50):
            total_reward = run_episode(env, agent, rpm)
            episode += 1
        logger.info('episode:{}, Test reward:{}'.format(episode, total_reward))
        # test部分
        eval_reward, e_r = evaluate(env, agent, render=True)  # render=True 查看显示效果
        logger.info('episode:{}, e_greed:{}, lr:{}, Test reward:{}, end reward:{}'.format(
            episode, agent.e_greed, algorithm.lr, eval_reward, e_r))

        """
        if algorithm.lr >= 0.0005: # 这里是动态调整学习率，每50个episode更新一次，可用可不用
            algorithm.lr -= 1e-6
        """

        if e_r == 5:  # 我这里根据测试集是否都在旗子之间降落判断是否完成任务，也可以根据平均eval_reward>200(我记得官方完成任务是200还是多少分来着)来判断是否完成任务
            break

    # 训练结束，保存模型
    save_path = './ddqn_LunarLander.ckpt'
    agent.save(save_path)
    evaluate(env, agent, render=True) # 保存完模型再玩一遍

小结

到这里就结束了，这个训练一般在1000-2000个episode就能训练好，作为一个新手项目很值得一试
目前我也是新手，自己在空余时间摸索，现在在弄这个算法完马里奥，可惜一直无法通关，学会了如何跳墙和跳坑又不知道怎么避小怪了，希望有大佬指点一二。

感谢：飞浆《世界冠军带你从零实践强化学习》提供的学习资料，感谢科科老师教学(b站@科科磕盐)