俺のブログ

学習メモや日記をつらつらと書きます。

夏休みDay n(nは任意の整数)強化学習でFlappy_Birdやってみた

強化学習でFlappy_Birdやってみた(Q学習編)


Playing Flappy Bird by Q_Learning

↑2000回学習させたもの

忙しい人のためにコードはコチラ

記事を読む上での注意

  • 細かい強化学習の説明はできないため書いてありません
  • 勉強中のため色んな本やサイトを参考にコードを書いています
  • 下手な説明
  • 汚いコード
  • DeepLearningとかではないです

以上のことを許容できる方は読んでみてください。

目標

Flappy Birdを強化学習を使って攻略させること。
今回は強化学習の中でもQ学習を使って取り組みます。
以降DQNやDDQN、A3Cなどの他の手法も使って比較していこうと思っています。

概要

OpenAI Gymである程度遊んで飽きてしまい、自分でOpenAI Gym以外のタスクに取り組んでみたかったことからPLE(PyGame Learning Environment)に取り組もうと考えました。PLEとはOpenAI Gymのような強化学習用のシミュレーション環境のことです。今回はその中でもFlappy Birdというゲームに取り組みます。

Flappy Bird Nguyen Ha Dongによるスマートフォン向けゲームアプリ。 2013年5月に公開されて以来、その難易度の高さと中毒性から人気を博していたが、「収入は増えた代わりに日常生活が台無しになった」として削除された。

引用 : Hatena Keyword

使用した環境

環境 Version
OS mac OS X Sierra
Python 3.6.1
Anaconda Anaconda 4.4.0 (x86_64)
numpy 1.15.0
matplotlib 2.0.2
PLE 3.10
pygame 1.9.4
Pillow 4.1.1

PLEが公式にサポートしているのはPython2.7.6になります(2018年8月16日現在)。
他のバージョンでは動作は保証できません。

環境構築

Python, Anacondaの環境はある前提です。ない方は他のサイトなりを参考にしてください。

PLEのインストール

公式の説明はココ(英語)

PLEを実行する上で下記のライブラリが必要となります。

Anacondaを入れられている人はnumpyは入っているはずなので以下のライブラリをそれぞれインストールしましょう。

$ pip install Pillow
$ pip install pygame

以上のライブラリをインストールできたらGitHubリポジトリからcloneしましょう。

$ git clone https://github.com/ntasfi/PyGame-Learning-Environment

cloneが終わったら以下のコマンドを実行します。

$ cd PyGame-Learning-Environment
$ sudo pip install -e .

これでインストールが完了です!

公式のReferenceページにはQuickStartといってPLEのサンプルコードがありますが、Python3系ではライブラリ側に問題があって動きそうにないのでQuickStartするのは諦めましょう。
どうしてもFlappy Birdで遊びたい人はココからFlappy BirdのClone環境をインストールしてきましょう。

コードの解説

コードはココにあるので適宜参考にしてください。

ライブラリのインポート

import math
import copy
import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline
import os
os.environ["SDL_VIDEODRIVER"] = "dummy"  # ポップアウトウィンドウを表示しない
from ple.games.flappybird import FlappyBird
from ple import PLE
from collections import defaultdict
from ple.games.flappybird import FlappyBird

他のタスクを実行したい場合はここのFlappyBirdの部分を変えましょう。
他のタスク一覧

アニメの作成

moviepyというライブラリを使用しているため以下を実行してください。依存環境としてffmpegが必要です。

$ pip install moviepy
def make_anim(images, fps=60, true_image=False):
    duration = len(images) / fps
    import moviepy.editor as mpy

    def make_frame(t):
        try:
            x = images[int(len(images) / duration * t)]
        except:
            x = images[-1]

        if true_image:
            return x.astype(np.uint8)
        else:
            return ((x + 1) / 2 * 255).astype(np.uint8)

    clip = mpy.VideoClip(make_frame, duration=duration)
    clip.fps = fps
    return clip

特に説明はなし。サイトから丸パクリしました。

グラフの作成

累積報酬とプレイできた時間をグラフに表示します。

def make_graph(reward_per_epoch, lifetime_per_epoch):
    fig, (axL, axR) = plt.subplots(ncols=2, figsize=(10,4))
    axL.set_title('lifetime')
    axL.grid(True)
    axL.plot(lifetime_per_epoch)
    axR.set_title('reward')
    axR.grid(True)
    axR.plot(reward_per_epoch)
    fig.show()

定数の宣言

ETA = 0.5
GAMMA = 0.99
GOAL_FRAME = 1200 

学習率 ETA 0.5
時間割引率 GAMMA 0.99
目標フレーム数 GOAL_FRAME 1200

Agentクラス

実際にゲームをプレイするAgentクラスを宣言します。

class Agent:
    def __init__(self, num_actions):
        self.brain = Brain(num_actions)
    
    def update_Q_function(self, state, action, reward, state_prime):
        self.brain.update_policy(state, action, reward, state_prime)
    
    def get_action(self, state, episode):
        action = self.brain.decide_action(state, episode)
        return action

コンストラクタでAgentの脳みそとなるBrainクラスをインスタンス化します。
update_Q_functionではQ値を更新するupdate_policyを呼び出します。
get_actionではQ値から行動を選択します。

Brainクラス

Agentの脳みその役割を担うBrainクラスを宣言します。

class Brain:
    
    def __init__(self, num_actions):
        self.num_actions = num_actions
        self.q_table = defaultdict(lambda: np.zeros(num_actions))

    def decide_action(self, state, episode):
        #  ε-greedy
        state_idx = self.get_state_idx(state)  # 相対位置の取得
        epsilon = 0.5 * (1 / (episode + 1))
        if epsilon <= np.random.uniform(0, 1):
            action = np.argmax(self.q_table[state_idx])  # Q値が最大の行動を選択する
        else:
            action = np.random.choice(self.num_actions)  # ランダムな行動を選択する
        return action

    def update_policy(self, state, action, reward, state_prime):
        state_idx = self.get_state_idx(state)
        state_prime_idx = self.get_state_idx(state_prime)
        # Q学習を用いてQ値を更新する
        best_q = np.max(self.q_table[state_prime_idx])
        self.q_table[state_idx][action] += ETA * (
            reward + GAMMA * best_q - self.q_table[state_idx][action])
    
    bucket_range_per_feature = {
        'next_next_pipe_bottom_y': 40,
        'next_next_pipe_dist_to_player': 512,
        'next_next_pipe_top_y': 40,
        'next_pipe_bottom_y': 20,
        'next_pipe_dist_to_player': 20,
        'next_pipe_top_y': 20,
        'player_vel': 4,
        'player_y': 16
    }
    
    def get_state_idx(self, state):
        # パイプの絶対位置の代わりに相対位置を使用する
        state = copy.deepcopy(state)
        state['next_next_pipe_bottom_y'] -= state['player_y']
        state['next_next_pipe_top_y'] -= state['player_y']
        state['next_pipe_bottom_y'] -= state['player_y']
        state['next_pipe_top_y'] -= state['player_y']

        # アルファベット順に並び替える
        state_key = [k for k, v in sorted(state.items())]

        # 相対位置を返す
        state_idx = []
        for key in state_key:
            state_idx.append(int(state[key] / self.bucket_range_per_feature[key]))
        return tuple(state_idx)

コンストラクタでは行動の数を引数として、Qテーブルを宣言します。

self.q_table = defaultdict(lambda: np.zeros(num_actions))

ではdafultdictの引数にlambda: np.zeros(num_actions)を取っています。
今回のactionnの数は「ボタンを押してジャンプする」の一つだけなので、num_actionsは1となります。

関数decide_action()ではマルコフ性に従い現在の状態から行動を選択します。
基本的にはQテーブルから行動を決定します。ですが、初期の状態は良いQテーブルができているわけではないので、たまたまQ値が大きくなってしまっている場合もあります。それを避けるためにある確率εでランダムに行動をします(ε-greedy)。 また、εの値はエピソードが増えるにつれて小さくすることで、ランダムに行動を起こす確率を低くします。
\epsilon = 0.5 \times \frac{1}{episode + 1}

関数update_policy()ではQ学習の更新式にしたがってQ値を更新します。($$\eta$$は学習率, \gammaは時間割引率)
[tex:\displaystyle Q{best} = max(Q{前状態})]
[tex:\displaystyle \Q{new-value} += \eta * (reward + \gamma * Q_best - Q{current-value})]

関数get_state_idx()では学習を安定させるために自分(鳥)とパイプとの距離を相対距離とします。
こうすることで状態変数の数を減らすことができ、値の最大値と最小値の差も小さくなります。

Environmentクラス

ゲームをプレイする盤面を表現するEnvironmentクラスを宣言します。

class Environment:
    
    def __init__(self, graph=True):
        self.game = FlappyBird()
        self.env = PLE(self.game, fps=30, display_screen=False)
        self.num_actions = len(self.env.getActionSet()) # 1
        self.agent = Agent(self.num_actions)
        self.graph=graph
    
    def run(self):
        
        from IPython.display import Image, display

        reward_per_epoch = []
        lifetime_per_epoch = []
        PRINT_EVERY_EPISODE = 500
        SHOW_GIF_EVERY_EPISODE = 5000
        NUM_EPISODE = 50000
        for episode in range(0, NUM_EPISODE):
            # 環境のリセット
            self.env.reset_game()
            # record frame
            frames = [self.env.getScreenRGB()]

            # 状態の初期化
            state = self.game.getGameState()
            cum_reward = 0  # このエピソードにおける累積報酬の和
            t = 0

            while not self.env.game_over():

                # 行動の選択
                action = self.agent.get_action(state, episode)

                # 行動を実行し、報酬を得る
                reward = self.env.act(
                        self.env.getActionSet()[action])  
                        # パイプを超えれば、reward +=1 失敗したら reward  -= 5

                frames.append(self.env.getScreenRGB())

                # 累積報酬
                cum_reward += reward

                # 次状態を得る
                state_prime = self.game.getGameState() 

                # Agentの更新
                self.agent.update_Q_function(state, action, reward, state_prime)

                # 次のイテレーションの用意
                state = state_prime
                t += 1
            
            # 500エピソード毎にlogを出力
            if episode % PRINT_EVERY_EPISODE == 0:
                print("Episode %d finished after %f time steps" % (episode, t))
                print("cumulated reward: %f" % cum_reward)
                reward_per_epoch.append(cum_reward)
                lifetime_per_epoch.append(t)
                if len(frames) > GOAL_FRAME:
                    print("len frames:", len(frames))
                    clip = make_anim(frames, fps=60, true_image=True).rotate(-90)
                    display(clip.ipython_display(fps=60, autoplay=1, loop=1))
                    if self.graph == True:
                        make_graph(reward_per_epoch, lifetime_per_epoch)
                    break
                

            # 5000エピソード毎にアニメーションを作成
            if episode % SHOW_GIF_EVERY_EPISODE == 0:
                print("len frames:", len(frames))
                clip = make_anim(frames, fps=60, true_image=True).rotate(-90)
                display(clip.ipython_display(fps=60, autoplay=1, loop=1))

コンストラクタでゲームやAgentのインスタンス化を行います。

関数run()ではメインの学習ループを実行します。
学習グラフを表示するためのlistである、PRINT_EVERY_EPISODEとSHOW_GIF_EVERY_EPISODEを宣言します。 SHOW_GIF_EVERY_EPISODEとNUM_EPISODEはlogやアニメーションを出力するタイミングの設定です。

forループの中では主に環境と状態の初期化、ゲームのプレイを行っています。
ゲームのプレイの簡単な流れとしては、行動を選択->行動から報酬を得る->次状態を受け取り->Agentの更新を行っています。
また、初期設定では500ループ毎に何Stepプレイしたか、累積報酬の2つを出力します。 そのときに目標フレーム数(60fpsなので、20秒なら1200frame)より長いフレーム数プレイしたら学習をストップします。 それと同時に報酬の和とプレイ時間を学習グラフとして出力します。

f:id:tkr1205:20180818234602p:plain
学習グラフ

グラフは500回に一回データを取得しているので、x軸が10の場合5000回目の学習の値です。\n また、lifetimeはframe数なので、y軸÷60をすればプレイ時間(sec)となります。

そして、5000ループ毎にアニメーションを作成します。

実行

特に説明はなし。

flappybird_env = Environment()
flappybird_env.run()

結果

学習0回目

学習10000回目

学習20000回目


Playing Flappy Bird by Q_Learning

コードはGitHub

Jupyter Notebook形式でGitHubにあげているので見てみてね

まとめ・感想

今回OpenAI Gym以外のタスクに初めて挑戦してみました。
最初Q学習でFlappyBirdをプレイするのは難しいと思っていましたが、やってみると案外いけるものだなと思いました。一度20000回まで学習させたときにはメモリが足りなくなってしまい途中で学習が止まっていました。もっとメモリの使用をおさえたり学習の速度を高速にする事が今後の課題です。 強化学習教師あり学習と違って複雑なものが多いように思います。ですが、その分とても勉強していて楽しいです。 まだまだ強化学習も勉強し始めたばかりなので、色んな本屋サイトを参考にしましたがとても良い勉強になったと思います。今度はDQNで同じFlappyBirdのタスクに挑戦してみようかと考えています。

SNS

気軽にフォローしてください。泣いて喜びます。

Twitter:@tkr12051

インターン探しています

現在リモートで働かせていただけるインターン先を探しております。
機械学習分野について勉強中です。
しっかりと扱える言語はPythonのみですが、タスクを与えられれば自分で調べながら完成まで頑張ることをモットーにしています。 以前はインターンにてwebの知識ゼロの状態からFlaskを使ってwebAPIの作成を行いました。
もし興味を持っていただけましたらご連絡ください。

参考文献

つくりながら学ぶ! 深層強化学習 ~PyTorchによる実践プログラミング~
OpenAI Gym
PLE
Q学習
強化学習のキホン
PythonのDefaultDictについて
PLEのインストール
FlappyBirdの動かしかたとQ学習の参考