External Memory

プログラミング周辺知識の備忘録メイン

Deep Q-Networks (DQN)

強化学習は普通の深層学習より学習が能動的という意味でAlっぽいので面白そうではある。以下の有名そうな論文を読んで勉強のためのとっかかりとした。

Playing Atari with Deep Reinforcement Learning
https://arxiv.org/pdf/1312.5602.pdf

強化学習の全体像を知ったわけではないが、
最適化をニューラルネットワークに任せるという意味では、
政策やアルゴリズムをあまり考える手間が少なそうなのでおそらく取っ付きやすいのではないかと思う。


DQNでは最適化された行動価値関数Qを近似するためにCNNを用いる。
CNNはSGDを用いた重み更新を用いたQ-learningアルゴリズムの変形で学習される。
この時、相関データと非定常な分布を緩和するために以前の遷移をランダムにサンプリングするexperience replayを用いる。

最適化されたQは報酬の期待値が最大化され、time t、状態s、action a、報酬Rを用いて以下のように表される。

Q^*(s,a)=\max_{\pi}\mathbb{E}[R_t|s_t=s,a_t=a,\pi]


またこれはBellman equationと等価である。

Q^*(s,a)=\mathbb{E}_{s'\sim \mathcal{E}}[r+\gamma \max_{a'}Q(s',a')|s,a]
s',a'は次ステップの状態と行動、γは割引率である。

ニューラルネットワーク非線形近似を行うと(Q(s,a;\theta)\approx Q^*(s,a):θは重み)、
loss関数はiteration iを用いて以下のようになる。

L_i(\theta_i)=\mathbb{E}_{s,a\sim \rho(\cdot)}[(y_i-Q(s,a;\theta_i))^2]

ここでy_i=\mathbb{E}_{s'\sim \mathcal{E}}[r+\gamma \max_{a'}Q(s',a'|\theta_{i-1})|s,a]となり、
前回の重みθ_{i-1}を用いているが、文献中のAlgorithm 1ではそうはなっていない。
なぜi-1としているのかわからないが、step一回ごとに更新するのでs'、a'と行動分布を揃えるためだろうか。


"
追記11/1
Human-Level Control through Deep Reinforcement Learning
https://storage.googleapis.com/deepmind-media/dqn/DQNNaturePaper.pdf
を読むと、やっぱり一回前のパラメータを使っている。
Qの変動が抑えられるのでactionの選択に関する影響に遅延が生じるため、発散や振動が発生しにくくなる。
"


このニューラルネットワークを用いた手法は環境に依存せず強化学習を行え、また政策を勘案する必要がない。よってatari gameの種類ごとにモデルやアルゴリズムの変更を行う必要がない。


Deep Q-learningの手続きは文献中Algorithm 1に記載がある。
ここでreplay memoryは最近のN個を保持しランダムサンプリングを行う。
これにより学習サンプルの強い相関や最大化行動の影響によるサンプルと偏りを緩和する。φは入力の前処理である。


前処理はRGBをグレースケールとしスケールを約半分にし次いでcroppingし正方形とイメージとし、履歴の最近4フレームにこの前処理を行い入力のための蓄積を行っている。
CNNにおけるQのパラメータ化においては、入力を状態パラメータ、出力を行動予測に対応するQ値としている。
これにより一回の順方向伝播ですべての行動のQ値を計算することが出来る。
CNNはconv2層、Fully-connected1層である。
conv 8*8 16ch(stride4) + conv 4*4 32ch(stride2) + FC256units + output(num of action 4-18units)。

ハイパーパラメータは全てのgameで報酬の設定を除き同じである。
報酬はスケールの関係上全てのプラス報酬を1、マイナス報酬を-1、その他0としている。

batchsize32のRMSProp optimaizerを用い、ε-greedyでは1 millionステップごとに1-0.1まで線形減衰させている。
テスト時はε=0.05である。
また計算コストの関係からスペースインベーダ以外のゲームでは4ステップごとに行動を選択し(インベーダは3)、そのステップ間はその行動を維持する。

また学習中に評価においてはQ値のトータル平均報酬を用いるのでなく、ある状態の集合からの最大Q値の平均を用いる。
これは重みの小さな変化によりサンプリングする状態分布が大きく変化するので非常にnoisyになり、学習性能の追跡がやりづらいからである。

全てのゲームで他のRL手法より高い平均スコアであり、
スペースインベーダ、Q*bert、Seaquest以外のゲームで人間より高いまた近い平均スコアを示している。インベーダなどは長期的なスケールでの戦略が要求されるからである。



Algorithm 1はpythonコードではおそらくこのような感じになる。
enviromentは空で、εの減衰などいろいろやってないし、当然動かない。
CNN部分は代わりにリカレントネットワークとかもあり得そうな気がするが。

import tensorflow as tf
import numpy as np

class Environment(object):
    def __init__(self):
        self.actions =(0,1)
    
    def newgame(self):
        pass
    
    def observe(self):
        pass
    
    def preprocess(self,state):
        pass
    
    def excute(self,action):
        pass
    

class Agent(object):
    
    def __init__(self,env):
        self.env = env
        self.replay_memory =[]
        self.max_memory_size =1000
        self.current_pos = 0
        self.actions = env.actions
        self.max_step = 1000
        self.discount_factor = 0.9
        self.batch_size =32
        self.y = self.nn_inference()
        
    def predict(self,state,epsilon=0.1,is_training=True):
        if not is_training:
            epsilon=0.05
            
        if np.random.rand() <= epsilon:
            action = np.random.choice(self.actions)
        else:
            action = self.actions[np.argmax(self.q_eval(state))]
        
        return action
    
    def store_transition(self,state,action,reward,n_state,terminal):
        self.replay_memory[self.current_pos] = (state,action,reward,n_state,terminal)
        self.current_pos = (self.current_pos +1) % self.max_memory_size

    def nn_inference(self):
        self.x  = tf.placeholder(tf.float32, [None, 84*84])
        
        W1 = tf.Variable(tf.truncated_normal([8, 8, 4, 16],stddev=0.05))
        b1 = tf.Variable(tf.zeros([16]))
        conv1 = tf.nn.relu(tf.nn.conv2d(self.x, W1, strides=[1,4,4,1], padding='SAME')+ b1)
        
        W2 = tf.Variable(tf.truncated_normal([4, 4, 16, 32],stddev=0.05))
        b2 = tf.Variable(tf.zeros([32]))
        conv2 = tf.nn.relu(tf.nn.conv2d(conv1, W2, strides=[1,2,2,1], padding='SAME')+ b2)
        
        W3 = tf.Variable(tf.truncated_normal([11*11*32, 256],stddev=0.05))
        b3 = tf.Variable(tf.zeros([256]))
        fc1 = tf.reshape(conv2, [-1, 256])
        fc1 = tf.nn.relu(tf.matmul(fc1, W3)+ b3)
        
        W4 = tf.Variable(tf.truncated_normal([256, self.actions],stddev=0.05))
        b4 = tf.Variable(tf.zeros([self.actions]))
        output = tf.matmul(fc1, W4)+ b4
        
        return output
    
    def q_eval(self,state):
        return self.sess.run(self.y, feed_dict={self.x:[state]})[0]
    
    def train_update(self):
        state_batch = []
        labels = []
        r_mem_length = len(self.replay_memory)
        
        batchsize = min(r_mem_length,self.batch_size)
        r_mem_indexes = np.random.randint(0,r_mem_length,batchsize)
        
        for mem_index in r_mem_indexes:
            state,action,reward,n_state,terminal = self.replay_memory[mem_index]
            state_batch.append(state)

            q_vals= self.q_eval(state)
            
            if terminal:
                q_vals[action] = reward
            else:
                q_vals[action] = reward + self.discount_factor * np.max(self.q_eval(n_state))
            
            labels.append(q_vals)
            
        self.sess.run(self.train_op,feed_dict={self.x:state_batch,self.y_:labels})
        
    def train(self,epoch):
        
        self.y_ = tf.placeholder(tf.float32, [None, self.actions])
        loss = tf.reduce_mean(tf.square(self.y_ - self.y))
        
        self.train_op = tf.train.RMSPropOptimizer(self.learning_rate).minimize(loss)
        
        self.sess = tf.InteractiveSession()
        tf.global_variables_initializer().run()
        
        for i in range(epoch):
            self.env.newgame()
            state, reward, terminal = self.env.observe()
            step = 0
            
            while not terminal or step == self.max_step:
                #phi = env.preprocess(state)                    
                action = agent.predict(state)
                #for i in range(4):
                self.env.excute(action)                
                next_state, reward, terminal = self.env.observe()
                
                #n_phi = env.preprocess(next_state)
                agent.store_transition(state,action,reward,next_state,terminal)
                
                agent.train_update()
                
                state = next_state
                step += 1
                
        saver = tf.train.Saver()
        saver.save(self.sess,"tmp/dqn",global_step = epoch)
    
    
if __name__ == '__main__':

    env = Environment()
    agent = Agent(env)
    
    agent.train(10000)
    
    test_play()