基于PPO的强化学习超级马里奥自动通关

一、环境准备

二、训练思路

1.训练初期：

2.思路整理及改进：

思路一：

思路二：

思路三：

思路四：

3.训练效果：

三、结果分析

四、完整代码

训练代码：

测试代码：

本文将基于强化学习中的PPO算法训练一个自动玩超级马里奥的智能体，用于强化学习的项目实践

一、环境准备

所需环境如下：

pip install nes-py
pip install gym-super-mario-bros
pip install setuptools==65.5.0 "wheel<0.40.0"
pip install gym==0.21.0
pip install stable-baselines3【extra】==1.6.0
pip install torch==1.13.1+cu116 torchvision==0.14.1+cu116 torchaudio==0.13.1 --extra-index-url https://download.pytorch.org/whl/cu116

注意： 在环境配置方面，nes-py库安装的先决条件是安装Microsoft Visual C++，其下载地址为：Microsoft C++ Build Tools - Visual Studio

在安装Microsoft Visual C++时需选择桌面开发：

二、训练思路

1.训练初期：

使用了最简单的训练框架，并选择PPO算法中较简单的的CnnPolicy网络（可以尝试MlpPolicy和MultiInputPolicy网络）以及马里奥操控中的SIMPLE_MOVEMENT操作模块：

自然，效果是不尽人意的，马里奥在所选关卡的第三根水管处（即最高的那个水管）不断尝试跳跃，直至时间耗尽也未能通过。

2.思路整理及改进：

思路一：

既然训练效果不佳，是否跟训练轮数有关？固将总训练轮数增加至3000000，并尝试训练。跑出来的模型有所改进，马里奥在成功越过所有水管后，遇到了新的难题——越过两个断崖。至此，无论如何增加轮数，马里奥似乎到了一个瓶颈，固继续进行修改。

思路二：

在增加训练轮数的基础上，选择对关卡的环境图像进行预处理——使用GrayScaleObservation转换为灰度观察，并保留通道维度。同时，我们对训练参数进行调整：

尝试训练后，能够得到一个不稳定越过断崖的新模型，但对断崖之后的环境似乎有些陌生，陷入了前半段关卡的“局部最优解”。

思路三：

由于之前的训练过程中使用了较小的学习率（1e-9），进而使得马里奥在关卡中陷入了局部最优，所以选择对学习率进行微调，使其在最开始的训练阶段使用较大的学习率，在后期减小学习率，从而达到先快速探索参数空间并加速收敛，再提高模型的稳定性和收敛精度。

至此，训练出来的测试模型，奖励反馈有所增长，但实际测试效果与调整前相差不多。

思路四：

在上述尝试无明显效果后，猜测效果的好坏是否与马里奥的奖励机制有关，固在查阅奖励部分代码后，对“抵达终点”的奖励予以提高，希望对效果有所改善。

然结果并没有明显改观，更换调整方向。分别尝试马里奥的三套运动方式：

经过对比，complex_movement的效果远超另外两套，且在前面思路的改动下模型质量有显著提升，固整理上述调整方案，进行底模训练。

3.训练效果：

以奖励折扣率gamma = 0.9、gae_lambda = 0.9、clip_range = 0.2、步长n_steps = 7168，并用1e-3作为开始训练的学习率，并在训练过程中使其动态地在1e-5，1e-7中调整，修改抵达终点的奖励反馈，同时设置训练轮数为4000000，训练动作组为complex_movement进行训练。得到基础奖励回报为1520的底模，并将其继续用于迁移学习，得到2300的新模型。在实际测试后发现，模型确有改观，固继续将新模型用于训练，最终得到3200的最终模型，其能顺利到达终点并进入关卡的下一阶段。

三、结果分析

与之前的训练经验相比，使用复杂的动作组未必比简单的动作组训练出的效果差，学习率的调整也是必要的，先用较大学习率打好基础，再有小学习率继续细化模型。同时，要给足够的训练轮数（足够的训练时间）。若是能够把奖励机制更进一步细化增加奖励细节，对其的训练是会更有帮助的。

四、完整代码

训练代码：

from nes_py.wrappers import JoypadSpace
import time
import os
import numpy as np
from datetime import datetime
from matplotlib import pyplot as plt
import gym_super_mario_bros
from gym_super_mario_bros.actions import SIMPLE_MOVEMENT, COMPLEX_MOVEMENT, RIGHT_ONLY
from gym.wrappers import GrayScaleObservation
from gym import Wrapper
from stable_baselines3.common.monitor import Monitor
from stable_baselines3.common.vec_env import DummyVecEnv
from stable_baselines3.common.vec_env import VecFrameStack
from stable_baselines3 import PPO
from stable_baselines3.common.results_plotter import load_results, ts2xy
from stable_baselines3.common.callbacks import BaseCallback# 定义自定义奖励包装器
class CustomRewardWrapper(Wrapper):def __init__(self, env):super(CustomRewardWrapper, self).__init__(env)self.curr_score = 0def step(self, action):state, reward, done, info = self.env.step(action)# 自定义的奖励reward += (info["score"] - self.curr_score) / 40.self.curr_score = info["score"]if done:if info["flag_get"]:reward += 50else:reward -= 50return state, reward / 10., done, infoclass SaveOnBestTrainingRewardCallback(BaseCallback):"""Callback for saving a model (the check is done every ``check_freq`` steps)based on the training reward (in practice, we recommend using ``EvalCallback``).:param check_freq: (int):param log_dir: (str) Path to the folder where the model will be saved.It must contains the file created by the ``Monitor`` wrapper.:param verbose: (int)"""def __init__(self, check_freq, save_model_dir, verbose=1):super(SaveOnBestTrainingRewardCallback, self).__init__(verbose)self.check_freq = check_freqself.save_path = os.path.join(save_model_dir, './')self.best_model_subdir = os.path.join(self.save_path, 'best_model')self.best_mean_reward = -np.infself.best_model_path = Noneself.best_score_model_path = os.path.join(self.save_path, 'pass_customs_model.zip')  # 增加通关模型路径# def _init_callback(self) -> None:def _init_callback(self):# Create folder if neededif self.save_path is not None:os.makedirs(self.save_path, exist_ok=True)# def _on_step(self) -> bool:def _on_step(self):if self.n_calls % self.check_freq == 0:print('self.n_calls: ', self.n_calls)model_path1 = os.path.join(self.save_path, 'model_{}'.format(self.n_calls))self.model.save(model_path1)# Save the best modelx, y = ts2xy(load_results(monitor_dir), 'timesteps')if len(x) > 0:mean_reward = np.mean(y[-self.check_freq:])if self.verbose > 0:print("Num timesteps: {}, Best mean reward: {:.2f}, Last mean reward: {:.2f}".format(self.n_calls, self.best_mean_reward, mean_reward))if mean_reward > self.best_mean_reward:if self.best_model_path is not None:try:os.remove(self.best_model_path)  # Delete the old best modelexcept OSError:passself.best_mean_reward = mean_reward# Update path for the new best modelself.best_model_path = os.path.join(self.save_path, 'best_model.zip')# Save the new best modelself.model.save(self.best_model_path)if self.verbose > 0:print("New best mean reward: {:.2f} - saving best model".format(mean_reward))# Save the best mean reward to a filereward_record_file = './Mario_model_save/model/mario_model/best_mean_reward.txt'with open(reward_record_file, 'a') as file:# 将最佳平均奖励值和时间戳一同写入文件file.write("New best mean reward: {:.2f} - Recorded at {}\n".format(mean_reward, datetime.now()))return True# 总的训练timesteps
my_total_timesteps = 4000000
# 需要改变学习率的timestep
change_lr_timestep = 2000000# 学习率调度函数
def learning_rate_schedule(progress_remaining):"""参数 progress_remaining 表示剩下的训练进度(从1开始降低到0)。通过训练进度来动态调整学习率。"""current_timestep = my_total_timesteps * (1 - progress_remaining)if current_timestep < change_lr_timestep:return 1e-3  # 1e-3elif change_lr_timestep <= current_timestep <= int(change_lr_timestep * 1.5):return 1e-5else:return 1e-7env = gym_super_mario_bros.make('SuperMarioBros-1-2-v0')
env = JoypadSpace(env, COMPLEX_MOVEMENT)  # 使用复杂的按键映射env = CustomRewardWrapper(env)  # 应用自定义奖励包装器monitor_dir = r'./Mario_model_save/monitor_log/'
os.makedirs(monitor_dir, exist_ok=True)
env = Monitor(env, monitor_dir)  # 将环境包装为监视器env = GrayScaleObservation(env, keep_dim=True)  # 转换为灰度观察，并保留通道维度
env = DummyVecEnv([lambda: env])  # 创建虚拟环境
env = VecFrameStack(env, 4, channels_order='last')  # 将最近4帧堆叠在一起best_params = {'n_steps': 7168,  # 7168'gamma': 0.9,# 'learning_rate': 1e-3,   # 1e-3, 1e-4, 1e-5'clip_range': 0.2,'gae_lambda': 0.9,
}# 更新best_params中的learning_rate参数
best_params.update({'learning_rate': learning_rate_schedule})tensorboard_log = r'./Mario_model_save/tensorboard_log/'
# 正常训练
model = PPO("CnnPolicy", env, verbose=1,tensorboard_log=tensorboard_log,**best_params)
'''
# 加载预训练模型
pretrained_model_path = r'D:\python_project\Mario\model\mario_model\pretraining_model_4.zip'
model = PPO.load(pretrained_model_path, env=env, tensorboard_log=tensorboard_log, **best_params)'''# 保存模型位置
save_model_dir = r'./Mario_model_save/model/mario_model/'
callback1 = SaveOnBestTrainingRewardCallback(10000, save_model_dir)model.learn(total_timesteps=my_total_timesteps, callback=callback1)
# model.save("mario_model")

测试代码：

from nes_py.wrappers import JoypadSpace
import gym_super_mario_bros
from gym_super_mario_bros.actions import SIMPLE_MOVEMENT, RIGHT_ONLY, COMPLEX_MOVEMENT
import time
from matplotlib import pyplot as plt
from gym.wrappers import GrayScaleObservation
from stable_baselines3.common.monitor import Monitor
from stable_baselines3.common.vec_env import DummyVecEnv
from stable_baselines3.common.vec_env import VecFrameStack
import os
from stable_baselines3 import PPOfrom stable_baselines3.common.results_plotter import load_results, ts2xy
import numpy as np
from stable_baselines3.common.callbacks import BaseCallbackenv = gym_super_mario_bros.make('SuperMarioBros-v0')
env = JoypadSpace(env, COMPLEX_MOVEMENT)monitor_dir = r'./Mario/monitor_log/'
os.makedirs(monitor_dir, exist_ok=True)
env = Monitor(env, monitor_dir)env = GrayScaleObservation(env, keep_dim=True)
env = DummyVecEnv([lambda: env])
env = VecFrameStack(env, 4, channels_order='last')save_model_dir = r'model/mario_model/pretraining_model_5.zip'
# save_model_dir = r'./Mario/model/mario_model/pretraining_model.zip'model = PPO.load(save_model_dir)obs = env.reset()
obs = obs.copy()
done = True
while True:if done:state = env.reset()action, _states = model.predict(obs)obs, rewards, done, info = env.step(action)obs = obs.copy()# time.sleep(0.01)env.render()env.close()