目录
前言
关于tensorforce的背景可以从他的名字看出来,tensor*必然是师从tensorflow框架的,事实上也确实是由tensorflow框架来的。使用tensorforce可以快速的构建强化学习代码,同时还可以利用其中现有的强化学习框架,例如常见的AC,A2C,A3C,PPO等。并且可以很好地结合OpenAIGym等主流开源环境和框架。
tensorforce的要点
tensorforce的要点,或者说强化学习的要点大致可以分为两大部分,一是环境(Environment),二是代理(Agent)。
Environment
tensorforce当中的环境可以使用开源框架的自带环境(如果你希望在OpenAIGym等环境上进行实验操作),或者自定义环境。
开源框架环境
支持的开源环境框架有以下这些:
+ Arcade Learning Environment 一个简单的面向对象的框架,允许研究人员和业余爱好者为Atari 2600游戏开发AI代理。
+ CARLA, 是用于自动驾驶研究的开源模拟器。
+ OpenAI Gym, 一种用于开发和比较强化学习算法的工具包,该工具包支持教学代理从走路到玩游戏(例如Pong或Pinball)的所有操作。
+ OpenAI Retro, 可让您将经典的视频游戏转换为Gym环境以进行强化学习,并提供约1000种游戏的集成。
+ OpenSim, 包含肌肉骨骼模型在内的人体模型仿真环境。
+ PyGame 可以快速开始使用Python进行强化学习的环境。
+ ViZDoom, 允许开发者仅使用视觉信息玩毁灭战士的AI机器人。
开源环境的创建相对于通用环境的创建简单,只需要少数语句和关键词就可以实现。以OpenAIGym为例:
class tensorforce.environments.OpenAIGym(level, visualize=False, import_modules=None, min_value=None, max_value=None, terminal_reward=0.0, reward_threshold=None, drop_states_indices=None, visualize_directory=None, **kwargs)
创建只需要
env=OpenAIGym(level='LunarLanderv2',visualize=True)
就可以实现LunarLanderv2可视化的着陆游戏。并使用
env.states()以及env.actions()
查看状态空间和动作空间
通用的环境:
通用的环境的创建,采用.create()方法创建
#static mrthod
Environment.create(environment=None, max_episode_timesteps=None, remote=None, blocking=False, host=None, port=None, **kwargs)
- 其中environment参数必须为以下几种:JSON文件,特定关键词(如ac,a2c等,可在官方文档上查到),配置字典(同JSON文件),库模块,环境对象或Gym.Env
- max_episode_timesteps为每个episode的最大迭代步数
另外的重要方法有:
.close(), .states(), .actions(), .execute()
自己采用gym平台编写环境
# 保存为customEnv.py
import gym
from gym import spaces
import numpy as np
class customEnv(gym.Env):
# metadata 是修改渲染时的参数,如果自定义的环境不用图形化,那么无需设置这个参数
# metadata = {
# 'render.modes': ['human', 'rgb_array'],
# 'video.frames_per_second': 2
# }
def __init__(self):
self.observation_space_name = np.array(['hglass', 'hito', 'hmapbi3', 'hpcbm', 'hpedot']) # 状态空间的名字,针对离散空间而言
self.state_lb=[1,1.5,16,1,1] #状态空间的下限
self.state_ub=[10,8,25,5,5] #状态空间的上限
self.action_space = spaces.Discrete(10) #构建离散动作的,一共10个离散动作
self.observation_space = spaces.Box(np.array(self.state_lb), np.array(self.state_ub)) # 使用space.Box方法搭建状态空间,设置lb下限和ub上限
self.state = None #此处不设置初始状态
def step(self, action):
assert self.action_space.contains(action) ,'Invalid action!' #对于超出范围的动作幅度显示无效动作
#以下为环境对传入动作进行响应,返回state,reward,done,info四个信息
last_state=self.state
#以下为对action进行响应的一个示例
if np.mod(action,2)==1: #对相应的位置进行减去操作
self.state[int((action-1)/2)]=self.state[int((action-1)/2)]-self.precision_level
else:
self.state[int((action) / 2)] = self.state[int((action) / 2)] + self.precision_level
if not self.observation_space.contains(self.state):
self.state=last_state
reward=-10
done=True
info='out of bound'
else:
last_eff=strcture2effi(last_state,self.model)
curr_eff=strcture2effi(self.state,self.model)
if curr_eff<last_eff:
reward=-10
done=True
info='current_eff less than last_eff'
else:
if curr_eff>0:
reward = 50
done = False
info = 'current_eff more than 0 and last_eff'
else:
reward = 5
done = False
info = 'current_eff more than last_eff but less than 0'
return self.state,reward,done,info
def reset(self):
'''
清空环境
:return:
'''
self.state=np.array([10,2,23,1.5,2]) #请空环境时将状态设置为某些值
return self.state
def render(self, mode='human'):
return None # 渲染画面,不需要的话无需写这个功能
def close(self):
return None # 关闭环境
将自己编写的环境接入Tensorforce框架
- 首先需要注册自定义的gym环境。首先找到gym环境的文件夹位置。如果使用的是anaconda,路径是~\Anaconda\Lib\site-packages\gym\gym\envs。新建一个文件夹user并进入。将上方的customEnv.py放入user文件夹当中。并增加入口文件init.py,内容为:
# __init__.py
from gym.envs.user.Car2D import Car2DEnv
- 再次进入到~\Anaconda\Lib\site-packages\gym\envs文件夹中。向入口文件init.py填写如下内容:
register(
id='customEnv-v0', # 自定义id,别名
entry_point='gym.envs.user:customEnv',
max_episode_steps=1000, # 最大episode
reward_threshold=35.0, # reward的最小值
)
如果在tensorforce中不希望环境运行之后被擦除,可以在tensorforce当中的runner.py源码当中把以下两句注释掉:
self.agent.close()
self.environment.close()
- 以上步骤之后就可以使用tensorforce进行训练
import numpy as np
from tensorforce.agents import PPOAgent
from tensorforce.execution import Runner
from tensorforce.contrib.openai_gym import OpenAIGym
env = OpenAIGym('customEnv-v0', visualize=False) #第一个参数就是自定义的id参数
#Network as list of layers
network_spec = [
dict(type='dense', size=32, activation='tanh'),
dict(type='dense', size=32, activation='tanh')
] #自定义 Policy network 的结构,list的层叠结构
agent = PPOAgent(
states_spec=env.states, #这里会使用我们之前的定义的state
actions_spec=env.actions, # 以及action
network_spec=network_spec, #Policy network的结构,强化学习中不宜太复杂,否则不容易收敛
batch_size=32, #batch size建议2的幂
# BatchAgent
keep_last_timestep=True,
# PPOAgent
step_optimizer=dict(
type='adam',
learning_rate=1e-3
),
optimization_steps=10,
# Model
scope='ppo',
discount=0.99,
# DistributionModel
distributions_spec=None,
entropy_regularization=0.01,
# PGModel
baseline_mode=None,
baseline=None,
baseline_optimizer=None,
gae_lambda=None,
# PGLRModel
likelihood_ratio_clipping=0.2,
summary_spec=None,
distributed_spec=None
)
#Create the runner
def episode_finished(r):
print("Finished episode {ep} after {ts} timesteps (reward: {reward})".format(
ep=r.episode, ts=r.episode_timestep, reward=r.episode_rewards[-1]))
return True
runner = Runner(agent=agent, environment=env)
#Callback function printing episode statistics
def episode_finished(r):
print("Finished episode {ep} after {ts} timesteps (reward: {reward})".format(ep=r.episode, ts=r.episode_timestep,
reward=r.episode_rewards[-1]))
return True
#Start learning
runner.run(episodes=1000, max_episode_timesteps=200, episode_finished=episode_finished)
#Print statistics
print("Learning finished. Total episodes: {ep}. Average reward of last 100 episodes: {ar}.".format(ep=runner.episode,ar=np.mean(runner.episode_rewards[-100:])))
while True:
agent.reset()
state, done = env.reset(), False
episode_reward = 0
while not done:
action = agent.act(state, deterministic = True)
state, done, reward = env.execute(action)
agent.observe(done, reward)
episode_reward += reward
print([episode_reward])
关于PPOAgent的参数详细介绍可以在官网查看