強化学習について調べていたところ深層強化学習:ピクセルから『ポン』という面白い記事を見つけましたので、Rubyでやってみました。
4500回ぐらい学習させたところの動画。右側の緑が学習させたAI
まだまだ弱いです。。
Andrej karpathy さんの書いた元記事のPythonのコード (何と132行)を、Ruby/NArray にほぼ直訳しています。NArrayとはRubyでは最速の行列計算ライブラリです。
OpenAI Gym の Pong はPython製なので、Rubyとはパイプで通信することにしました。最初は自作の簡易ニューラルネットワークを流用しようと試みましたが、RMSProp や報酬の正規化などやらないと思うように学習してくれないことが判明したので、諦めて直訳しました。コード量はRubyの部分だけでも270行と膨大な量(100行以上のコードは膨大だという思想)になりましたが、理解のためにごちゃごちゃとコメントを入れたりPythonとの接続で増加した分がほとんどです。エンジニアじゃない人が適当に書いてるので間違ってる部分も多数残ってるかと思いますが、何となく学習できていそうなのでよしとします。
Python側のコードです。pong.py
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
# coding: utf-8 # The original source is available at https://gist.github.com/karpathy/a4166c7fe253700972fcbc77e4ea32c5 import numpy as np import gym import sys render = False # 画面を表示するかどうか判定 render_flag = sys.stdin.readline() if "true" in render_flag: render = True D = 80 * 80 # 前処理 def prepro(I): """ prepro 210x160x3 uint8 frame into 6400 (80x80) 1D float vector """ I = I[35:195] # crop I = I[::2,::2,0] # downsample by factor of 2 I[I == 144] = 0 # erase background (background type 1) I[I == 109] = 0 # erase background (background type 2) I[I != 0] = 1 # everything else (paddles, ball) just set to 1 return I.astype(np.float).ravel() env = gym.make("Pong-v0") observation = env.reset() prev_x = None # used in computing the difference frame while True: if render: env.render() cur_x = prepro(observation) x = cur_x - prev_x if prev_x is not None else np.zeros(D) prev_x = cur_x # Ruby に向けて(observation)を書き込み print(x.tobytes()) # Ruby の返信を読み込む action_ruby = sys.stdin.readline() action_ruby = int(action_ruby) observation, reward, done, info = env.step(action_ruby) # Ruby に向けてreward,doneを送信(infoは無視) print(reward) print(done) if done: observation = env.reset() # reset env prev_x = None |
次にRuby側のコードです。rl.rb
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 |
# OpenAI Gym/ https://gym.openai.com/envs/Pong-v0 # Written in Ruby with Numo/NArray. # This program is a translation of pg-pong.py written by Andrej karpathy # The original source is available at https://gist.github.com/karpathy/a4166c7fe253700972fcbc77e4ea32c5 # 行列計算ライブラリ require 'numo/narray' # OpenBlasで計算を高速化 require 'numo/linalg' # 標準入力・標準出力にパイプをつなぐ require 'open3' # コマンドラインのオプション require 'optparse' # ターミナルに色をつける require 'rainbow' # import numpy as np と同じ N = Numo D = 80 * 80 # 入力層のユニット数 H = 200 # 隠れ層のユニット数 BATCH_SIZE = 10 # いくつのゲームで重み更新するか LEARNING_RATE = 1e-3 # 学習率 GAMMA = 0.99 # 割引率(discount factor) DECAY_RATE = 0.99 # RMSProp の係数 render = false resume = false # オプション解析 opt = OptionParser.new opt.on('-d', '--display', 'ゲーム画面を表示する') {|v| render = v } opt.on('-r', '--resume', '再開(ファイルから重みを読み込む)') {|v| resume = v } opt.parse!(ARGV) RENDER = render # ゲーム画面表示するか RESUME = resume # 重み読み込むか # Pongのクラス # 強化学習における"環境" class PyPong def initialize(render) # OpenAI Gymを実行する # Python のエラーを端末に表示するようにpopen3ではなくopen2を使用する # 標準出力をバッファするとタイムラグが生じるので -u オプション をつける @python_stdin, @python_stdout = *Open3.popen2('python', '-u', 'pong.py') # 画像を表示するかどうかをPython側に通知 self.puts render # puts は下に定義 end # "状態観測"を(Python側で前処理済みのゲーム画面)を取得する def get_observation N::DFloat.from_string(readline) # Numo::DFloat [6400] end # "報酬"(reward)を取得する def get_reward readline.to_f # Float end # ゲームの終了フラグを取得する def get_done readline.chomp == "True" # true/false end # OpenAI Gym の標準入力へ書き込む def puts(x) @python_stdin.puts(x) end # OpenAI Gym の標準出力を読み込む def readline @python_stdout.readline end end # 活性化関数 sigmoid def sigmoid(x) 1.0 / (1.0 + N::NMath.exp(-x)) end # OpenBlas使った方が高速 def matmul(a,b) N::Linalg.matmul(a,b) # a.dot(b) numo/linalg 使えない環境の時など。こんなんでちゃんと動くかは不明 end # 2層ニューラルネットワーク # レイヤーの構成は固定 # 学習率の調節はRMSPropを使用している class PongNet def initialize(d,h) @model = { # 2016/8/21 githubから最新の numo/narray をインストールする必要あり w1: N::DFloat.new(h,d).rand_norm / Math.sqrt(d), # 重みw1 w2: N::DFloat.new(h).rand_norm / Math.sqrt(h) # 重みw2 # この初期化を Xavier initialization と言うらしい } @grad_buffer, @rmsprop_cache = {}, {} # 初期化 @model.each do |k, v| @grad_buffer[k] = N::DFloat.zeros(v.shape) @rmsprop_cache[k] = N::DFloat.zeros(v.shape) end end attr_accessor :model, :grad_buffer, :rmsprop_cache # 順伝播 def policy_forward(x) h = matmul(model[:w1], x) h[h<0] = 0 # ReLU logp = matmul(model[:w2], h) p = sigmoid(logp) [p, h] end # 逆伝播 # こちらはforwardと異なりミニバッチであることに要注意 def policy_backward(epx, eph, epdlogp) dw2 = matmul(eph.transpose, epdlogp).flatten dh = epdlogp.expand_dims(1) * model[:w2].flatten #numpy.outer のかわり dh[eph <= 0] = 0 dw1 = matmul(dh.transpose, epx) grad_buffer[:w1] += dw1 grad_buffer[:w2] += dw2 {w1: dw1, w2: dw2} # 実際には戻り値使わない(上でbufferに溜め込んでるから) end # 重みを更新する # RMSProp という手法 # ・ガチャガチャと朝三暮四で更新される項目は、更新を抑制させる # ・あまり更新されない項目が手を上げた時は、優先的に更新させる # というイメージ def update model.each do |k,v| g = grad_buffer[k] rmsprop_cache[k] = DECAY_RATE * rmsprop_cache[k] + (1 - DECAY_RATE) * (g**2) model[k] += LEARNING_RATE * g / (N::NMath.sqrt(rmsprop_cache[k]) + 1e-5) grad_buffer[k] = N::DFloat.zeros(v.shape) end end # 重みをファイルに保存する def save model.each do |k, v| File.write(k.to_s, v.to_string) end end # 重みをファイルから読み込みこむ def resume model.each do |k, v| data = File.read(k.to_s) model[k] = N::DFloat.from_string(data).reshape(*v.shape) end end end # 報酬をディスカウントする # 過去よりも最近の行動を重視する def discount_rewards(r) discounted_r = N::DFloat.zeros(r.shape) running_add = 0 r.size.times do |t1| t = r.size - t1 - 1 running_add = 0 if r[t] != 0 # 得点が入った段階で初期化 running_add = running_add * GAMMA + r[t] discounted_r[t] = running_add end discounted_r end # ゲーム pong を生成。強化学習における"環境" @pong = PyPong.new(RENDER) # ニューラルネットを生成 @nn = PongNet.new(D,H) # RESUMEが真なら、ニューラルネットに重みを読み込ませる @nn.resume if RESUME # 記録用 # 本当はhsはPongNetクラスに組み込むべきか? # xs : 入力層の状態 # hs : 隠れ層の状態 # dlogps : 教師ラベルもどきとの勾配 # drs : 報酬 xs,hs,dlogps,drs = [],[],[],[] running_reward = nil # 今までの報酬の移動平均。表示用。 reward_sum = 0 # 現在のゲームの報酬の合計。表示用。 episode_number = 0 # ゲームの回数。表示用。 # メインループ loop do # 環境から観測状態を取得する x = @pong.get_observation # aprob 行動の確率 h 隠れ層 aprob, h = @nn.policy_forward(x) # 行動選択 "方策" # ときどき普段とは違う行動をしてみるのが強化学習 action = (rand < aprob) ? 2 : 3 # 教師ラベルもどき y = (action == 2) ? 1 : 0 # 行動する @pong.puts action # 環境から報酬を取得する reward = @pong.get_reward # エピソードの終了を判定 done = @pong.get_done # 各種記録する xs << x.to_a # 入力層の状態 hs << h.to_a # 隠れ層の状態 dlogps << (y - aprob) # 教師ラベルもどきとの誤差 drs << reward # 報酬 reward_sum += reward # 1ゲーム中の報酬の合計 # ゲームセットしたタイミングで逆伝播させる if done episode_number += 1 # Ruby配列 → NArray配列 に変換 # numpy.vstackがないのでRuby配列を経由してみました # もっと簡単な方法があるかも xs, hs, dlogps, drs = [xs, hs, dlogps, drs].map{|ar| N::DFloat[*ar]} discounted_epr = discount_rewards(drs) # 正規化的な処理をする discounted_epr -= drs.mean discounted_epr /= drs.stddev dlogps *= discounted_epr # 初期の連戦連敗の状況にあっても学習するための工夫か。 # 大失敗ばかりの時は、ラリーもプラス評価になる。 # しかし、連戦連勝になった時は、逆に単なるラリーがマイナス評価に。 # ミニバッチで逆伝播させる @nn.policy_backward(xs, hs, dlogps) # エピソードがBATCH_SIZEに達した時に、重みを更新する @nn.update if episode_number % BATCH_SIZE == 0 # 端末に成績を表示する running_reward ||= reward_sum # 初回用 running_reward = running_reward * 0.99 + reward_sum * 0.01 # 1ゲームあたり報酬合計の移動平均 puts "環境をリセットします。報酬: " + Rainbow(reward_sum).purple + " 報酬の移動平均: " + Rainbow(running_reward).purple # 重みを保存する @nn.save if episode_number % 100 == 0 reward_sum = 0 xs,hs,dlogps,drs = [],[],[],[] end # 報酬が発生(どちらかが得点)したときに端末に表示する l = " ゲーム " + Rainbow(episode_number).green + " 報酬: " if reward == 1.0 puts l + Rainbow(" #{reward}").white.bg(:red) elsif reward == -1.0 puts l + Rainbow(reward).white.bg(:blue) end end |
何時間も実行するとだんだんとメモリの使用量が増えていくので、どこかまずい部分があると思われます。
感想
・policy gradient は実装が簡単で素晴らしい。
・強化学習はディープラーニングを用いた分類器よりも奥が深い技術のような気がする。世間の人がいう「人工知能」はこっちに近い。
・強化学習はアルゴリズムが人間の哲学的な領域にまで踏み込んでくる怖さがどこかにある。
・人工知能じゃ実現できないなと思っていたようなことも案外実現できてしまいそうな気がする。
・ボールそのものよりも相手の動きをみてボールの位置を判定しているフシがあり、敵プレーヤーが代わっても実力を発揮できるか少し疑問。
・やはりRubyでも機械学習は可能なのだ。