kaggleで強化学習をやってみた - 機械学習 Memo φ(・ω・ )

概要

現在、kaggle に Connect X という強化学習の Getting Started コンペがあります。このコンペを通じて強化学習を少し勉強したので、その内容を記載したいと思います。

こちらの書籍をもとに強化学習について理解したことと、Connect Xコンペでの実装を解説した記事になります。間違いがあれば、コメントいただけたら嬉しいです。

bookclub.kodansha.co.jp

強化学習とは

強化学習とは、行動から報酬が得られる環境において、各状況で報酬に繋がるような行動を出力するように、モデルを作成すること。

教師あり学習との違いは連続した行動によって得られる報酬を最大化させるという点です。囲碁を考えた時、ある局面で悪手に見えた一手が、先々進めると実は良い手だった、といった場合のその一手を選択できるようにするのが強化学習になります。

Connect X と強化学習

いわゆる四目並べゲームです。対戦相手より先に、自分のピースを縦・横・斜めのいずれかで、４つ揃えられたら勝ちになります。

f:id:YukoIshizaki:20200401191514p:plain:w250
Connect X | Kaggle

提出するファイルは通常のようなcsvファイルではなく、エージェントの振る舞いが記載されているPythonファイルを提出します。

Connect X のルールをふまえ、強化学習での考えを整理します。

エージェント

四目並べを行うプレーヤー

行動 Action

ピースを入れること
$A = \{a_1, a_2, a_3, ...\}$
ConnectXでは、ピースは「チェッカー」、列を選ぶことを「ドロップ」と表現。

状態 State

ゲームボード上のチェッカーの配置。
$S = \{s_1, s_2, s_3, ...\}$

(以降の記載では、 $s$ が現在の状態、 $s'$ が次のSTEPの状態と表している)

報酬 Reward

ゲーム終了時に勝つと 1 が、負けると 0 が、どちらでもない場合 (引き分け・勝負がついていない) だと 0.5 が報酬として得られます。

行動後すぐに得られる報酬を即時報酬と呼びます。

また、時間割引された報酬の総和を以下のように表します。

$G_t:= r_{t+1} + \gamma r_{t+2} + \gamma r^2_{t+3} + \cdots + \gamma r^{T-t-1}_T= \displaystyle \sum_{k=0}^{T-t-1} \gamma^kr_{t+k+1}$
$t$ は時間 (手/ステップ)、 $\gamma$ $(\in [0, 1] )$ が時間割引率

10手で勝利した場合と、 20手で勝利した場合では、前者の方がより良いものと評価したいため。

これは再帰的に表すことが可能。
$G_t:= r_{t+1}+\gamma G_{t+1}$

報酬関数 Reward Function

報酬を返す関数。
$R(s, s')$

遷移関数 Transition Function

現在の状態と行動から、ある状態になる確率と、遷移先を返す関数。
$T(s, a)$

遷移関数が状態遷移確率 $P_a(s, s')$ を出力し、遷移先は状態遷移確率の高いものとなる。
Connect X では、ゲーム上選択可能なActionをした場合、必ず想定通りの状態に遷移するので考慮しないものとします。

戦略 Policy

ある状態 $s$ で次の行動 $a$ を決める関数。
$\pi(a|s)$

遷移関数と似ていますが、Policyは実際に起こす行動を決めるもので、その行動を起こすとどのような状態になるのかを定めているのが遷移関数です。

強化学習の種類

モデルベース

遷移関数と報酬関数をベースに学習することをモデルベースといいます。
ある状態 $s$ で戦略 $\pi(a|s)$ に基づいて行動することで得られる価値 $V_{\pi}(s)$ を、以下のように表すことができます。

$\begin{eqnarray} V_{\pi}(S_t) &=& E_{\pi}[r_{t+1} + \gamma V_{\pi}(S_{t+1})] \\ &=& \displaystyle \sum_{a}{\pi(a|s)} \displaystyle \sum_{s'}{T(s'|s,a)(R(s, s')+ \gamma V_{\pi}(s'))} \end{eqnarray}$

期待値 $E$ は、行動確率 (戦略から決まる) と遷移確率をかけることで導き出すことができます。

価値が最大になるような行動を常に選択する方法を Value ベースといい、行動の評価方法のみを学習します。それとは別に、戦略によって行動を決定し、その戦略の評価と更新に行動評価を使う方法を Policy ベースといいます。

上記の式において、次のSTEPにおける価値 $V(s')$ が計算済みでないといけないわけですが、全ての行動に対する価値を計算するのはパターンが多い場合は容易ではないため、動的計画法 DP が用いられます。

モデルベースではエージェントが一歩も動くことなく、環境の情報のみで最適な計画 (戦略) を導くことができます。ただし、これは遷移関数と報酬関数が既知 (もしくは推定が可能) である必要があります。そのため、一般的にはモデルベースではなくモデルフリーが使われます。今回の Connect X でもモデルフリーでのアプローチになるため、モデルベースの詳細については割愛します。

モデルフリー

エージェントが自ら動き、その経験を使って学習することをモデルフリーといいます。

経験とは、見積もっていた価値 $V(s)$ と、実際に行動してみた時の価値 $r + \gamma V(s')$ の差分のことです。

代表的なものに、モンテカルロ法とTD法があります。TD法は1STEP進んだら、誤差 (TD誤差) を小さくする更新を行い、モンテカルロ法はエピソード終了までSTEPを進めてから、誤差を小さくする更新を行います。

TD法の $V$ の更新の仕方
$V(s_t) \leftarrow V(s_t) + \alpha(r_{t+1} + \gamma V(s_{t+1}) - V(s_t))$

モンテカルロ法の $V$ の更新の仕方
$V(s_t) \leftarrow V(s_t) + \alpha ( ( r_{t+1} + \gamma r_{t+2} + \gamma^{2} r_{t+3} + \cdots + \gamma^{T-t-1} r_{T-t})- V(s_t))$

TD法の代表的なものにQ-learningがあります。ある状態におけるある行動をすることの価値を $Q(s, a)$ と表しQ値と言います。Q-learningは戦略を使用せずに、価値が最大となる状態に遷移する行動をとり、価値評価を更新するため Off-Policy (戦略がない)と言います。これに対し、SARSAという方法は行動の決定が戦略に基づくものであり、戦略を更新するため、On-Policy と言います。戦略をActorが担当し、価値評価をCriticが担当して交互に更新を行うActor Critic法というものもあります。

Connect X

強化学習について大まかに理解したところで、Connect X の環境を触ってみたいと思います。

インストール

ConnectX コンペの環境が使えるよう、以下のライブラリをインストールします。

>> pip install kaggle-environments

ライブラリの使い方

make でゲーム環境のインスタンを生成し、render でゲームボードの状態を表示することができます。

from kaggle_environments import make, utils

env = make("connectx", debug=True)
env.render()

f:id:YukoIshizaki:20200404024256p:plain:w250

configuration に、ゲームの構成情報があります。列が 7 で行が 6 のボードでチェッカーを 4 つ揃えたら良いことがわかります。

print(env.configuration)

>> {'timeout': 5, 'columns': 7, 'rows': 6, 'inarow': 4, 'steps': 1000}

エピソードが終了すると、done が True を返します。
対戦相手をランダムとして、トレーナーを作成し、ゲームを初期化 (リセット) し、毎回 0 列目にドロップしてみます。

trainer = env.train([None, "random"])
state = trainer.reset()
print(f"board: {state.board}\n"\
      f"mark: {state.mark}")

while not env.done:
    state, reward, done, info = trainer.step(0)
    print(f"reward: {reward}, done: {done}, info: {info}")
    board = state.board
env.render(mode="ipython", width=350, height=300)

>> board: [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
>> mark: 1
>> reward: 0.5, done: False, info: {}
>> reward: 0.5, done: False, info: {}
>> reward: 0.5, done: False, info: {}
>> reward: 1, done: True, info: {}

state.board には、ボード上の配置がシリアル化された配列が得られます
state.mark で自分のチェッカーが 1 か 2 か判別できます
trainer.step() に自分がドロップする列を渡すと、相手もドロップした後の state とreward 、ゲームの終了判定フラグが得られます
すでに6つチェッカーが配置されている列にドロップすると、Invalid Action となり reward Nan でゲーム終了となります
renderのmode を ipython にすると jupyter notebook 上でプレイ動画の再生ができます

f:id:YukoIshizaki:20200404120829g:plain:w300

評価指標

ガウス分布 $N(\mu, \sigma^2)$ でモデル化され、 $\mu$ の値がスキル評価としてLBに反映されています。サブミットすると、 $\mu$ は 600で初期化されて、全エージェントのプールに入れられます。各エージェントは 1 日最大 8 エピソード分、自分の評価と近しいものと対戦を行います。その対戦で負けると $\mu$ の値が小さくなり、勝つと $\mu$ の値が大きくなり、引き分けだと両者の平均となります。値の更新は、それぞれの偏差を考慮した値になり $\sigma$ も更新されます。また、新しいエージェントの場合は、レートを少し上げて出来るだけ早く、適切な値になるように調整しているそうです。

新たなエージェントを作成したとき、サブミット前に現在のLBのおける $\mu$ 値の計算をするのは難しいですが、いずれにせよ、強いエージェントは徐々に LB を登っていき、負け続けると下がっていくようになっています。

エージェントの作成

Connect X コンペでは、エージェントの振る舞いが記載された Python ファイルを提出する必要があるので、エージェントを作成して提出してみます。

一番上が 0 (空) である列の中から、ランダムに 1 つ選ぶだけのエージェントを作成します。

from random import choice
def my_agent(state, configuration):
    return choice([c for c in range(configuration.columns) if state.board[c] == 0])

evaluate に、ゲーム名とエージェントとエピソード数を渡すと、対戦結果が得られます。
以下の出力だと 2 勝 1 敗です。

from kaggle_environments import evaluate
print(evaluate("connectx", [my_agent, "random"], num_episodes=3))

>> [[1, 0], [0, 1], [1, 0]]

submission.py ファイルに my_agent を出力します。

import inspect
import os

def write_agent_to_file(function, file):
    with open(file, "a" if os.path.exists(file) else "w") as f:
        f.write(inspect.getsource(function))

write_agent_to_file(my_agent, "submission.py")

これは提出ファイルのエージェントが正常に動作するかの確認コードです。サブミットする前に、確認しておきます。

import sys
out = sys.stdout
submission = utils.read_file("{提出ファイルPath}")
agent = utils.get_last_callable(submission)
sys.stdout = out

env = make("connectx", debug=True)
env.run([agent, agent])
print("Success" if env.state[0].status == env.state[1].status == "DONE" else "Failed")

>> Success

ファイルが出力されたら、いつもと同じようにファイルをアップロードします。
通常と同じく、kernelから提出することも、APIで提出することもできます。

f:id:YukoIshizaki:20200404142714p:plain:w700

LB上のディスプレイアイコンをクリックすると、LB上での対戦動画がみれます！このような他のコンペとは違うところは、面白いですね。

Q-Learning の実装

ある状態である行動を行うことの価値をQ値 $Q(s, a)$ と表し、そのQ値を学習する方法である、Q-Learning を Connect X に用に実装してみます。

Qテーブル

Q値を格納しておくQテーブルの実装

Q : Qテーブルをdictで、keyに状態を, valueに全actionのQ値を配列で格納しておく
get_state_key : Qテーブルのkeyである、状態 (自分がどちらのチェッカーかも加味) を state_key (16進数)で表す
get_q_values : ある状態での全actionのQ値を配列 (0 ~ 6: ドロップする列順) で返す関数
update : ある状態におけるあるアクションに対して更新をかける

class QTable():
    def __init__(self, actions):
        self.Q = {} # Qテーブル
        self.actions = actions
    
    def get_state_key(self, state):
        # 16進数で状態のkeyを作る
        board = state.board[:]
        board.append(state.mark)
        state_key = np.array(board).astype(str) 
        return hex(int(''.join(state_key), 3))[2:]        
        
    def get_q_values(self, state):
        # 状態に対して、全actionのQ値の配列を出力
        state_key = self.get_state_key(state)
        if state_key not in self.Q.keys(): 
            # 過去にその状態になったことがない場合
            self.Q[state_key] = [0] * len(self.actions)
        return self.Q[state_key]
    
    def update(self, state, action, add_q):
        # Q値を更新
        state_key = self.get_state_key(state)
        self.Q[state_key] = [q + add_q if idx == action else q for idx, q in enumerate(self.Q[state_key])]

Agent の実装

policy function : Qテーブルをもとに、ある状態におけるQ値が最大なactionを選択する
custom_reward : Qテーブルの作成がよりうまくいくように報酬関数をカスタマイズ
learn : エピソードごとにQテーブルを更新して学習させる
q_table : 状態 x 行動に対して、価値を格納しおく Q テーブル
reward_log : 報酬の履歴

パラメータ

episode_cnt : 学習に使うエピソード数
epsilon : 探索を行う(Q値に従わない)ようにする確率, はじめは大きくて徐々に小さくなるように実装
gamma : 時間割引率
learn_rate : 学習率

env = make("connectx", debug=True)
trainer = env.train([None, "random"])

class QLearningAgent():
    def __init__(self, env, epsilon=0.99):
        self.env = env
        self.actions = list(range(self.env.configuration.columns))
        self.q_table = QTable(self.actions)
        self.epsilon = epsilon
        self.reward_log = []    
        
    def policy(self, state):
        if np.random.random() < self.epsilon:
            # epsilonの割合で、ランダムにactionを選択する
            return choice([c for c in range(len(self.actions)) if state.board[c] == 0])
        else:
            # ゲーム上選択可能で、Q値が最大なactionを選択する
            q_values = self.q_table.get_q_values(state)
            selected_items = [q if state.board[idx] == 0 else -1e7 for idx, q in enumerate(q_values)]
            return int(np.argmax(selected_items))
        
    def custom_reward(self, reward, done):
        if done:
            if reward == 1: # 勝ち
                return 20
            elif reward == 0: # 負け
                return -20
            else: # 引き分け
                return 10
        else:
            return -0.05 # 勝負がついてない
        
    def learn(self, trainer, episode_cnt=10000, gamma=0.6, 
              learn_rate=0.3, epsilon_decay_rate=0.9999, min_epsilon=0.1):
        for episode in tqdm(range(episode_cnt)):
            # ゲーム環境リセット
            state = trainer.reset() 
            # epsilonを徐々に小さくする
            self.epsilon = max(min_epsilon, self.epsilon * epsilon_decay_rate) 
            while not env.done:
                # どの列にドロップするか決めるて実行する
                action = self.policy(state) 
                next_state, reward, done, info = trainer.step(action)
                reward = self.custom_reward(reward, done)
                # 誤差を計算してQテーブルを更新する
                gain = reward + gamma * max(self.q_table.get_q_values(next_state))
                estimate = self.q_table.get_q_values(state)[action]
                self.q_table.update(state, action, learn_rate * (gain - estimate)) 
                state = next_state
      
            self.reward_log.append(reward)

結果

# 学習
qa = QLearningAgent(env)
qa.learn(trainer)

# ゲーム終了時に得られた報酬の移動平均
import seaborn as sns
sns.set(style='darkgrid')
pd.DataFrame({'Average Reward': qa.reward_log}).rolling(500).mean().plot(figsize=(10,5))
plt.show()

更新された q_table に学習で得られた Q 値が、 reward_log に報酬の履歴 (勝敗) が得られます。
報酬の移動平均をみると、徐々に勝率が上がっているのが確認できます。ちゃんと学習できているようです！

f:id:YukoIshizaki:20200405200836p:plain:w500

Pythonファイルへの出力

また、エージェントの振る舞いをする1つの関数としてPythonファイルへ出力するため、Qテーブルのデータを文字列に変換し、以下のコードでPythonファイルに書き込む際にdictとして扱えるようにして出力します。

tmp_dict_q_table = qa.q_table.Q.copy()
dict_q_table = dict()

# 学習したQテーブルで、一番Q値の大きいActionに置き換える
for k in tmp_dict_q_table:
    if np.count_nonzero(tmp_dict_q_table[k]) > 0:
        dict_q_table[k] = int(np.argmax(tmp_dict_q_table[k]))

my_agent = '''def my_agent(observation, configuration):
    from random import choice
    # 作成したテーブルを文字列に変換して、Pythonファイル上でdictとして扱えるようにする
    q_table = ''' \
    + str(dict_q_table).replace(' ', '') \
    + '''
    board = observation.board[:]
    board.append(observation.mark)
    state_key = list(map(str, board))
    state_key = hex(int(''.join(state_key), 3))[2:]
    # Qテーブルに存在しない状態の場合
    if state_key not in q_table.keys():
        return choice([c for c in range(configuration.columns) if observation.board[c] == 0])
    # Qテーブルから最大のQ値をとるActionを選択
    action = q_table[state_key]
    # 選んだActionが、ゲーム上選べない場合
    if observation.board[action] != 0:
        return choice([c for c in range(configuration.columns) if observation.board[c] == 0])
    return action
    '''

with open('submission.py', 'w') as f:
    f.write(my_agent)

Qテーブルの作り方・ファイル出力の仕方はこちらのkernelを参考にしました.
ConnectX with Q-Learning | Kaggle

Deep Q-Net の実装

強化学習にディープラーニングを使った代表的なDeep Q-Netについて、Connect X 用に実装してみます。
基本的な考え方はQ-learningと同じで、Qテーブルで行なっていた価値の評価に、CNNを用います。
inputは状態 $s$ で、outputはactionの価値で、Loss関数でTD誤差を最小化するするように実装します。

また、うまく学習を行うための 3 つのテクニックがあります。

Experience Replay

エージェントの行動履歴を貯めておき、そこからサンプリングして学習に利用します。行動履歴とは [ 状態, 行動, 報酬, 遷移先の状態, エピソードの終了フラグ ] のまとまりになります。さまざまなエピソードの異なるタイミングのデータが使えることで、学習を安定させることができます。

Fixed Target Q-Network

遷移先の価値を計算する場合、現在の更新しているモデル(CNN)と同じものを使用すると重みを更新するたびに違った値になってしまい、TD誤差が安定しないものになってしまいます。一定期間、更新していないCNNモデルから遷移先の価値を計算し、あるタイミングで更新をかける、といった方法をとります。価値の評価のために更新し続けているCNNと遷移先の価値計算用のCNN、2 つを使って学習します。

Clipping

報酬を、成功が 1 , 失敗が -1 , それ以外は 0 に統一します。

CNN の実装

価値評価を行うためのCNNを実装します。上記、Fixed Target Q-Network を使うため、価値評価用のCNNと遷移先価値計算用のCNN、両方このCNNを使います。

今回は、四目並べという小さいゲームボードなので、ネットワーク構成を畳み込み2層の小さいCNNにしてみました。input は状態のゲームボードのチェッカーの配置を2次元 (7, 6) でそのまま入れてます。output は action の value (7) です。

import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F

class CNN(nn.Module):
    def __init__(self, outputs=7):
        super(CNN, self).__init__()
        self.conv1 = nn.Conv2d(1, 16, 3)
        self.bn1 = nn.BatchNorm2d(16)
        self.conv2 = nn.Conv2d(16, 32, 3)
        self.bn2 = nn.BatchNorm2d(32)
        self.fc = nn.Linear(192, 32)
        self.head = nn.Linear(32, outputs)

    def forward(self, x):
        x = F.relu(self.bn1(self.conv1(x)))
        x = F.relu(self.bn2(self.conv2(x)))
        x = x.view(x.size()[0], -1)
        x = self.fc(x)
        x = self.head(x)
        return x

Deep Q-Net の Agent の実装

エージェントの実装をします。Q-lerningでの実装の違いは、以下の 4 点です。

見積もり価値と、実際に行動価値の誤差(TD誤差)を最小化するところをCNNにする
CNNに入れられるように、チェッカーの配置を (1, 7, 6) の Tensorに変換するところと
自分のチェッカーを 1 、相手のチェッカーを 0.5 にしたこと
上記のテクニック Experience Replay, Fixed Target Q-Network, Clipping を使用すること

class DeepQNetworkAgent():
    def __init__(self, env, lr=1e-2, min_experiences=100, max_experiences=10_000, channel=1):
        self.env = env
        self.model = CNN() # 価値評価用のCNN
        self.teacher_model = CNN() # 遷移先価値評価用のCNN
        self.optimizer = optim.Adam(self.model.parameters(), lr=lr)
        self.criterion = nn.MSELoss()
        self.experience = {'s': [], 'a': [], 'r': [], 'n_s': [], 'done': []} # 行動履歴
        self.min_experiences = min_experiences
        self.max_experiences = max_experiences
        self.actions = list(range(self.env.configuration.columns))
        self.col_num = self.env.configuration.columns
        self.row_num = self.env.configuration.rows
        self.channel = channel
        
    def add_experience(self, exp):
        # 行動履歴の更新
        if len(self.experience['s']) >= self.max_experiences:
            # 行動履歴のサイズが大きすぎる時は古いものを削除
            for key in self.experience.keys():
                self.experience[key].pop(0)
        for key, value in exp.items():
            self.experience[key].append(value)
        
        
    def preprocess(self, state):
        # 状態は自分のチェッカーを1, 相手のチェッカーを0.5とした7x6多次元配列で表す
        result = np.array(state.board[:])
        result = result.reshape([self.col_num, self.row_num])
        if state.mark == 1:
            return np.where(result == 2, 0.5, result)
        else:
            result = np.where(result == 2, 1, result)
            return np.where(result == 1, 0.5, result)
    
    def estimate(self, state):
        # 価値の計算
        return self.model(
            torch.from_numpy(state).view(-1, self.channel, self.col_num, self.row_num).float()
        ) 
    
    def future(self, state):
        # 遷移先の価値の計算
        return self.teacher_model(
            torch.from_numpy(state).view(-1, self.channel, self.col_num, self.row_num).float()
        )
    
    def policy(self, state, epsilon):
        # 状態から、CNNの出力に基づき、次の行動を選択
        if np.random.random() < epsilon:
            # 探索
            return int(np.random.choice([c for c in range(len(self.actions)) if state.board[c] == 0]))
        else:
            # Actionの価値を取得
            prediction = self.estimate(self.preprocess(state))[0].detach().numpy()
            for i in range(len(self.actions)):
                # ゲーム上選択可能なactionに絞る
                if state.board[i] != 0:
                    prediction[i] = -1e7
            return int(np.argmax(prediction))
    
    def update(self, gamma):
        # 行動履歴が十分に蓄積されているか
        if len(self.experience['s']) < self.min_experiences:
            return 
        
        # 行動履歴から学習用のデータのidをサンプリングする
        ids = np.random.randint(low=0, high=len(self.experience['s']), size=32)
        states = np.asarray([self.preprocess(self.experience['s'][i]) for i in ids])
        states_next = np.asarray([self.preprocess(self.experience['n_s'][i]) for i in ids])
        
        # 価値の計算
        estimateds = self.estimate(states).detach().numpy() # 見積もりの価値
        future = self.future(states_next).detach().numpy() # 遷移先の価値
        target = estimateds.copy()
        for idx, i in enumerate(ids):
            a = self.experience['a'][i]
            r = self.experience['r'][i]
            d = self.experience['done'][i]
            reward = r
            if not d:
                reward += gamma * np.max(future[idx])
        # TD誤差を小さくするようにCNNを更新
        self.optimizer.zero_grad()
        loss = self.criterion(torch.tensor(estimateds, requires_grad=True),
                              torch.tensor(target, requires_grad=True))
        loss.backward()
        self.optimizer.step()
        
    def update_teacher(self):
        # 遷移先の価値の更新
        self.teacher_model.load_state_dict(self.model.state_dict())

Deep Q-Net の Trainer の実装

基本的に、Q-learning と変わりません。
行動履歴をためていく処理と、一定の間隔で価値評価用のCNNのパラメータを遷移先価値計算用のCNNにコピーしている処理が追加されています。

class DeepQNetworkTrainer():
    def __init__(self, env):
        self.epsilon = 0.9
        self.env = env
        self.agent = DeepQNetworkAgent(env)
        self.reward_log = []
    
    def custom_reward(self, reward, done):
        # Clipping
        if done:
            if reward == 1: # 勝ち
                return 1
            elif reward == 0: # 負け
                return -1
            else: # 引き分け
                return 0
        else:
            return 0 # 勝負がついてない
        
    def train(self, trainer,epsilon_decay_rate=0.9999, min_epsilon=0.1, episode_cnt=100, gamma=0.6):
        iter = 0
        for episode in tqdm(range(episode_cnt)):
            rewards = []
            state = trainer.reset() # ゲーム環境リセット
            self.epsilon = max(min_epsilon, self.epsilon * epsilon_decay_rate) # epsilonを徐々に小さくする
            while not env.done: 
                # どの列にドロップするか決める
                action = self.agent.policy(state, self.epsilon)
                prev_state = state
                state, reward, done, _ = trainer.step(action) 
                reward = self.custom_reward(reward, done)
                # 行動履歴の蓄積
                exp = {'s': prev_state, 'a': action, 'r': reward, 'n_s': state, 'done': done}
                self.agent.add_experience(exp)
                # 価値評価の更新
                self.agent.update(gamma)
                iter += 1
                if iter % 100 == 0:
                    # 遷移先価値計算用の更新
                    self.agent.update_teacher()
            self.reward_log.append(reward)

結果

実際に Deep Q-Net Agentで学習してみます。

dq = DeepQNetworkTrainer(env)
dq.train(trainer, episode_cnt=30000)

# 結果の描画
import seaborn as sns
sns.set()

sns.set_palette("winter", 8)
sns.set_context({"lines.linewidth": 1})
pd.DataFrame({'Average Reward': dq.reward_log}).rolling(300).mean().plot(figsize=(10,5))

f:id:YukoIshizaki:20200405195449p:plain:w500

報酬の履歴から勝敗の移動平均をみてみると、徐々に勝てるようになっていて、うまく学習できていそうです。(さきほどのQ-learningとは報酬関数が異なるので、y軸のスケールが異なります)

今回、20,000エピソード学習させましたが、他の方のkernelを見ると3000エピソードぐらいでうまく学習させられている人もいるので、CNNやパラメータを調整して上手く早く学習できるように工夫した方が良いのかもしれません。

おわり

強化学習初心者の勉強の場として、kaggle の Connect X は最適だと思いました！kaggle の notebook を立ち上げればすぐにエージェントを動かせる環境が整うのはとても便利です。学習済みエージェントをどう記載するかという悩ましい問題はあるのですが(外部ファイルの読み込み、学習したモデルの読み込みができない)、Getting Started コンペなので、気軽に参加できて楽しかったです。

Connect X の実装がメインになり、強化学習の理論についてはまだ勉強不足なので、引き続き学んでいきたいです。

勉強会のお知らせ

Wantedly では毎週木曜日18:30から機械学習の勉強会を開いていますが、現在、社員が原則リモートワークのためオンライン (hangouts) で開催しています！オンラインだからこそ参加しやすいかと思いますので、興味がある方は、是非！

github.com

また、カジュアル面談 (現在オンライン)・インターンも募集しています！
www.wantedly.com

www.wantedly.com

概要