強化トレーニングの概要：多腕バンディットから本格的なRLエージェントまで

こんにちは、Habr！ 強化学習は、機械学習の最も有望な分野の1つです。 これにより、人工知能は、ロボット工学やビデオゲームから消費者の行動や医療のモデリングまで、幅広い問題を解決できます。 この入門記事では、強化学習の主なアイデアを研究し、独自の自己学習ボットをゼロから構築します。

はじめに

強化学習と従来の機械学習の主な違いは、人工知能が履歴データではなく環境との相互作用のプロセスでトレーニングされることです。 複雑な相互作用を復元するニューラルネットワークの能力と強化学習におけるエージェントの自己学習を組み合わせることで、マシンは大成功を収め、最初にいくつかのAtariビデオゲームで勝利し、その後世界チャンピオンになりました。



教師と一緒に学習タスクを操作することに慣れている場合は、強化学習の場合、わずかに異なるロジックが適用されます。 強化トレーニングでは、「要因-正解」のペアのセットから学習するアルゴリズムを作成する代わりに、これらのペアを独立して生成することにより、環境と対話するようエージェントに教える必要があります。 それから彼は、観察（観察）、勝利（報酬）、行動（行動）のシステムを通してそれらについて訓練されます。



明らかに、今では常に正しい答えが得られないため、タスクは少し複雑になります。 この一連の記事では、強化学習エージェントを作成してトレーニングします。 エージェントの最も単純なバージョンから始めましょう。強化学習の主なアイデアは非常に明確であるため、より複雑なタスクに進みます。

多腕バンディット

強化を伴う学習のタスクの最も簡単な例は、多腕バンディットの問題です（特に、 こことここでHabréで広く取り上げられています ）。 問題のステートメントでは、n個のスロットマシンがあり、それぞれが一定の勝率を持っています。 次に、エージェントの目標は、予想されるゲインが最も高いスロットマシンを見つけて、常に選択することです。 簡単にするために、スロットマシンは4つしかなく、そこから選択する必要があります。



実際、このクラスのタスクは次のプロパティによって特徴付けられるため、このタスクは強化学習に確実に起因する可能性があります。

• 異なるアクションは異なる勝利につながります。 たとえば、迷路の中で宝物を探すとき、左に曲がるとダイヤモンドの束を意味し、右に曲がると毒蛇の穴を意味します。
• エージェントは時間遅延のある勝利を受け取ります。 これは、迷路の中で左に曲がっても、これが正しい選択であることをすぐに理解しないことを意味します。
• ゲインは、システムの現在の状態に依存します。 上記の例を続けると、迷路の現在の部分で左折が正しい場合がありますが、必ずしも残りの部分ではそうではありません。

多腕バンディットの問題では、2番目も3番目の条件も存在しないため、大幅に簡素化され、可能なすべてのオプションから最適なアクションを特定することにのみ集中できます。 強化学習では、これは「行動規則」（ポリシー）を見つけることを意味します。 ポリシー勾配と呼ばれる方法を使用します。この方法では、ニューラルネットワークが次のように動作ルールを更新します。エージェントがアクションを実行し、環境からフィードバックを受け取り、それに基づいて勾配降下によりモデルの重みを調整します。



強化学習の分野では、エージェントが価値関数を訓練する別のアプローチがあります。 エージェントは、現在の状態で最適なアクションを見つけるのではなく、この状態になってどれだけ収益性があるかを予測し、このアクションを実行することを学習します。 どちらのアプローチでも良い結果が得られますが、ポリシーの勾配のロジックはより明白です。

ポリシーの勾配

すでにわかっているように、この場合、各スロットマシンの期待されるゲインは、環境の現在の状態に依存しません。 ニューラルネットワークは、それぞれが1つのスロットマシンに対応する重みのセットのみで構成されることがわかります。 これらの重みは、最大の勝利を得るためにどのハンドルを引く必要があるかを決定します。 たとえば、すべてのウェイトを1に初期化すると、エージェントはすべてのスロットマシンでの勝利について同様に楽観的になります。



モデルの重みを更新するには、e-greedyの動作ラインを使用します。 つまり、ほとんどの場合、エージェントは期待されるゲインを最大化するアクションを選択しますが、場合によっては（確率がeに等しい）アクションがランダムになります。 これにより、可能なすべてのオプションが選択されるようになり、ニューラルネットワークはそれぞれのオプションについて「学習」することができます。



アクションの1つを完了すると、エージェントはシステムからフィードバックを受け取ります：勝ったかどうかに応じて1または-1。 次に、この値を使用して損失関数を計算します。

$$

$$損失= -log（n）* A$$

（利点）は、すべての強化学習アルゴリズムの重要な要素です。 アクションがベースラインよりもどれだけ優れているかを示しています。 将来的には、より複雑なベースラインを使用しますが、今のところは0に等しいと想定しています。つまり、 Aは各アクションの報酬（1または-1）に等しいだけです。 nは動作の規則、現在のステップで選択したスロットマシンのハンドルに対応するニューラルネットワークの重みです。



直感的に、損失関数は、ゲインにつながるアクションの重みが増加し、損失につながるアクションの重みが減少するような値をとる必要があります。 その結果、ウェイトが更新され、エージェントは、最終的には常にそれを選択するまで、勝つ可能性が最も高い固定確率を持つスロットマシンを選択します。

アルゴリズムの実装

盗賊 まず、私たちは山賊を作成します（日常生活では、スロットマシンは山賊と呼ばれます）。 この例では4になります。pullBandit関数は標準正規分布から乱数を生成し、それをバンディットの値と比較してゲームの結果を返します。 ギャングがリストにあるほど、エージェントは彼を選択することで勝つ可能性が高くなります。 したがって、エージェントに常に最後のギャングを選択する方法を学習してほしい。

import tensorflow as tf import numpy as np #  .  №4    . bandits = [0.2,0,-0.2,-5] num_bandits = len(bandits) def pullBandit(bandit): #   result = np.random.randn(1) if result > bandit: # return 1 else: # return -1
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

エージェント 以下のコードは、盗賊の一連の値で構成される単純なエージェントを作成します。 各値は、1つまたは別の盗賊の選択に応じて勝ち/負けに対応します。 エージェントの重みを更新するには、ポリシーの勾配を使用します。つまり、損失関数を最小化するアクションを選択します。

 tf.reset_default_graph() # 2   feed-forward  .     . weights = tf.Variable(tf.ones([num_bandits])) chosen_action = tf.argmax(weights,0) # 6    .        ,        . reward_holder = tf.placeholder(shape=[1],dtype=tf.float32) action_holder = tf.placeholder(shape=[1],dtype=tf.int32) responsible_weight = tf.slice(weights,action_holder,[1]) loss = -(tf.log(responsible_weight)*reward_holder) optimizer = tf.train.GradientDescentOptimizer(learning_rate=0.001) update = optimizer.minimize(loss)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

エージェントトレーニング。 特定のアクションを選択し、勝ち/負けを受け取ることでエージェントを訓練します。 取得した値を使用して、モデルの重みを更新して、予想される大きな見返りのある盗賊をより頻繁に選択する方法を知っています。

 total_episodes = 1000 #   total_reward = np.zeros(num_bandits) #     0 e = 0.1 #   init = tf.global_variables_initializer() #  tensorflow with tf.Session() as sess: sess.run(init) i = 0 while i < total_episodes: #        if np.random.rand(1) < e: action = np.random.randint(num_bandits) else: action = sess.run(chosen_action) #  ,     reward = pullBandit(bandits[action]) #  _,resp,ww = sess.run([update,responsible_weight,weights], feed_dict={reward_holder:[reward],action_holder:[action]}) #     total_reward[action] += reward if i % 50 == 0: print("     " + str(num_bandits) + " bandits: " + str(total_reward)) i+=1 print(" ,   №" + str(np.argmax(ww)+1) + " ...") if np.argmax(ww) == np.argmax(-np.array(bandits)): print("...  !") else: print("...   !")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

結果：

    4   : [-1. 0. 0. 0.]    4   : [ -1. -1. 0. 45.]    4   : [ -2. -1. 1. 91.]    4   : [ -3. -1. 1. 138.]    4   : [ -2. 0. 2. 183.]    4   : [ 0. 0. 2. 229.]    4   : [ -1. 0. 3. 277.]    4   : [ -1. -1. 4. 323.]    4   : [ -2. -1. 2. 370.]    4   : [ -2. -2. 3. 416.]    4   : [ -2. -2. 3. 464.]    4   : [ -3. -1. 3. 508.]    4   : [ -7. 1. 4. 549.]    4   : [ -9. 0. 3. 593.]    4   : [ -9. 0. 4. 640.]    4   : [ -9. 0. 5. 687.]    4   : [ -9. -2. 5. 735.]    4   : [ -9. -4. 7. 781.]    4   : [ -10. -4. 8. 829.]    4   : [ -13. -4. 7. 875.]  ,   №4 ... ...  !
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

Jupyter Notebookの完全版はこちらからダウンロードできます 。

強化された本格的な学習問題の解決

いくつかの可能なソリューションから最適なソリューションを選択できるエージェントを作成する方法がわかったので、完全な強化学習の例となるより複雑なタスクに移りましょう：システムの現在の状態を評価するため、エージェントはゲインを最大化するアクションを選択する必要があります今だけでなく、将来的にも。



強化学習に対処できるシステムは、マルコフ決定プロセス（MDP）と呼ばれます。 このようなシステムは、ある状態から別の状態への移行を保証するゲインとアクションによって特徴付けられ、これらのゲインは、システムの現在の状態とエージェントがこの状態で行う決定に依存します。 勝ちは遅れて獲得できます。



正式には、マルコフの意思決定プロセスは次のように定義できます。 MDPは、考えられるすべての状態SとアクションAのセットで構成され、各瞬間に状態sになり、これらのセットからアクションaを実行します。 したがって、タプル（s、a）が与えられ 、 T（s、a）が決定されます-新しい状態への遷移の確率s 'およびR（s、a） -ゲイン。 その結果、MDPの任意の時点で、エージェントは状態sになり、決定を下し、それに応じて新しい状態s 'とゲインrを受け取ります。



たとえば、ドアを開くプロセスでさえ、マルコフの意思決定プロセスとして表すことができます。 状態は、ドアに対する私たちの視界であり、世界における私たちの身体とドアの位置です。 私たちができるすべての可能な体の動きはセットAであり、勝利はドアの正常な開放です。 特定のアクション（たとえば、ドアへのステップ）は、目標の達成に近づきますが、ドア自体を直接開くことによってのみ提供されるため、それ自体は何の利益ももたらしません。 結果として、エージェントは遅かれ早かれ問題の解決につながるようなアクションを実行する必要があります。

倒立振子を安定させるタスク

ゲームとアルゴリズムテストを使用してAIボットを開発およびトレーニングするためのプラットフォームであるOpenAI Gymを使用し、そこから古典的な問題、つまり倒立振子またはカートポールの安定化問題を取り上げます。 私たちの場合、問題の本質は、トロリーを水平方向に動かして、できるだけ長くロッドを直立させておくことです。





多腕バンディット問題とは異なり、このシステムには以下があります。

• 観察 エージェントは、ロッドの現在の位置と角度を知る必要があります。 ニューラルネットワークはこの観測を使用して、アクションの可能性を評価します。
• 繰延賞金。 現時点および将来の両方で有益であるような方法でトロリーを動かす必要があります。 これを行うには、「観測-アクション」のペアとゲインの調整値を比較します。 修正は、時間の経過に伴うアクションの重み付けを行う関数によって実行されます。

ゲインの時間遅延を考慮するには、いくつかの修正を加えたポリシー勾配法を使用する必要があります。 まず、時間単位ごとに複数のケースを持つエージェントを更新する必要があります。 これを行うには、バッファー内のすべての観測値を収集し、それらを同時に使用してモデルの重みを更新します。 単位時間あたりのこの観測セットは、 割引後のゲインと比較されます。



したがって、エージェントの各アクションは、即時の賞金だけでなく、その後のすべての賞金も考慮して行われます。 また、調整されたゲインを損失関数の要素A（利点）の推定値として使用します。

アルゴリズムの実装

ライブラリをインポートし、Cart-Poleタスク環境をロードします。

 import tensorflow as tf import tensorflow.contrib.slim as slim import numpy as np import gym import matplotlib.pyplot as plt %matplotlib inline try: xrange = xrange except: xrange = range env = gym.make('CartPole-v0') #  
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

エージェント 最初に、現時点で後続のすべての賞金を割り引く関数を作成します。

 gamma = 0.99 #   def discount_rewards(r): """     ,    """ discounted_r = np.zeros_like(r) running_add = 0 for t in reversed(xrange(0, r.size)): running_add = running_add * gamma + r[t] discounted_r[t] = running_add return discounted_r
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次に、エージェントを作成します。

 class agent(): def __init__(self, lr, s_size,a_size,h_size): #  feed-forward  . #       self.state_in= tf.placeholder(shape=[None,s_size],dtype=tf.float32) hidden = slim.fully_connected(self.state_in,h_size, biases_initializer=None,activation_fn=tf.nn.relu) self.output = slim.fully_connected(hidden,a_size, activation_fn=tf.nn.softmax,biases_initializer=None) self.chosen_action = tf.argmax(self.output,1) #   # 6    . #      #   , #       . self.reward_holder = tf.placeholder(shape=[None],dtype=tf.float32) self.action_holder = tf.placeholder(shape=[None],dtype=tf.int32) self.indexes = tf.range(0, tf.shape(self.output)[0])*tf.shape(self.output)[1] + self.action_holder self.responsible_outputs = tf.gather(tf.reshape(self.output, [-1]), self.indexes) #  self.loss = -tf.reduce_mean(tf.log(self.responsible_outputs)* self.reward_holder) tvars = tf.trainable_variables() self.gradient_holders = [] for idx,var in enumerate(tvars): placeholder = tf.placeholder(tf.float32,name=str(idx)+'_holder') self.gradient_holders.append(placeholder) self.gradients = tf.gradients(self.loss,tvars) optimizer = tf.train.AdamOptimizer(learning_rate=lr) self.update_batch = optimizer.apply_gradients(zip(self.gradient_holders, tvars))
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

エージェントトレーニング。 最後に、エージェントのトレーニングに移りましょう。

 tf.reset_default_graph() #  tensorflow myAgent = agent(lr=1e-2,s_size=4,a_size=2,h_size=8) #  total_episodes = 5000 #   max_ep = 999 update_frequency = 5 init = tf.global_variables_initializer() #  tensorflow with tf.Session() as sess: sess.run(init) i = 0 total_reward = [] total_lenght = [] gradBuffer = sess.run(tf.trainable_variables()) for ix,grad in enumerate(gradBuffer): gradBuffer[ix] = grad * 0 while i < total_episodes: s = env.reset() running_reward = 0 ep_history = [] for j in range(max_ep): #    ,   a_dist = sess.run(myAgent.output,feed_dict={myAgent.state_in:[s]}) a = np.random.choice(a_dist[0],p=a_dist[0]) a = np.argmax(a_dist == a) s1,r,d,_ = env.step(a) #     ep_history.append([s,a,r,s1]) s = s1 running_reward += r if d == True: #  ep_history = np.array(ep_history) ep_history[:,2] = discount_rewards(ep_history[:,2]) feed_dict = {myAgent.reward_holder:ep_history[:,2], myAgent.action_holder:ep_history[:,1], myAgent.state_in:np.vstack(ep_history[:,0])} grads = sess.run(myAgent.gradients, feed_dict=feed_dict) for idx,grad in enumerate(grads): gradBuffer[idx] += grad if i % update_frequency == 0 and i != 0: feed_dict = dictionary = dict(zip(myAgent.gradient_holders, gradBuffer)) _ = sess.run(myAgent.update_batch, feed_dict=feed_dict) for ix,grad in enumerate(gradBuffer): gradBuffer[ix] = grad * 0 total_reward.append(running_reward) total_lenght.append(j) break #   if i % 100 == 0: print(np.mean(total_reward[-100:])) i += 1
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

結果：

 16.0 21.47 25.57 38.03 43.59 53.05 67.38 90.44 120.19 131.75 162.65 156.48 168.18 181.43
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ここで見ることができる完全なJupyterノートブック。 強化学習の研究を続ける次の記事でお会いしましょう！