【強化学習】DQNを秒速で扱える『ChainerRL』使い方メモ

f:id:imslotter:20170320221632p:plain

こんにちは、ほけきよです！

強化学習って知ってますか？
「AlphaGoが、碁の世界王者を破った」
などと最近脚光を浴びている機械学習技術ですね。

私のブログでも何回か関連記事を出しているのですが、今回は、Chainerで強化学習を簡単に扱えるツール、「ChainerRL」を使ってみました！

なかなか便利だったので使い方を解説やまとめ、Tipsを加えながらメモしてみました。 (コードはjupyter準拠なので、上から順番にコピペすると基本は動くはずです)

これから強化学習やってみたいという人は、これで強化学習がどんなものか試してみるのもいいかもしれません！

強化学習って？
chainerrl
Setup
必要なライブラリをimportする
environmentの設定
- environmentに必要な要件(最低限)
- 今回使うもの
Agentの設定
最適化手法, パラメータの設定
実行(学習)
テストする
学習、テストを凝った設定じゃなく簡単に
おまけ1 : GPUを使ってみる
- エラー
- 実行結果
おまけ2 obsからの可視化
- 可視化結果
終わりに
関連
- 関連記事
- 関連書籍

強化学習って？

この記事が一番わかりやすくて初めには良い。

qiita.com

chainerrl

chainerの強化学習用モジュール
既存のchainerのネットワークを使いながら、最新の強化学習(DQN)を使える.

quickstartに色々と調べたことを加えながら、実際に動かしてみる。

Setup

pip install chainerrl

もしくはソースコードから。ソースコードはここ

github.com

git clone https://github.com/pfnet/chainerrl.git
cd chainerrl
python setup.py install

必要なライブラリをimportする

chainer : Deeplearning用のライブラリ
chainerrl : chainer強化学習用ライブラリ
gym : 強化学習の実験環境

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

environmentの設定

ChainerRLを使うためには、環境モデルを"environments"として保存しておく必要がある
OpenAI Gymにあるものはそのままgym.make(hogehoge)で使えるようになっている
- hogehogeの中にいれられるリストはこのページ OpenAI/envs

environmentに必要な要件(最低限)

observation space　: ある時刻での状態、入力
action space : ある時刻tで選ぶ行動
2つのmethod (resetとstep)
- env.reset : 初期化
- env.step : 実行
  - 行動を実行し、次の状態へ移る
  - 4つの値(次の観測, 報酬, 試行の打ち切りか否か, 追加情報)

今回使うもの

CartPole－V　

倒立振り子。誰もが子供の頃にやった、ほうきを手に乗せて、そのまま保つ。的なやつ。元ネタ論文はこちら。たぶん論文は購読しないと読めない？

observation : [cartの速度, 位置, poleの角速度, 角度]
action : 右に進む or 左に進む

env = gym.make(“CartPole-v0”)

print("observation space   : {}".format(env.observation_space))
print("action space        : {}".format(env.action_space))
obs = env.reset() #初期化
#env.render()#レンダリングした環境を見せてくれる
print("initial observation : {}".format(obs))

action = env.action_space.sample()
obs, r, done, info = env.step(action)

### どんな値が入っているのか確認！
print('next observation    : {}'.format(obs))
print('reward              : {}'.format(r))
print('done                : {}'.format(done))
print('info                : {}'.format(info))

env.render()を使うと、こんな感じの動画が出てくる。OpenAI gymの環境から持ってきたやつなら、デフォで使えるみたいなので、確認したい人はおすすめ。 f:id:imslotter:20170320210923p:plain

Agentの設定

環境をつくれたので、次は環境上を動くagentを作っていく

ChainerRLでデフォで実装されているagent

ほぼ最新と言えるアルゴリズムが揃っている & 開発が盛んなので、state of the artが出れば取り入れてくれる。

対応しているものの表(READMEより引用)

Algorithm	Discrete Action	Continous Action	Recurrent Model	CPU Async Training
DQN (including DoubleDQN etc.)	o	o (NAF)	o	x
DDPG	x	o	o	x
A3C	o	o	o	o
ACER	o	o	o	o
NSQ (N-step Q-learning)	o	o (NAF)	o	o
PCL (Path Consistency Learning)	o	o	o	o

今回はDQNを使う。DQNはAlphaGoでも使われたやつ。

Q関数の設計

強化学習を使うのに重要なQ関数(今の状態と行った行動の結果で、どのくらいの報酬が見込めるか)を決める必要がある。 DQNなどでは、入力からQ関数をニューラルネットワークで近似する

chaineRLではchainer.LinkとしてQ関数を定義することができる
出力はchainerrl.action_value.DiscreteActionValueでラップされている

観測した入力(次元数obs_size)から、次の行動(n_actions)を決定する関数の設計

class QFunction(chainer.Chain):
    def __init__(self, obs_size, n_actions, n_hidden_channels=50):
        #super(QFunction, self).__init__(##python2.x用
        super().__init__(#python3.x用
            l0=L.Linear(obs_size, n_hidden_channels),
            l1=L.Linear(n_hidden_channels,n_hidden_channels),
            l2=L.Linear(n_hidden_channels, n_actions))
        
    def __call__(self, x, test=False): 
        """
        x ; 観測#ここの観測って、stateとaction両方？
        test : テストモードかどうかのフラグ
        """
        h = F.tanh(self.l0(x)) #活性化関数は自分で書くの？
        h = F.tanh(self.l1(h))
        return chainerrl.action_value.DiscreteActionValue(self.l2(h))

obs_size = env.observation_space.shape[0]
n_actions = env.action_space.n
q_func = QFunction(obs_size, n_actions)
# q_func.to_gpu(0) ## GPUを使いたい人はこのコメントを外す

(参考)predifined Q-functions

予め設計されてているQ関数を使うことも可能

_q_func = chainerrl.q_functions.FCStateQFunctionWithDiscreteAction(
    obs_size, n_actions,
    n_hidden_layers=2, n_hidden_channels=50)

最適化手法, パラメータの設定

AgentをDQNで動かすための種々の設定をする

optimizer 何を使って最適化するか。chainerにいろいろと組み込まれている。optimizersリストはこちら
gamma 報酬の割引率.過去の結果をどのくらい重要視するか
explorer 次の戦略を考えるときの方法
replay_buffer Experience Replayを実行するかどうか

optimizer = chainer.optimizers.Adam(eps=1e-2)
optimizer.setup(q_func) #設計したq関数の最適化にAdamを使う
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**6)
phi = lambda x:x.astype(np.float32, copy=False)##型の変換(chainerはfloat32型。float64は駄目)

agent = chainerrl.agents.DoubleDQN(
    q_func, optimizer, replay_buffer, gamma, explorer,
    replay_start_size=500, update_frequency=1,
    target_update_frequency=100, phi=phi)

実行(学習)

環境、Agent及びそれを更新するDQNが完成したので、あとは実行していく

import time
n_episodes = 200
max_episode_len = 200
start = time.time()
for i in range(1, n_episodes + 1):
    obs = env.reset()
    reward = 0
    done = False
    R = 0  # return (sum of rewards)
    t = 0  # time step
    while not done and t < max_episode_len:
        # 動きを見たければここのコメントを外す
        # env.render()
        action = agent.act_and_train(obs, reward)
        obs, reward, done, _ = env.step(action)
        R += reward
        t += 1
    if i % 10 == 0:
        print('episode:', i,
              'R:', R,
              'statistics:', agent.get_statistics())
    agent.stop_episode_and_train(obs, reward, done)
print('Finished, elapsed time : {}'.format(time.time()-start))

テストする

trainingは完了したので、testを実際にやってみる。テストなので、最後agent.stop_episode_and_train(obs, reward, done)は呼ばない。

for i in range(10):
    obs = env.reset()
    done = False
    R = 0
    t = 0
    while not done and t < 200:
        # env.render()
        action = agent.act(obs)
        obs, r, done, _ = env.step(action)
        R += r
        t += 1
    print('test episode:', i, 'R:', R)
    agent.stop_episode()

ここまでで一通りの流れは完了です！！モデルを保存したければ、agent.save("hoge")とすれば保存ができます。

学習、テストを凝った設定じゃなく簡単に

学習・テストをいちいち書くのも面倒なので、そういう時は↓の関数を打つ

これだけで、一発で実行してくれる。

chainerrl.experiments.train_agent_with_evaluation(
    agent, env,
    steps=2000,           # 2000step学習
    eval_n_runs=10,       #  評価(テスト)を10回する
    max_episode_len=200,  # それぞれの評価に対する長さの最大(200s)
    eval_frequency=1000,  # テストを1000stepの学習ごとに実施
    outdir='result')      # 'result'フォルダに保存

おまけ1 : GPUを使ってみる

一行でGPUを使うことができる ↓を、q_funcを定義した後に挟み込む

q_func.to_gpu(0)

エラー

実行時にこんなエラーが出るかも

OSError: Failed to run `nvcc` command. Check PATH environment variable: [Errno 2] No such file or directory: 'nvcc'

たぶん、CUDAを使うPATHが通っていないのが原因なので、チェックしてパスを追加してやる。 python内でパスを追加するならこちら

import os
print(os.environ["PATH"]) #チェック。PATHにCUDAがないことを確認
os.environ["PATH"] +=  ":/usr/local/cuda/bin/" #自分の環境でCUDAが入っているパス

実行結果

おんなじタスクをGPU使ってやる

	CPU	GPU
実行時間	561s	558s

??? どうやら、入力次元が小さすぎて、並列化するところがなさ過ぎるのでだめっぽい？むしろcommunication costがかかっちゃっているのかも

おまけ2 obsからの可視化

env.render()はenvが設計されていないとできないので、観測から普通に可視化ができるか試してみた。

※1回分のテスト結果を可視化

import pylab as plt
import numpy as np
import matplotlib.animation as animation
fig = plt.figure(figsize=(10,5))
ims = []
l = 1.0
obs = env.reset()
R,t,done = 0, 0, False
while not done and t < 200:
    action = agent.act(obs)
    print("t:{} obs:{} action:{}".format(t,obs,action))
    im = plt.plot([-2,2],[0,0],color="black")
    im = plt.plot([obs[1],obs[1]+l*np.sin(obs[3])],[0,l*np.cos(obs[3])],
                  "o-",color="blue",lw=4,label="Pole")
    ims.append(im)
    obs, r, done, _ = env.step(action)
    R += r
    t += 1
#     print("test episode : {} R: {}".format(i,R))
agent.stop_episode()
plt.legend()
plt.xlim(-2.0,2.0)
plt.ylim(-1.0,1.0)
ani = animation.ArtistAnimation(fig, ims, interval=100)
ani.save("animation.gif", writer="imagemagick")