强化学习笔记(4)-深度Q学习

以下为学习《强化学习：原理与python实现》这本书的笔记。

在之前学习到的强度学习方法中，每次更新价值函数只更新某个状态动作对的价值估计。但是有些情况下状态动作对的数量非常大，不可能对所有的状态动作对逐一更新。函数近似方法用参数化的模型来近似整个状态价值函数，并在每次学习时更新整个函数，这样，对于没有被访问过的状态动作对的价值估计也能得到更新。

函数近似方法采用一个参数为w的函数 $q(s,a;w) (s\in S, a\in A(s))$ 来近似动作价值。函数的形式不定，可以是线性函数，也可以是神经网络。在学习过程中通过不断优化w参数值来找到最优的策略。

同策回合的函数近似法

对于用函数近似方法来对同策回合的价值估计进行更新的算法如下：

1. （初始化）任意初始化参数w

2. 逐回合执行以下操作：

2.1 （采样）用环境和当前动作价值估计q导出的策略（如 $\epsilon$ 柔性策略）生成轨迹样本 $S_{0},A_{0},R_{1},S_{1},A_{1},R_{2},...,S_{T-1},A_{T-1},R_{T},S_{T}$

2.2（初始化回报） $G\leftarrow0$

2.3（逐步更新）对 $t\leftarrow T-1,T-2,...,0$ ，执行以下步骤：

2.3.1（更新回报） $G\leftarrow \gamma G+R_{t+1}$

2.3.2（更新动作价值函数）更新参数w以减小 $[G-q(S_{t},A_{t};w)]^{2}$ （如 $w\leftarrow w+\alpha [G-q(S_{t},A_{t};w)]\triangledown q(S_{t},A_{t};w)$ ）

半梯度下降算法

对于动态规划和时序差分学习，采用“自益”来估计回报，回报的估计值与w相关，是存在偏差的。在试图减小每一步的回报估计 $U_{t}$ 和动作价值估计 $q(S_{t},A_{t};w)$ 的差别时，定义每一步的损失为 $[U_{t}-q(S_{t},A_{t};w)]^{2}$ ，整个回合的损失为 $\sum_{t=0}^{T}[U_{t}-q(S_{t},A_{t})]^{2}$ 。在更新参数w以减小损失时，应注意不对回报的估计求梯度，只对动作价值的估计求关于w的梯度，这就是半梯度下降算法。以下是半梯度下降算法求解最优策略的算法：

1. （初始化）任意初始化参数w

2. 逐回合执行以下操作

2.1（初始化状态动作对）选择状态S，再根据输入策略 $\pi$ 选择动作A

2.2 如果回合未结束，执行以下操作

2.2.1 用当前动作价值估计 $q(S,\cdot ;w)$ 导出的策略（如 $\epsilon$ 柔性策略）确定动作A

2.2.2（采样）执行动作A，观测得到奖励R和新状态S'

2.2.3 用当前动作价值估计 $q(S,\cdot ;w)$ 导出的策略确定动作A'

2.2.4（计算回报的估计值）如果是动作价值评估，则 $U\leftarrow R+\gamma v(S';w)$ 。如果是期望SARSA算法，则 $U\leftarrow R+\gamma\sum_{a}^{}\pi(a|S';w)q(S',a;w)$ 。如果是Q学习则 $U\leftarrow R+\gamma max_{a}q(S',a;w)$

2.2.5（更新动作价值函数）如果是期望SARSA算法或Q学习，更新参数w以减小 $[U-q(S,A;w)]^{2}$ ，如 $w\leftarrow w+\alpha [U-q(S,A;w)]\triangledown q(S,A;w)$ 。注意此步不可以重新计算U

2.2.6 $S\leftarrow S'$

深度Q学习

深度Q学习将深度学习与强化学习相结合，是第一个深度强化学习算法。其核心是用一个神经网络来替代动作价值函数。因为当同时出现异策，自益和函数近似时，无法保证收敛性，出现训练困难的问题，可以进行经验回放（存储经验，并在经验中按一定规则采样）和目标网络（修改网络更新方式，不把刚学习到的权重用于后续的自益过程）两种方式来改进。以下是带经验回放和目标网络的深度Q学习最优策略的求解算法：

1.（初始化）初始化评估网络 $q(\cdot,\cdot;w)$ 的参数w，目标网络 $q(\cdot, \cdot;w_{target})$ 的参数 $w_{target}\leftarrow w$

2. 逐回合执行以下操作

2.1（初始化状态动作对）选择状态S

2.2 如果回合未结束，执行以下操作

2.2.1（采样）根据 $q(S,\cdot;w)$ 选择动作A并执行，观测得到奖励R和新状态S'

2.2.2（经验存储）将经验(S,A,R,S')存入经验库D中

2.2.3（经验回放）从经验库D中选取一批经验 $(S_{i},A_{i},R_{i},S_{i}^{'}) (i \in B)$

2.2.4（计算回报的估计值） $U_{i}\leftarrow R_{i}+\gamma max_{a}q(S'_{i},a;w_{target}) (i \in B)$

2.2.5（更新动作价值函数）更新以减小 $\frac{1}{|B|}\sum_{i\in B}^{}[U_{i}-q(S_{i},A_{i};w)]^{2}$ ，如 $w\leftarrow +\alpha \frac{1}{|B|}\sum_{i\in B}^{}[U_{i}-q(S_{i},A_{i};w)]\triangledown q(S_{i},A_{i};w)$

小车上山案例

下面将以gym库中的小车上山（MountainCar-v0）为例，用深度Q学习来求解最优的策略。

这个例子里面，小车的位置范围是[-1.2, 0.6]，速度范围是[-0.07,0.07]。智能体可以对小车施加三种动作中的一种：向左施力，不施力，向右施力。当小车的水平位置大于0.5时，控制目标成功达成，回合结束。一般来说，如果在连续100回合中的平均步数<=110，就认为问题解决了。

下面代码将查看相关的环境信息

import gym
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import tensorflow as tf
import tensorflow.keras as keras
from keras.initializers import GlorotUniform
import tqdm
%matplotlib inline
from IPython import displayenv = gym.make("MountainCar-v0")
env = env.unwrapped
print("observe space = {}".format(env.observation_space))
print("action space = {}".format(env.action_space))
print("Position range = {}, {}".format(env.min_position, env.max_position))
print("Speed range = {}, {}".format(-env.max_speed, env.max_speed))
print("Target position = {}".format(env.goal_position))

输出如下：

observe space = Box(-1.2000000476837158, 0.6000000238418579, (2,), float32)
action space = Discrete(3)
Position range = -1.2, 0.6
Speed range = -0.07, 0.07
Target position = 0.5

如果一直对小车施加向右的力，小车是无法到达目标的，可以用以下代码检验：

positions, velocities = [], []
observation = env.reset()
for _ in range(200):positions.append(observation[0])velocities.append(observation[1])next_observation, reward, done, _ = env.step(2)if done:breakobservation = next_observation
if next_observation[0] > 0.5:print("success")
else:print("fail")fig, ax = plt.subplots()
ax.plot(positions, label='position')
ax.plot(velocities, label='velocity')
ax.legend()
fig.show()

结果如下，可见小车的位置在-0.5到-0.25之间震荡：

下面将实现深度Q学习的代码，首先是实现经验回放的功能，如以下代码：

class DQNReplayer:def __init__(self, capacity):self.memory = pd.DataFrame(index=range(capacity),columns=['observation', 'action', 'reward', 'next_observation', 'done'])self.i = 0self.count = 0self.capacity = capacitydef store(self, *args):self.memory.loc[self.i] = argsself.i = (self.i+1)%self.capacityself.count = min(self.count+1, self.capacity)def sample(self, size):indices = np.random.choice(self.count, size=size)return (np.stack(self.memory.loc[indices, field]) for field in self.memory.columns)

以下代码实现一个智能体的类，包括了建立神经网络（输入为observation的维度，中间层为一个64维的层，输出为状态动作价值估计）和进行训练的代码：

class DQNAgent:def __init__(self, env, net_kwargs={}, gamma=0.99, epsilon=0.001, replayer_capacity=10000, batch_size=64):observation_dim = env.observation_space.shape[0]self.action_n = env.action_space.nself.gamma = gammaself.epsilon = epsilonself.batch_size = batch_sizeself.replayer = DQNReplayer(replayer_capacity)self.train_steps = 0self.evaluate_net = self.build_network(input_size=observation_dim, output_size=self.action_n, **net_kwargs)self.target_net = self.build_network(input_size=observation_dim, output_size=self.action_n, **net_kwargs)self.target_net.set_weights(self.evaluate_net.get_weights())def build_network(self, input_size, hidden_sizes, output_size, activation=tf.nn.relu, output_activation=None, learning_rate=0.01):model = keras.Sequential()for layer, hidden_size in enumerate(hidden_sizes):kwargs = dict(input_shape=(input_size,)) if not layer else {}model.add(keras.layers.Dense(units=hidden_size, activation=activation, kernel_initializer=GlorotUniform(seed=0), **kwargs))model.add(keras.layers.Dense(units=output_size, activation=output_activation, kernel_initializer=GlorotUniform(seed=0)))optimizer = keras.optimizers.Adam(lr=learning_rate)model.compile(loss='mse', optimizer=optimizer)return modeldef learn(self, observation, action, reward, next_observation, done):self.replayer.store(observation, action, reward, next_observation, done)self.train_steps += 1observations, actions, rewards, next_observations, dones = self.replayer.sample(self.batch_size)next_qs = self.target_net.predict(next_observations)next_max_qs = next_qs.max(axis=-1)us = rewards + self.gamma*(1.-dones)*next_max_qstargets = self.evaluate_net.predict(observations)targets[np.arange(us.shape[0]), actions] = usself.evaluate_net.fit(observations, targets, verbose=0)if done:self.target_net.set_weights(self.evaluate_net.get_weights())def decide(self, observation):if np.random.rand() < self.epsilon:return np.random.randint(self.action_n)qs = self.evaluate_net.predict(observation[np.newaxis])  #增加一维，原维度是1维return np.argmax(qs)net_kwargs = {'hidden_sizes':[64,], 'learning_rate':0.01}
agent = DQNAgent(env, net_kwargs=net_kwargs)

通过以下方法来进行100个回合的训练：

def play_qlearning(env, agent, train=False, render=False):episode_reward = 0observation = env.reset()step = 0img = plt.imshow(env.render(mode='rgb_array'))while True:step += 1if render:plt.title("%s | Step: %d" % ("Moutain Car",step))plt.axis('off')img.set_data(env.render(mode='rgb_array'))display.display(plt.gcf())display.clear_output(wait=True)action = agent.decide(observation)next_observation, reward, done, _ = env.step(action)episode_reward += rewardif train:agent.learn(observation, action, reward, next_observation, done)if done:breakobservation = next_observationreturn episode_rewardepisodes = 100
episode_rewards = []
for episode in tqdm.trange(episodes):episode_reward = play_qlearning(env, agent, True, False)episode_rewards.append(episode_reward)agent.evaluate_net.save('evaluate.h5')
agent.target_net.save('target.h5')

训练完成后，通过运行以下代码来查看效果：

agent.evaluate_net = keras.models.load_model('evaluate.h5')
agent.target_net = keras.models.load_model('target.h5')episode_reward = play_qlearning(env, agent, False, True)

结果如下：

MoutainCar_Trim