強化学習：ピクセルから『ポン』をRubyで書いてみた – 1

強化学習について調べていたところ深層強化学習：ピクセルから『ポン』という面白い記事を見つけましたので、Rubyでやってみました。

4500回ぐらい学習させたところの動画。右側の緑が学習させたAI

まだまだ弱いです。。

Andrej karpathy さんの書いた元記事のPythonのコード（何と132行）を、Ruby/NArray にほぼ直訳しています。NArrayとはRubyでは最速の行列計算ライブラリです。
OpenAI Gym の Pong はPython製なので、Rubyとはパイプで通信することにしました。最初は自作の簡易ニューラルネットワークを流用しようと試みましたが、RMSProp や報酬の正規化などやらないと思うように学習してくれないことが判明したので、諦めて直訳しました。コード量はRubyの部分だけでも270行と膨大な量（100行以上のコードは膨大だという思想）になりましたが、理解のためにごちゃごちゃとコメントを入れたりPythonとの接続で増加した分がほとんどです。エンジニアじゃない人が適当に書いてるので間違ってる部分も多数残ってるかと思いますが、何となく学習できていそうなのでよしとします。

Python側のコードです。pong.py

# 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

# 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

# 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

# ２層ニューラルネットワーク
# レイヤーの構成は固定
# 学習率の調節は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 # １ゲームあたり報酬合計の移動平均
    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

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

# 活性化関数 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

# ２層ニューラルネットワーク

# レイヤーの構成は固定

# 学習率の調節は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

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

# 重みをファイルに保存する

def save

model.each do |k, v|

File.write(k.to_s, v.to_string)

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

# 報酬をディスカウントする

# 過去よりも最近の行動を重視する

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 # １ゲームあたり報酬合計の移動平均

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

何時間も実行するとだんだんとメモリの使用量が増えていくので、どこかまずい部分があると思われます。

感想

・policy gradient は実装が簡単で素晴らしい。
・強化学習はディープラーニングを用いた分類器よりも奥が深い技術のような気がする。世間の人がいう「人工知能」はこっちに近い。
・強化学習はアルゴリズムが人間の哲学的な領域にまで踏み込んでくる怖さがどこかにある。
・人工知能じゃ実現できないなと思っていたようなことも案外実現できてしまいそうな気がする。
・ボールそのものよりも相手の動きをみてボールの位置を判定しているフシがあり、敵プレーヤーが代わっても実力を発揮できるか少し疑問。
・やはりRubyでも機械学習は可能なのだ。

感想

コメントを残す コメントをキャンセル

コメントを残すコメントをキャンセル