機械学習入門者向け ChainerRLでブロック崩しの学習

深層強化学習を用いたブロック崩しの学習

このアカデミーでは、深層強化学習向けライブラリのChainerRLを使用して、ブロック崩しの学習を体験します。この記事のコードは、chainerrlのquickstart.ipynb ChainerRL Quickstart Guideの一部を改変して作成しました。
扱うゲームを

• CartPore→Breakout
• Q関数で使用するネットワークの変更
• agentに過去4時刻分のゲーム画面を渡す

などの点を変更しました。

ChainerRLのインストールと環境構築

機械学習を用いた画像分類でインストールしたAnacondaが入っている前提で導入を進めます。
Anacondaインストール方法は 機械学習を用いた画像分類 で解説しています。

homeディレクトリに移動します。

$ cd ~

以降の環境導入に必要なパッケージのインストールをします。

$ sudo apt install git
$ sudo apt install cmake
$ sudo apt install zlib1g-dev

今回使用するpythonのライブラリ(chainerとchainerrl)をインストールします。

$ sudo pip3 install chainer==2.1.0
$ sudo pip3 install chainerrl

OpenAIGymの環境構築を行います。

$ git clone https://github.com/openai/gym
$ cd gym
$ pip3 install -e .
$ pip3 install gym[atari]

Jupyter notebookを使ってプログラムの実行

ここからは、実際にプログラムを実行していきます。
Anacondaをインストールしたときに入っている、Jupyter notebook形式での実行を推奨します。
こちらからBreakoutRetrain.zipをダウンロード してください。

上記のzipファイルをダウンロードしたら、ファイルを解凍し、

$ cd ~/Downloads
$ unzip BreakoutRetrain.zip
$ mv ~/Downloads/BreakoutRetrain ~/Documents
$ cd ~/Documents/BreakoutRetrain
$ jupyter notebook BreakoutRetrain.ipynb

でダウンロードしたjupyter notebookを開きましょう。
以降はshift+Enterで各処理ごとのcellを実行していきましょう。
ここからは、プログラムの解説となります。

必要なライブラリのインポート

プログラムを実行するのに必要なライブラリをインポートします。

import chainer
import chainer.functions as F
import chainer.links as L
import chainerrl
import gym
import numpy as np

from scipy import misc
import matplotlib.pyplot as plt
%matplotlib inline
import time

ゲーム環境の作成と確認

画像 : wikipedia

ゲーム環境の作成

ゲーム環境を作成し、OpenAIGymにおいてゲームの情報がどのように扱われているかを確認します。
まず、ブロック崩しのゲーム環境を作成します。また、ゲームの画面描画を4フレームごとに設定します。その後、ゲームから与えられる環境情報と、行動空間を確認します。

env = gym.make('Breakout-v0')
env.frameskip = 4
print('observation space:', env.observation_space)
print('action space:', env.action_space)

次に、ゲーム環境を初期化し、変数obsにゲーム画面の情報が代入されます。
このとき代入されたゲーム画面を84*84にリサイズするための処理を行います。

obs = env.reset()
obs = obs[:,:,0]
obs = (misc.imresize(obs, (110, 84)))[110-84-8:110-8,:]

確認

リサイズされた画像の確認をします。

plt.imshow(obs, cmap='gray')

ゲーム内での行動をランダムに選択し、中身を確認します(行動を表す数値が入っている)。

action = env.action_space.sample()

実行

ゲーム内で行動を実行します。
その結果として行動後のゲーム画面( obs )、報酬値( r, 今回はブロックを崩すと+1 )、ゲームの終了判定( done, Trueで終了 )、学習には使用しない付加情報(info)が返ってきます。

obs, r, done, info = env.step(action)
plt.figure()
plt.imshow(obs)
print('reward:', r)
print('done:', done)
print('info', info)

Q関数の定義

Q関数の定義をします。今回は、84×84にリサイズしたゲーム画面×4時刻分を入力とし、出力にゲームでとることのできる行動を出力します。Q関数で使用するネットワークは、畳み込み層×3と全結合層×2から構成される畳み込みニューラルネットワーク(CNN)を用いています。各層の活性化関数にはReLU関数を採用しています。

class QFunction(chainer.Chain):
def __init__(self, n_actions):
super().__init__(
L0=L.Convolution2D(4 , 32, ksize=8, stride=4),
L1=L.Convolution2D(32, 64, ksize=4, stride=2),
L2=L.Convolution2D(64, 64, ksize=3, stride=1),
L3=L.Linear(3136, 512),
L4=L.Linear(512, n_actions))

def __call__(self, x, test=False):
h = F.relu(self.L0(x))
h = F.relu(self.L1(h))
h = F.relu(self.L2(h))
h = F.relu(self.L3(h))
return chainerrl.action_value.DiscreteActionValue(self.L4(h))

ブロック崩しにおいて取ることのできる行動の種類数を引数に渡して、Q関数を初期化します。

n_actions = env.action_space.n
q_func = QFunction(n_actions)

ネットワーク(Q関数)の最適化手法の選択

勾配の最適化手法にAdamを選択し、optimizerにQ関数をセットします。

optimizer = chainer.optimizers.Adam(eps=1e-2)
optimizer.setup(q_func)

各種パラメータの設定

学習用いる各種パラメータの設定を行います。ここでは学習率は0.95、探索手法はε-greedy法、過去の経験を参照するために保持しておくreplay_bufferのサイズを1000とします。また、np.float32形式ではないデータが発生したときにデータ形式を修正する関数を定義し、ここまでで定義した各種パラメータを引数で渡してDQNのエージェントを初期化します。

gamma = 0.95

explorer = chainerrl.explorers.ConstantEpsilonGreedy(
epsilon=0.3, random_action_func=env.action_space.sample, )

replay_buffer = chainerrl.replay_buffer.ReplayBuffer(capacity=10 ** 4)

phi = lambda x: x.astype(np.float32, copy=False)

agent = chainerrl.agents.DQN(
q_func, optimizer, replay_buffer, gamma, explorer,
replay_start_size=500, update_interval=1,
target_update_interval=100, phi=phi)

実際にゲームの学習を行う

ここからは実際にゲームの学習を行います。
実行時間の関係で、今回は事前に180ゲーム学習した状態から学習を再開します。

• n_episodesで学習するゲーム数を設定します。
• forループの中で実際に学習を実行します。
• forループの中では1ゲームごとにゲーム環境や報酬の合計値などを初期化しています。
• whileループの中では、1ゲームが終了するまでの間(今回はブロック崩しのボールを5個失うまで)、過去4時刻分のリサイズしたゲーム画面と報酬値をもとにゲームの学習を行います。
• 1ゲーム終了するごとに、現在のゲーム数とそのゲーム内での報酬値の合計(崩したブロックの総数)、学習にかかった累計時間を出力します。学習には約30分前後かかります。

agent.load('Breakout180')

n_episodes = 20
start = time.time()
for i in range(181, n_episodes + 181):
obs4steps = np.zeros((4,84,84), dtype=np.float32)
obs = env.reset()
obs = obs[:,:,0]
obs = (misc.imresize(obs, (110, 84)))[110-84-8:110-8,:]
obs4steps[0] = obs
reward = 0
done = False
R = 0
t = 0
while not done:
action = agent.act_and_train(obs4steps, reward)
obs, reward, done, _ = env.step(action)
obs = obs[:,:,0]
obs = (misc.imresize(obs, (110, 84)))[110-84-8:110-8,:]
obs4steps = np.roll(obs4steps, 1, axis=0)
obs4steps[0] = obs
R += reward
t += 1
print('episode:', i,
'R:', R)
print('time:', time.time() - start)
agent.stop_episode_and_train(obs4steps, reward, done)
print('Finished')

学習結果の保存

先ほど学習した結果に名前を付けて保存します。

agent.save(‘Breakout200’)

学習結果の読み出しと確認

最後に学習ゲーム数が50、180、200の性能を比べてみましょう。

ゲーム数50のスコア平均

各ゲームごとの報酬値の合計を保持するための変数を用意します。
学習前のモデルを読み込み、実際に50ゲームテストプレイをします。

プログラムの流れはほとんど学習の時と同じですが、
この学習段階では、ボールを降らせる為の操作(action=1)を学習できていない状態に遷移すると、テストプレイ時に固まってしまうので、それを回避するためのコードが追加してあります。
また、#を消してenv.render()のコメントアウトを外すと、ゲームの様子が別ウインドウで開き確認することができます。

score50 = [0 for i in range(50)]
agent.load('Breakout50')
for i in range(50):
obs4steps = np.zeros((4,84,84))
obs = env.reset()
obs = obs[:,:,0]
obs = (misc.imresize(obs, (110,84)))[110-84-8:110-8, :]
obs4steps[0] = obs
done = False
R = 0
t = 0
while not done:
#env.render()
action = agent.act(obs4steps)
obs, r, done, _ = env.step(action)
obs = obs[:,:,0]
obs = (misc.imresize(obs, (110,84)))[110-84-8:110-8, :]
obs4steps = np.roll(obs4steps, 1, axis=0)
obs4steps[0] = obs
R += r
t += 1
if np.allclose(obs4steps[0], obs4steps[3]):
env.step(1)

score50[i] = R
print('test episode:', i, 'R:', R)

最後に、スコアの平均値を表示します。

print(np.mean(score50))

ゲーム数180,200のスコア平均

先ほどののソースコードとほぼ同じなので載せませんが、Jupyter notebookにて実行してみてください。ゲーム数50の段階よりは打ち返せていると思います。

180と200については、視覚的な動きとしての差は少ないかもしれませんが、各ゲームごとのスコア平均をとると改善されていることが確認できるはずです（学習の進み方はランダムなので100%ではないですが、20回の学習で改善されやすい学習済みモデルを用意しました）。

学習結果の動画