本篇是深度强化学习动手系列文章，自MyEncyclopedia公众号文章深度强化学习之：DQN训练超级玛丽闯关发布后收到不少关注和反馈，这一期，让我们实现目前主流深度强化学习算法PPO来打另一个红白机经典游戏1942。

相关文章链接如下：

强化学习开源环境集

视频论文解读：PPO算法

视频论文解读：组合优化的强化学习方法

解读TRPO论文，深度强化学习结合传统优化方法

解读深度强化学习基石论文：函数近似的策略梯度方法

NES 1942 环境安装

红白机游戏环境可以由OpenAI Retro来模拟，OpenAI Retro还在 Gym 集成了其他的经典游戏环境，包括Atari 2600，GBA，SNES等。

不过，受到版权原因，除了一些基本的rom，大部分游戏需要自行获取rom。

环境准备部分相关代码如下

1

pip install gym-retro

1

python -m retro.import /path/to/your/ROMs/directory/

OpenAI Gym 输入动作类型

在创建 retro 环境时，可以在retro.make中通过参数use_restricted_actions指定 action space，即按键的配置。

1

env = retro.make(game='1942-Nes', use_restricted_actions=retro.Actions.FILTERED)

可选参数如下，FILTERED，DISCRETE和MULTI_DISCRETE 都可以指定过滤的动作，过滤动作需要通过配置文件加载。

1
2
3
4
5
6
7
8

class Actions(Enum):
    """
    Different settings for the action space of the environment
    """
    ALL = 0  #: MultiBinary action space with no filtered actions
    FILTERED = 1  #: MultiBinary action space with invalid or not allowed actions filtered out
    DISCRETE = 2  #: Discrete action space for filtered actions
    MULTI_DISCRETE = 3  #: MultiDiscete action space for filtered actions

DISCRETE和MULTI_DISCRETE 是 Gym 里的 Action概念，它们的基类都是gym.spaces.Space，可以通过 sample()方法采样，下面具体一一介绍。

• Discrete：对应一维离散空间，例如，Discrete(n=4) 表示 [0, 3] 范围的整数。

1
2
3

from gym.spaces import Discrete
space = Discrete(4)
print(space.sample())

输出是

1

3

• Box：对应多维连续空间，每一维的范围可以用 [low，high] 指定。 举例，Box(low=-1.0, high=2, shape=(3, 4,), dtype=np.float32) 表示 shape 是 [3, 4]，每个范围在 [-1, 2] 的float32型 tensor。

1
2
3
4

from gym.spaces import Box
import numpy as np
space = Box(low=-1.0, high=2.0, shape=(3, 4), dtype=np.float32)
print(space.sample())

输出是

1
2
3

[[-0.7538084   0.96901214  0.38641307 -0.05045208]
 [-0.85486996  1.3516271   0.3222616   1.2540635 ]
 [-0.29908678 -0.8970335   1.4869047   0.7007356 ]]

• MultiBinary: 0或1的多维离散空间。例如，MultiBinary([3,2]) 表示 shape 是3x2的0或1的tensor。

  1 2 3

  from gym.spaces import MultiBinary space = MultiBinary([3,2]) print(space.sample())

输出是

1
2
3

[[1 0]
 [1 1]
 [0 0]]

• MultiDiscrete：多维整型离散空间。例如，MultiDiscrete([5,2,2]) 表示三维Discrete空间，第一维范围在 [0-4]，第二，三维范围在[0-1]。

1
2
3

from gym.spaces import MultiDiscrete
space = MultiDiscrete([5,2,2])
print(space.sample())

输出是

1

[2 1 0]

• Tuple：组合成 tuple 复合空间。举例来说，可以将 Box，Discrete，Discrete组成tuple 空间：Tuple(spaces=(Box(low=-1.0, high=1.0, shape=(3,), dtype=np.float32), Discrete(n=3), Discrete(n=2)))

1
2
3
4

from gym.spaces import *
import numpy as np
space = Tuple(spaces=(Box(low=-1.0, high=1.0, shape=(3,), dtype=np.float32), Discrete(n=3), Discrete(n=2)))
print(space.sample())

输出是

1
2

(array([ 0.22640526,  0.75286865, -0.6309239 ], dtype=float32), 0, 1)

• Dict：组合成有名字的复合空间。例如，Dict({'position':Discrete(2), 'velocity':Discrete(3)})

  1 2 3

  from gym.spaces import * space = Dict({'position':Discrete(2), 'velocity':Discrete(3)}) print(space.sample())

输出是

1

OrderedDict([('position', 1), ('velocity', 1)])

NES 1942 动作空间配置

了解了 gym/retro 的动作空间，我们来看看1942的默认动作空间

1
2

env = retro.make(game='1942-Nes')
print("The size of action is: ", env.action_space.shape)

1

The size of action is:  (9,)

表示有9个 Discrete 动作，包括 start, select这些控制键。

从训练1942角度来说，我们希望指定最少的有效动作取得最好的成绩。根据经验，我们知道这个游戏最重要的键是4个方向加上 fire 键。限定游戏动作空间，官方的做法是在创建游戏环境时，指定预先生成的动作输入配置文件。但是这个方式相对麻烦，我们采用了直接指定按键的二进制表示来达到同样的目的，此时，需要设置 use_restricted_actions=retro.Actions.FILTERED。

下面的代码限制了6种按键，并随机play。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

action_list = [
    # No Operation
    [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
    # Left
    [0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0],
    # Right
    [0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0],
    # Down
    [0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0],
    # Up
    [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0],
    # B
    [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
]

def random_play(env, action_list, sleep_seconds=0.01):
    env.viewer = None
    state = env.reset()
    score = 0
    for j in range(10000):
        env.render()
        time.sleep(sleep_seconds)
        action = np.random.randint(len(action_list))

        next_state, reward, done, _ = env.step(action_list[action])
        state = next_state
        score += reward
        if done:
            print("Episode Score: ", score)
            env.reset()
            break
            
env = retro.make(game='1942-Nes', use_restricted_actions=retro.Actions.FILTERED)
random_play(env, action_list)

来看看其游戏效果，全随机死的还是比较快。

图像输入处理

一般对于通过屏幕像素作为输入的RL end-to-end训练来说，对图像做预处理很关键。因为原始图像较大，一方面我们希望能尽量压缩图像到比较小的tensor，另一方面又要保证关键信息不丢失，比如子弹的图像不能因为图片缩小而消失。另外的一个通用技巧是将多个连续的frame合并起来组成立体的frame，这样可以有效表示连贯动作。

下面的代码通过 pipeline 将游戏每帧原始图像从shape (224, 240, 3) 转换成 (4, 84, 84)，也就是原始的 width=224，height=240，rgb=3转换成 width=84，height=240，stack_size=4的黑白图像。具体 pipeline为

1. MaxAndSkipEnv：每两帧过滤一帧图像，减少数据量。

2. FrameDownSample：down sample 图像到指定小分辨率 84x84，并从彩色降到黑白。

3. FrameBuffer：合并连续的4帧，形成 (4, 84, 84) 的图像输入

1
2
3
4
5
6
7

def build_env():
    env = retro.make(game='1942-Nes', use_restricted_actions=retro.Actions.FILTERED)
    env = MaxAndSkipEnv(env, skip=2)
    env = FrameDownSample(env, (1, -1, -1, 1))
    env = FrameBuffer(env, 4)
    env.seed(0)
    return env

观察图像维度变换

1
2
3
4
5

env = retro.make(game='1942-Nes', use_restricted_actions=retro.Actions.FILTERED)
print("Initial shape: ", env.observation_space.shape)

env = build_env(env)
print("Processed shape: ", env.observation_space.shape)

确保shape 从 (224, 240, 3) 转换成 (4, 84, 84)

1
2

Initial shape:  (224, 240, 3)
Processed shape:  (4, 84, 84)

FrameDownSample实现如下，我们使用了 cv2 类库来完成黑白化和图像缩放

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

class FrameDownSample(ObservationWrapper):
    def __init__(self, env, exclude, width=84, height=84):
        super(FrameDownSample, self).__init__(env)
        self.exclude = exclude
        self.observation_space = Box(low=0,
                                     high=255,
                                     shape=(width, height, 1),
                                     dtype=np.uint8)
        self._width = width
        self._height = height

    def observation(self, observation):
        # convert image to gray scale
        screen = cv2.cvtColor(observation, cv2.COLOR_RGB2GRAY)

        # crop screen [up: down, left: right]
        screen = screen[self.exclude[0]:self.exclude[2], self.exclude[3]:self.exclude[1]]

        # to float, and normalized
        screen = np.ascontiguousarray(screen, dtype=np.float32) / 255

        # resize image
        screen = cv2.resize(screen, (self._width, self._height), interpolation=cv2.INTER_AREA)
        return screen

MaxAndSkipEnv，每两帧过滤一帧

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

class MaxAndSkipEnv(Wrapper):
    def __init__(self, env=None, skip=4):
        super(MaxAndSkipEnv, self).__init__(env)
        self._obs_buffer = deque(maxlen=2)
        self._skip = skip

    def step(self, action):
        total_reward = 0.0
        done = None
        for _ in range(self._skip):
            obs, reward, done, info = self.env.step(action)
            self._obs_buffer.append(obs)
            total_reward += reward
            if done:
                break
        max_frame = np.max(np.stack(self._obs_buffer), axis=0)
        return max_frame, total_reward, done, info

    def reset(self):
        self._obs_buffer.clear()
        obs = self.env.reset()
        self._obs_buffer.append(obs)
        return obs

FrameBuffer，将最近的4帧合并起来

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

class FrameBuffer(ObservationWrapper):
    def __init__(self, env, num_steps, dtype=np.float32):
        super(FrameBuffer, self).__init__(env)
        obs_space = env.observation_space
        self._dtype = dtype
        self.observation_space = Box(low=0, high=255, shape=(num_steps, obs_space.shape[0], obs_space.shape[1]), dtype=self._dtype)

    def reset(self):
        frame = self.env.reset()
        self.buffer = np.stack(arrays=[frame, frame, frame, frame])
        return self.buffer

    def observation(self, observation):
        self.buffer[:-1] = self.buffer[1:]
        self.buffer[-1] = observation
        return self.buffer

最后，visualize 处理后的图像，同样还是在随机play中，确保关键信息不丢失

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

def random_play_preprocessed(env, action_list, sleep_seconds=0.01):
    import matplotlib.pyplot as plt

    env.viewer = None
    state = env.reset()
    score = 0
    for j in range(10000):
        time.sleep(sleep_seconds)
        action = np.random.randint(len(action_list))

        plt.imshow(state[-1], cmap="gray")
        plt.title('Pre Processed image')
        plt.pause(sleep_seconds)

        next_state, reward, done, _ = env.step(action_list[action])
        state = next_state
        score += reward
        if done:
            print("Episode Score: ", score)
            env.reset()
            break

matplotlib 动画输出

CNN Actor & Critic

Actor 和 Critic 模型相同，输入是 (4, 84, 84) 的图像，输出是 [0, 5] 的action index。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

class Actor(nn.Module):
    def __init__(self, input_shape, num_actions):
        super(Actor, self).__init__()
        self.input_shape = input_shape
        self.num_actions = num_actions

        self.features = nn.Sequential(
            nn.Conv2d(input_shape[0], 32, kernel_size=8, stride=4),
            nn.ReLU(),
            nn.Conv2d(32, 64, kernel_size=4, stride=2),
            nn.ReLU(),
            nn.Conv2d(64, 64, kernel_size=3, stride=1),
            nn.ReLU()
        )

        self.fc = nn.Sequential(
            nn.Linear(self.feature_size(), 512),
            nn.ReLU(),
            nn.Linear(512, self.num_actions),
            nn.Softmax(dim=1)
        )

    def forward(self, x):
        x = self.features(x)
        x = x.view(x.size(0), -1)
        x = self.fc(x)
        dist = Categorical(x)
        return dist

PPO核心代码

先计算 \(r_t(\theta)\)，这里采用了一个技巧，对 \(\pi_\theta\) 取 log，相减再取 exp，这样可以增强数值稳定性。

1
2
3
4

dist = self.actor_net(state)
new_log_probs = dist.log_prob(action)
ratio = (new_log_probs - old_log_probs).exp()
surr1 = ratio * advantage

surr1 对应PPO论文中的 \(L^{CPI}\)

然后计算 surr2，对应 \(L^{CLIP}\) 中的 clip 部分，clip可以由 torch.clamp 函数实现。\(L^{CLIP}\) 则对应 actor_loss。

1
2

surr2 = torch.clamp(ratio, 1.0 - self.clip_param, 1.0 + self.clip_param) * advantage
actor_loss = - torch.min(surr1, surr2).mean()

最后，计算总的 loss \(L_t^{CLIP+VF+S}\)，包括 actor_loss，critic_loss 和 policy的 entropy。

1
2
3
4

entropy = dist.entropy().mean()

critic_loss = (return_ - value).pow(2).mean()
loss = actor_loss + 0.5 * critic_loss - 0.001 * entropy

上述完整代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

for _ in range(self.ppo_epoch):
    for state, action, old_log_probs, return_, advantage in sample_batch():
        dist = self.actor_net(state)
        value = self.critic_net(state)

        entropy = dist.entropy().mean()
        new_log_probs = dist.log_prob(action)

        ratio = (new_log_probs - old_log_probs).exp()
        surr1 = ratio * advantage
        surr2 = torch.clamp(ratio, 1.0 - self.clip_param, 1.0 + self.clip_param) * advantage

        actor_loss = - torch.min(surr1, surr2).mean()
        critic_loss = (return_ - value).pow(2).mean()

        loss = actor_loss + 0.5 * critic_loss - 0.001 * entropy

        # Minimize the loss
        self.actor_optimizer.zero_grad()
        self.critic_optimizer.zero_grad()
        loss.backward()
        self.actor_optimizer.step()
        self.critic_optimizer.step()

补充一下 GAE 的计算，advantage 根据公式

可以转换成如下代码

1
2
3
4
5
6
7
8
9

def compute_gae(self, next_value):
    gae = 0
    returns = []
    values = self.values + [next_value]
    for step in reversed(range(len(self.rewards))):
        delta = self.rewards[step] + self.gamma * values[step + 1] * self.masks[step] - values[step]
        gae = delta + self.gamma * self.tau * self.masks[step] * gae
        returns.insert(0, gae + values[step])
    return returns

外层 Training 代码

外层调用代码基于随机 play 的逻辑，agent.act()封装了采样和 forward prop，agent.step() 则封装了 backprop 和参数学习迭代的逻辑。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

for i_episode in range(start_epoch + 1, n_episodes + 1):
    state = env.reset()
    score = 0
    timestamp = 0

    while timestamp < 10000:
        action, log_prob, value = agent.act(state)
        next_state, reward, done, info = env.step(action_list[action])
        score += reward
        timestamp += 1

        agent.step(state, action, value, log_prob, reward, done, next_state)
        if done:
            break
        else:
            state = next_state

训练结果

让我们来看看学习的效果吧，注意我们的飞机学到了一些关键的技巧，躲避子弹；飞到角落尽快击毙敌机；一定程度预测敌机出现的位置并预先走到位置。

YouTube

BiliBili

导读：本论文由Berkeley 的几位大神于2015年发表于 JMLR（Journal of Machine Learning Research）。深度强化学习算法例如DQN或者PG（Policy Gradient）都无法避免训练不稳定的问题：在训练过程中效果容易退化并且很难恢复。针对这个通病，TRPO采用了传统优化算法中的trust region方法，以保证每一步迭代能够获得效果提升，直至收敛到局部最优点。

本篇论文涉及到的知识点比较多，不仅建立在强化学习领域经典论文的结论：Kakade & Langford 于2002 年发表的 Approximately Optimal Approximate Reinforcement Learning 关于优化目标的近似目标和重要性采样，也涉及到传统优化方法 trust region 的建模和其具体的矩阵近似数值算法。读懂本论文，对于深度强化学习及其优化方法可以有比较深入的理解。本论文附录的证明部分由于更为深奥和冗长，在本文中不做具体讲解，但是也建议大家能够仔细研读。

阅读本论文需要注意的是，这里解读的版本是arxiv的版本，这个版本带有附录，不同于 JMLR的版本的是，arxiv版本中用reward函数而后者用cost函数，优化方向相反。

arxiv 下载链接为 https://arxiv.org/pdf/1502.05477.pdf

0. 论文框架

本论文解决的目标是希望每次迭代参数能保证提升效果，具体想法是利用优化领域的 trust region方法（中文可以翻译成置信域方法或信赖域方法），通过参数在trust region范围中去找到一定能提升的下一次迭代。

本论文框架如下

1. 首先，引入Kakade & Langford 论文 Approximately Optimal Approximate Reinforcement Learning 中关于近似优化目标的结论。（论文第二部分）

2. 基于 Kakade 论文中使用mixture policy保证每一步效果提升的方法，扩展到一般随机策略，引入策略分布的total variation divergence作为约束。（论文第三部分）

3. 将total variation divergence约束替换成平均 KL divergence 约束，便于使用蒙特卡洛方法通过采样来生成每一步的具体优化问题。（论文第四，五部分）

4. 给出解决优化问题的具体算法，将优化目标用first order来近似，约束项用second order 来近似，由于second order涉及到构造Hessian matrix，计算量巨大，论文给出了 conjugate gradient + Fisher information matrix的近似快速实现方案。（论文第六部分）

5. 从理论角度指出，Kakade 在2002年提出的方法natrual policy gradient 和经典的policy gradient 都是TRPO的特别形式。（论文第七部分）

6. 评价TRPO在两种强化学习模式下的最终效果，一种是MuJoCo模拟器中能得到真实状态的模式，一种是Atari游戏环境，即观察到的屏幕像素可以信息完全地表达潜在真实状态的模式。（论文第八部分）

本文下面的小结序号和论文小结序号相同，便于对照查阅。

1. 介绍

TRPO 第一次证明了最小化某种 surrogate 目标函数且采用non-trivial的步长，一定可以保证策略提升。进一步将此 surrogate 目标函数转换成trust region约束下的优化问题。TRPO是一种on-policy 的算法，因为每一步迭代，需要在新的策略下通过采样数据来构建具体优化问题。

2. 已有理论基础

第二部分主要回顾了 Kakade & Langford 于2002 年的论文 Approximately Optimal Approximate Reinforcement Learning 中的一系列结论。

先来定义几个重要概念的数学定义

\(\eta(\pi)\) 是策略 \(\pi\) 的目标，即discounted reward 和的期望。

2. 然后是策略的Q值和V值

3. 最后是策略的advantage函数

接着，开始引入 Kakade & Langford 论文结论，即下式（公式1）。

公式1表明，下一次迭代策略的目标可以分解成现有策略的目标 \(\eta(\pi)\) 和现有advantage 函数在新策略trajectory分布下的期望。

公式1可以很容易从trajectory分布转换成新策略在状态的访问频率，即公式2

状态的访问频率或稳定状态分布定义成

注意到公式2中状态的期望依然依赖于新策略 \(\rho_{\widetilde\pi}\) 的稳定状态分布，不方便实现。原因如下，期望形式有利于采样来解决问题，但是由于采样数据源于 on-policy \(\pi\) 而非 \({\widetilde\pi}\) ，因此无法直接采样未知的策略 \({\widetilde\pi}\)。

幸好，Kakade 论文中证明了，可以用 \(\rho_{\pi}\) 的代替 \(\rho_{\widetilde\pi}\) 并且证明了这种代替下的近似目标函数 \(L_{\pi}\) 是原来函数的一阶近似

\[ L_{\pi}(\widetilde\pi) \approx \eta(\widetilde\pi) \]

即满足

\(L_{\pi}\) 具体定义表达式为

\(L_{\pi}(\widetilde\pi)\) 是一阶近似意味着在小范围区域中一定是可以得到提升的，但是范围是多大，是否能保证 \(\eta\) 的提升？Kakade的论文中不仅给出了通过mix新老策略的提升方式，还给出了这个方式对原目标 \(\eta\) 较 \(L_{\pi}(\widetilde\pi)\) 的提升下届。

策略更新规则如下

公式6为具体提升下届为

3. 扩展到随机策略

论文的这一部分将Kakade的mix policy update 扩展到一般的随机策略，同时依然保证每次迭代能得到目标提升。

首先，每次策略迭代必须不能和现有策略变化太大，因此，引入分布间常见的TV divergence，即 total variation divergence。

有了两个分布距离的定义，就可以定义两个策略的距离。离散状态下，一个策略是状态到动作分布的 map 或者 dict，因此，可以定义两个策略的距离为所有状态中最大的动作分布的 \(D_{TV}\)，即

进一步将 TV divergence 转换成 KL divergence，转换成KL divergence 的目的是为了后续使用传统且成熟的 trust region 蒙特卡洛方法和 conjugate gradient 的优化近似解法。

由于上面两种距离的大小关系，可以推导出用KL divergence表示的 \(\eta\) 较 \(L_{\pi}(\widetilde\pi)\) 的提升下届

根据公式9，就可以形成初步的概念上的算法一，通过每一步形成无约束优化问题，同时保证每次迭代的 \(\pi_i\) 对应的 \(\eta\) 是递增的。

4. Trust Region Policy Optimization

看到这里已经不容易了，尽管算法一给出了一个解决方案，但是本论文的主角TRPO 还未登场。TRPO算法的作用依然是近似！

算法一对于下面的目标函数做优化，即每次找到下一个 \(\theta_i\) 最大化下式，\(\eta\) 每一步一定能得到提升。

将 \(D^{max}_{KL}\) 放入约束项中符合trust region 这种传统优化解法。

关于 \(D^{max}_{KL}\) 约束，再补充两点

1. 其定义是两个策略中所有状态中最大的动作分布的 \(D_{TV}\) ，因此它约束了所有状态下新老策略动作分布的KL散度，也就意味着有和状态数目相同数量的约束项，海量的约束项导致算法很难应用到实际中。

2. 约束项的 trust region 不是参数 \(\theta\) 的空间，而是其KL散度的空间。

基于第一点，再次使用近似法，在约束项中用KL期望来代替各个状态下的KL散度，权重为on-policy 策略的分布 \(\rho(\theta_{old})\)

最终，得到TRPO在实际中的优化目标（12式）：

5. 用采样方法来Trust Region约束优化

论文第五部分，将TRPO优化目标12式改写成期望形式，引入两种蒙特卡洛方法 single path 和 vine 来采样。

具体来说，\(L_{\theta_{old}}\) 由两项组成 \[ L_{\theta_{old}} = \eta(\theta_{old}) + \sum_s \rho_{\theta_{old}}(s)\sum_a {\pi_{\theta}}(a |s) A_{\theta_{old}}(s,a) \]

第一项是常量，只需优化第二项，即优化问题等价为13式

随后，为了可以适用非 on-policy \(\pi_{\theta_{old}}\) 的动作分布来任意采样，引入采样的动作分布 \(q(a|s)\)，将13式中的 \(\sum_a\) 部分通过重要性采样改成以下形式：

再将13式中的 \(\sum_s \rho(s)\) 改成期望形式 \(\mathbb{E}_{s \sim \rho}\) ，并将 \(A\) 改成 \(Q\) 值，得14式。

至此，我们得到trust region优化的期望形式：优化目标中期望的状态空间是基于 on-policy \(\pi_{\theta_{old}}\)，动作空间是基于任意采样分布 \(q(a|s)\)，优化约束中的期望是基于 on-policy \(\pi_{\theta_{old}}\)。

5.1 Single path采样

根据14式，single path 是最基本的的蒙特卡洛采样方法，和REINFORCE算法一样， 通过on-policy \(\pi_{\theta_{old}}\)生成采样的 trajectory数据： \(s_0, a_0, s_1, a_1, ..., a_{T-1}, s_{T}\)，然后代入14式。注意，此时 \(q(a|s) = \pi_{\theta_{old}}(a|s)\)，即用现有策略的动作分布直接代替采样分布。

5.2 Vine 采样

虽然single path方法简单明了，但是有着online monte carlo方法固有的缺陷，即variance较大。Vine方法通过在一个状态多次采样来改善此缺陷。Vine的翻译是藤，寓意从一个状态多次出发来采样，如下图，\(s_n\) 状态下采样多个rollouts，很像植物的藤长出多分叉。当然，vine方法要求环境能restart 到某一状态，比如游戏环境通过save load返回先前的状态。

具体来说，vine 方法首先通过生成多个on-policy 的trajectories来确定一个状态集合 \(s_1, s_2, ..., s_N\)。对于状态集合的每一个状态 \(s_n\) 采样K个动作，服从 $ a_{n, k} q(s_{n}) $ 。接着，对于每一个 \((s_n, a_{n,k})\) 再去生成一次 rollout 来估计 \(\hat{Q}_{\theta_{i}}\left(s_{n}, a_{n, k}\right)\) 。试验证明，在连续动作空间问题中，\(q\left(\cdot \mid s_{n}\right)\) 直接使用 on-policy 可以取得不错效果，在离散空间问题中，使用uniform分布效果更好。

6. 转换成具体优化问题

再回顾一下现在的进度，12式定义了优化目标，约束项是KL divergence空间的trust region 形式。14式改写成了等价的期望形式，通过两种蒙特卡洛方法生成 state-action 数据集，可以代入14式得到每一步的具体数值的优化问题。论文这一部分简单叙述了如何高效但近似的解此类问题，详细的一些步骤在附录中阐述。我们把相关解读都放在下一节。

7. 和已有理论的联系

7.1 简化成 Natural Policy Gradient

再回到12式，即约束项是KL divergence空间的trust region 形式

对于这种形式的优化问题，一般的做法是通过对优化目标做一阶函数近似，即 \[ L_{\theta_{old}}(\theta) \approx L_{\theta_{old}}\left(\theta_{old}\right)+g^{T}\left(\theta-\theta_{old}\right) \]

\[ \left.g \doteq \nabla_{\theta} L_{\theta_{old}}(\theta)\right|_{\theta_{old}} \]

并对约束函数做二阶函数近似，因为约束函数在 \(\theta_{old}\) 点取到极值，因此一阶导为0。 \[ \bar{D}_{K L}\left(\theta \| \theta_{old}\right) \approx \frac{1}{2}\left(\theta-\theta_{old}\right)^{T} H\left(\theta-\theta_{old}\right) \]

\[ \left.H \doteq \nabla_{\theta}^{2} \bar{D}_{K L}\left(\theta \| \theta_{old}\right)\right|_{\theta_{old}} \]

12式的优化目标可以转换成17式

对应参数迭代更新公式如下

这个方法便是Kakade在2002年发表的 natrual policy gradient 论文。

7.2 简化成 Policy Gradient

注意，\(L_{\theta_{old}}\)的一阶近似的梯度 \[ \left.\nabla_{\theta} L_{\theta_{\text {old }}}(\theta)\right|_{\theta=\theta_{\text {old }}} \cdot\left(\theta-\theta_{\text {old }}\right) \]

即PG定理 \[ \frac{\partial \rho}{\partial \theta}=\sum_{s} d^{\pi}(s) \sum_{a} \frac{\partial \pi(s, a)}{\partial \theta} Q^{\pi}(s, a) \]

7.3 近似数值解法

这里简单描述关于17式及其参数更新规则中的大矩阵数值计算近似方式。

$ {D}_{}^{} $ 二阶近似中的 \(A\) 是 Hessian 方形矩阵，维度为 \(\theta\) 个数的平方。

直接构建 \(A\) 矩阵或者其逆矩阵 \(A^{-1}\)都是计算量巨大的， 注\(A^{-1}\)出现在natural policy update \(\theta\) 更新公式中，\(A^{-1} \nabla_{\theta} L(\theta)\) 。

一种方法是通过构建Fisher Information Matrix，引入期望形式便于采样 \[ \mathbf{A}=E_{\pi_{\theta}}\left[\nabla_{\theta} \log \pi_{\theta}(\mathbf{a} \mid \mathbf{s}) \nabla_{\theta} \log \pi_{\theta}(\mathbf{a} \mid \mathbf{s})^{T}\right] \] 另一种方式是使用conjugate gradient 方法，通过矩阵乘以向量快速计算法迭代逼近 \(A^{-1} \nabla_{\theta} L(\theta)\)。

8. 试验结果

另一种模式是完全信息下的Atari游戏环境，这种环境下观察到的屏幕像素可以信息完全地表达潜在真实状态。

导读：这篇式1999 年Richard Sutton 在强化学习领域中的经典论文，论文证明了策略梯度定理和在用函数近似 Q 值时策略梯度定理依然成立，本文奠定了后续以深度强化学习策略梯度方法的基石。理解熟悉本论文对 Policy Gradient，Actor Critic 方法有很好的指导意义。

论文分成四部分。第一部分指出策略梯度在两种期望回报定义下都成立（定理一）。第二部分提出，如果 \(Q^{\pi}\) 被函数 \(f_w\) 近似时且满足兼容（compatible）条件，以 \(f_w\) 替换策略梯度中的 \(Q^{\pi}\)公式也成立（定理二）。第三部分举Gibbs分布的策略为例，如何应用 \(Q^{\pi}\)近似函数来实现策略梯度算法。第四部分证明了近似函数的策略梯度迭代法一定能收敛到局部最优解。附录部分证明了两种定义下的策略梯度定理。

1. 策略梯度定理

对于Agent和环境而言，可以分成episode和non-episode，后者的时间步骤可以趋近于无穷大，但一般都可以适用两种期望回报定义。一种是单步平均reward ，另一种是指定唯一开始状态并对trajectory求 \(\gamma\)-discounted 之和，称为开始状态定义。两种定义都考虑到了reward的sum会趋近于无穷大，通过不同的方式降低了此问题的概率。

A. 平均reward定义

目标函数 \(\rho(\pi)\) 定义成单步的平均reward，这种情况下等价于稳定状态分布下期望值。

稳定状态分布定义成无限次数后状态的分布。

此时，\(Q^{\pi}\) 定义为无限步的reward sum 减去累积的单步平均 reward \(\rho(\pi)\)，这里减去\(\rho(\pi)\)是为了一定程度防止 \(Q^{\pi}\)没有上界。

B. 开始状态定义

在开始状态定义方式中，某指定状态\(s_0\)作为起始状态，\(\rho(\pi)\) 的定义为 trajectory 的期望回报，注意由于时间步骤 t 趋近于无穷大，必须要乘以discount 系数 \(\gamma < 1\) 保证期望不会趋近无穷大。

\(d^{\pi}\) 依然为无限次数后状态的稳定分布。

策略梯度定理

论文指出上述两种定义都满足策略梯度定理，即目标 \(\rho\) 对于参数 \(\theta\) 的偏导不依赖于 \(d^{\pi}\) 对于 \(\theta\) 偏导，仅取决

关于策略梯度定理的一些综述，可以参考。

论文中还提到策略梯度定理公式和经典的William REINFORCE算法之间的联系。REINFORCE算法即策略梯度的蒙特卡洛实现。

联系如下：

首先，根据策略梯度定理，如果状态 s 是通过 \(\pi\) 采样得到，则下式是$$ 的无偏估计。注意，这里action的summation和 \(\pi\) 是无关的。

2. 函数近似的策略梯度

论文第二部分，进一步引入 \(Q_{\pi}\) 的近似函数 \(f_w\): $ $。

如果我们有\(Q_{\pi}(s_t, a_t)\)的无偏估计，例如 \(R_t\)，很自然，可以让 \(\partial f_w \over \partial w\) 通过最小化 \(R_t\) 和 \(f_w\)之间的差距来计算。

当拟合过程收敛到局部最优时，策略梯度定理中右边项对于 \(w\) 求导为0，可得(3)式。

至此，引出策略梯度定理的延续，即定理2：当 \(f_w\) 满足(3)式同时满足(4)式（称为compatible条件时），可以用 \(f_w(s, a)\)替换原策略梯度中的 \(Q_{\pi}(s,a)\)

3. 一个应用示例

假设一个策略用features的线性组合后的 Gibbs分布来生成，即：

上式和advantage 函数 \(A^{\pi}(s, a)\) 定义一致，因此可以认为 \(f_w\) 的意义是 \(A^{\pi}\) 的近似。

\(A^{\pi}\)具体定义如下

4. 函数近似的策略梯度收敛性证明

这一部分证明了在满足一定条件后，\(\theta\) 可以收敛到局部最优点。

条件为

1. Compatible 条件，公式（4）
2. 任意两个 \(\partial \pi \over \partial \theta\) 偏导是有限的，即

3. 步长数列满足如下条件

4. 环境的 reward 是有限的

   此时，当 \(w_k\) 和 \(\theta_k\) 按如下方式迭代一定能收敛到局部最优。

5. 策略梯度定理的两种情况下的证明

下面简单分解策略梯度的证明步骤。

A. 平均reward 定义下的证明

根据定义，将 \(\theta\) 导数放入求和号中，并分别对乘积中的每项求导。

将\(Q_{\pi}\)的定义代入第二项 \(Q^{\pi}\) 对 \(\theta\) 求偏导中，引入环境reward 随机变量 \(R^a_s\)，环境dynamics \(P\) 和 \(\rho(\pi)\)

\(\theta\) 偏导进一步移入，\(R^a_s\)， \(P\) 不依赖于\(\theta\)。

\(\rho(\pi)\) 对于 \(\theta\) 偏导整理到等式左边

两边同时乘以 \(\sum d^{\pi}\)

由于 \(d^{\pi}\) 是状态在 \(\pi\) 下的平稳分布，\(\sum \pi \sum P\) 项表示 agent 主观 \(\pi\) 和环境客观 \(P\) 对于状态分布的影响，因此可以直接去除。

整理证得。

B. Start-state 定义下的证明

根据定义，将 \(\theta\) 导数放入求和号中，并分别对乘积中的每项求导。

将\(Q_{\pi}\)的定义代入第二项 \(Q^{\pi}\) 对 \(\theta\) 求偏导中，引入环境reward 随机变量 \(R^a_s\)，环境dynamics \(P\)

\(\theta\) 偏导进一步移入，\(R^a_s\)， \(P\) 不依赖于\(\theta\)。注意，此式表示从状态 \(s\) 出发一步之后的能到达的所有 \(s^{\prime}\) ，将次式反复unroll \(V^{\pi}\) 成 \(Q^{\pi}\) 之后得到

\(\operatorname{Pr}(s \rightarrow x, k, \pi)\) 表示 k 步后 状态 s 能到达的所有状态 x

根据定义，\(\rho = V^{\pi}(s_0)\)

将 \(V^{\pi}(s_0)\) 替换成unroll 成 \(Q^{\pi}\) 的表达式

\(\operatorname{Pr}(s \rightarrow x, k, \pi)\) 即 \(d^{\pi}\)

Policy gradient 定理作为现代深度强化学习的基石，同时也是actor-critic的基础，重要性不言而喻。但是它的推导和理解不是那么浅显，不同的资料中又有着众多形式，不禁令人困惑。本篇文章MyEncyclopedia试图总结众多资料背后的一些相通的地方，并写下自己的一些学习理解心得。

引入 Policy Gradient

那么问题来了，首先一个如何定义 \(J(\theta)\)，其次，如何求出或者估计 $ J()$。

第一个问题比较直白，用value function或者广义的expected return都可以。

这里列举一些常见的定义。对于episodic 并且初始都是 \(s_0\)状态的情况，直接定义成v值，即Sutton教程中的episodic情况下的定义

其中，状态平稳分布 \(d^{\pi}(s)\) 定义为

即$ $ 是一次trajectory，服从以 \(\theta\) 作为参数的随机变量

\(J(\theta)\) 对于所有的可能的 \(\tau\) 求 expected return。这种视角下对于finite 和 infinite horizon来说也有变形。

当环境dynamics未知时，如何再去求 $ J()$ 呢。还有就是如果涉及到状态的分布也是取决于环境dynamics的，计算 $ J()$ 也面临同样的问题。

幸好，policy gradient定理完美的解答了上述问题。我们先来看看它的表述内容。

Policy Gradient Theorem

Policy Gradient 定理的伟大之处在于等式右边并没有 \(d^{\pi}(s)\)，或者环境transition model \(p\left(s^{\prime}, r \mid s, a\right)\)！同时，等式右边变换成了最利于统计采样的期望形式，因为期望可以通过样本的平均来估算。

但是，这里必须注意的是action space的期望并不是基于 $(a S_{t}, ) $ 的权重的，因此，继续改变形式，引入 action space的 on policy 权重 $(a S_{t}, ) $ ，得到 2.3式。

将 \(q_{\pi}\)替换成 \(G_t\)，由于

得到2.5式

Policy Gradient Theorem - Trajectory Form

和监督学习的联系

Policy Gradient中的 \(\nabla_{\theta} \log \pi\) 广泛存在在机器学习范畴中，被称为 score function gradient estimator。RL 在supervised learning settings 中有 imitation learning，即通过专家的较优stochastic policy \(\pi_{\theta}(a|s)\) 收集数据集

对照Policy Graident RL，on-policy \(\pi_{\theta}(a|s)\) 产生数据集

目标是最大化on-policy \(\pi_{\theta}\) 分布下的expected return

对比 4.3 和 4.2，发现此时4.3中只多了一个权重系数 \(G_n\)。

关于 $G_{n} {} {}(a_{n} s_{n}) $ 或者 \(G_{t} \frac{\nabla \pi\left(A_{t} \mid S_{t}, \boldsymbol{\theta}_{t}\right)}{\pi\left(A_{t} \mid S_{t}, \boldsymbol{\theta}_{t}\right)}\) 有一些深入的理解。

Policy Gradient 工作的机制大致如下

首先，根据现有的 on-policy \(\pi_{\theta}\) 采样出一些动作 action 产生trajectories，这些trajectories最终得到反馈 \(R(\tau)\)

最后，由于采样到的action分布服从于\(a \sim \pi_{\theta}(a)\) ，除掉 \(\pi_{\theta}\) ：

\(G_{t} \frac{\nabla \pi\left(A_{t} \mid S_{t}, \boldsymbol{\theta}_{t}\right)}{\pi\left(A_{t} \mid S_{t}, \boldsymbol{\theta}_{t}\right)}\)

此时，采样的均值可以去无偏估计2.2式中的Expectation。

上一期 MyEncyclopedia公众号文章 从Q-Learning 演化到 DQN，我们从原理上讲解了DQN算法，这一期，让我们通过代码来实现任天堂游戏机中经典的超级玛丽的自动通关吧。本文所有代码在 https://github.com/MyEncyclopedia/reinforcement-learning-2nd/tree/master/super_mario。

DQN 算法回顾

上期详细讲解了DQN中的两个重要的技术：Target Network 和 Experience Replay，正是有了它们才使得 Deep Q Network在实战中容易收敛，以下是Deepmind 发表在Nature 的 Human-level control through deep reinforcement learning 的完整算法流程。

超级玛丽 NES OpenAI 环境

安装基于OpenAI gym的超级玛丽环境执行下面的 pip 命令即可。

1

pip install gym-super-mario-bros

我们先来看一下游戏环境的输入和输出。下面代码采用随机的action来和游戏交互。有了 组合游戏系列3: 井字棋、五子棋的OpenAI Gym GUI环境 对于OpenAI Gym 接口的介绍，现在对于其基本的交互步骤已经不陌生了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

import gym_super_mario_bros
from random import random, randrange
from gym_super_mario_bros.actions import RIGHT_ONLY
from nes_py.wrappers import JoypadSpace
from gym import wrappers

env = gym_super_mario_bros.make('SuperMarioBros-v0')
env = JoypadSpace(env, RIGHT_ONLY)

# Play randomly
done = False
env.reset()

step = 0
while not done:
    action = randrange(len(RIGHT_ONLY))
    state, reward, done, info = env.step(action)
    print(done, step, info)
    env.render()
    step += 1

env.close()

游戏render效果如下

。。。

注意我们在游戏环境初始化的时候用了参数 RIGHT_ONLY，它定义成五种动作的list，表示仅使用右键的一些组合，适用于快速训练来完成Mario第一关。

1
2
3
4
5
6
7

RIGHT_ONLY = [
    ['NOOP'],
    ['right'],
    ['right', 'A'],
    ['right', 'B'],
    ['right', 'A', 'B'],
]

观察一些 info 输出内容，coins表示金币获得数量，flag_get 表示是否取得最后的旗子，time 剩余时间，以及 Mario 大小状态和所在的 x，y位置。

1
2
3
4
5
6
7
8
9
10
11
12

{
   "coins":0,
   "flag_get":False,
   "life":2,
   "score":0,
   "stage":1,
   "status":"small",
   "time":381,
   "world":1,
   "x_pos":594,
   "y_pos":89
}

游戏图像处理

Deep Reinforcement Learning 一般是 end-to-end learning，意味着游戏的 screen image 作为observation直接视为真实状态，喂给神经网络训练。于此相反的另一种做法是，通过游戏环境拿到内部状态，例如所有相关物品的位置和属性作为模型输入。这两种方式的区别有两点。第一点，用观察到的屏幕像素代替真正的状态 s，在partially observable 的环境时可能因为 non-stationarity 导致无法很好的工作，而拿内部状态利用了额外的作弊信息，在partially observable环境中也可以工作。第二点，第一种方式屏幕像素维度比较高，输入数据量大，需要神经网络的大量训练拟合，第二种方式，内部真实状态往往维度低得多，训练起来很快，但缺点是因为除了内部状态往往还需要游戏相关规则作为输入，因此generalization能力不如前者强。

这里，我们当然采样屏幕像素的 end-to-end 方式了，自然首要任务是将游戏帧图像有效处理。超级玛丽游戏环境的屏幕输出是 (240, 256, 3) shape的 numpy array，通过下面一系列的转换，尽可能的在不影响训练效果的情况下减小采样到的数据量。

1. MaxAndSkipFrameWrapper：每4个frame连在一起，采取同样的动作，降低frame数量。

2. FrameDownsampleWrapper：将原始的 (240, 256, 3) down sample 到 (84, 84, 1)

3. ImageToPyTorchWrapper：转换成适合 pytorch 的 (1, 84, 84) shape

4. FrameBufferWrapper：保存最后4次屏幕采样

5. NormalizeFloats：Normalize 成 [0., 1.0] 的浮点值

1
2
3
4
5
6
7
8
9

def wrap_environment(env_name: str, action_space: list) -> Wrapper:
    env = make(env_name)
    env = JoypadSpace(env, action_space)
    env = MaxAndSkipFrameWrapper(env)
    env = FrameDownsampleWrapper(env)
    env = ImageToPyTorchWrapper(env)
    env = FrameBufferWrapper(env, 4)
    env = NormalizeFloats(env)
    return env

CNN 模型

模型比较简单，三个卷积层后做 softmax输出，输出维度数为离散动作数。act() 采用了epsilon-greedy 模式，即在epsilon小概率时采取随机动作来 explore，大于epsilon时采取估计的最可能动作来 exploit。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

class DQNModel(nn.Module):
    def __init__(self, input_shape, num_actions):
        super(DQNModel, self).__init__()
        self._input_shape = input_shape
        self._num_actions = num_actions

        self.features = nn.Sequential(
            nn.Conv2d(input_shape[0], 32, kernel_size=8, stride=4),
            nn.ReLU(),
            nn.Conv2d(32, 64, kernel_size=4, stride=2),
            nn.ReLU(),
            nn.Conv2d(64, 64, kernel_size=3, stride=1),
            nn.ReLU()
        )

        self.fc = nn.Sequential(
            nn.Linear(self.feature_size, 512),
            nn.ReLU(),
            nn.Linear(512, num_actions)
        )

    def forward(self, x):
        x = self.features(x).view(x.size()[0], -1)
        return self.fc(x)

    def act(self, state, epsilon, device):
        if random() > epsilon:
            state = torch.FloatTensor(np.float32(state)).unsqueeze(0).to(device)
            q_value = self.forward(state)
            action = q_value.max(1)[1].item()
        else:
            action = randrange(self._num_actions)
        return action

Experience Replay 缓存

实现采用了 Pytorch CartPole DQN 的官方代码，本质是一个最大为 capacity 的 list 保存采样的 (s, a, r, s', is_done) 五元组。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

Transition = namedtuple('Transition', ('state', 'action', 'reward', 'next_state', 'done'))

class ReplayMemory:

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

    def push(self, *args):
        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):
        return len(self.memory)

DQNAgent

我们将 DQN 的逻辑封装在 DQNAgent 类中。DQNAgent 成员变量包括两个 DQNModel，一个ReplayMemory。

train() 方法中会每隔一定时间将 Target Network 的参数同步成现行Network的参数。在td_loss_backprop()方法中采样 ReplayMemory 中的五元组，通过minimize TD error方式来改进现行 Network 参数 \(\theta\)。Loss函数为：

\[ L\left(\theta_{i}\right)=\mathbb{E}_{\left(s, a, r, s^{\prime}\right) \sim \mathrm{U}(D)}\left[\left(r+\gamma \max _{a^{\prime}} Q_{target}\left(s^{\prime}, a^{\prime} ; \theta_{i}^{-}\right)-Q\left(s, a ; \theta_{i}\right)\right)^{2}\right] \]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

class DQNAgent():

    def act(self, state, episode_idx):
        self.update_epsilon(episode_idx)
        action = self.model.act(state, self.epsilon, self.device)
        return action

    def process(self, episode_idx, state, action, reward, next_state, done):
        self.replay_mem.push(state, action, reward, next_state, done)
        self.train(episode_idx)

    def train(self, episode_idx):
        if len(self.replay_mem) > self.initial_learning:
            if episode_idx % self.target_update_frequency == 0:
                self.target_model.load_state_dict(self.model.state_dict())
            self.optimizer.zero_grad()
            self.td_loss_backprop()
            self.optimizer.step()

    def td_loss_backprop(self):
        transitions = self.replay_mem.sample(self.batch_size)
        batch = Transition(*zip(*transitions))

        state = Variable(FloatTensor(np.float32(batch.state))).to(self.device)
        action = Variable(LongTensor(batch.action)).to(self.device)
        reward = Variable(FloatTensor(batch.reward)).to(self.device)
        next_state = Variable(FloatTensor(np.float32(batch.next_state))).to(self.device)
        done = Variable(FloatTensor(batch.done)).to(self.device)

        q_values = self.model(state)
        next_q_values = self.target_net(next_state)

        q_value = q_values.gather(1, action.unsqueeze(-1)).squeeze(-1)
        next_q_value = next_q_values.max(1)[0]
        expected_q_value = reward + self.gamma * next_q_value * (1 - done)

        loss = (q_value - expected_q_value.detach()).pow(2)
        loss = loss.mean()
        loss.backward()

外层 Training 代码

最后是外层调用代码，基本和以前文章一样。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

def train(env, args, agent):
    for episode_idx in range(args.num_episodes):
        episode_reward = 0.0
        state = env.reset()

        while True:
            action = agent.act(state, episode_idx)
            if args.render:
                env.render()
            next_state, reward, done, stats = env.step(action)
            agent.process(episode_idx, state, action, reward, next_state, done)
            state = next_state
            episode_reward += reward
            if done:
                print(f'{episode_idx}: {episode_reward}')
                break

Berkeley 2017年联合了DeepMind 以及 OpenAI 举办了一个大咖云集的深度强化学习训练营，是难得的前沿深度强化学习佳品，本公众号 MyEncyclopedia 用代码实现了权威教材 Sutton & Barto 第二版强化学习的基础部分之后，会大致沿着这个训练营的思路，从原理到代码逐步揭示强化深度学习面纱，并结合各种有意思的游戏环境来演示。

如果没有耐心的同学可以直接跳到文末的百度云盘下载链接，内容涵盖所有视频和slide。

此次训练营主讲的强化学习领域专家包括

• Pieter Abbeel，前Berkeley 机器人学习实验室主任，伯克利人工智能研究(BAIR)实验室联合主任

• Andrej Karpathy，前 OpenAI研究科学家、现特斯拉AI总监

• Vlad Mnih，Deepmind 研究科学家

• John Schulman，Deepmind 研究科学家，OpenAI共同创建人

• Sergey Levine，Berkeley 计算机副教授

课程列表

• Core Lecture 1 Intro to MDPs and Exact Solution Methods -- Pieter Abbeel
• Core Lecture 2 Sample-based Approximations and Fitted Learning -- Rocky Duan
• Core Lecture 3 DQN + Variants -- Vlad Mnih
• Core Lecture 4a Policy Gradients and Actor Critic -- Pieter Abbeel
• Core Lecture 4b Pong from Pixels -- Andrej Karpathy
• Core Lecture 5 Natural Policy Gradients, TRPO, and PPO -- John Schulman
• Core Lecture 6 Nuts and Bolts of Deep RL Experimentation -- John Schulman
• Core Lecture 7 SVG, DDPG, and Stochastic Computation Graphs -- John Schulman
• Core Lecture 8 Derivative-free Methods -- Peter Chen
• Core Lecture 9 Model-based RL -- Chelsea Finn
• Core Lecture 10a Utilities -- Pieter Abbeel
• Core Lecture 10b Inverse RL -- Chelsea Finn
• Frontiers Lecture I: Recent Advances, Frontiers and Future of Deep RL -- Vlad Mnih
• Frontiers Lecture II: Recent Advances, Frontiers and Future of Deep RL -- Sergey Levine
• TAs Research Overviews

前两讲总结了强化学习基础理论方面，包括用动态规划求精确解，采样与环境交互的传统基本方法。第三四讲覆盖了主流的深度强化学习的几种模式：DQN，PG和AC。第五到七讲深入了深度强化学习的各种前沿方法。值得一提的是第六讲，很好的从实践中总结了各种调试诊断方法。余下的若干讲涉及到了非主流的剩余强化学习领域。

下载方法

关注 MyEncyclopedia 公众号，输入 rl-bootcamp-ucb-2017 即可获得百度云盘链接

上一期 MyEncyclopedia公众号文章 SARSA、Q-Learning和Expected SARSA时序差分算法训练CartPole中，我们通过CartPole的OpenAI Gym环境实现了Q-learning算法，这一期，我们将会分析Q-learning算法面临的maximization bias 问题和提出double learning算法来改进。接着，我们将tabular Q-learning算法扩展到用带参函数来近似 Q(s, a)，这就是Deepmind 在2015年Nature上发表的Deep Q Network （DQN）思想：用神经网络结合Q-learning算法实现超越人类玩家打Atari游戏的水平。

Q-Learning 回顾

在SARSA、Q-Learning和Expected SARSA时序差分算法训练CartPole 中，我们实现了同样基于 \(\epsilon\)-greedy 策略的Q-learning算法和SARSA算法，两者代码上的区别确实不大，但本质上Q-learning是属于 off-policy 范畴而 SARSA却属于 on-policy 范畴。一种理解方式是，Q-learning相比于SARSA少了第二次从 \(\epsilon\)-greedy 策略采样出下一个action，即S, A, R', S', A' 五元组中最后一个A'，而直接通过max操作去逼近 \(q^{*}\)。如此，Q-learning并没有像SARSA完成一次完整的GPI（Generalized Policy Iteration），缺乏on-policy的策略迭代的特点，故而 Q-learning 属于off-policy方法。我们也可以从另一个角度来分析两者的区别。注意到这两个算法不是一定非要使用 \(\epsilon\)-greedy 策略的。对于Q-learning来说，完全可以使用随机策略，理论上已经证明，只要保证每个action以后依然有几率会被探索下去，Q-learning 最终会收敛到最优策略。Q-learning使用 \(\epsilon\)-greedy 是为了能快速收敛。对于SARSA算法来说，则无法使用随机策略，因为随机策略无法形成策略提升。而 \(\epsilon\)-greedy 策略却可以形成策略迭代，完成策略提升，当然，\(\epsilon\)-greedy 策略在 SARSA 算法中也可以保证快速收敛。因此，尽管两者都使用 \(\epsilon\)-greedy 策略再借由环境产生reward和state，它们的作用并非完全一样。至此，我们可以体会到on-policy和off-policy本质的区别。

收敛条件

一种方式是将 \(\alpha\)设置成 (s, a)访问次数的倒数：\(\alpha_{n}(s,a) = 1/ n(s,a )\)

则整体更新公式为

\[ Q(s,a) \leftarrow Q(s,a) + \alpha_n(s, a)[R+\gamma \max_{a^{\prime}}Q(s^{\prime}, a^{\prime}) - Q(s, a)] \]

Q-Learning 最大化偏差问题

Q-Learning 会产生最大化偏差问题（Maximization Bias，在Sutton 教材6.7节），它的原因是用估计值中取最大值去估计真实值中最大是有偏的。这个可以做如下试验来模拟，若有5个 [-3, 3] 的离散均匀分布 \(d_i\)，\(\max(\mathbb{E}[d_i]) = 0\)，但是若我们用单批采样 \(x_i \sim d_i\)来估算 \(\mathbb{E}[d_i]\)在取max的话，\(\mathbb{E}[{\max(x_i)]}\) 是有bias的。但是如果我们将这个过程分解成选择最大action和评估其值两步，每一步用独立的采样集合就可以做到无偏，这个改进方法称为double learning。具体过程为第一步在\(Q_1\)集合中找到最大的action，第二步在\(Q_2\)中返回此action值，即：

则无限模拟后结果是无偏的：\(\mathbb{E}[Q_2(A^{\star})] = q(A^{\star})\) 下面是简单模拟试验两种方法的均值比较

试验完整代码如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

import random
from math import floor
import numpy as np
import pandas as pd
import seaborn as sns


def uniform(a: int, b: int) -> int:
    u = random.random()
    return a + floor((b - a + 1) * u)


if __name__ == "__main__":
    total_max_bias = 0
    avgs_max_bias = []
    total_double_sampling = 0
    avgs_double_sampling = []

    for e in range(1, 100):
        samples = np.array([uniform(-3, 3) for _ in range(5)])
        max_sample = max(samples)
        total_max_bias += max_sample
        avgs_max_bias.append(total_max_bias / e)

        samples2 = np.array([uniform(-3, 3) for _ in range(5)])
        total_double_sampling += samples2[np.argmax(samples)]
        avgs_double_sampling.append(total_double_sampling / e)

    df = pd.DataFrame({'Max of Samples': avgs_max_bias, 'Double Samples': avgs_double_sampling})
    import matplotlib.pyplot as plt
    sns.lineplot(data=df)
    plt.show()

回到Q-learning 中使用的 \(\epsilon\)-greedy策略，Q-learning可以保证随着\(\epsilon\) 的减小，最大化偏差会 asymptotically 趋近于真实值，但是double learning 可以更快地趋近于真实值。

下面是Sutton 强化学习第二版6.7节中完整的Double Q-learning算法。

更详细内容，可以参考 Hado V. Hasselt 的 Double Q-learning paper [3]。

Gradient Q-Learning

Tabular Q-learning由于受制于维度爆炸，无法扩展到高维状态空间，一般近似解决方案是用 approximating function来逼近Q函数。即我们将状态抽象出一组特征 \(s = \vec x= [x_1, x_2, ..., x_n]^T\)，Q 用一个 x 的函数来近似表达 \(Q(s, a) \approx g(\vec x; \theta)\)，如此，就联系起了深度神经网络。有了函数表达，深度学习还必须的元素是损失函数，这个很自然的可以用 TD error。至此，问题转换成深度学习的几个要素均已具备，Q-learning算法改造成了深度学习中的有监督问题。

估计值：\(Q\left(s, a ; \theta\right)\)

目标值：\(r+\gamma \max _{a^{\prime}} Q\left(s^{\prime}, a^{\prime} ; \theta\right)\)

损失函数：

\[ L\left(\theta\right)=\mathbb{E}_{\left(s, a, r, s^{\prime}\right) \sim \mathrm{U}(D)}\left[\left(r+\gamma \max _{a^{\prime}} Q\left(s^{\prime}, a^{\prime} ; \theta\right)-Q\left(s, a ; \theta\right)\right)^{2}\right] \]

收敛性分析

首先明确一点，至此 gradient q-learning 和 tabular Q-learning 一样，都是没有记忆的，即对于一个新的环境产生的 sample 去做 stochastic online update。

若Q函数是状态特征的线性函数，即 \(Q(s, a; \theta) = \Sigma_i w_i x_i\) ，那么线性Gradient Q-learning的收敛条件和Tabular Q-learning 一样，也为

若Q函数是非线性函数，即使符合上述条件，也无法保证收敛，本质上源于改变 \(\theta\) 使得 Q 值在 (s, a) 点上减小误差会影响 (s, a) 周边点的误差。

DQN减少不收敛的两个技巧

1. \(\theta_{i-1} \rightarrow \theta_{i}\) 改变导致max中的估计值和目标值中的Q同时变化，面临着 chasing its own tail 的问题。解决的方法是使用不同的参数来parameterize两个Q，并且目标值的Q网络参数固定一段时间产生一批固定策略下的环境采样。这个技巧称为 Target Network。引入这个 trick 后深度学习的要素变成

估计值：\(Q\left(s, a ; \theta_{i}\right)\)

目标值：\(r+\gamma \max _{a^{\prime}} Q\left(s^{\prime}, a^{\prime} ; \theta_i^{-}\right)\)

损失函数，DQN在Nature上的loss函数： \[ L\left(\theta_{i}\right)=\mathbb{E}_{\left(s, a, r, s^{\prime}\right) \sim \mathrm{U}(D)}\left[\left(r+\gamma \max _{a^{\prime}} Q\left(s^{\prime}, a^{\prime} ; \theta_{i}^{-}\right)-Q\left(s, a ; \theta_{i}\right)\right)^{2}\right] \]

2. 尽管目标值的 \(Q(;\theta^{-})\)固定了，但是\(\theta_{i-1} \rightarrow \theta_{i}\) 还会使得估计值的 \(Q(s, a;\theta_i)\) 在变化的同时影响其他的 \(Q(s_k, a_j;\theta_i)\)，让之前训练过的 (s, a)的点的损失值发生变化，解决的办法是将 online stochastic 改成 batch gradient，也就是将最近的一系列采样值保存下来，这个方法称为 experience replay。

有了这两个优化，Deep Q Network投入实战效果就容易收敛了，以下是Deepmind 发表在Nature 的 Human-level control through deep reinforcement learning [1] 的完整算法流程。

DQN with Double Q-Learning

DQN 算法和 Double Q-Learning 能不能结合起来呢？Hado van Hasselt 在 Deep Reinforcement Learning with Double Q-learning [4] 中提出参考 Double Q-learning 将 DQN 的目标值改成如下函数，可以进一步提升最初DQN的效果。

目标值：\(r+\gamma Q(s^{\prime}, \max _{a^{\prime}} Q\left(s^{\prime}, a^{\prime}; \theta_t\right); \theta_t^{-})\)

参考资料

1. Human-level control through deep reinforcement learning Volodymyr Mnih, Koray Kavukcuoglu, David Silver (2015)

2. CS885 Reinforcement Learning Lecture 4b: May 11, 2018

3. Double Q-learning Hado V. Hasselt (2010)

4. Deep Reinforcement Learning with Double Q-learning Hado van Hasselt, Arthur Guez, David Silver (2015)

这一期我们进入第六章：时序差分学习（Temporal-Difference Learning）。TD Learning本质上是加了bootstrapping的蒙特卡洛（MC），也是model-free的方法，但实践中往往比蒙特卡洛收敛更快。我们选取OpenAI Gym中经典的CartPole环境来讲解TD。更多相关内容，欢迎关注 本公众号 MyEncyclopedia。

CartPole OpenAI 环境

如图所示，小车上放了一根杆，杆会根据物理系统定理因重力而倒下，我们可以控制小车往左或者往右，目的是尽可能地让杆保持树立状态。

CartPole 观察到的状态是四维的float值，分别是车位置，车速度，杆角度和杆角速度。下表为四个维度的值范围。给到小车的动作，即action space，只有两种：0，表示往左推；1，表示往右推。

离散化连续状态

从上所知，CartPole step() 函数返回了4维ndarray，类型为float32的连续状态空间。对于传统的tabular方法来说第一步必须离散化状态，目的是可以作为Q table的主键来查找。下面定义的State类型是离散化后的具体类型，另外 Action 类型已经是0和1，不需要做离散化处理。

1
2

State = Tuple[int, int, int, int]
Action = int

离散化处理时需要考虑的一个问题是如何设置每个维度的分桶策略。分桶策略会决定性地影响训练的效果。原则上必须将和action以及reward强相关的维度做细粒度分桶，弱相关或者无关的维度做粗粒度分桶。举个例子，小车位置本身并不能影响Agent采取的下一动作，当给定其他三维状态的前提下，因此我们对小车位置这一维度仅设置一个桶（bucket size=1）。而杆的角度和角速度是决定下一动作的关键因素，因此我们分别设置成6个和12个。

以下是离散化相关代码，四个维度的 buckets=(1, 2, 6, 12)。self.q是action value的查找表，具体类型是shape 为 (1, 2, 6, 12, 2) 的ndarray。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

class CartPoleAbstractAgent(metaclass=abc.ABCMeta):
    def __init__(self, buckets=(1, 2, 6, 12), discount=0.98, lr_min=0.1, epsilon_min=0.1):
        self.env = gym.make('CartPole-v0')

        env = self.env
        # [position, velocity, angle, angular velocity]
        self.dims_config = [(env.observation_space.low[0], env.observation_space.high[0], 1),
                            (-0.5, 0.5, 1),
                            (env.observation_space.low[2], env.observation_space.high[2], 6),
                            (-math.radians(50) / 1., math.radians(50) / 1., 12)]
        self.q = np.zeros(buckets + (self.env.action_space.n,))
        self.pi = np.zeros_like(self.q)
        self.pi[:] = 1.0 / env.action_space.n

    def to_bin_idx(self, val: float, lower: float, upper: float, bucket_num: int) -> int:
        percent = (val + abs(lower)) / (upper - lower)
        return min(bucket_num - 1, max(0, int(round((bucket_num - 1) * percent))))

    def discretize(self, obs: np.ndarray) -> State:
        discrete_states = tuple([self.to_bin_idx(obs[d], *self.dims_config[d]) for d in range(len(obs))])
        return discrete_states

train() 方法串联起来 agent 和 env 交互的流程，包括从 env 得到连续状态转换成离散状态，更新 Agent 的 Q table 甚至 Agent的执行policy，choose_action会根据执行 policy 选取action。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

def train(self, num_episodes=2000):
    for e in range(num_episodes):
        print(e)
        s: State = self.discretize(self.env.reset())

        self.adjust_learning_rate(e)
        self.adjust_epsilon(e)
        done = False

        while not done:
            action: Action = self.choose_action(s)
            obs, reward, done, _ = self.env.step(action)
            s_next: State = self.discretize(obs)
            a_next = self.choose_action(s_next)
            self.update_q(s, action, reward, s_next, a_next)
            s = s_next

choose_action 的默认实现为基于现有 Q table 的 \(\epsilon\)-greedy 策略。

1
2
3
4
5

def choose_action(self, state) -> Action:
    if np.random.random() < self.epsilon:
        return self.env.action_space.sample()
    else:
        return np.argmax(self.q[state])

抽象出公共的基类代码 CartPoleAbstractAgent 之后，SARSA、Q-Learning和Expected SARSA只需要复写 update_q 抽象方法即可。

1
2
3
4

class CartPoleAbstractAgent(metaclass=abc.ABCMeta):
    @abc.abstractmethod
    def update_q(self, s: State, a: Action, r, s_next: State, a_next: Action):
        pass

TD Learning的精髓

在上一期，本公众号 MyEncyclopedia 的21点游戏的蒙特卡洛On-Policy控制介绍了Monte Carlo方法，知道MC需要在环境中模拟直至最终结局。若记\(G_t\)为t步以后的最终return，则 MC online update 版本更新为：

\[ V(S_t) \leftarrow V(S_t) + \alpha[G_{t} - V(S_t)] \]

可以认为 \(V(S_t)\) 向着目标为 \(G_t\) 更新了一小步。

而TD方法可以只模拟下一步，得到 \(R_{t+1}\)，而余下步骤的return，\(G_t - R_{t+1}\) 用已有的 \(V(S_{t+1})\) 来估计，或者统计上称作bootstrapping。这样 TD 的更新目标值变成 \(R_{t+1} + \gamma V(S_{t+1})\)，整体online update 公式则为： \[ V(S_t) \leftarrow V(S_t) + \alpha[R_{t+1} + \gamma V(S_{t+1})- V(S_t)] \]

概念上，如果只使用下一步 \(R_{t+1}\) 值然后bootstrap称为 TD(0)，用于区分使用多步后的reward的TD方法。另外，变化的数值 \(R_{t+1} + \gamma V(S_{t+1})- V(S_t)\) 称为TD error。

另外一个和Monte Carlo的区别在于一般TD方法保存更精细的Q值，\(Q(S_t, A_t)\)，并用Q值来boostrap，而MC一般用V值也可用Q值。

SARSA: On-policy TD 控制

SARSA的命名源于一次迭代产生了五元组 \(S_t，A_t，R_{t+1}，S_{t+1}，A_{t+1}\)。SARSA利用五个值做 action-value的 online update：

\[ Q(S_t,A_t) \leftarrow Q(S_t,A_t) + \alpha[R_{t+1}+\gamma Q(S_{t+1}, A_{t+1}) - Q(S_t,A_t)] \]

对应的Q table更新实现为：

1
2
3
4

class SarsaAgent(CartPoleAbstractAgent):

    def update_q(self, s: State, a: Action, r, s_next: State, a_next: Action):
        self.q[s][a] += self.lr * (r + self.discount * (self.q[s_next][a_next]) - self.q[s][a])

SARSA 在执行policy 后的Q值更新是对于针对于同一个policy的，完成了一次策略迭代（policy iteration），这个特点区分于后面的Q-learning算法，这也是SARSA 被称为 On-policy 的原因。下面是完整算法伪代码。

SARSA 训练分析

SARSA收敛较慢，1000次episode后还无法持久稳定，后面的Q-learning 和 Expected Sarsa 都可以在1000次episode学习长时间保持不倒的状态。

Q-Learning: Off-policy TD 控制

Q-Learning 是深度学习时代前强化学习领域中的著名算法，它的 online update 公式为： \[ Q(S_t,A_t) \leftarrow Q(S_t,A_t) + \alpha[R_{t+1}+\gamma \max_{a}Q(S_{t+1}, a) - Q(S_t,A_t)] \]

对应的 update_q() 方法具体实现

1
2
3
4

class QLearningAgent(CartPoleAbstractAgent):

    def update_q(self, s: State, a: Action, r, s_next: State, a_next: Action):
        self.q[s][a] += self.lr * (r + self.discount * np.max(self.q[s_next]) - self.q[s][a])

Q-Learning 被称为 off-policy 的原因是它并没有完成一次policy iteration，而是直接用已有的 Q 来不断近似 \(Q_{*}\)。

对比下面的Q-Learning 伪代码和之前的 SARSA 版本可以发现，Q-Learning少了一次模拟后的 \(A_{t+1}\)，这也是Q-Learning 中执行policy和预估Q值（即off-policy）分离的一个特征。

Q-Learning 训练分析

Q-Learning 1000次episode就可以持久稳定住。

SARSA 改进版 Expected SARSA

Expected SARSA 改进了 SARSA 的地方在于考虑到了在某一状态下的现有策略动作分布，以此来减少variance，加快收敛，具体更新规则为：

注意在实现中，update_q() 不仅更新了Q table，还显示更新了执行policy \(\pi\)。

1
2
3
4
5
6
7
8
9

class ExpectedSarsaAgent(CartPoleAbstractAgent):

    def update_q(self, s: State, a: Action, r, s_next: State, a_next: Action):
        self.q[s][a] = self.q[s][a] + self.lr * (r + self.discount * np.dot(self.pi[s_next], self.q[s_next]) - self.q[s][a])
        # update pi[s]
        best_a = np.random.choice(np.where(self.q[s] == max(self.q[s]))[0])
        n_actions = self.env.action_space.n
        self.pi[s][:] = self.epsilon / n_actions
        self.pi[s][best_a] = 1 - (n_actions - 1) * (self.epsilon / n_actions)

同样的，Expected SARSA 1000次迭代也能比较好的学到最佳policy。

这期继续Sutton强化学习第二版，第五章蒙特卡洛方法。在上期21点游戏的策略蒙特卡洛值预测学习了如何用Monte Carlo来预估给定策略 \(\pi\) 的 \(V_{\pi}\) 值之后，这一期我们用Monte Carlo方法来解得21点游戏最佳策略 \(\pi_{*}\)。

蒙特卡洛策略提升

回顾一下，在Grid World 策略迭代和值迭代中由于存在Policy Improvement Theorem，我们可以利用环境dynamics信息计算出策略v值，再选取最greedy action的方式改进策略，形成策略提示最终能够不断逼近最佳策略。 \[ \pi_{0} \stackrel{\mathrm{E}}{\longrightarrow} v_{\pi_{0}} \stackrel{\mathrm{I}}{\longrightarrow} \pi_{1} \stackrel{\mathrm{E}}{\longrightarrow} v_{\pi_{1}} \stackrel{\mathrm{I}}{\longrightarrow} \pi_{2} \stackrel{\mathrm{E}}{\longrightarrow} \cdots \stackrel{\mathrm{I}}{\longrightarrow} \pi_{*} \stackrel{\mathrm{E}}{\longrightarrow} v_{*} \] Monte Carlo Control方法搜寻最佳策略 \(\pi{*}\)，是否也能沿用同样的思路呢？答案是可行的。不过，不同于第四章中我们已知环境MDP就知道状态的前后依赖关系，进而从v值中能推断出策略 \(\pi\)，在Monte Carlo方法中，环境MDP是未知的，因而我们只能从action-value中下手，通过海量Monte Carlo试验来近似 \(q_{\pi}\)。有了策略 Q 值，再和MDP策略迭代方法一样，选取最greedy action的策略，这种策略提示方式理论上被证明了最终能够不断逼近最佳策略。 \[ \pi_{0} \stackrel{\mathrm{E}}{\longrightarrow} q_{\pi_{0}} \stackrel{\mathrm{I}}{\longrightarrow} \pi_{1} \stackrel{\mathrm{E}}{\longrightarrow} q_{\pi_{1}} \stackrel{\mathrm{I}}{\longrightarrow} \pi_{2} \stackrel{\mathrm{E}}{\longrightarrow} \cdots \stackrel{\mathrm{I}}{\longrightarrow} \pi_{*} \stackrel{\mathrm{E}}{\longrightarrow} q_{*} \]

但是此方法有一个前提要满足，由于数据是依据策略 \(\pi_{i}\) 生成的，理论上需要保证在无限次的模拟过程中，每个状态都必须被无限次访问到，才能保证最终每个状态的Q估值收敛到真实的 \(q_{*}\)。满足这个前提的一个简单实现是强制随机环境初始状态，保证每个状态都能有一定概率被生成。这个思路就是 Monte Carlo Control with Exploring Starts算法，伪代码如下：

下面我们实现21点游戏的Monte Carlo ES 算法。21点游戏只有200个有效的状态，可以满足算法要求的生成episode前先随机选择某一状态的前提条件。

相对于上一篇，我们增加 ActionValue和Policy的类型定义，ActionValue表示 \(q(s, a)\) ，是一个State到动作分布的Dict，Policy 类型也一样。Actions为一维ndarray，维数是离散动作数量。

1
2
3
4
5
6

State = Tuple[int, int, bool]
Action = bool
Reward = float
Actions = np.ndarray
ActionValue = Dict[State, Actions]
Policy = Dict[State, Actions]

下面代码示例如何给定 Policy后，依据指定状态state的动作分布采样，决定下一动作。

1
2
3

policy: Policy
A: ActionValue = policy[state]
action = np.random.choice([0, 1], p=A/sum(A))

整个算法的 python 代码实现如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

def mc_control_exploring_starts(env: BlackjackEnv, num_episodes, discount_factor=1.0) \
        -> Tuple[ActionValue, Policy]:
    states = list(product(range(10, 22), range(1, 11), (True, False)))
    policy = {s: np.ones(env.action_space.n) * 1.0 / env.action_space.n for s in states}
    Q = defaultdict(lambda: np.zeros(env.action_space.n))
    returns_sum = defaultdict(float)
    returns_count = defaultdict(float)

    for episode_i in range(1, num_episodes + 1):
        s0 = random.choice(states)
        reset_env_with_s0(env, s0)
        episode_history = gen_custom_s0_stochastic_episode(policy, env, s0)

        G = 0
        for t in range(len(episode_history) - 1, -1, -1):
            s, a, r = episode_history[t]
            G = discount_factor * G + r
            if not any(s_a_r[0] == s and s_a_r[1] == a for s_a_r in episode_history[0: t]):
                returns_sum[s, a] += G
                returns_count[s, a] += 1.0
                Q[s][a] = returns_sum[s, a] / returns_count[s, a]
                best_a = np.argmax(Q[s])
                policy[s][best_a] = 1.0
                policy[s][1-best_a] = 0.0

    return Q, policy

在MC Exploring Starts 算法中，我们需要指定环境初始状态，一种做法是env.reset()时接受初始状态，但是考虑到不去修改第三方实现的 BlackjackEnv类，采用一个取巧的办法，在调用reset()后直接改写env 的私有变量，这个逻辑封装在 reset_env_with_s0 方法中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

def reset_env_with_s0(env: BlackjackEnv, s0: State) -> BlackjackEnv:
    env.reset()
    player_sum = s0[0]
    oppo_sum = s0[1]
    has_usable = s0[2]

    env.dealer[0] = oppo_sum
    if has_usable:
        env.player[0] = 1
        env.player[1] = player_sum - 11
    else:
        if player_sum > 11:
            env.player[0] = 10
            env.player[1] = player_sum - 10
        else:
            env.player[0] = 2
            env.player[1] = player_sum - 2
    return env

算法结果的可视化和理论对比

Exploring Starts 蒙特卡洛控制改进

为了避免Monte Carlo ES Control在初始时必须访问到任意状态的限制，教材中介绍了一种改进算法，On-policy first-visit MC control for \(\epsilon \text{-soft policies}\) ，它同样基于Monte Carlo 预估Q值，但用 \(\epsilon \text{-soft}\) 策略来代替最有可能的action策略作为下一次迭代策略，\(\epsilon \text{-soft}\) 本质上来说就是对于任意动作都保留 \(\epsilon\) 小概率的访问可能，权衡了exploration和exploitation，由于每个动作都可能被无限次访问到，Explorting Starts中的强制随机初始状态就可以去除了。Monte Carlo ES Control 和 On-policy first-visit MC control for \(\epsilon \text{-soft policies}\) 都属于on-policy算法，其区别于off-policy的本质在于预估 \(q_{\pi}(s,a)\)时是否从同策略\(\pi\)生成的数据来计算。一个比较subtle的例子是著名的Q-Learning，因为根据这个定义，Q-Learning属于off-policy。

伪代码对应的 Python 实现如下。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

def mc_control_epsilon_greedy(env: BlackjackEnv, num_episodes, discount_factor=1.0, epsilon=0.1) \
        -> Tuple[ActionValue, Policy]:
    returns_sum = defaultdict(float)
    returns_count = defaultdict(float)

    states = list(product(range(10, 22), range(1, 11), (True, False)))
    policy = {s: np.ones(env.action_space.n) * 1.0 / env.action_space.n for s in states}
    Q = defaultdict(lambda: np.zeros(env.action_space.n))

    def update_epsilon_greedy_policy(policy: Policy, Q: ActionValue, s: State):
        policy[s] = np.ones(env.action_space.n, dtype=float) * epsilon / env.action_space.n
        best_action = np.argmax(Q[s])
        policy[s][best_action] += (1.0 - epsilon)

    for episode_i in range(1, num_episodes + 1):
        episode_history = gen_stochastic_episode(policy, env)

        G = 0
        for t in range(len(episode_history) - 1, -1, -1):
            s, a, r = episode_history[t]
            G = discount_factor * G + r
            if not any(s_a_r[0] == s and s_a_r[1] == a for s_a_r in episode_history[0: t]):
                returns_sum[s, a] += G
                returns_count[s, a] += 1.0
                Q[s][a] = returns_sum[s, a] / returns_count[s, a]
                update_epsilon_greedy_policy(policy, Q, s)

    return Q, policy