深層強化学習の実装例のご紹介

この記事は約39分で読めます。

Googleの子会社であるDeepMindにより開発された、人工知能を搭載したコンピューター囲碁プログラムAlphaGo(アルファ碁)の活躍により、深層学習や強化学習の注目度はさらに増しています。

今回は、このDeepMindが考案した深層強化学習(Deep Q-Network: DQN)の実装例を紹介しようと思います。
それにあたり、まずは基本的な強化学習(Q-Learning)と強化学習用オープンプラットフォームOpen AI Gymを紹介し、その後DQNへの発展を見てこうと思います。
また最後に、DQN以降に拡張された深層強化学習についても簡単に触れようと思います。

つい先日発売された以下の書籍では、今回紹介する深層強化学習の手法について触れていますので、参考になると思います。
深層強化学習だけでなく、そもそもの強化学習の歴史的な発展についても詳しく解説されている書籍ですので、強化学習の勉強におすすめな本です。

強化学習とQ-learningについて

強化学習(Reinforcement learning)は、ある環境内におけるエージェントが、現在の状態を観測し、取るべき行動を決定する問題を扱う、機械学習の一種の手法です。
よく機械学習の書籍では、機械学習の種類として「教師なし学習」と「教師あり学習」に加えて、3つ目に「強化学習」がよく挙げられていますよね。

ある環境内に置かれたエージェントは、環境の状態ごとに施策(行動を決定するルール)に基づいて行動を決定して、次の状態に遷移して報酬を受け取ります。
これを繰り返して、より多くの累積報酬を得られるような施策を学習する方法が強化学習です。

強化学習にも色々種類があるのですが、今回は、有限マルコフ決定過程な環境において、より多い累積報酬を得る施策として、行動価値関数Qを用いた学習方法を実装します。

エージェントが各状態において選択できる行動の中で、最も行動価値関数の値が高い行動を選択するように学習する方法をQ学習と言います。
学習は、エージェントが行動を選択し、遷移した状態とその時に得られた報酬を用いて、その時の行動価値関数Qの値を更新することを繰り返すことで行います。

例えば、 エージェントの学習率が\(\alpha\)、割引率が\(\gamma\)であるとして、状態\(s_t\)のエージェントが行動\(a_t\)を選択し、報酬\(r\)を得て、状態が\(s_{t+1}\)に遷移したとすると、このときの行動価値関数Qの値\(Q(s_t,a_t)\)を次の式で更新します。

\(Q_{s_t,a_t}{\leftarrow}Q_{s_t,a_t}+{\alpha}r_{t+1}+\gamma\max_{\alpha_t+1}Q(s_{t+1},a_{t+1})-Q(s_t,a_t)\)

学習率は、エージェントが行動の価値をどのくらいの割合で学習するかを表すパラメータであり、割引率は、将来もらえる報酬を現在の価値としてどのくらいの割合とするかを表すパラメータになります。

行動の選択は、今回は、基本的には行動価値関数の価値が高い行動を選びつつ、一定確率εでランダムに行動を選ぶように行動を選択させる、ε-greedy法という手法を用います。
他にもボルツマン分布に従う確率で行動を選択する方法などもあるようです。

前述の通り、このような強化学習を実行するには、強化学習で解く問題(環境)をプログラム上で用意しなければならないことになります。
そういった強化学習用の環境を提供するプラットフォームとしてOpen AI Gymというものがあります。

Open AI Gym : https://gym.openai.com/

Open AI Gymは環境(行動空間、状態空間、報酬)のみを提供してくれるライブラリです。
環境の種類は色々含まれています。
これらに対して自身で強化学習アルゴリズムを実装して環境に試してみることで、強化学習アルゴリズムの評価を行うことができます。

FrozenLakeをQ-learningで解いてみる例

それでは実際に、Open AI Gymで提供されている環境をQ学習で解いてみる例をご紹介します。
今回はToy textの「FrozenLake-v0」という環境にしました。

簡単に環境の説明をすると、下記のような、4×4のマスを想定し、Sがエージェントの現在地、Fが道、Hが穴(落ちるとゲームオーバーでマイナス報酬を得る)、Gがゴール(ここに来て初めてプラス報酬を得る)で、Gに向かうようエージェントを行動(移動)させるといった、経路探索をする環境です。

SFFF
FHFH
FFFH
HFFG

PythonでQ-learningで解いてみるコード例が以下になります。

env = gym.make("FrozenLake-v0")
n_obs = env.observation_space.n
n_act = env.action_space.n

q = np.zeros([n_obs, n_act])

epoch_cnt = 20000
max_steps = 200
epsilon = 0.001
gamma = 0.9
alpha = 0.9
 
rewards = np.zeros(epoch_cnt)

for epoch in tqdm(range(epoch_cnt)):
    
    pobs = env.reset()
    done = False
    
    for step in range(max_steps):
        
        pact = np.argmax(q[pobs, :])
        pact = np.random.choice(np.where(q[pobs, :] == q[pobs, pact])[0])
        if np.random.rand() <= epsilon:
            pact = env.action_space.sample()
            
        obs, reward, done, _ = env.step(pact) # return observation, reward, done, info
        
        if not done:
            q[pobs, pact] += alpha * (reward - q[pobs, pact] + gamma * np.max(q[obs, :]))
        else:
            q[pobs, pact] += alpha * (reward - q[pobs, pact])
            
        pobs = obs
        rewards[epoch] = reward
        if done:
            break
            
rates = np.average(rewards.reshape([epoch_cnt//1000, 1000]), axis = 1)
plt.plot(rates)
plt.show()
100%|██████████| 20000/20000 [00:20<00:00, 960.95it/s] 

最初はすぐに穴に落ちたりして報酬が低くゴールにたどり着けていませんが、エピソード数を重ねるごとにだんだんとゴールするようになって報酬を得るように学習していることが分かります。

Deep Q-Network: DQNについて

Q-learningでは前述の通り、\(Q_{s_t,a_t}\)をより良いものに更新していき、このテーブルの値を使って行動を選択しているわけですが、例えば、状態数や行動数がとても大きい数であったり、あるいは連続的であるなどの場合には、組み合わせの数が非常に多くなってしまい、計算させるのが困難なものになってしまいます。

そこで、DQNではこういった場合に対処するため、\(Q\)をテーブルではなく、関数(ニューラルネットワークor深層学習)で近似するということを考えます。
ただし、ここで近似するネットワークは、状態×行動を入力とするのではなく、状態を入力して行動の\(Q\)値を出力するネットワークを構成します。

Q-learningと同様に、\(Q(s_t,a_t)\)を\(r+\gamma\max_{a_t+1}Q(s_{t+1},a_{t+1})\)に近づけたいと考えるので、後者を教師信号\(target\)として、現在の\(Q\)との誤差関数\(L(s,a)\)を使ってネットワークを学習させます。

\(target=r+\gamma\max_{a_t+1}Q(s_{t+1},a_{t+1})\)
\(L(s,a)=\displaystyle\frac{1}{2}\Bigl(target-Q(s,a)\Bigr)^2\)

またDQNでは、以上の学習アルゴリズムの仕組みに加えて、下記のような工夫を加えることで、ネットワークを効率的に学習させています。

  • Experience replay
    • エージェントが繰り返し行動して得られた経験は時系列的に獲得するが、経験に相関があるとネットワークが過学習してしまう。これを防ぐため、経験をまずは保存しておいてからランダムに選んで学習させる
  • Freezing the target network
    • TD誤差の目標値に古い\(Q\)(target \(Q\)-network)を使う
      • \(target=r+\gamma\max_{a_t+1}Q_{\theta^{-1}}(s_{t+1},a_{t+1})\)
      • \(L_{\theta}(s,a)=\displaystyle\frac{1}{2}\Bigl(target-Q_{\theta}(s,a)\Bigr)^2\)
      • 一定周期で学習中の\(Q_\theta\)のパラメータと同期(\(\theta^{-1}\leftarrow\theta\))
  • Clipping rewards
    • 報酬のスケールを正スコア\(+1\)、負スコア\(-1\)といったように統一する
  • Skipping frames
    • 毎フレームで行動選択を行うと計算コストがかかる上、毎フレームで行動をする必要はない。よって数フレームおきに行動選択を行うようにする

FrozenLakeをDQNで解いてみる例

それでは、実際に先ほどのFrozenLakeの環境に対してDQNでアプローチをしてみるコード例を記します。
もちろん、FrozenLakeはDQNを使うほどの環境ではないのですが、実装の勉強のため簡単な問題でまずは見てみます。
ニューラルネットワークの実装にはChainerを使いました。
実装が以下の通り。

import copy
import time
import numpy as np
import matplotlib
import matplotlib.pyplot as plt
import chainer
import chainer.links as L
import chainer.functions as F
import gym

# NNクラス定義
class NN(chainer.Chain):
    
    def __init__(self):
        super(NN, self).__init__(
            xc = L.Linear(16, 100),
            ch = L.Linear(100, 100),
            hy = L.Linear(100, 4)
        )
        
    def __call__(self, x):
        
        x = chainer.Variable(x)
        h = F.relu(self.xc(x))
        h = F.relu(self.ch(h))
        y = F.relu(self.hy(h))
        return y
    
    def reset(self):
        
        self.zerograds()

# 状態変換関数定義
def convert(obs):
    
    tmp = np.zeros(16)
    tmp[obs] = 1
    obs = np.array(tmp, dtype="float32")
    return obs

# 環境
env = gym.make("FrozenLake-v0")

# モデル
Q = NN() # 近似Q関数
#chainer.serializers.load_npz("./******.npz", q) # 重みファイル読み込み
Q_ast = copy.deepcopy(Q)
optimizer = chainer.optimizers.Adam()
optimizer.setup(Q)

# 定数
EPOCH_NUM = 1000 # エポック数
MEMORY_SIZE = 1000 # メモリサイズいくつで学習を開始するか
BATCH_SIZE = 100 # バッチサイズ
EPSILON = 1 # ε-greedy法
EPSILON_DECREASE = 0.005 # εの減少値
EPSILON_MIN = 0.01 # εの下限
START_REDUCE_EPSILON = 1000 # εを減少させるステップ数
TRAIN_FREQ = 10 # Q関数の学習間隔
UPDATE_TARGET_Q_FREQ = 20 # Q関数の更新間隔
GAMMA = 0.99

total_step = 0 # 総ステップ(行動)数
memory = [] # メモリ
total_rewards = [] # 累積報酬記録用リスト
total_losses = [] # 累積損失記録用リスト

# 学習開始
print("\t".join(["epoch", "EPSILON", "reward", "total_step", "loss", "elapsed_time"]))
start = time.time()

for epoch in range(EPOCH_NUM):
    
    pobs = env.reset() # 環境初期化
    pobs = convert(pobs)
    done = False # ゲーム終了フラグ
    total_reward = 0 # 累積報酬
    total_loss = 0 # 累積損失
    
    while not done:
        
        # 行動選択
        pact = env.action_space.sample()
        if np.random.rand() > EPSILON: # ε-greedy法
            pact = Q(pobs.reshape((1, 16))) # 最適な行動を予測 # batchsize, channel, height, width
            pact = np.argmax(pact.data)
            
        # 行動
        obs, reward, done, _ = env.step(pact)
        obs = convert(obs)
        
        # メモリに蓄積
        memory.append((pobs, pact, reward, obs, done)) # 変換済みの行動前状態ベクトル、未変換の行動ラベル、報酬、変換済みの行動後状態ベクトル、ゲーム終了フラグ
        if len(memory) > MEMORY_SIZE: # メモリサイズを超えていれば消していく
            memory.pop(0)
            
        # 学習
        if len(memory) == MEMORY_SIZE: # メモリサイズ分溜まっていれば学習
            
            # 経験リプレイ
            if total_step % TRAIN_FREQ == 0:
                np.random.shuffle(memory)
                memory_idx = range(len(memory))
                for i in memory_idx[::BATCH_SIZE]:
                    batch = np.array(memory[i:i+BATCH_SIZE]) # 経験ミニバッチ
                    pobss = np.array(batch[:,0].tolist(), dtype="float32").reshape((BATCH_SIZE, 16))
                    pacts = np.array(batch[:,1].tolist(), dtype="int32")
                    rewards = np.array(batch[:,2].tolist(), dtype="int32")
                    obss = np.array(batch[:,3].tolist(), dtype="float32").reshape((BATCH_SIZE, 16))
                    dones = np.array(batch[:,4].tolist(), dtype="bool")
                    # set y
                    q = Q(pobss)
                    maxq = list(map(np.max, Q_ast(obss).data)) # maxQ
                    target = copy.deepcopy(q.data)
                    for j in range(BATCH_SIZE):
                        target[j, pacts[j]] = rewards[j]+GAMMA*maxq[j]*(not dones[j])
                    # Perform a gradient descent step
                    Q.reset()
                    loss = F.mean_squared_error(q, chainer.Variable(target))
                    total_loss += loss.data
                    loss.backward()
                    optimizer.update()
                    
            # Q関数の更新
            if total_step % UPDATE_TARGET_Q_FREQ == 0:
                Q_ast = copy.deepcopy(Q)
                
        # εの減少
        if EPSILON > EPSILON_MIN:
            if total_step > START_REDUCE_EPSILON:
                EPSILON -= EPSILON_DECREASE
                
        # 次の行動へ
        total_reward += reward
        total_step += 1
        pobs = obs
        
    total_rewards.append(total_reward) # 累積報酬を記録
    total_losses.append(total_loss) # 累積損失を記録
    #serializers.save_npz("./******.npz", q) # 重みファイル出力
    
    if (epoch+1) % 100 == 0:
        
        elapsed_time = time.time()-start
        r = sum(total_rewards[((epoch+1)-10):(epoch+1)])
        l = sum(total_losses[((epoch+1)-10):(epoch+1)])
        print("\t".join(map(str,[epoch+1, EPSILON, r, total_step, l, str(elapsed_time)+"[sec]"]))) # ログ出力
        start = time.time()

plt.figure(figsize=(10,5))
resize = (len(total_rewards)//10, 10)
tmp = np.array(total_rewards, dtype="float32").reshape(resize)
tmp = np.average(tmp, axis=1)
plt.plot(tmp)
plt.show()
epoch	EPSILON	reward	total_step	loss	elapsed_time
100	1	0.0	787	0	0.03460097312927246[sec]
200	0.00999999999999918	2.0	3009	1.1925165348802693	17.631765127182007[sec]
300	0.00999999999999918	9.0	7854	0.5742537427868228	33.66534638404846[sec]
400	0.00999999999999918	8.0	12133	0.43673591697006486	42.14092016220093[sec]
500	0.00999999999999918	5.0	16754	0.5355612131825183	47.29596662521362[sec]
600	0.00999999999999918	7.0	20616	0.4204226514848415	37.15161466598511[sec]
700	0.00999999999999918	6.0	24720	0.6047423291311134	37.86865234375[sec]
800	0.00999999999999918	5.0	28593	0.3329712890263181	40.22569251060486[sec]
900	0.00999999999999918	7.0	32296	0.6885635958751664	42.5173978805542[sec]
1000	0.00999999999999918	4.0	36414	0.7366322512098122	49.55789756774902[sec]

無事に学習をしました。
Q-learningの時と同様に、この環境についてはトータル報酬はせいぜい0.6程度までなのかもしれません。

CartPoleをDQNで解いてみる例

この章では、やはり前章のような離散的な状態を返す簡単な問題ではなく、もう少し難しめの連続的な状態を表す環境にも適用してみようと思います。

FrozenLakeは迷路探索のような問題でしたので、状態が「プレイヤーはどこにいるのか」といった離散的な値で表すことができました。
今回扱う環境は CartPole というゲームになります。
強化学習の界隈では古典的によく使われてきた環境のようで、Open AI Gymにも環境が用意されています。
Open AI Gymによる説明とゲームの様子が下記になります。

CartPole-v0
A pole is attached by an un-actuated joint to a cart, which moves along a frictionless track. The system is controlled by applying a force of +1 or -1 to the cart. The pendulum starts upright, and the goal is to prevent it from falling over. A reward of +1 is provided for every timestep that the pole remains upright. The episode ends when the pole is more than 15 degrees from vertical, or the cart moves more than 2.4 units from the center.

Open AI Gym

つまり、台車の上に棒(ポール)を立て、その棒が倒れてしまわないように、台車を右左に動かして、バランスよく立て続けるといったゲームです。
よく小学生の男子が箒を手のひらに乗せてバランスを保ち続けるといった遊びをやることがあると思いますが、あれと一緒です。

ゲームのルールについてまとめると下記のようになります。

  • 状態は、棒の (横軸座標、横軸速度、ポール角度、ポール角速度) が連続値として観測できる
  • 行動は、台車を左で動かすか、右へ動かすかの2値
  • 棒が一定角度傾いてしまったらゲーム終了
  • 台車が中心から一定距離を離れてしまってもゲーム終了
  • 報酬は棒が立っている限り得られ続ける
  • 棒が一定角度傾いてしまったら報酬なし(ただし、台車の位置に関しては問われないので、一定距離離れてしまっても報酬としては得られている状態になる)

実際にOpen AI Gymから環境を読み込んで、ランダムに行動を取り続けてみたときの各値が下記のようになります。

env = gym.make("CartPole-v0")
print("observation space num: ", env.observation_space.shape[0])
print("action space num: ", env.action_space.n)
print("-"*50)
pobs = env.reset()
done = False
while not done:
    act = env.action_space.sample()
    obs, reward, done, _ = env.step(act)
    print(pobs, act, reward, obs, done)
    pobs = obs
observation space num:  4
action space num:  2
--------------------------------------------------
[ 0.0415392  -0.00341152  0.00436514 -0.04859381] 0 1.0 [ 0.04147097 -0.19859579  0.00339326  0.24546316] False
[ 0.04147097 -0.19859579  0.00339326  0.24546316] 1 1.0 [ 0.03749906 -0.00352247  0.00830253 -0.04614752] False
[ 0.03749906 -0.00352247  0.00830253 -0.04614752] 1 1.0 [ 0.03742861  0.19147945  0.00737958 -0.33619941] False
[ 0.03742861  0.19147945  0.00737958 -0.33619941] 0 1.0 [ 0.0412582  -0.00374674  0.00065559 -0.04119852] False
[ 0.0412582  -0.00374674  0.00065559 -0.04119852] 1 1.0 [ 4.11832606e-02  1.91365807e-01 -1.68383229e-04 -3.33674533e-01] False
[ 4.11832606e-02  1.91365807e-01 -1.68383229e-04 -3.33674533e-01] 1 1.0 [ 0.04501058  0.38649015 -0.00684187 -0.62641055] False
[ 0.04501058  0.38649015 -0.00684187 -0.62641055] 1 1.0 [ 0.05274038  0.58170694 -0.01937008 -0.92124037] False
[ 0.05274038  0.58170694 -0.01937008 -0.92124037] 1 1.0 [ 0.06437452  0.77708521 -0.03779489 -1.21994726] False
[ 0.06437452  0.77708521 -0.03779489 -1.21994726] 1 1.0 [ 0.07991622  0.97267339 -0.06219384 -1.52422905] False
[ 0.07991622  0.97267339 -0.06219384 -1.52422905] 1 1.0 [ 0.09936969  1.16848876 -0.09267842 -1.83565743] False
[ 0.09936969  1.16848876 -0.09267842 -1.83565743] 1 1.0 [ 0.12273947  1.36450518 -0.12939157 -2.15562869] False
[ 0.12273947  1.36450518 -0.12939157 -2.15562869] 0 1.0 [ 0.15002957  1.17086919 -0.17250414 -1.9055378 ] False
[ 0.15002957  1.17086919 -0.17250414 -1.9055378 ] 0 1.0 [ 0.17344695  0.97798019 -0.2106149  -1.67096363] True

それでは先ほどと同様にDQNを使って学習させてみます。
今度は深層学習のライブラリはPyTorchを使ってみました。
実装のコードは以下の通り。

import copy
import time
import numpy as np
import matplotlib
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.autograd import Variable
import gym
from gym import wrappers

# 環境
MONITOR = False
env = gym.make("CartPole-v0")
if MONITOR:
    env = wrappers.Monitor(env, "./result", force=True)

obs_num = env.observation_space.shape[0]
acts_num = env.action_space.n
HIDDEN_SIZE = 100

class NN(nn.Module):
    
    def __init__(self):
        
        super(NN, self).__init__()
        self.fc1 = nn.Linear(obs_num, HIDDEN_SIZE)
        self.fc2 = nn.Linear(HIDDEN_SIZE, HIDDEN_SIZE)
        self.fc3 = nn.Linear(HIDDEN_SIZE, HIDDEN_SIZE)
        self.fc4 = nn.Linear(HIDDEN_SIZE, acts_num)
    
    def __call__(self, x):
        
        h = F.relu(self.fc1(x))
        h = F.relu(self.fc2(h))
        h = F.relu(self.fc3(h))
        y = F.relu(self.fc4(h))
        return y

# 定数
EPOCH_NUM = 3000 # エポック数
STEP_MAX = 200 # 最高ステップ数
MEMORY_SIZE = 200 # メモリサイズいくつで学習を開始するか
BATCH_SIZE = 50 # バッチサイズ
EPSILON = 1.0 # ε-greedy法
EPSILON_DECREASE = 0.001 # εの減少値
EPSILON_MIN = 0.1 # εの下限
START_REDUCE_EPSILON = 200 # εを減少させるステップ数
TRAIN_FREQ = 10 # Q関数の学習間隔
UPDATE_TARGET_Q_FREQ = 20 # Q関数の更新間隔
GAMMA = 0.97 # 割引率
LOG_FREQ = 1000 # ログ出力の間隔

# モデル
Q = NN() # 近似Q関数
Q_ast = copy.deepcopy(Q)
optimizer = optim.RMSprop(Q.parameters(), lr=0.00015, alpha=0.95, eps=0.01)

total_step = 0 # 総ステップ(行動)数
memory = [] # メモリ
total_rewards = [] # 累積報酬記録用リスト

# 学習開始
print("\t".join(["epoch", "epsilon", "reward", "total_step", "elapsed_time"]))
start = time.time()
for epoch in range(EPOCH_NUM):
    
    pobs = env.reset() # 環境初期化
    step = 0 # ステップ数
    done = False # ゲーム終了フラグ
    total_reward = 0 # 累積報酬
    
    while not done and step < STEP_MAX:
        if MONITOR:
            env.render()
        # 行動選択
        pact = env.action_space.sample()
        # ε-greedy法
        if np.random.rand() > EPSILON:
            # 最適な行動を予測
            pobs_ = np.array(pobs, dtype="float32").reshape((1, obs_num))
            pobs_ = Variable(torch.from_numpy(pobs_))
            pact = Q(pobs_)
            maxs, indices = torch.max(pact.data, 1)
            pact = indices.numpy()[0]
            
        # 行動
        obs, reward, done, _ = env.step(pact)
        if done:
            reward = -1
            
        # メモリに蓄積
        memory.append((pobs, pact, reward, obs, done)) # 状態、行動、報酬、行動後の状態、ゲーム終了フラグ
        if len(memory) > MEMORY_SIZE: # メモリサイズを超えていれば消していく
            memory.pop(0)
            
        # 学習
        if len(memory) == MEMORY_SIZE: # メモリサイズ分溜まっていれば学習
            # 経験リプレイ
            if total_step % TRAIN_FREQ == 0:
                memory_ = np.random.permutation(memory)
                memory_idx = range(len(memory_))
                for i in memory_idx[::BATCH_SIZE]:
                    batch = np.array(memory_[i:i+BATCH_SIZE]) # 経験ミニバッチ
                    pobss = np.array(batch[:,0].tolist(), dtype="float32").reshape((BATCH_SIZE, obs_num))
                    pacts = np.array(batch[:,1].tolist(), dtype="int32")
                    rewards = np.array(batch[:,2].tolist(), dtype="int32")
                    obss = np.array(batch[:,3].tolist(), dtype="float32").reshape((BATCH_SIZE, obs_num))
                    dones = np.array(batch[:,4].tolist(), dtype="bool")
                    # set y
                    pobss_ = Variable(torch.from_numpy(pobss))
                    q = Q(pobss_)
                    obss_ = Variable(torch.from_numpy(obss))
                    maxs, indices = torch.max(Q_ast(obss_).data, 1)
                    maxq = maxs.numpy() # maxQ
                    target = copy.deepcopy(q.data.numpy())
                    for j in range(BATCH_SIZE):
                        target[j, pacts[j]] = rewards[j]+GAMMA*maxq[j]*(not dones[j]) # 教師信号
                    # Perform a gradient descent step
                    optimizer.zero_grad()
                    loss = nn.MSELoss()(q, Variable(torch.from_numpy(target)))
                    loss.backward()
                    optimizer.step()
            # Q関数の更新
            if total_step % UPDATE_TARGET_Q_FREQ == 0:
                Q_ast = copy.deepcopy(Q)
                
        # εの減少
        if EPSILON > EPSILON_MIN and total_step > START_REDUCE_EPSILON:
            EPSILON -= EPSILON_DECREASE
            
        # 次の行動へ
        total_reward += reward
        step += 1
        total_step += 1
        pobs = obs
        
    total_rewards.append(total_reward) # 累積報酬を記録
    
    if (epoch+1) % LOG_FREQ == 0:
        r = sum(total_rewards[((epoch+1)-LOG_FREQ):(epoch+1)])/LOG_FREQ # ログ出力間隔での平均累積報酬
        elapsed_time = time.time()-start
        print("\t".join(map(str,[epoch+1, EPSILON, r, total_step, str(elapsed_time)+"[sec]"]))) # ログ出力
        start = time.time()
        
if MONITOR:
    env.render(close=True)
epoch	epsilon	reward	total_step	elapsed_time
1000	0.0999999999999992	68.873	70873	149.7089443206787[sec]
2000	0.0999999999999992	164.292	237165	254.5188992023468[sec]
3000	0.0999999999999992	170.694	409859	166.58216881752014[sec]

得られる累積報酬がエポックを重ねるごとに増えていっていることが確認できます。
200ステップまでしか行動しませんので、この環境での最高累積報酬は200です。

また、この環境では以下の通り、動画として保存することができます。
以下はエポックがまだ最初の方のプレイ動画です。
まだ学習が進んでいなく、すぐに棒が傾いてしまってゲームがすぐ終了(そして動画もすぐ終了)する状態です。

次に以下が、学習が進んだ状態のプレイ動画が下記になります。
台車を小刻みに動かしながらバランスをとって棒を立て続けられるようになっています。

このようにして学習できていることが確認できました。
実は前章のFrozenLakeのときよりも、ハイパーパラメータの調整に苦労しました。
DQNに関しては、勾配法のアルゴリズムの選択も含め、論文などで良いハイパーパラメータ値についての研究がされています。
最初はなかなかうまく学習しませんでしたが、アルゴリズムやハイパーパラメータをそのような研究を参考に設定し直すことで、うまく学習するようになりました。
この辺りの試行錯誤をする必要があることを念頭に置いておいた方が良いでしょう。

その他の深層強化学習の発展の紹介

上記のDQNが登場したのが2013年頃のことで、その後、こういった深層強化学習の分野の研究も盛んに行われるようになってから、様々なDQNの改良が登場しています。
最近DQNの周辺の研究の中からいくつか簡単に触れてみようと思います。

Double DQN

論文は下記になります。

DQNでは、TD誤差計算時に目標値に大きすぎる値が設定されると、前の状態を過大評価してしまうという問題がありました。

そこで、2つの\(Q\)関数を混ぜて学習をさせることで、TD誤差の計算の安定化を図ったものがDouble DQNです。
具体的には、DQNにおける教師信号出力用のネットワークを\(Q_{\theta^{-1}}\)、そのコピーを\(Q_\theta\)とすると、\(s_{t+1}\)でとるべき行動を\(Q_\theta\)で決定し、その評価値を\(Q_{\theta^{-1}}\)で出力して\(Q_\theta\)を更新します。

Dueling Double DQN

論文は下記になります。

ここまでの方法では、\(Q\)関数において、1回の更新で1つの行動に対する価値しか更新ができません。
そこでこの研究では、\(Q\)関数を状態価値関数\(V(s)\)とAdvantage(行動優位)関数\(A(s,a)\)に分解して学習させます。

\(Q_{\theta}(s_t,a_t)=V_{\theta,\beta}+\Bigl(A_{\theta,\alpha}(s_t,a_t)+\displaystyle\frac{1}{|A|}\sum_{\alpha_{t+1}}A_{\theta,\alpha}(s_t,a_{t+1})\Bigr)\)

こうすることで、\(V\)については毎回更新ができるので、TD誤差の計算の伝播が早くなるようです。
単純に線形和で分解してしまうと、状態価値関数、Advantage関数がともに誤差を含めて学習してしまう恐れがあるため、これを防ぐために\(\displaystyle\frac{1}{|A|}\sum_{a_{t+1}}A_{\theta,\alpha}(s_t,a_{t+1})\)を加えるということをするようです。

Prioritized Experience Replay

論文は下記になります。

DQNでは経験からランダムに選んで学習してきているので、より学習に役立つ経験を優先して学習させるようにします。
具体的には、経験サンプルの重要性\(p_i\)を、TD誤差の絶対値\(|\delta_i|\)(パラメータの更新幅とみなせる)を用いて表し、確率とした上で経験サンプリングをします。

\(P(i)=\displaystyle\frac{p^{\alpha}_i}{\sum_{k}p^{\alpha}_k},\hspace{0.5em}p_i=|\delta_i|+\epsilon\)

こうすることで、TD誤差の大きい経験を優先して学習させられるようになります。

Gorila DQN

論文は下記になります。

「GOogle ReInforcement Learning Architecture」を略して「Gorila」です。
ちょっと遊んでいますね笑

こちらは、これまでの学習アルゴリズムの改善とは少し違い、学習を並列分散させるということをします。
Memory(経験)を集めるActorと、誤差計算を行うLearnerに分けて、並列処理をすることで、DQNよりも高速に学習できたことが報告されています。

A3C

論文は下記になります。

Google DeepMindにより、2016年に発表された論文です。
Asynchronous Advantage Actor-Critic(A3C)は、これまでのDQNの学習アルゴリズムとは大きく変わっており、こちらも並列処理をさせてGPUを使ったDQNよりも高速に学習させているという点と、これまでベースとしていた強化学習手法であるQ-Learningとは違い、Actor-Criticを採用しています。

主な特徴は以下の2点です。

  • Asynchronous

CPUのマルチスレッドで同時に複数のエージェントを並列で走らせ、パラメータを非同期に更新します。
つまり、

  1. \(\theta_{local}\)(スレッドごとに別々のパラメータ)←\(\theta\)(グローバルなパラメータ)(同期)
  2. \(\theta_{local}\)を使って勾配\(d\theta\)を計算
  3. \(d\theta\)で\(\theta\)を更新
  4. 上記を繰り返す

ということをします。
この時にパラメータだけでなく、最適化アルゴリズムRMSpropの変数についてもグローバルに共有するようです。

  • Advantage Actor-Critic

Actor-Criticは、政策\(\pi\)と、状態価値関数\(V\)をそれぞれ独立に推定しながら学習を行う強化学習手法です。
それぞれ独立に行うことにより、行動が連続的な場合でも学習させやすいといったメリットがあります。

ただし、状態価値関数は、報酬\(r_{t+1}\)と次の状態価値\(V(s_{t+1})\)を使って更新されます。
したがって、次の状態価値が正しくなければ、現在の状態価値も正しく推定されないといった問題も起こります。

そこで、Actor-Criticによって時系列にまま学習できる性質を生かし、\(k\)ステップ先までの報酬を考慮した推定値(Advantage)を使います。
これにより現在の状態価値が、より確からしい推定値となり、学習が早く進むようです。

以上、これらの工夫によって、これまでのDQNよりも学習速度が大幅に向上することが報告されています。

また、A3Cは、実は深層強化学習のライブラリとして公開されているChainerRLにすでに実装されています。
単純に学習を試したいということであれば、こちらを試してみるのが良いかもしれません。

まとめ

今回は、強化学習(Q-learning)と深層強化学習(DQN)の実装方法とその発展手法について紹介しました。

Q-learningは価値ベースの手法であり、DQNは深層学習を用いてQ-learningを発展させた手法です。DQNは高次元・複雑な問題にも対応できるため、幅広い応用が期待されます。

また、DQNの発展手法として、Double DQNやDueling DQNなどの最新手法についても紹介しました。

強化学習は実用的な問題への適用が増えている分野であり、本記事を参考にしながらさまざまな応用にも挑戦していただければ幸いです。