深層強化学習の簡単な例〜Double DQN適用〜 - 化学系エンジニアがAIを学ぶ

はじめに

以前の記事で簡単な問題を対象にDQN、Experience Replayを適用してみたが、今回はさらにDouble DQNを追加してみる。なお下記サイトを参考にした。

Let’s make a DQN: Double Learning and Prioritized Experience Replay – ヤロミル

https://www.renom.jp/ja/notebooks/product/renom_rl/ddqn/notebook.html

Double DQN

f:id:schemer1341:20190911214824p:plain:w500

ポイントは $\gamma {\rm max}Q(s_{t+1}, a)$ で、次の状態の最大のQ値を求めるのに自分自身のネットワークを使用している点である。もし仮にQ値の初期値が変に偏っていて、ある行動のQ値が他に比べ極端に高い場合、同じネットワークで更新を繰り返すと、その行動の価値が過大に評価されてしまうことになる。

これを緩和するために提案された方法がDouble DQNである。この方法では、メインのネットワークとは別のネットワーク（Target network)を作り、ネットワーク更新時に次の状態における行動はメインネットワークで決めるが、その行動の価値評価は別のネットワークで行う。なお別のネットワークには過去のメインネットワークを用いる。この方法を用いる場合のネットワーク更新式は次の形となる。

f:id:schemer1341:20190911011958p:plain:w250

f:id:schemer1341:20190911012155p:plain:w570

対象とする問題

前回記事と同一の報酬払出装置を考える。前回のプログラムを修正してDouble DQNを適用する。

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

Double DQN の適用

次の通り、メインネットワークに加え、メインとは別のネットワーク(Target network)を準備する。

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

class DQN(nn.Module):
    def __init__(self):
        super(DQN, self).__init__()
        self.l1 = nn.Linear(1, 3)
        self.l2 = nn.Linear(3, 3)
        self.l3 = nn.Linear(3, 2)

    def forward(self, x):
        x = F.relu(self.l1(x))
        x = F.relu(self.l2(x))
        x = self.l3(x)
        return x

dqn = DQN()   # メインネットワーク
dqn_t = DQN() # もう１つのネットワーク(Target network)
optimizer = torch.optim.SGD(dqn.parameters(), lr=0.01)
criterion = nn.MSELoss()

Double DQNを適用する場合のネットワーク更新は下記の形となる。通常のDQNと異なるのは下記「Q値の算出」にあるa_m、max_q_nextを求める部分だけである。（なおここではExperience Replayを利用している。）

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値
    a_m = dqn(next_state).max(-1)[1].unsqueeze(1) # メインネットワークで決めた次のaction,[1]でindexを得る
    max_q_next = dqn_t(next_state).gather(-1, a_m) # Target networkから上で決めたactionの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の確認のために返す

学習時に所定の計算ステップごとにTarget networkにメインネットワークをコピーして更新される形とする（下記コードの一番下）。更新されるまでは学習時に過去のメインネットワークを参照する形となる。

NUM_EPISODES = 1200
NUM_STEPS = 5

log = [] # 結果のプロット用

for episode in range(NUM_EPISODES):
    env = Dispenser(0)
    total_reward = 0 # 1エピソードでの報酬の合計を保持する

    for s in range(NUM_STEPS):
        ## 現在の状態を確認
        state = env.state
        ## 行動を決める
        action = decide_action(state, episode)
        ## 決めた行動に従いステップを進める。次の状態、報酬を得る
        next_state, reward = env.step(action)
        ## 結果をReplayMemoryに記録
        memory.push(state, action, next_state, reward)
        ## DQNを更新
        loss = update_dqn(memory)
        total_reward += reward

    log.append([total_reward, loss])
    if (episode+1)%100==0: print(episode+1, loss)
    ## 2エピソード = 計算10ステップ 毎の頻度でTarget networkにメインネットワークをコピー
    if episode%2==0:
        dqn_t.load_state_dict(dqn.state_dict())