前言

关于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框架

  1. 首先需要注册自定义的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
  1. 再次进入到~\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()
  1. 以上步骤之后就可以使用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的参数详细介绍可以在官网查看

类似文章