OpenAI Gym で自前の環境をつくる

この記事はBrainPad AdventCalendar 20174日目の記事です。

私の所属するチームでは、最近強化学習の実応用に関する研究に取り組んでいます。この記事では、強化学習を使って独自の問題に取り組む最初の一歩を紹介します。

OpenAI Gym と Environment

OpenAI Gym は、非営利団体 OpenAI の提供する強化学習の開発・評価用のプラットフォームです。
強化学習は、与えられた環境(Environment)の中で、エージェントが試行錯誤しながら価値を最大化する行動を学習する機械学習アルゴリズムです。そのため、エージェントの学習アルゴリズムそのものでなく、環境もとても大事な要素となります。OpenAI Gym では強化学習でよく使われる様々な環境が共通のインターフェイスで利用でき、実験した結果をアップロードして他人のアルゴリズムと比較したりすることも出来るようになっています。
とてもありがたいのですが、強化学習を実用するには、OpenAI Gym では提供されていない、独自の環境を準備する必要があります。そこで、このエントリーでは、OpenAI Gym における環境の作り方をまとめようと思います。

OpenAI Gym のインストール

何はともあれ、 OpenAI Gym をインストールしなければいけません。

$ pip install git+https://github.com/openai/gym

環境(Environment) づくりの基本

OpenAI Gym では、以下の手順で独自の環境を構築します。

1. gym.Env を継承し、必要な関数を実装する
2. gym.envs.registration.register 関数を使って gym に登録する

それでは、 1. から具体的に見ていきます。

gym.Env を継承したクラスを実装する

gym.Env の子クラスは、以下のメソッドとプロパティを実装する必要があります。

メソッド	解説
_step(self, action)	action を実行し、結果を返す
_reset(self)	状態を初期化し、初期の観測値を返す
_render(self, mode='human', close=False)	環境を可視化する
_close(self)	環境を閉じて後処理をする
_seed(self, seed=None)	ランダムシードを固定する

プロパティ	解説
action_space	行動(Action)の張る空間
observation_space	観測値(Observation)の張る空間
reward_range	報酬の最小値と最大値のリスト

_close と _seed は必須ではありません。
また、_renderの引数にmodeがありますが、これは任意の文字列です。(class毎に受け取れる文字列を指定します。)ただし、慣習として

• human: 人間の為にコンソールか画面に表示。戻り値なし。画像の表示には gym.envs.classic_control.rendering.SimpleImageViewer が使える。
• rgb_array: 画面のRGB値をnumpy.array(形は(x, y, 3)) で返す
• ansi:　文字列もしくは StringIO を返す

と言うものがあります。(全てを実装する必要はありません)

具体例を挙げます。この例では、MyEnv.MAP のスタートからゴールまで、勇者が歩くことを想定しています。途中、敵との遭遇(簡単のため、遭遇するとダメージ10を受ける)や毒沼(一歩進む毎に1のダメージ)がある中で、如何にゴールまでたどり着くか、を学習させます。

myenv/env.py

import sys

import gym
import numpy as np
import gym.spaces


class MyEnv(gym.Env):
    metadata = {'render.modes': ['human', 'ansi']}
    FIELD_TYPES = [
        'S',  # 0: スタート
        'G',  # 1: ゴール
        '~',  # 2: 芝生(敵の現れる確率1/10)
        'w',  # 3: 森(敵の現れる確率1/2)
        '=',  # 4: 毒沼(1step毎に1のダメージ, 敵の現れる確率1/2)
        'A',  # 5: 山(歩けない)
        'Y',  # 6: 勇者
    ]
    MAP = np.array([
        [5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5],  # "AAAAAAAAAAAA"
        [5, 5, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2],  # "AA~~~~~~~~~~"
        [5, 5, 2, 0, 2, 2, 5, 2, 2, 4, 2, 2],  # "AA~S~~A~~=~~"
        [5, 2, 2, 2, 2, 2, 5, 5, 4, 4, 2, 2],  # "A~~~~~AA==~~"
        [2, 2, 3, 3, 3, 3, 5, 5, 2, 2, 3, 3],  # "~~wwwwAA~~ww"
        [2, 3, 3, 3, 3, 5, 2, 2, 1, 2, 2, 3],  # "~wwwwA~~G~~w"
        [2, 2, 2, 2, 2, 2, 4, 4, 2, 2, 2, 2],  # "~~~~~~==~~~~"
    ])
    MAX_STEPS = 100

    def __init__(self):
        super().__init__()
        # action_space, observation_space, reward_range を設定する
        self.action_space = gym.spaces.Discrete(4)  # 東西南北
        self.observation_space = gym.spaces.Box(
            low=0,
            high=len(self.FIELD_TYPES),
            shape=self.MAP.shape
        )
        self.reward_range = [-1., 100.]
        self._reset()

    def _reset(self):
        # 諸々の変数を初期化する
        self.pos = self._find_pos('S')[0]
        self.goal = self._find_pos('G')[0]
        self.done = False
        self.damage = 0
        self.steps = 0
        return self._observe()

    def _step(self, action):
        # 1ステップ進める処理を記述。戻り値は observation, reward, done(ゲーム終了したか), info(追加の情報の辞書)
        if action == 0:
            next_pos = self.pos + [0, 1]
        elif action == 1:
            next_pos = self.pos + [0, -1]
        elif action == 2:
            next_pos = self.pos + [1, 0]
        elif action == 3:
            next_pos = self.pos + [-1, 0]

        if self._is_movable(next_pos):
            self.pos = next_pos
            moved = True
        else:
            moved = False

        observation = self._observe()
        reward = self._get_reward(self.pos, moved)
        self.damage += self._get_damage(self.pos)
        self.done = self._is_done()
        return observation, reward, self.done, {}

    def _render(self, mode='human', close=False):
        # human の場合はコンソールに出力。ansiの場合は StringIO を返す
        outfile = StringIO() if mode == 'ansi' else sys.stdout
        outfile.write('\n'.join(' '.join(
                self.FIELD_TYPES[elem] for elem in row
                ) for row in self._observe()
            ) + '\n'
        )
        return outfile

    def _close(self):
        pass

    def _seed(self, seed=None):
        pass

    def _get_reward(self, pos, moved):
        # 報酬を返す。報酬の与え方が難しいが、ここでは
        # - ゴールにたどり着くと 100 ポイント
        # - ダメージはゴール時にまとめて計算
        # - 1ステップごとに-1ポイント(できるだけ短いステップでゴールにたどり着きたい)
        # とした
        if moved and (self.goal == pos).all():
            return max(100 - self.damage, 0)
        else:
            return -1

    def _get_damage(self, pos):
        # ダメージの計算
        field_type = self.FIELD_TYPES[self.MAP[tuple(pos)]]
        if field_type == 'S':
            return 0
        elif field_type == 'G':
            return 0
        elif field_type == '~':
            return 10 if np.random.random() < 1/10. else 0
        elif field_type == 'w':
            return 10 if np.random.random() < 1/2. else 0
        elif field_type == '=':
            return 11 if np.random.random() < 1/2. else 1

    def _is_movable(self, pos):
        # マップの中にいるか、歩けない場所にいないか
        return (
            0 <= pos[0] < self.MAP.shape[0]
            and 0 <= pos[1] < self.MAP.shape[1]
            and self.FIELD_TYPES[self.MAP[tuple(pos)]] != 'A'
        )

    def _observe(self):
        # マップに勇者の位置を重ねて返す
        observation = self.MAP.copy()
        observation[tuple(self.pos)] = self.FIELD_TYPES.index('Y')
        return observation

    def _is_done(self):
        # 今回は最大で self.MAX_STEPS までとした
        if (self.pos == self.goal).all():
            return True
        elif self.steps > self.MAX_STEPS:
            return True
        else:
            return False

    def _find_pos(self, field_type):
        return np.array(list(zip(*np.where(
        self.MAP == self.FIELD_TYPES.index(field_type)
    ))))
...

gym に登録する

上記のクラスはそのままでも使えるのですが、以下の様に gym.envs.registration.register を使うと、gym.make('....') で自作の環境を呼び出すことができます。プログラム実行初期に呼び出したいため、__init__.py 内に記述します。

myenv/__init__.py

from gym.envs.registration import register

register(
    id='myenv-v0',
    entry_point='myenv.env:MyEnv'
)

ここで、id は<環境名>-v<バージョン>という形式である必要があります。
これにより、以下のようなコードで、自分の環境MyEnvを呼び出せるようになります。

import myenv
import gym

env = gym.make('myenv-v0')
...

Keras-RLを使って、自前の環境で強化学習を行う

さて、自前の環境ができたら、強化学習をしてみたくなります。
OpenAI は baseline というリポジトリで有名なアルゴリズムの実装を公開しているので、それを使ってもよいのですが、本記事では、お手軽さの観点から Keras-RL を紹介します。
(Keras-RL以外には、ChainerRL などのライブラリも OpenAI Gym の環境に対応しています。)

Keras-RL のインストール

Keras-RL は pip を使って簡単にインストール出来ます。

$ git clone https://github.com/matthiasplappert/keras-rl.git
$ python keras-rl/setup.py install

Keras-RLの動作確認

まずはサンプルを動かしてみましょう。

$ python keras-rl/examples/dqn_cartpole.py

以下のような画面があらわれたら成功です。

独自の環境を使ってみる

dqn_cartpole.py の冒頭部分を下記のように書き換えると、独自の環境で学習をしてくれます

keras-rl/examples/dqn_cartpole.py

import myenv  # これを読み込んでおく
import numpy as np
import gym

from keras.models import Sequential
from keras.layers import Dense, Activation, Flatten
from keras.optimizers import Adam

from rl.agents.dqn import DQNAgent
from rl.policy import BoltzmannQPolicy
from rl.memory import SequentialMemory


ENV_NAME = 'myenv-v0'  # register で使った ID
...

Keras-RL の基本

最後に、keras-rl/examples/dqn_cartpole.py を例に、 Keras-RL の基本的な流れを紹介します。といっても、とても簡単なので、 dqn_cartpole.py を読んでいただければすぐに分かるかと思います。

1. モデルの定義

Keras-RLでは、まずモデルを定義します。ここで言うモデルは、Kerasのモデルのことで、observation を入力とし、Q値を推定するネットワークです。keras-rl/examples/dqn_cartpole.pyでは、隠れ層が3層のMLPを使っています。

# Next, we build a very simple model.
model = Sequential()
model.add(Flatten(input_shape=(1,) + env.observation_space.shape))
model.add(Dense(16))
model.add(Activation('relu'))
model.add(Dense(16))
model.add(Activation('relu'))
model.add(Dense(16))
model.add(Activation('relu'))
model.add(Dense(nb_actions))
model.add(Activation('linear'))
print(model.summary())

2. エージェントを定義

次にエージェントを定義します。Keras-RLでは、DQNなど、価値ベースのアルゴリズムが既に実装されているので、それを利用します。コンストラクタの引数は、エージェント毎に違いますが、それぞれexampleが用意されているので、まずはそれを真似するのが良いでしょう。

# Finally, we configure and compile our agent. You can use every built-in Keras optimizer and
# even the metrics!
memory = SequentialMemory(limit=50000, window_length=1)
policy = BoltzmannQPolicy()
dqn = DQNAgent(model=model, nb_actions=nb_actions, memory=memory, nb_steps_warmup=10,
               target_model_update=1e-2, policy=policy)
dqn.compile(Adam(lr=1e-3), metrics=['mae'])

3. コンパイル・学習・適用

学習や適用は、普通のKerasのモデルと似た感じで実装できます。Kerasのモデルを利用しているので、コンパイルも必要です。

dqn.compile(Adam(lr=1e-3), metrics=['mae'])

# Okay, now it's time to learn something! We visualize the training here for show, but this
# slows down training quite a lot. You can always safely abort the training prematurely using
# Ctrl + C.
dqn.fit(env, nb_steps=50000, visualize=True, verbose=2)

# After training is done, we save the final weights.
dqn.save_weights('dqn_{}_weights.h5f'.format(ENV_NAME), overwrite=True)

# Finally, evaluate our algorithm for 5 episodes.
dqn.test(env, nb_episodes=5, visualize=True)

まとめ

本記事では、強化学習の第一歩として、OpenAI Gym の独自環境の作り方と、Keras-RLを紹介しました。
強化学習は、今非常にホットな分野で、新しいアルゴリズムがどんどん出てきています。実は Keras-RL にはA3Cを始めとする最近のアルゴリズムは実装されていませんが、DQNなどのアルゴリズムを非常に簡単に使うことができるので「強化学習に触れてみる」という目的にはとても良いと思います。