OpenAI Gym 入门

口仆

发布于 2020-09-03 16:32:47

4.8K1

发布于 2020-09-03 16:32:47

1 简介

这一部分参考官网提供的文档[1]，对 Gym 的运作方式进行简单的介绍。Gym 是一个用于开发和比较强化学习算法的工具包，其对「代理」（agent）的结构不作要求，还可以和任意数值计算库兼容（如 Tensorflow 和 Pytorch）。Gym 提供了一系列用于交互的「环境」，这些环境共享统一的接口，以方便算法的编写。

1.1 环境

首先我们可以通过如下代码调用并展示（可视化）一个环境：

import gym
env = gym.make('CartPole-v0')
env.reset()
for _ in range(1000):
    env.render() # 可视化环境
    env.step(env.action_space.sample()) # 选择随机动作
env.close()

该代码创建了一个著名的 CartPole 环境，用于控制小车使上面的杆保持竖直不倒，如下图所示。在每一次迭代中，我们从动作空间中采样了一个随机动作（本环境中只有「向左」和「向右」两个动作）并执行。

执行代码后我们会发现，小车并不能如上图所示维持住平衡，而会直接滚出屏幕外。这是因为我们并没有根据环境的反馈而采取正确的动作。

1.2 观测

为了做出更加合适的动作，我们需要先了解环境的反馈。环境的 step 函数可以返回我们想要的值，其总共返回如下四个值：

observation（「object」）：一个环境特定的对象以表示当前环境的观测状态，如相机的像素数据，机器人的关节角度和速度，桌游中的即时战况等
reward（「float」）：前一个动作所获得的奖励值，其范围往往随着环境的变化而各不相同，但目标一般都是提升总奖励值
done（「boolean」）：是否需要重置（reset）环境，不同的环境会有不同的终止条件，包括执行动作的次数限制、状态的变化阈值等
info（「dict」）：输出学习过程中的相关信息，一般用于调试

通过上述函数，我们可以实现经典的「代理-环境循环」，在每个时间步，代理选择一个动作，环境返回一个观察（状态）和一个奖励：

基于环境的反馈，我们可以对代码进行如下修改，达到终止条件时即退出循环：

import gym
env = gym.make('CartPole-v0')
for i_episode in range(20):
    observation = env.reset()
    for t in range(100):
        env.render()
        print(observation)
        action = env.action_space.sample()
        observation, reward, done, info = env.step(action)
        if done:
            print("Episode finished after {} timesteps".format(t+1))
            break
env.close()

在 CartPole 环境中，根据官方 wiki[2]，「观测状态」有四维：小车位置、小车速度、杆的角度和杆顶端的速度（初始状态每个值均在 ±0.05 区间内取随机值）；「终止条件」有三条：杆的角度大于 ±12 度，小车的位置超过 ±2.4，以及迭代次数超过 200（v1 为 500）；每个步骤的「奖励」均为 1（包括终止步）。

1.3 空间

在 Gym 中，状态和动作都是通过 Space 类型来表示的，其可以定义连续或离散的子空间。最常用的两种 Space 是 Box 和 Discrete，在 CartPole 环境中状态空间和动作空间就分别对应这两种 Space：

import gym
env = gym.make('CartPole-v0')
print(env.action_space)
#> Discrete(2)
print(env.observation_space)
#> Box(4,) # 注意其第二维未指定

Discrete 定义了一个从 0 开始取值的离散空间，而 Box 则可以表示一个 m*n 维的连续空间，需要为每个维度设置上下界。我们可以通过如下方式新建空间：

from gym import spaces
space = spaces.Discrete(8) # 包含 {0, 1, 2, ..., 7} 八个元素的集合

# 每一维相同的上下界
space_box_1 = Box(low=-1.0, high=2.0, shape=(3, 4), dtype=np.float32)
# Box(3, 4)

# 每一维不同的上下界
space_box_2 = Box(low=np.array([-1.0, -2.0]), high=np.array([2.0, 4.0]), dtype=np.float32)
# Box(2,)

除了上述两种空间外，Gym 还提供了一些其他的空间，包括多维离散空间、字典空间等，具体可以参考官方源码[3]。

2 自定义环境

Gym 内置了许多强化学习的经典环境，包括经典控制、雅达利游戏、算法、机器人控制等。我们可以通过下述代码查看所有可以直接调用的环境：

from gym import envs
print(envs.registry.all())

但在实际应用中，面对一个全新的场景，我们往往需要自定义一个环境来训练我们的算法。本节将介绍如何自定义一个环境。

2.1 文件结构

根据官方说明[4]，创建一个新的环境需要建立如下结构的 PIP 包：

gym-foo/
├── README.md
├── setup.py # 声明模块的相关信息和依赖
└── gym_foo/
    ├── __init__.py # 通过register注册自定义环境
    └── envs/
        ├── __init__.py # 导入自定义环境模块
        └── foo_env.py # 实现环境所需的各种方法

其中 gym-foo/setup.py 应包含如下代码：

from setuptools import setup

setup(name='gym_foo',
      version='0.0.1',
      install_requires=['gym']  # 声明包依赖
)

gym-foo/gym-foo/__init__py 应包含如下代码：

from gym.envs.registration import register

register(
    id='foo-v0',
    entry_point='gym_foo.envs:FooEnv',
)

gym-foo/gym-foo/envs/__init__py 应包含如下代码：

from gym_foo.envs.foo_env import FooEnv

gym-foo/gym_foo/envs/foo_env.py 应包含如下代码：

import gym
from gym import error, spaces, utils
from gym.utils import seeding

class FooEnv(gym.Env):
  metadata = {'render.modes': ['human']} # 只是示例，并不是必要语句

  def __init__(self):
    ...
  def step(self, action):
    ...
  def reset(self):
    ...
  def render(self, mode='human'):
    ...
  def close(self):
    ...

当全部定义完成后，我们可以在一级目录下通过 pip install -e . 来安装自定义环境（-e 表示本地可编辑的代码，可以快速更新改动），然后即可在程序中调用该环境：

import gym
import gym_foo
env = gym.make('foo-v0')

2.2 案例

下面通过经典的「井字棋」（Tic-Tac-Toe）游戏来说明环境的自定义方法。井字棋的玩法如下（就是简化版五子棋）：

在井字棋环境中，状态即当前棋盘的局面，动作则是每一回合玩家的走棋。这里假定玩家为先手，电脑为后手。奖励基于玩家的胜负情况制定。具体代码如下：

class CdmEnv(gym.Env):
    def __init__(self):
        self.state = [['-' for _ in range(3)] for _ in range(3)]
        self.counter = 0 # 当前步数
        self.done = 0 # 对局是否结束
        self.add = [0, 0] # 表示胜负情况，1代表获胜
        self.reward = 0
  
    def check(self): # 检查当前棋盘是否有人可以获得胜利
        if self.counter < 5:
            return 0 # 步数小于五步，无人获胜
        for i in range(3): # 判断横竖三行是否有连子，有则返回对应胜者
            if self.state[i][0] != '-' and self.state[i][1] == self.state[i][0] and self.state[i][1] == self.state[i][2]:
                if self.state[i][0] == 'o': # o代表第一名选手的棋子
                    return 1
                else:
                    return 2
            if self.state[0][i] != '-' and self.state[1][i] == self.state[0][i] and self.state[1][i] == self.state[2][i]:
                if self.state[0][i] == 'o': # o代表第一名选手的棋子
                    return 1
                else:
                    return 2
        # 判断两条对角线上是否有连子
        if self.state[0][0] != '-' and self.state[1][1] == self.state[0][0] and self.state[1][1] == self.state[2][2]:
            if self.state[0][0] == 'o': # o代表第一名选手的棋子
                    return 1
            else:
                return 2
        if self.state[0][2] != '-' and self.state[1][1] == self.state[0][2] and self.state[1][1] == self.state[2][0]:
            if self.state[0][2] == 'o': # o代表第一名选手的棋子
                return 1
            else:
                return 2
        # 所有其他情况均无胜者
        return 0

    def step(self, action): # 执行动作，返回状态、奖励、是否重置及相关信息
        x, y = action[0], action[1]# 这里用1-9的数字表示棋盘上点的坐标
        if self.done == 1:
            print("游戏结束")
            return [self.state, self.reward, self.done, self.add]
        elif self.state[x][y] != '-':
            print("非法步骤")
            return [self.state, self.reward, self.done, self.add]
        else:
            if self.counter % 2 == 0: # 偶数步表示第一名选手
                self.state[x][y] = 'o'
            else:
                self.state[x][y] = 'x'
            self.counter += 1
            if self.counter == 9: # 棋盘放满则游戏结束
                self.done = 1
            # self.render() # 可视化棋盘
        win = self.check() # 判断当前的胜负情况
        if win: # 如果可以分出胜负
            self.done = 1
            print("玩家 ", win, " 号获胜", sep = '', end = '\n')
            self.add[win - 1] = 1
            if win == 1: # 这里1号玩家赢了给予奖励，2号赢了则给予惩罚（即环境希望1号获胜）
                self.reward = 1
            else:
                self.reward = -1
        return [self.state, self.reward, self.done, self.add]

    def reset(self): # 重置环境
        self.state = [['-' for _ in range(3)] for _ in range(3)]
        self.counter = 0
        self.done = 0
        self.add = [0, 0]
        self.reward = 0
        return self.state

    def render(self):
        for i in range(3):
            for j in range(3):
                print(self.state[i][j], end = " ")
            print("") # 换行
        print("")
    
    def getPosition(self, x, y):
        item = self.state[x][y]
        return item

    def close(self):
        print("env closed")

实际上，上述环境存在许多缺陷。首先是状态空间的表示，我们无法直接判断当前位置是否包含棋子，需要在 agent 中去记录，这显然是不合理的；其次是获胜条件与奖励函数的制定，我们希望环境不去区分玩家与电脑，而是针对每一步给出当前玩家应该受到的奖励。因此，可以考虑将奖励改为当前步下获胜以及防止下一步对方获胜的奖励，相应的胜负判断条件也需要进行修改。一个比较完善的代码（基于 Q 学习）可以参考这里[5]。

以上就是 OpenAI Gym 的相关介绍及自定义环境的简单示范。