keras 使って DQN で迷路を解いてみた

世界観をつかめるぐらいには機械学習やっておきたいと思い、とりあえず何かしらのお題がないと興味が続かなさそうなので、二次元の盤面上で何かしらの行動をする、ローグライクのモンスターのエージェントを作るのを目標にしようと思う。自分がゲーム作るとき、大抵エージェントのルール作る段階で飽きてくるので。

今回の記事は、迷路を解くところまで。

学習資料

雰囲気を掴むのに、ニコ動の解説動画わかりやすかった。

よく使われてる OpenAI Gym 、見た目は派手だが、環境変数が多すぎていまいち理解の助けにならない + 次元が多すぎて収束が遠いので、すごい単純なゲームルールを自分で作って、それを自分で解く形式を探した。

その結果、迷路を解かせるこのコードがわかりやすかったので、これをリファクタしながら勉強した。

https://github.com/shibuiwilliam/maze_solver

こっちは自分のリファクタ後のコード。迷路生成部分を含めて268行。Solver の定義 + 訓練は 150行ぐらい。

maze_dqn_solver.py

keras tensorflow numpy が入ってれば動くはず。元コードは pandas 使ってたが、リファクタしてたら不要になった。

勉強過程で、コピペでポンと動くやつがなかなかなくて困ったので、シングルファイルでポンと動くことを意識してる。環境構築は前回の記事参照。

mizchi.hatenablog.com

ゲームルール

10x10のランダム生成の迷路を S から G までゴールできれば50点
踏むと -1点されるマスが散らばってる
一番高いスコアでゴールできたモデルの勝ち

こんな感じ(@@ はエージェントの位置)

.
      #   #   #   S   #   #   #   #   #   #
      #  -1   #   0  -1   #   0   0  -1   #
      #  -1   #  -1   0   0  -1  -1  -1   #
      #   0  -1   #   #   0  -1  -1  -1   #
      #   0   0   0   0   0   0   #   0   #
      #   0  -1  -1   0   0  -1  -1   #   #
      #  -1  -1   0   0   #  -1  -1   0   #
      #  -1   #  -1   0  -1  -1  -1  -1   #
      #   0   0   0  -1  -1  @@  -1  -1   #
      #   #   #   #   #   #  50   #   #   #

環境はいわゆるゲームって感じの実装で、機械学習は全然関係ない。

DQN Solver の考え方

点p(x1, y1) から点p(x2, y2) へ遷移する際のスコアを表にする(マルコフ決定過程)

[p(0, 1) => p(0, 2)] => score(p(0, 1), p(0, 2))
[p(0, 1) => p(1, 1)] => score(p(0, 1), p(1, 1))
[p(1, 1) => p(1, 2)] => score(p(1, 1), p(1, 2))
...

score の評価関数は、そのときのスコアと、さらにその次で取りうる一歩の中うち、最大のスコアになるものにγを掛けたものを足す。(今回は0.9)

その部分のコード (target_f がそれ)

            if done:
                target_f = reward
            else:
                next_rewards = []
                for a in next_movables:
                    np_next_s_a = np.array([[next_state, a]])
                    next_rewards.append(self.model.predict(np_next_s_a))
                np_n_r_max = np.amax(np.array(next_rewards))
                target_f = reward + GAMMA * np_n_r_max

次の一歩を先読みしてるスコアになっているので、何度も繰り返すうちに、最終的にプラスになるものが各セルの評価スコアとして伝搬されるはず。次の一歩の評価を含んでいるので、中間状態でマイナスになっていてもそれも打ち消される。

学習ステップ

とりあえず最初はゴールするまでランダムに歩かせる
行動履歴の中から、ランダムで(最大)32個の状態を取り出して、p(x1, y1) => p(x2, y2) => reward のスコアの組を model.fit (訓練) する
ε-greedy で訓練された予測スコアに少しずつ従うようになる

訓練部分

    def replay_experience(self, batch_size):
        batch_size = min(batch_size, len(self.memory))
        minibatch = random.sample(self.memory, batch_size)
        x = []
        y = []
        for i in range(batch_size):
            state, action, reward, next_state, next_movables, done = minibatch[
                i]
            input_action = [state, action]
            if done:
                target_f = reward
            else:
                next_rewards = []
                for a in next_movables:
                    np_next_s_a = np.array([[next_state, a]])
                    next_rewards.append(self.model.predict(np_next_s_a))
                np_n_r_max = np.amax(np.array(next_rewards))
                target_f = reward + GAMMA * np_n_r_max
            x.append(input_action)
            y.append(target_f)

        self.model.fit(np.array(x), np.array([y]).T, epochs=1, verbose=0)

行動選択部分

    def choose_action(self, at, movables):
        if self.epsilon >= random.random():
            return random.choice(movables)
        else:
            return self.choose_best_action(at, movables)

    def choose_best_action(self, at, movables):
        best_actions = []
        max_act_value = -100
        for a in movables:
            np_action = np.array([[at, a]])
            act_value = self.model.predict(np_action)
            if act_value > max_act_value:
                best_actions = [
                    a,
                ]
                max_act_value = act_value
            elif act_value == max_act_value:
                best_actions.append(a)
        return random.choice(best_actions)

訓練されるモデルの定義

def build_maze_solver():
    model = Sequential()
    model.add(Dense(128, input_shape=(2, 2), activation='tanh'))
    model.add(Flatten())
    model.add(Dense(128, activation='tanh'))
    model.add(Dense(128, activation='tanh'))
    model.add(Dense(1, activation='linear'))
    model.compile(loss="mse", optimizer=RMSprop(lr=LEARNING_RATE))
    return model

最初の層の input_shape が (2, 2) なのは (x1, y1), (x2, y2) のような形になるため。最後の activation する層が、予測される reward の出力。中間層は二層。なんで二層かはよくわかってない。たしかにこれで収束する。

10000回ゴールするまで実行する。(たぶん、収束するまで10000回じゃ足りないのだが、何度もデバッグするのが面倒だったので減らしている)

EPISODE_COUNT = 10000
MAX_WALK_COUNT = 1000

solver = Solver()

for e in range(EPISODE_COUNT):
    at = field.start_point
    score = 0
    for time in range(MAX_WALK_COUNT):
        movables = field.get_actions(at)
        action = solver.choose_action(at, movables)
        reward, done = field.get_value(action)
        score = score + reward
        next_state = action
        next_movables = field.get_actions(next_state)
        solver.remember_memory(at, action, reward, next_state, next_movables,
                               done)
        if done or time == (MAX_WALK_COUNT - 1):
            if e % 500 == 0:
                print("episode: {}/{}, score: {}, e: {:.2} \t @ {}".format(
                    e, EPISODE_COUNT, score, solver.epsilon, time))
            break
        at = next_state

    solver.replay_experience(32)

結果

episode: 500/10000, score: -486.0, e: 0.95    @ 999
episode: 1000/10000, score: -9.0, e: 0.9     @ 126
episode: 1500/10000, score: 20.0, e: 0.86    @ 40
episode: 2000/10000, score: -108.0, e: 0.82      @ 280
episode: 2500/10000, score: 17.0, e: 0.78    @ 54
episode: 3000/10000, score: 25.0, e: 0.74    @ 46
episode: 3500/10000, score: 23.0, e: 0.7     @ 34
episode: 4000/10000, score: 20.0, e: 0.67    @ 50
episode: 4500/10000, score: 31.0, e: 0.64    @ 26
episode: 5000/10000, score: 10.0, e: 0.61    @ 56
episode: 5500/10000, score: 23.0, e: 0.58    @ 48
episode: 6000/10000, score: 35.0, e: 0.55    @ 20
episode: 6500/10000, score: 31.0, e: 0.52    @ 32
episode: 7000/10000, score: 41.0, e: 0.5     @ 16
episode: 7500/10000, score: 38.0, e: 0.47    @ 24
episode: 8000/10000, score: 36.0, e: 0.45    @ 24
episode: 8500/10000, score: 38.0, e: 0.43    @ 24
episode: 9000/10000, score: 37.0, e: 0.41    @ 30
episode: 9500/10000, score: 38.0, e: 0.39    @ 22

score が徐々に上がって、辿り着くまでのステップ数も減っていってる

中間層の数でどのような影響があるか

全部抜いてみた

def build_maze_solver():
    model = Sequential()
    model.add(Dense(128, input_shape=(2, 2), activation='tanh'))
    model.add(Flatten())
    model.add(Dense(1, activation='linear'))
    model.compile(loss="mse", optimizer=RMSprop(lr=LEARNING_RATE))
    return model

結果

episode: 500/10000, score: 7.0, e: 0.95   @ 75
episode: 1000/10000, score: -51.0, e: 0.9    @ 257
episode: 1500/10000, score: -59.0, e: 0.86   @ 349
episode: 2000/10000, score: -343.0, e: 0.82      @ 999
episode: 2500/10000, score: -262.0, e: 0.78      @ 999
episode: 3000/10000, score: -277.0, e: 0.74      @ 999
...

初回は偶然うまくいったが、全く収束しなくて笑った

一つだけ抜いてみた

def build_maze_solver():
    model = Sequential()
    model.add(Dense(128, input_shape=(2, 2), activation='tanh'))
    model.add(Flatten())
    model.add(Dense(128, activation='tanh'))
    model.add(Dense(1, activation='linear'))
    model.compile(loss="mse", optimizer=RMSprop(lr=LEARNING_RATE))
    return model

結果

episode: 500/10000, score: -117.0, e: 0.95    @ 413
episode: 1000/10000, score: 6.0, e: 0.9      @ 83
episode: 1500/10000, score: -54.0, e: 0.86   @ 195
episode: 2000/10000, score: 8.0, e: 0.82     @ 73
episode: 2500/10000, score: 35.0, e: 0.78    @ 39
episode: 3000/10000, score: 31.0, e: 0.74    @ 61
episode: 3500/10000, score: 27.0, e: 0.7     @ 57
episode: 4000/10000, score: 34.0, e: 0.67    @ 31
episode: 4500/10000, score: 32.0, e: 0.64    @ 39
episode: 5000/10000, score: 33.0, e: 0.61    @ 31
episode: 5500/10000, score: 42.0, e: 0.58    @ 27
episode: 6000/10000, score: 40.0, e: 0.55    @ 19
episode: 6500/10000, score: 40.0, e: 0.52    @ 27
episode: 7000/10000, score: 37.0, e: 0.5     @ 27
episode: 7500/10000, score: 39.0, e: 0.47    @ 27
episode: 8000/10000, score: 39.0, e: 0.45    @ 19
episode: 8500/10000, score: 32.0, e: 0.43    @ 35
episode: 9000/10000, score: 42.0, e: 0.41    @ 21
episode: 9500/10000, score: 43.0, e: 0.39    @ 17

収束してる。精度はあまり変わってないような気がする。

中間層を2つ足した

def build_maze_solver():
    model = Sequential()
    model.add(Dense(128, input_shape=(2, 2), activation='tanh'))
    model.add(Flatten())
    model.add(Dense(128, activation='tanh'))
    model.add(Dense(128, activation='tanh'))
    model.add(Dense(128, activation='tanh'))
    model.add(Dense(128, activation='tanh'))
    model.add(Dense(1, activation='linear'))
    model.compile(loss="mse", optimizer=RMSprop(lr=LEARNING_RATE))
    return model

結果

episode: 500/10000, score: -447.0, e: 0.95    @ 965
episode: 1000/10000, score: -24.0, e: 0.9    @ 137
episode: 1500/10000, score: 27.0, e: 0.86    @ 47
episode: 2000/10000, score: 24.0, e: 0.82    @ 45
episode: 2500/10000, score: 19.0, e: 0.78    @ 73
episode: 3000/10000, score: 41.0, e: 0.74    @ 27
episode: 3500/10000, score: 32.0, e: 0.7     @ 27
episode: 4000/10000, score: 41.0, e: 0.67    @ 21
episode: 4500/10000, score: 30.0, e: 0.64    @ 39
episode: 5000/10000, score: 40.0, e: 0.61    @ 23
episode: 5500/10000, score: 37.0, e: 0.58    @ 23
episode: 6000/10000, score: 40.0, e: 0.55    @ 25
episode: 6500/10000, score: 45.0, e: 0.52    @ 15
episode: 7000/10000, score: 40.0, e: 0.5     @ 19
episode: 7500/10000, score: 38.0, e: 0.47    @ 27
episode: 8000/10000, score: 40.0, e: 0.45    @ 27
episode: 8500/10000, score: 39.0, e: 0.43    @ 25
episode: 9000/10000, score: 39.0, e: 0.41    @ 19
episode: 9500/10000, score: 44.0, e: 0.39    @ 21

収束はしてるけど、これ以上増やしても無意味みたいな点がありそう

これ最終的なモデル実行時間が遅くなったりするんだろうか。

ゲーム用のエージェントを作るなら

ダイクストラと A* は何度も書いてるので、今回の題材は自分には理解しやすかった。

今回の迷宮(環境)は固定だが、与えられた任意の迷宮を解きたければ、たぶん、自分自身を中心としてゴール含むN*Nの相対座標系でクリッピングした行列を入力値に含む必要がありそう。そうするとめっちゃ入力値多い…

移動アクションの他、攻撃アクションを追加してみるとか、その際の彼我の HP の変動とかを入力に含んでみるとゲームっぽくなりそう。

どのへんまで環境変数増やすと計算爆発するとか収束しないとか、そのへんの勘所がないので色々試してみないとまだわかんなさそう。DOTA II のエージェントを作った論文があるので、それを読むといいんだろうか。

今の時点では、正直 keras の便利さがよくわかってなくて、直接 tensorflow 使うのとそんなに変わんなさそう。

python 久しぶりに書いた感じだと、スコープのガバガバさでイラッとする場面が多く、 let のような変数宣言がほしい気持ちになりがちだった。