dqn在训练过程中loss越来越大_强化学习笔记：OpenAI Gym+DQN+Tensorflow2实现

兜里没有糖了

2493人浏览 · 2021-01-18 14:33:54

兜里没有糖了 · 2021-01-18 14:33:54 发布

参考了一些文章，针对OpenAI gym环境，使用tf2.x实现了DQN算法；加上了一些没有太大必要（？）的小功能，比如：自动保存视频，保存训练日志从而利用TensorBoard实现数据可视化，保存和读取训练记忆，保存和读取DQN权重。适用于CartPole环境，只需少量改动也可以用于MountainCar环境。代码下载及本文参考见文末。

本篇目录

代码框架
- 主循环
- 单局游戏
- 自定义reward
- 自动录制视频
DQN算法主体
- 神经网络
- DQN
- 效果
- TensorBoard可视化
- 关于Reward
代码下载
参考

CartPole环境

代码框架

主循环

-------------初始化---------------
- 创建Gym环境实例
- 设置DQN参数
- 创建DQN实例
  - 载入DQN记忆（可选）
  - 载入DQN权重（可选）
- 创建tensorflow summary writer
- 设置训练参数
--------------训练----------------
- 循环N次
|     运行一局,得到该局内的reward和losses
|     记录数据到summary
|     显示数据到命令行
- 结束循环
--------------结束----------------
  - 保存DQN记忆
  - 保存DQN权重
  - 录制视频
- 关闭环境

代码：

from cart_pole import MyModel
from cart_pole import DQN
from cart_pole import play_game
from cart_pole import make_video

import numpy as np
import tensorflow as tf
import gym
import os
import datetime
from statistics import mean
from gym import wrappers

def main():
    ####################初始化#####################

    env = gym.make('CartPole-v0')
    gamma = 0.9
    num_states = len(env.observation_space.sample())
    num_actions = env.action_space.n
    hidden_units = [20, 20]
    max_experiences = 2000
    min_experiences = 1000
    batch_size = 32
    lr = 0.01
    e_greedy = 0.9
    e_greedy_increment = 1.001
    replace_target_iter = 50

    DQN_ = DQN(num_states, num_actions, hidden_units, gamma, max_experiences, 
            min_experiences, batch_size, lr,e_greedy, replace_target_iter,e_greedy_increment)

    DQN_.loadModel()
    DQN_.loadMemory()
    
    current_time = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
    log_dir = 'logs/dqn/' + current_time
    summary_writer = tf.summary.create_file_writer(log_dir)

    N = 10000 # 总训练轮次
    Dispaly_interval = 100
    total_rewards = np.empty(N)
 
    ####################主循环#####################
    for n in range(N):
        total_reward, losses = play_game(env, DQN_)
        total_rewards[n] = total_reward
        avg_rewards = total_rewards[max(0, n - Dispaly_interval):(n + 1)].mean()
        with summary_writer.as_default():
            tf.summary.scalar('episode reward', total_reward, step=n)
            tf.summary.scalar('running avg reward(100)', avg_rewards, step=n)
            tf.summary.scalar('average loss)', losses, step=n)
        if n % Dispaly_interval == 0:
            print("episode:", n, "eps:", DQN_.epsilon, "avg reward (last ",Dispaly_interval,"):", avg_rewards,
                  "episode loss: ", losses)


    ####################结束#####################

    make_video(env, DQN_)
    # DQN_.saveModel()
    # DQN_.saveMemory()
    env.close()

单局游戏

DQN的主算法

在此基础上，做了一些更改

增加每一局内总reward和loss的记录，用于可视化
使用自定义的reward计算，而非gym环境返回的reward
Epsilon-greedy的系数e每局递增，这一步也可以放进DQN的训练函数中，放在这里意味着“每一局游戏”epsilon递增一次，放在DQN的训练函数中意味着"每一次训练"递增一次，一局游戏中包含多次训练，相同的系数下，后者会让epsilon增大的特别快。
将梯度下降的过程和Q_target的更新都包装进了DQN.train() 函数中

def play_game(env, DQN):
    rewards = 0
    done = False
    observations = env.reset()
    losses = list()
    DQN.epsilon_incre()
    while not done:
        # env.render() #注释掉则不显示GUI，训练加速
        
        action = DQN.get_action(observations)
        prev_observations = observations
        observations, reward, done, _ = env.step(action)
        reward = My_reward(env,observation)
        rewards += reward
        if done:
            reward = -10 #适用于cart_pole环境
            env.reset()

        exp = {'s': prev_observations, 'a': action, 'r': reward, 's2': observations, 'done': done}
        DQN.add_experience(exp)

        loss = DQN.train()
        if isinstance(loss, int): #当记忆库数量不足时loss=0
            losses.append(loss)
        else:
            losses.append(loss.numpy()) #类型转换

    return rewards, mean(losses)

自定义reward

gym环境的done的定义是位置不超过边界，角度不超过某一阈值（0.2左右）

gym环境的reward的定义是，只要done=false->reward=1，done=true之后reward=0，这样的设置会让训练很低效。

在我们有环境、预期目标和被控对象的一定知识之后，把这份知识通过reward的方式“告诉“被控对象，能达到加速训练的效果，这也就是reward shaping。

在这个例子中，我们的目标是杆子一直立着，越久越好，同时车也越靠近中心越好。因此，使用这样的函数自定义reward

def My_reward(env,observations):

    x, _, theta, _ = observations # observations = [x,x导数,theta,theta导数]
    rp = (env.x_threshold - abs(x))/env.x_threshold - 0.8 # reward position
    ra = (env.theta_threshold_radians - abs(theta))/env.theta_threshold_radians - 0.5 # reward angle
    reward = rp + ra 
 return reward

env.x_threshold和env.theta_threshold_radians是gym环境自带的参数，代表失败的阈值，这里的函数定义可以理解为：

横轴为角度或位置，纵轴为reward，位置和角度越靠近0，reward越大，超出一定范围reward为负，图示并非实际数值

同时，以上的讨论依旧局限于Done=false的时刻，当Done=true时，即车超出边界or棒子与竖直的夹角大于一定阈值，需要令reward等于一个比较大的负数，从而让被控对象以后避免这个导致Done为true的action，这其实也是reward shaping的一部分，不加这里的设置，系统照样会收敛，不过加上之后，学习会更加高效，系统会倾向于更加谨慎，避免done=true。但也带来了一些代价。

实现这个的代码并没有写进My_reward函数里，而是写在play_game的循环里了。具体的原因在本文的最后一部分给了说明。

自动录制视频

为了加速，在训练过程中默认是不显示窗口界面的，如果需要可以使用env.render()函数调出窗口，为了录制训练结束之后的系统表现，使用warppers.Monitor函数

在录制时，DQN的epsilon参数被设置为1，这意味着不会随机选择动作，epsilon的目的在于一定程度上避免系统陷入局部最优，适用于学习的过程，在测试的过程中没有意义。

def make_video(env, DQN):
    env = wrappers.Monitor(env, r"D:桌面RLvideos", force=True)
    rewards = 0
    steps = 0
    done = False
    observation = env.reset()
    DQN.epsilon = 1 #epsilon = 1代表严格按照DQN输出行动，没有随机因素
 while not done:
        env.render() #调出并更新窗口
        action = DQN.get_action(observation)
        observation, reward, done, _ = env.step(action)
        reward = My_reward(env,observations)
        steps += 1
        rewards += reward
 print("Testing steps: {} rewards {}: ".format(steps, rewards))

DQN算法主体

这里使用keras.Model的自定义子类作为Q神经网络的实现，但完全不一定非要这么写，也可以用keras.Model，Tensorlayer之类的其他方式，毕竟gym的这几个问题用到的神经网路结构还是相对很简单的。

神经网络

# keras.model的自定义子类
class MyModel(tf.keras.Model):
    def __init__(self, num_states, hidden_units, num_actions):
        super(MyModel, self).__init__()
        self.input_layer = tf.keras.layers.InputLayer(input_shape=(num_states,))
        self.hidden_layers = []
        for i in hidden_units:
            self.hidden_layers.append(tf.keras.layers.Dense(
                i, activation='tanh', kernel_initializer='RandomNormal'))
        self.output_layer = tf.keras.layers.Dense(
            num_actions, activation='linear', kernel_initializer='RandomNormal')
        
    
    @tf.function
    def call(self, inputs):
        z = self.input_layer(inputs)
        for layer in self.hidden_layers:
            z = layer(z)
        output = self.output_layer(z)
        return output

    def predict(self, inputs):
        return self(np.atleast_2d(inputs.astype('float32')))

DQN

import numpy as np
import tensorflow as tf
import gym
import os
import datetime
from statistics import mean
from gym import wrappers
import pickle

class DQN:
    def __init__(self, num_states, num_actions, hidden_units, gamma, max_experiences, min_experiences,
                 batch_size, lr,e_greedy, replace_target_iter = 50,e_greedy_increment=None):
        self.num_actions = num_actions
        self.batch_size = batch_size
        self.optimizer = tf.optimizers.Adam(lr)
        self.gamma = gamma
        self.model = MyModel(num_states, hidden_units, num_actions)
        self.target = MyModel(num_states, hidden_units, num_actions)
        self.experience = {'s': [], 'a': [], 'r': [], 's2': [], 'done': []} #字典形式的记忆库
        self.max_experiences = max_experiences
        self.min_experiences = min_experiences
        self.learn_step_counter = 0
        self.replace_target_iter = replace_target_iter

        self.epsilon_increment = e_greedy_increment
        self.epsilon_max = e_greedy
        self.epsilon = 0.1 if self.epsilon_increment is not None else self.epsilon_max


    # 以Target net作为参考训练
    def train(self):
        if self.learn_step_counter % self.replace_target_iter == 0:
            self.learn_step_counter = 0
            self.copy_weights()

        if len(self.experience['s']) < self.min_experiences:
            return 0 # 0 代表啥也没学到


        ids = np.random.randint(low=0, high=len(self.experience['s']), size=self.batch_size)
        # {'s': array([-0.02874029, -0.58186904,  0.00809103,  0.89257765]), 
        # 'a': 1, 'r': 1.0, 's2': array([-0.04037767, -0.38685777,  0.02594258,  0.60244905]), 'done': False}
        # {'s': array([-0.13791506, -0.78732364,  0.19086538,  1.4356326 ]),
        #  'a': 0, 'r': -200, 's2': array([-0.15366153, -0.98421693,  0.21957803,  1.78138363]), 'done': True}
        #  state = x, x_dot, theta, theta_dot
        states = np.asarray([self.experience['s'][i] for i in ids])
        actions = np.asarray([self.experience['a'][i] for i in ids])
        rewards = np.asarray([self.experience['r'][i] for i in ids])
        states_next = np.asarray([self.experience['s2'][i] for i in ids])
        dones = np.asarray([self.experience['done'][i] for i in ids])

        value_next = np.max(self.target.predict(states_next), axis=1)
        # np.where(condition, x, y)
        # 满足条件(condition)，输出x，不满足输出y。
        actual_values = np.where(dones, rewards, rewards+self.gamma*value_next)

        with tf.GradientTape() as tape:
            selected_action_values = tf.math.reduce_sum(
                self.model.predict(states) * tf.one_hot(actions, self.num_actions), axis=1)
            loss = tf.math.reduce_mean(tf.square(actual_values - selected_action_values))
        variables = self.model.trainable_variables
        gradients = tape.gradient(loss, variables)
        self.optimizer.apply_gradients(zip(gradients, variables))


        self.learn_step_counter += 1   
        return loss 

    # 选取行动
    def get_action(self, states):
        if np.random.random() > self.epsilon:
            return np.random.choice(self.num_actions)
        else:
            return np.argmax(self.model.predict(np.atleast_2d(states))[0])

    # 记忆库
    def add_experience(self, exp):
        if len(self.experience['s']) >= self.max_experiences:
            for key in self.experience.keys():
                self.experience[key].pop(0) # 满了就删除一个
        for key, value in exp.items():
            self.experience[key].append(value)

    # 更新fronzen/target net
    def copy_weights(self):
        variables1 = self.target.trainable_variables
        variables2 = self.model.trainable_variables
        for v1, v2 in zip(variables1, variables2):
            v1.assign(v2.numpy())

    def epsilon_incre(self):
        self.epsilon = self.epsilon * self.epsilon_increment if self.epsilon < self.epsilon_max else self.epsilon_max

    def saveModel(self):
        self.model.save_weights(r'D:桌面RLDQNeval_net.h5')
        self.target.save_weights(r'D:桌面RLDQNtarget_net.h5')
        print('Weights Saved.')

    def loadModel(self):
        # 模型没有被调用过则没有权重，调用一下生成初始权重
        self.model.predict(np.array([1,1,1,1]))
        self.target.predict(np.array([1,1,1,1]))
        self.model.load_weights(r'D:桌面RLDQNeval_net.h5')
        self.target.load_weights(r'D:桌面RLDQNtarget_net.h5')
        print('Weights loaded.')

    def saveMemory(self):
        f = open(r'D:桌面RLMemory.data', 'wb')
        # 将变量存储到目标文件中区
        pickle.dump(self.experience, f)
        # 关闭文件
        f.close()
        print('Memory saved.')

    def loadMemory(self):
        fr = open(r'D:桌面RLMemory.data','rb')
        self.experience  = pickle.load(fr)
        print('Memory loaded.')

效果

训练200回合之后的效果

用两层每层20个节点的神经网络，epsilon初值0.1，每回合递乘1.02，最大值0.9，在200回合后的效果

TensorBoard可视化

打开日志存放的文件夹，打开终端，运行tensorboard --logdir=日志所在的文件夹名，例如：

tensorboard --logdir=20201112-111458
#或者
#tensorboard --logdir=D:RLlogsdqn20201112-111458

打开输出的链接，即可进入TensorBoard界面

loss曲线

loss代表了神经网络的学习程度，在没有过拟合的时候，loss越带代表神经网络的”学到的东西“越多，可以看到前100回合内loss有一个高峰，代表“学到了很多东西”，然后慢慢降低，代表”学完了“

平均reward曲线

2000回合左右就有很好的性能，之后有过拟合的倾向

这里用了相同参数，但是更长训练时间的结果，可以看到DQN的性能远非能达到的最好，在2k之后有过拟合倾向，越学越拉跨，但是毕竟只使用了20*2个中间神经元，想要获得更好的训练效果，可以考虑增加神经元节点（hidden unit）数、层数、降低epsilon的递增率、增加记忆池大小，更高层的改动包括改进神经网络结构，改进DQN算法，使用Double DQN等等，但是训练时间自然也会相应地提高。

关于Reward

除此之外，其实本代码的reward的处理并不严谨，gym环境在一定时间过去之后，无论系统状态如何，Done也会被设置为true，这是为了给单局时间设置一个上界，避免每次时间无限地递增，从而避免训练时间无限地递增。

在每局游戏时间结束将导致一个reward在done=true时被设置为-10，这其实相当于：在系统通过学习，能够撑到时间终点之后，在每一局结束时，为DQN的[状态，动作]映射，加上一些随机的噪声。更通俗地讲：假设游戏在第300步结束，done在此时被设置为true，那么第299步的reward会被设置为-10，系统会认为“是299步的这个[状态,动作]映射对带来了惩罚”，因此会降低这个映射对的权重，但其实这个映射对并没有任何错误，只是时间到了而已。

不过这个随机的噪声对系统的性能影响依旧是十分有限的，因为在每局完整的比赛中，假设会记录300次动作，只有1个[状态，动作]映射对受到污染。

但从本质上讲，这个问题的原因在于DQN并没有把时间当作输入，只能感受到位置和角度的状态，因此解决方法也有三个：

0：更改reward shaping的方式，不对done=true时的reward做任何修改，或者干脆就不reward shaping，用gym原生的reward.

1：在DQN的神经网络输入层中加入这一时间维度，这虽然能让DQN感知到时间这个维度了，但是这个方法很明显太笨了，会让DQN有更复杂的结构，需要更长的时间收敛，需要更多的训练，仅仅是为了避免稳态下约1/300的噪声，并不可取。

2：在reward处理中考虑时间，即如果done=true是由角度和位置的偏移带来的，那么给与对应的惩罚：reward=-10，而如果done=true是由该局游戏结束带来的，那么reward=reward，即不做出更改，依旧按照自定义函数给出的值记录进记忆。这样做需要增加一个判断的流程，代码会更加复杂，好处是系统收敛先验地更快。

代码下载

MUCZ/Gym_DQN

参考

https://mofanpy.com/tutorials/machine-learning/reinforcement-learning/DQN1/

https://towardsdatascience.com/deep-reinforcement-learning-build-a-deep-q-network-dqn-to-play-cartpole-with-tensorflow-2-and-gym-8e105744b998

https://blog.csdn.net/november_chopin/article/details/107913103

天工造物局

更多推荐

肌肉骨骼仿真如何重塑机器人控制？探索MyoSuite的强化学习革命

想象一下，让机器人像人类一样自然地抓取物体、行走、甚至打乒乓球——这不再是科幻电影中的场景。MyoSuite，这个基于MuJoCo物理引擎和OpenAI Gym API的开源肌肉骨骼仿真套件，正在重新定义机器人控制的可能性。在过去的几年里，强化学习在机器人控制领域取得了显著进展，但大多数研究都集中在刚性机器人上。MyoSuite则专注于更接近生物系统的肌肉骨骼模型，为研究者提供了一个探索生物力学控