【飞桨开发者说】王子瑞，四川大学电气工程学院2018级自动化专业本科生，飞桨开发者技术专家PPDE，RoboMaster川大火锅战队成员，强化学习爱好者
超级马里奥兄弟作为几代人的童年回忆，陪伴了我们的成长。如今，随着深度强化学习的发展，越来越多的游戏已经被AI征服。今天，我们将以超级马里奥为例子，展示如何用深度强化学习试着通关游戏。
马里奥游戏环境简介
游戏环境只给予3次机会通关，即玩家或AI需要在3次机会内通过游戏的32关。环境提供了 RIGHT_ONLY、SIMPLE_MOVEMENT和COMPLEX_MOVEMENT三种难度的操作模式。人们只需要对环境输入各种动作所代表的数值，就能实现对马里奥的各种操作。
马里奥游戏环境链接：
https://pypi.org/project/gym-super-mario-bros/
PPO算法简介
相信了解强化学习的各位一定听说过近端策略优化PPO算法吧。PPO算法是一种新型的 Policy Gradient算法，Policy Gradient算法对步长十分敏感。在训练过程中，若没有选择到合适的步长，新旧策略的变化可能会出现差异过大的现象，不利于模型的收敛。PPO提出了新的目标函数，可以在多个训练步骤中实现小幅度的更新，解决了Policy Gradient算法中步长难以确定的问题。
PPO算法论文链接：
https://arxiv.org/abs/1707.06347
基于飞桨框架2.0实现PPO
在此之前，我们先看看模型结构。模型是Actor-Critic结构，但是我们对模型结构做了一点简化，Actor和Critic只在输出层有所区别。由于模型处理的是图像信息，故我们在全连接层前加入了卷积层。下面就让我们用飞桨框架2.0实现PPO算法吧！
-
class MARIO(Layer):
-
def __init__(self, input_num, actions):
-
super(MARIO,
self).__init_
_()
-
self.num_input = input_num
-
self.channels =
32
-
self.kernel =
3
-
self.stride =
2
-
self.padding =
1
-
self.fc =
32 *
6 *
6
-
self.conv
0 = Conv2D(out_channels=
self.channels,
-
kernel_size=
self.kernel,
-
stride=
self.stride,
-
padding=
self.padding,
-
dilation=[
1,
1],
-
groups=
1,
-
in_channels=input_num)
-
self.relu
0 = ReLU()
-
self.conv1 = Conv2D(out_channels=
self.channels,
-
kernel_size=
self.kernel,
-
stride=
self.stride,
-
padding=
self.padding,
-
dilation=[
1,
1],
-
groups=
1,
-
in_channels=
self.channels)
-
self.relu1 = ReLU()
-
self.conv2 = Conv2D(out_channels=
self.channels,
-
kernel_size=
self.kernel,
-
stride=
self.stride,
-
padding=
self.padding,
-
dilation=[
1,
1],
-
groups=
1,
-
in_channels=
self.channels)
-
self.relu2 = ReLU()
-
self.conv3 = Conv2D(out_channels=
self.channels,
-
kernel_size=
self.kernel,
-
stride=
self.stride,
-
padding=
self.padding,
-
dilation=[
1,
1],
-
groups=
1,
-
in_channels=
self.channels)
-
self.relu3 = ReLU()
-
self.linear
0 = Linear(in_features=int(
self.fc), out_features=
512)
-
self.linear1 = Linear(in_features=
512, out_features=actions)
-
self.linear2 = Linear(in_features=
512, out_features=
1)
-
-
def forward(self, x):
-
x = paddle.to_tensor(data=x)
-
x =
self.conv
0(x)
-
x =
self.relu
0(x)
-
x =
self.conv1(x)
-
x =
self.relu1(x)
-
x =
self.conv2(x)
-
x =
self.relu2(x)
-
x =
self.conv3(x)
-
x =
self.relu3(x)
-
x = paddle.reshape(x, [x.shape[
0], -
1])
-
x =
self.linear
0(x)
-
logits =
self.linear1(x)
-
value =
self.linear2(x)
-
return logits, value
本文的PPO属于在线学习，大致分为以下三个模块：
-
获取动作轨迹
-
计算优势函数
-
数据采样与模型参数更新
由于PPO是Policy Gradient算法，我们的智能体需要生成一个类别分布，即一个包含每个动作发生概率的向量。然后根据向量中的概率，选择我们的动作。最后与环境交互，并将返回的各种状态信息以及奖励存入列表当中备用。下面是获取动作轨迹模块的部分代码：
-
for _
in range(num_local_steps):
-
logits, value = model(curr_states)
-
values.append(
value.squeeze())
-
policy = F.softmax(logits, axis=
1)
-
old_m = Categorical(policy)
# 生成类别分布
-
action = old_m.sample([
1])
# 采样
-
old_log_policy = old_m.log_prob(action)
-
old_log_policies.append(old_log_policy)
-
[
agent_conn.send(("step", act)) for agent_conn, act in zip(envs.agent_conns, action.numpy().astype("int8"))]
-
state, reward, done, info = zip(*[agent_conn.recv()
for agent_conn
in envs.agent_conns])
接着上面的代码，我们需要计算优势函数。具体来说，优势函数指当前状态s采用动作a的收益与当前状态s平均收益的差。优势越大，动作a收益就越高，同一状态下采用该动作的概率也就应该更高。
这里，我们用到了广义优势估计GAE（Generalized Advantage Estimator），几乎所有最先进的Policy Gradient算法实现都使用了该技术。这项技术主要用来修正我们Critic模型提供的价值，使其成为方差最小的无偏估计。
-
for
value, reward,
done in list(zip(values, rewards, dones))[::-1]:
-
gae = gae * gamma * tau
-
gae = gae + reward + gamma * next_value.detach().numpy() * (
1.0 - done) -
value.detach().numpy()
-
next_value =
value
-
R.append(paddle.to_tensor(gae +
value.detach().numpy()))
-
advantages = R - values
最后，我们使用PPO算法更新模型参数。这里并没有计算KL散度，而是通过截断的方式实现小幅度的更新。
-
for i in range(num_epochs):
-
indice = paddle.randperm(num_local_steps * num_processes)
-
for j in range(batch_size):
-
batch_indices = indice[
-
int(j * (num_local_steps * num_processes / batch_size)): int((j +
1) * (
-
num_local_steps * num_processes / batch_size))]
-
logits, value = model(paddle.gather(states, batch_indices, axis=
0))
-
new_policy = F.softmax(logits, axis=
1)
-
new_m = Categorical(new_policy)
-
new_log_policy = new_m.log_prob(paddle.gather(actions, batch_indices, axis=
0))
-
ratio = paddle.exp(new_log_policy - paddle.gather(old_log_policies, batch_indices, axis=
0))
-
advantages = paddle.gather(advantages, batch_indices, axis=
0)
-
actor_loss = paddle.to_tensor(list((ratio * advantages).numpy() + (paddle.clip(ratio,
1.
0 - epsilon,
1.
0 + epsilon) * advantages).numpy()))
-
actor_loss = -paddle.mean(paddle.min(actor_loss, axis=
0))
-
critic_loss = F.smooth_l
1_loss(paddle.gather(R, batch_indices), value)
-
entropy_loss = paddle.mean(new_m.entropy())
-
total_loss = actor_loss + critic_loss - beta * entropy_loss
-
clip_grad = paddle.nn.ClipGradByNorm(clip_norm=
0.
25)
-
optimizer = paddle.optimizer.Adam(learning_rate=lr, parameters=model.parameters(), grad_clip=clip_grad)
-
optimizer.clear_grad()
-
total_loss.backward()
-
optimizer.step()
由于篇幅有限，此部分只呈现了简要思路与部分删减后的代码，感兴趣的同学可以直接查看源码。
通关小技巧
马里奥的通关小技巧有很多，这里主要给大家提供三个方向的思路：
-
原始输入图像预处理
-
奖励函数重设置
-
多线程/并行训练
原始输入图像预处理：简化图像特征，叠合连续4帧图像作为输入，可以起到捕捉游戏环境的动态性的作用。
奖励函数重设置：不同的奖励函数所鼓励的行为是不同的，例如提高踩怪的奖励，就可以使模型更倾向于踩怪。本文重新分配了一下各种奖励的权重，对于通关也有更丰厚的额外奖励。
-
def step(self, action):
-
state, reward, done, info =
self.env.step(action)
-
if
self.
monitor:
-
self.monitor.record(state)
-
state = process_frame(state)
-
reward += (info[
"score"] -
self.curr_score) /
40.
-
self.curr_score = info[
"score"]
-
if
done:
-
if info[
"flag_get"]:
-
reward +=
50
-
else:
-
reward -=
50
-
self.env.reset()
-
return state, reward /
10., done, info
多线程/并行训练：并行化可以有效提高模型的训练效率，同时也是目前强化学习的趋势之一。本文通过 Python 的 multiprocess 模块实现并行化。
-
class MultipleEnvironments:
-
def __init__(self, world, stage, action_type, num_envs, output_path=None):
-
self.agent_conns, self.env_conns = zip(*[mp.Pipe()
for _
in range(num_envs)])
-
'''选择操作模式
-
'''
-
if action_type ==
"right":
-
actions = RIGHT_ONLY
-
elif action_type ==
"simple":
-
actions = SIMPLE_MOVEMENT
-
else:
-
actions = COMPLEX_MOVEMENT
-
'''创建多环境
-
'''
-
self.envs = [create_train_env(world, stage, actions, output_path=output_path)
for _
in range(num_envs)]
-
self.num_states = self.envs[
0].observation_space.shape[
0]
-
self.num_actions = len(actions)
-
-
'''创建多进程
-
'''
-
for index
in range(num_envs):
-
process = mp.Process(target=self.run, args=(index,))
-
process.start()
-
self.env_conns[index].close()
-
def run(self, index):
-
self.agent_conns[index].close()
-
while
True:
-
request, action = self.env_conns[index].recv()
-
if request ==
"step":
-
self.env_conns[index].send(self.envs[index].step(int(action)))
-
elif request ==
"reset":
-
self.env_conns[index].send(self.envs[index].reset())
-
else:
-
raise NotImplementedError
效果展示：本文以关卡1-1为例。目前多线程训练的马里奥已经正式通过测试。在 8 线程下，训练过程中我们的马里奥能够获得的Reward值变化趋势如下图所示。
最后，诚邀大家收看超级马里奥兄弟1-1通关全过程：
全文回顾
我们在这篇文章里，先简单介绍超级马里奥兄弟的游戏环境，然后补充一些与PPO算法有关的知识，并基于Paddle2.0进行实现了该算法。
除此之外，本文还总结了一些通关小技巧，并以游戏第一关为例，展示了训练过程。最后，我们向大家展现了训练完成后的效果。
训练代码项目链接：
https://aistudio.baidu.com/aistudio/projectdetail/1434971
通关展示项目链接：
https://aistudio.baidu.com/aistudio/projectdetail/1434950
2021年2月3日作者在飞桨PaddlePaddle B站直播间分享：《用深度强化学习轻松通关马里奥》
欢迎收看~ 哔哩哔哩直播，二次元弹幕直播平台 (bilibili.com)
转载：https://blog.csdn.net/PaddlePaddle/article/details/113604905