深層強化学習(Deep Q Network, DQN)の簡単な例〜Experience Replay追加〜

はじめに

前回の記事でOpenAI Gymを使わず非常に簡単な問題を対象にDQNを適用してみたが、"Experience Replay"を入れていなかった。今回は前回の問題にExperience Replayを追加してみる。なおこれを実施するにあたり、下記サイトを参考にした。

第15回 CartPole課題で深層強化学習DQNを実装｜Tech Book Zone Manatee

Reinforcement Learning (DQN) Tutorial — PyTorch Tutorials 1.5.1 documentation

Experience Replay

行動とその結果を経験として記録しておき、その過去の経験をサンプリングして学習データとする方法。通常の逐次で学習する方法に比べ、時間的に離れたデータを使用することができ、偏りを抑えた安定した学習ができるとのこと。複数データをまとめてニューラルネットワークにミニバッチ学習させることもできる。

具体的には「現在の状態」、「行動」、「行動後の状態」、「報酬」のデータの組をある一定の数だけメモリに残しておいて、その中から学習に使うデータをいくつかランダムにサンプリングし、これらを１つのバッチとしてニューラルネットワークに学習させる。

対象とする問題

前回記事と同一の報酬払出装置を考える。プログラムは前回のものにいくつか追加・修正してExperience Replayを適用する。

深層強化学習(Deep Q Network, DQN)の簡単な例 - 化学系エンジニアがAIを学ぶ

Experience Replayの実装

下記のpytorch.orgのチュートリアルにあるように、namedtupleを用いて経験を記録するクラスを追加する。

Reinforcement Learning (DQN) Tutorial — PyTorch Tutorials 1.5.1 documentation

ここでは500回分の行動の結果を記録することにしている(数は適当)。500を超えると古い方から順番に上書きされる形となっている。サンプリング時は指定したバッチサイズ数だけのデータが取り出される。

from collections import namedtuple

# 「現在の状態」、「行動」、「行動後の状態」、「報酬」を記録するnamedtuple
Transition = namedtuple('Transition', ('state', 'action', 'next_state', 'reward'))

import random

class ReplayMemory(object):
    def __init__(self, capacity):
        self.capacity = capacity # メモリの許容サイズ
        self.memory = []         # 経験を保存するリスト
        self.index = 0           # 保存した経験リストのindex

    def push(self, state, action, next_state, reward): # 経験をリストに追加(push)する
        if len(self.memory) < self.capacity: # 許容サイズ以下なら追加
            self.memory.append(None)
        self.memory[self.index] = Transition(state, action, next_state, reward)
        self.index = (self.index + 1) % self.capacity # indexを1つ進める。許容サイズ超えたら古い順から上書き

    def sample(self, batch_size): # バッチサイズ分の経験のランダムサンプリング
        return random.sample(self.memory, batch_size) 

    def __len__(self):
        return len(self.memory)

CAPACITY = 500

memory = ReplayMemory(CAPACITY)

DQNの学習

上記の行動を記録したクラスを利用できる形に、DQNの学習部分を修正する。

BATCH_SIZE = 50 # サンプリングするデータ数

def update_dqn(replay_memory): # 行動を記録したクラスを引数とする
    ## メモリがバッチサイズより小さいときはまだ学習しない
    if len(replay_memory) < BATCH_SIZE:
        return

    ## バッチ取得
    transitions = replay_memory.sample(BATCH_SIZE)
    ## (状態、行動、次の状態、報酬)✕バッチサイズ を (状態xバッチサイズ、行動✕バッチサイズ、、、)に変える
    batch = Transition(*zip(*transitions))
    
    ## 各値をtensorに変換
    state = torch.FloatTensor(batch.state).unsqueeze(1)
    action = torch.LongTensor(batch.action).unsqueeze(1)
    next_state = torch.FloatTensor(batch.next_state).unsqueeze(1)
    reward = torch.FloatTensor(batch.reward).unsqueeze(1)

    ## Q値の算出
    q_now = dqn(state).gather(-1, action) # 今の状態のQ値
    max_q_next = dqn(next_state).max(-1)[0].unsqueeze(1).detach() # 状態移行後の最大のQ値
    gamma = 0.9
    q_target = reward + gamma * max_q_next # 目標のQ値

    # DQNパラメータ更新
    optimizer.zero_grad()
    loss = criterion(q_now, q_target) # 今のQ値と目標のQ値で誤差を取る
    loss.backward()
    optimizer.step()
    
    return loss.item() # lossの確認のために返す