强化学习实战5——BaseLine3使用自定义环境训练【输入状态向量】

张

张建站

2026/4/9 6:45:05

10分钟阅读

简介想要学会将BL3接入自定义的环境需要做两件事第零你得对自己的任务熟悉知晓观测空间、动作空间、奖励函数如何设计根据需求决定要用的模型再开始写代码。第一明确Gym的环境配置了解需要自己手写哪些函数配置哪些参数第二套用训练模板根据你预先决定使用的模型抄模板就行官方是提供了这些常用模型的示例代码第零步我们这次教程使用之前编写的Predator环境动作空间是离散的状态空间也是离散的可以使用A2C模型也可以用DQN等。我们先完成第一步明确Gym的环境配置写出需要自定义的函数。【这部分内容之前讲过了】Gymnasium 环境简介https://gymnasium.farama.org/api/env/Gymnasium 环境Env的核心标准 API 非常简洁主要由4 个核心函数和2 个核心属性组成。这是所有强化学习任务交互的基础。 1. 核心交互函数这些是你在训练循环Training Loop中必须调用的函数。函数名参数/说明返回值 (Tuple).reset()(重置环境)作用在每个回合Episode开始前调用将环境恢复到初始状态。参数seed(可选用于复现实验结果),options(特定环境的额外参数)。1.observation初始环境状态符合observation_space定义。2.info辅助诊断信息字典。.step(action)(执行动作)作用这是强化学习的核心。将 Agent 的动作Action传入环境环境反馈下一步的状态和奖励。参数action(Agent 选择的动作)。1.observation执行动作后的新环境状态。2.reward该动作获得的奖励浮点数。3.terminated布尔值。True表示回合正常结束如到达目标/坠毁。4.truncated布尔值。True表示回合被强制截断如超时/出界。5.info辅助诊断信息。.render()(渲染画面)作用将环境的当前状态可视化。注意在gymnasium.make()时需指定render_mode如human,rgb_array。根据render_mode不同而不同*human通常返回None(直接在窗口显示)。*rgb_array返回图像帧np.ndarray。*ansi返回文本字符串。.close()(关闭环境)作用释放环境占用的资源如关闭 Pygame 窗口、数据库连接等。建议在脚本结束或训练完成后调用。None 2. 核心属性 (Spaces)在编写代码前你需要通过这两个属性来了解环境的输入输出规格.action_space含义定义了 Agent 可以采取的所有合法动作的范围。用途用于构建 Agent 的输出层。例如如果是Discrete(4)说明有 4 个离散动作如 Lunar Lander 的 0, 1, 2, 3。常用方法sample()(随机采样一个动作)。.observation_space含义定义了环境状态观测值的数据结构和范围。用途用于构建 Agent 的输入层。例如Box(4,)表示一个包含 4 个浮点数的数组。常用方法sample()(随机采样一个观测值常用于测试)。️ 3. 辅助属性与函数.metadata包含环境的元信息比如支持的渲染模式 (render_modes) 和帧率 (render_fps)。.spec环境的配置规格通常在通过gymnasium.make()创建时生成。.np_random环境内部的随机数生成器用于保证实验的可复现性 (Reproducibility)。总结代码模板一个标准的自定义环境类模板长这样https://stable-baselines3.readthedocs.io/en/master/guide/custom_env.htmlimport gymnasium as gym import numpy as np from gymnasium import spaces class CustomEnv(gym.Env): Custom Environment that follows gym interface. metadata {render_modes: [human], render_fps: 30} def __init__(self, arg1, arg2, ...): super().__init__() # Define action and observation space # They must be gym.spaces objects # Example when using discrete actions: self.action_space spaces.Discrete(N_DISCRETE_ACTIONS) # Example for using image as input (channel-first; channel-last also works): self.observation_space spaces.Box(low0, high255, shape(N_CHANNELS, HEIGHT, WIDTH), dtypenp.uint8) def step(self, action): ... return observation, reward, terminated, truncated, info def reset(self, seedNone, optionsNone): ... return observation, info def render(self): ... def close(self): ...我们要按要求修改我们的envCube类。首先可以看出自定义环境要继承gym.Env。其次需要指定action_space和observation_space。self.action_space gym.spaces.Discrete(ACTION_SPACE_VALUES) self.observation_space gym.spaces.Box(low-SIZE1, highSIZE-1, shape(4,), dtypeint)注意动作空间是9个是离散的因此使用Discrete。而观察空间其实也是离散的但是这里还是使用了连续的BoxBox有四个参数分别是连续区间的最小最大值状态向量的维度以及单个数据类型。我们设定的是SIZE*SIZE的棋盘状态用(x1,y1,x2,y2)这个(4,)tuple表示。注意一下更新后spaces.Discrete、spaces_Box都无法直接用要加前缀gym写出来应该如下def __init__(self,SIZE10, ACTION_SPACE_VALUES 9, RETURN_IMAGE True, MAX_STEP200, FOOD_REWARD 25, ENEMY_PENALITY -300, MOVE_PENALITY -1): super(envCube,self).__init__() self.SIZESIZE #self.OBSERVATION_SPACE_VALUES(SIZE,SIZE,3) self.OBSERVATION_SPACE_VALUES(4,) #self.ACTION_SPACE_VALUESACTION_SPACE_VALUES self.RETURN_IMAGE RETURN_IMAGE # 考虑返回值是否图像 self.MAX_STEPMAX_STEP self.FOOD_REWARD FOOD_REWARD # agent获得食物的奖励 self.ENEMY_PENALITY ENEMY_PENALITY # 遇上对手的惩罚 self.MOVE_PENALITY MOVE_PENALITY # 每移动一步的惩罚 self.action_space gym.spaces.Discrete(ACTION_SPACE_VALUES) self.observation_space gym.spaces.Box(low-SIZE1, highSIZE-1, shape(4,), dtypeint)然后我们验证一下为了检测我们的环境是否符合gym的要求可以使用这个函数验证肯定会遇到不少报错但都很好解决基本就是把observation改为np.array的形式而不是tuple。以及返回需要添加info的问题。下面这个是能跑通的class envCube(gym.Env): # 设定三个部分的颜色分别是蓝、绿、红 d {1: (255, 0, 0), # blue 2: (0, 255, 0), # green 3: (0, 0, 255)} # red PLAYER_N 1 FOOD_N 2 ENEMY_N 3 metadata {render_modes: [human], render_fps: 30} def __init__(self,SIZE10, ACTION_SPACE_VALUES 9, RETURN_IMAGE False, MAX_STEP200, FOOD_REWARD 25, ENEMY_PENALITY -300, MOVE_PENALITY -1): super(envCube,self).__init__() self.SIZESIZE #self.OBSERVATION_SPACE_VALUES(SIZE,SIZE,3) self.OBSERVATION_SPACE_VALUES(4,) #self.ACTION_SPACE_VALUESACTION_SPACE_VALUES self.RETURN_IMAGE RETURN_IMAGE # 考虑返回值是否图像 self.MAX_STEPMAX_STEP self.FOOD_REWARD FOOD_REWARD # agent获得食物的奖励 self.ENEMY_PENALITY ENEMY_PENALITY # 遇上对手的惩罚 self.MOVE_PENALITY MOVE_PENALITY # 每移动一步的惩罚 self.action_space gym.spaces.Discrete(ACTION_SPACE_VALUES) self.observation_space gym.spaces.Box(low-SIZE1, highSIZE-1, shape(4,), dtypeint) # 环境重置 def reset(self,seedNone, optionsNone): self.player Cube(self.SIZE) self.food Cube(self.SIZE) self.enemy Cube(self.SIZE) # 如果玩家和食物初始位置相同重置食物的位置直到位置不同 while self.player self.food: self.food Cube(self.SIZE) # 如果敌人和玩家或食物的初始位置相同重置敌人的位置直到位置不同 while self.player self.enemy or self.food self.enemy: self.enemy Cube(self.SIZE) # 判断观测是图像和数字 if self.RETURN_IMAGE: observation np.array(self.get_image()) else: observation (self.player - self.food)(self.player - self.enemy) observationnp.array(observation) self.episode_step 0 info{} return observation,info def step(self,action): self.episode_step1 self.player.action(action) self.food.move() self.enemy.move() # 分类讨论输出new_obs if self.RETURN_IMAGE: new_observation np.array(self.get_image()) else: new_observation (self.player - self.food)(self.player - self.enemy) new_observationnp.array(new_observation) #获取reward值 if self.player self.food: rewardself.FOOD_REWARD elif self.playerself.enemy: rewardself.ENEMY_PENALITY else: rewardself.MOVE_PENALITY #检测截止情况 terminated False truncated False # 4. 判断结束条件 if self.player self.food or self.player self.enemy: terminated True if self.episode_step self.MAX_STEP: truncated True info{} return new_observation,reward,terminated,truncated,info def render(self,modehuman): imgself.get_image() img img.resize((800,800)) cv2.imshow(Predator,np.array(img)) cv2.waitKey(1) def get_image(self): # 图像显示 env np.zeros(self.OBSERVATION_SPACE_VALUES,dtype np.uint8) env[self.food.x][self.food.y] self.d[self.FOOD_N] env[self.player.x][self.player.y] self.d[self.PLAYER_N] env[self.enemy.x][self.enemy.y] self.d[self.ENEMY_N] img Image.fromarray(env,RGB) return img第二步套用官方提供的代码接下来我们套用之前写的A2C模型的训练代码来自这一讲【A2C 再战登月器降落】这一讲中官方提供了示例代码https://stable-baselines3.readthedocs.io/en/master/modules/a2c.htmlfrom stable_baselines3 import A2C from stable_baselines3.common.env_util import make_vec_env # Parallel environments vec_env make_vec_env(CartPole-v1, n_envs4) model A2C(MlpPolicy, vec_env, verbose1) model.learn(total_timesteps25000) model.save(a2c_cartpole) del model # remove to demonstrate saving and loading model A2C.load(a2c_cartpole) obs vec_env.reset() while True: action, _states model.predict(obs) obs, rewards, dones, info vec_env.step(action) vec_env.render(human)我们参照着这个代码写出了#A2C版本代码 # Instantiate the agent model A2C(MlpPolicy, env, verbose1, tensorboard_log./logs, learning_rate5e-4, policy_kwargs{net_arch:[256, 256]} ) # Train the agent and display a progress bar model.learn(total_timestepsint(2.5e6), progress_barTrue,tb_log_nameA2C_Net256_2M_50) # Save the agent model.save(A2C_Net256_2M_50_lunar) del model # delete trained model to demonstrate loadingmodel DQN.load(dqn_Net256_lunar, envenv) model.set_env(env) mean_reward, std_reward evaluate_policy(model, model.get_env(),deterministicTrue,renderFalse, n_eval_episodes10)我们不需要改架构只需要改参数即可。这次我们打算训练50万次那么total_timesteps就要设定为5e6然后我们把训练的日志记录名称改为S10_A2C_Net256_50W。#A2C版本代码 # Instantiate the agent model A2C(MlpPolicy, env, verbose1, tensorboard_log./logs, learning_rate5e-4, policy_kwargs{net_arch:[256, 256]} ) # Train the agent and display a progress bar model.learn(total_timestepsint(2.5e6), progress_barTrue,tb_log_nameS10_A2C_Net256_50W)然后保存模型的名称也改一下A2C_Net256_50_lunar# Save the agent model.save(S10_A2C_Net256_50W_lunar) del model # delete trained model to demonstrate loading打开tensorboard可以看到训练出来的效果还是不错的奖励基本都收敛到0附近了而且步数也稳定在10步左右就能吃到食物。效果不好考虑修改模型架构现在我们想扩大一点棋盘改为SIZE20在20*20的情况下50万次的训练很明显是无法达到预期效果的仅有-200。很多时候训练步数给到50W也无法达到很好的效果这种情况下就需要我们修改模型架构了。model A2C(MlpPolicy, env, verbose1, tensorboard_log./logs, learning_rate5e-4, policy_kwargs{net_arch:[256, 256]} )核心就是这个policy_kwargs。如何修改呢我们可以参考https://stable-baselines3.readthedocs.io/en/master/guide/custom_policy.html架构的修改policy_kwargs核心参数详解1.features_extractor_class和features_extractor_kwargs用于指定如何从原始观测Observation中提取特征。用途如果你的输入是图像通常用CnnPolicy如果是向量Vector通常用MlpPolicy。2.net_arch(最常用的参数)这是设定神经网络层数和每层神经元数量的核心参数。它决定了 Actor策略网络和 Critic价值网络的隐藏层结构。设定规则net_arch接受一个列表List。列表中的每一个数字代表一个隐藏层的神经元数量。如果需要将策略网络和价值网络的结构分开定义可以使用字典Dict格式。两种主要设定格式格式 A统一架构当 Actor 和 Critic 使用相同的网络结构时使用。写法policy_kwargs dict(net_arch[128, 128])含义创建两个隐藏层每层都有 64 个神经元。这个结构将同时用于策略Policy和价值函数Value。格式 B分离架构当需要为 Actor 和 Critic 分别设定不同结构时使用。这是文档中重点展示的高级用法。写法policy_kwargs dict(net_archdict(pi[32, 32], vf[64, 64]))含义策略网络使用两层 64 个神经元的结构而价值网络使用两层 32 个神经元的结构。激活函数的修改在官方的示例中如果我们想修改激活函数默认tanh可以导入th库然后在这里设定activation_fnth.nn.ReLU我们也参考先共享一层再分开model A2C(MlpPolicy, env, verbose1, tensorboard_log./logs, learning_rate5e-4, policy_kwargs{ net_arch: dict(vf[32, 32], pi[32, 16]), # 修正了这里 activation_fn: th.nn.ReLU } ) model.policy在这个架构下policynet第一层是32第二层变成了16而valuenet第一层是32第二层也是32对应了dict(vf[32, 32], pi[32, 16])然后我们开始训练。训练结束发现红色线形势一片大好很快就收敛了只需要15步就能吃到食物而且奖励也达到了正数这样我们的模型架构修改是成功的总结这一节综合了前几节的内容。包括【强化学习实战4——编写符合Gym的环境】以及【强化学习实战2——A2C 再战登月器降落】要学会将BL3接入自定义的环境需要做两件事第零你得对自己的任务熟悉知晓观测空间、动作空间、奖励函数如何设计根据需求决定要用的模型再开始写代码。第一明确Gym的环境配置了解需要自己手写哪些函数配置哪些参数第二套用训练模板根据你预先决定使用的模型抄模板就行官方是提供了这些常用模型的示例代码第三当增加训练步数也无法达到预期效果时就要考虑修改模型架构了。修改有三个地方一个是“是否使用分离的AC模型”一个是“是否修改不同层级的神经元数量”一个是“是否修改激活函数”实际上这样的修改往往还不足够在下一节我们会讲如何自定义特征提取器。这一节我们仅仅是触及了net_arch的修改并没有涉及前面的特征提取器的修改。我们最终的目的是要能够自己定义Observation自定义特征提取器自定义模型架构。从而实现这张图。都可以做尝试。本节代码下载依赖库!pip install opencv-python !pip install pillow !pip install gymnasium !pip install matplotlib !pip install stable-baselines3[extra] !pip install tensorboard导入依赖库#将学习环境所需要的依赖库导入 import gymnasium as gym import numpy as np from gymnasium import spaces import cv2 from PIL import Image import time import torch as th import pickle #对象到文件的处理 import matplotlib.pyplot as plt from matplotlib import style from stable_baselines3.common.env_checker import check_env style.use(ggplot)设置环境参数#环境参数设定 #SIZE10 #3个对象在10*10的范围内 EPISODES30000 #智能体玩的游戏轮数 SHOW_EVERY3000 #每3000局展示一次游玩过程 #奖励与惩罚 #FOOD_REWARD25#吃食物的奖励 #ENEMY_PENALITY300 #被敌人抓住的惩罚 #MOVE_PENALITY1 #移动惩罚 #环境计算参数 epsilon0.6 #在强化学习时抽取随机动作的概率 40%使用最大价值期望动作 EPS_DECAY0.9998 #每玩一局游戏就让随机动作概率乘以这个数到最后基本就定性了 DISCOUNT0.95 #折扣回报未来奖励的折扣 LEARNING_RATE0.1 #学习率步长 #q_table_file q_table_save.pkl #q_table None q_table qtable_1775227149.pickle #d {1:(255,0,0), #蓝色——玩家 # 2:(0,255,0), #绿色——食物 # 3:(0,0,255)} #红色——敌人 #PLAYER_N1 #FOOD_N2 #ENEMY_N3编写Cube对象类#为三个对象创建类 class Cube: def __init__(self,size):#初始位置 self.sizesize self.xnp.random.randint(0,self.size-1) self.ynp.random.randint(0,self.size-1) def __str__(self): #打印当前位置 return f{self.x},{self.y} def __sub__(self,other):#这个类的另一个实体 return (self.x-other.x,self.y-other.y) def __eq__(self,other): return self.x other.x and self.y other.y def action(self,choise): if choise 0: self.move(x1,y1) elif choise 1: self.move(x-1,y1) elif choise 2: self.move(x1,y-1) elif choise 3: self.move(x-1,y-1) elif choise 4: self.move(x0,y1) elif choise 5: self.move(x0,y-1) elif choise 6: self.move(x1,y0) elif choise 7: self.move(x-1,y0) elif choise 8: self.move(x0,y0) def move(self,xFalse,yFalse): if not x:#如果x没有给值 self.x np.random.randint(-1,2) else: self.x x if not y:#如果y没有给值 self.y np.random.randint(-1,2) else: self.y y #考虑边界 if self.x0: self.x0 elif self.xself.size: self.xself.size-1 if self.y0: self.y0 elif self.yself.size: self.yself.size-1编写envCube环境类class envCube(gym.Env): # 设定三个部分的颜色分别是蓝、绿、红 d {1: (255, 0, 0), # blue 2: (0, 255, 0), # green 3: (0, 0, 255)} # red PLAYER_N 1 FOOD_N 2 ENEMY_N 3 metadata {render_modes: [human], render_fps: 30} def __init__(self,SIZE20, ACTION_SPACE_VALUES 9, RETURN_IMAGE False, MAX_STEP200, FOOD_REWARD 25, ENEMY_PENALITY -300, MOVE_PENALITY -1): super(envCube,self).__init__() self.SIZESIZE #self.OBSERVATION_SPACE_VALUES(SIZE,SIZE,3) self.OBSERVATION_SPACE_VALUES(4,) #self.ACTION_SPACE_VALUESACTION_SPACE_VALUES self.RETURN_IMAGE RETURN_IMAGE # 考虑返回值是否图像 self.MAX_STEPMAX_STEP self.FOOD_REWARD FOOD_REWARD # agent获得食物的奖励 self.ENEMY_PENALITY ENEMY_PENALITY # 遇上对手的惩罚 self.MOVE_PENALITY MOVE_PENALITY # 每移动一步的惩罚 self.action_space gym.spaces.Discrete(ACTION_SPACE_VALUES) self.observation_space gym.spaces.Box(low-SIZE1, highSIZE-1, shape(4,), dtypeint) # 环境重置 def reset(self,seedNone, optionsNone): self.player Cube(self.SIZE) self.food Cube(self.SIZE) self.enemy Cube(self.SIZE) # 如果玩家和食物初始位置相同重置食物的位置直到位置不同 while self.player self.food: self.food Cube(self.SIZE) # 如果敌人和玩家或食物的初始位置相同重置敌人的位置直到位置不同 while self.player self.enemy or self.food self.enemy: self.enemy Cube(self.SIZE) # 判断观测是图像和数字 if self.RETURN_IMAGE: observation np.array(self.get_image()) else: observation (self.player - self.food)(self.player - self.enemy) observationnp.array(observation) self.episode_step 0 info{} return observation,info def step(self,action): self.episode_step1 self.player.action(action) self.food.move() self.enemy.move() # 分类讨论输出new_obs if self.RETURN_IMAGE: new_observation np.array(self.get_image()) else: new_observation (self.player - self.food)(self.player - self.enemy) new_observationnp.array(new_observation) #获取reward值 if self.player self.food: rewardself.FOOD_REWARD elif self.playerself.enemy: rewardself.ENEMY_PENALITY else: rewardself.MOVE_PENALITY #检测截止情况 terminated False truncated False # 4. 判断结束条件 if self.player self.food or self.player self.enemy: terminated True if self.episode_step self.MAX_STEP: truncated True info{} return new_observation,reward,terminated,truncated,info def render(self,modehuman): imgself.get_image() img img.resize((200,200)) cv2.imshow(Predator,np.array(img)) if self.player self.food or self.player self.enemy or self.episode_step self.MAX_STEP: cv2.waitKey(1500) else: cv2.waitKey(1) def get_image(self): # 图像显示 env np.zeros(self.OBSERVATION_SPACE_VALUES,dtype np.uint8) env[self.food.x][self.food.y] self.d[self.FOOD_N] env[self.player.x][self.player.y] self.d[self.PLAYER_N] env[self.enemy.x][self.enemy.y] self.d[self.ENEMY_N] img Image.fromarray(env,RGB) return img创建环境对象envenvCube()检测环境是否符合Gym标准check_env(env)导入主函数依赖库import gymnasium as gym from stable_baselines3 import DQN,A2C from stable_baselines3.common.evaluation import evaluate_policy import os import stable_baselines3 as sb3 print(sb3.__version__) os.environ[KMP_DUPLICATE_LIB_OK]True构建模型model A2C(MlpPolicy, env, verbose1, tensorboard_log./logs, learning_rate5e-4, policy_kwargs{ net_arch: dict(vf[32, 32], pi[32, 16]), # 修正了这里 activation_fn: th.nn.ReLU } ) model.policy训练模型# Train the agent and display a progress bar model.learn(total_timestepsint(500000), progress_barTrue,tb_log_nameS20_A2C_Net32X32X16_50W)保存模型# Save the agent model.save(S20_A2C_Net32X32X16_50W_lunar) del model # delete trained model to demonstrate loading加载与评估模型model DQN.load(S20_A2C_Net32X32X16_50W_lunar, envenv) model.set_env(env) mean_reward, std_reward evaluate_policy(model, model.get_env(),deterministicTrue,renderFalse, n_eval_episodes10)