強化学習をやってみたいから骨組みをつくったり、、、

2021年3月28日2021年4月3日

強化学習を少しかじる

強化学習そういえばやったことないなと思ったので、少しだけかじってみる。
理論的な部分については、作りながら覚える＋わざわざ記事にする必要ないので、、、

とはいえ、巷に転がっているのは振り子が落としたイケナイゲームをなんかゴニョっておしまいで、
え？自作はどーするの？ってなったので、今後自作ができるようなSAMPLEを探してみた。

なんかいい感じに纏まっているものないかと検索していたところ、こんなものを発見したので、これに沿ってやってみる。必要あらば適宜加筆を行う

こちらですが、問題定義は以下の様になっています。かなりシンプルですが、癖がありますねｗ

状況：シャワーの温度最適化
できること：シャワーの温度を１度ずつ変更できる
シャワーは６０秒間で１秒に一回変更できる。

いざ実装

とはいえ、こぴりつつ補足あらば加筆します

実行環境構築

実行環境と強化学習の環境がぶる。。。
今回は特別なものは使いません

!pip install tensorflow==2.3.0
#  OpenAI gym: 強化学習プラットフォーム
!pip install gym
!pip install keras
# keras を　RL(強化学習)で使えるようにするlib
!pip install keras-rl2

# 基礎となる環境
from gym import Env
# 行動などで用いる変数の定義
from gym.spaces import Discrete, Box

#others
import numpy as np
import random

!pip install tensorflow==2.3.0

# OpenAI gym: 強化学習プラットフォーム

!pip install gym

!pip install keras

# keras を　RL(強化学習)で使えるようにするlib

!pip install keras-rl2

# 基礎となる環境

from gym import Env

# 行動などで用いる変数の定義

from gym.spaces import Discrete, Box

#others

import numpy as np

import random

環境をつくる

# Cutom Env
# inherit from openAI Env
class ShowEnv(Env):
  # initialize
  def __init__(self):
    # actionの取りうる値の空間
    # 0: down, 1: stay, 2: up
    self.action_space = Discrete(3)
    # 観測データの取りうる値の空間: 
    # continuous (multiple-value) value.  Used for images dataframe 
    self.observation_space =  Box(low=np.array([0]), high=np.array([100]))
    # state: temprature
    self.state = 38+ random.randint(-1,3)
    # iteration
    self.shower_length = 60

  def step(self, action):
    # Apply action
    # 0 -1 = -1 temperature
    # 1 -1 = 0 
    # 2 -1 = 1 temperature 
    self.state += action -1 
    # Reduce shower length by 1 second
    self.shower_length -= 1 
    
    # Calculate reward
    # very simple
    if self.state >=37 and self.state <=39: 
        reward =1 
    else: 
        reward = -1 
    
    # Check if shower is done ( check finish )
    if self.shower_length <= 0: 
        done = True
    else:
        done = False
    
    # Apply temperature noise
    # Havinng nonise is aid to simulate a real env
    # but it takes longer 
    self.state += random.randint(-1,1)
    # Set placeholder for info
    info = {}
    
    # Return step information
    return self.state, reward, done, info


  # visualize
  def render(self):
    # imfomative
    # フリコなら振り子が揺れるやつを画面表示できる。 
    pass

  # reset env
  def reset(self):
      # Reset shower temperature
      self.state = 38 + random.randint(-3,3)
      # Reset shower time
      self.shower_length = 60 
      return self.state

# Cutom Env

# inherit from openAI Env

class ShowEnv(Env):

# initialize

def __init__(self):

# actionの取りうる値の空間

# 0: down, 1: stay, 2: up

self.action_space = Discrete(3)

# 観測データの取りうる値の空間:

# continuous (multiple-value) value. Used for images dataframe

self.observation_space = Box(low=np.array([0]), high=np.array([100]))

# state: temprature

self.state = 38+ random.randint(-1,3)

# iteration

self.shower_length = 60

def step(self, action):

# Apply action

# 0 -1 = -1 temperature

# 1 -1 = 0

# 2 -1 = 1 temperature

self.state += action -1

# Reduce shower length by 1 second

self.shower_length -= 1

# Calculate reward

# very simple

if self.state >=37 and self.state <=39:

reward =1

else:

reward = -1

# Check if shower is done ( check finish )

if self.shower_length <= 0:

done = True

else:

done = False

# Apply temperature noise

# Havinng nonise is aid to simulate a real env

# but it takes longer

self.state += random.randint(-1,1)

# Set placeholder for info

info = {}

# Return step information

return self.state, reward, done, info

# visualize

def render(self):

# imfomative

# フリコなら振り子が揺れるやつを画面表示できる。

pass

# reset env

def reset(self):

# Reset shower temperature

self.state = 38 + random.randint(-3,3)

# Reset shower time

self.shower_length = 60

return self.state

報酬定義

今回はかなりシンプルに作っていますが、
実際はどのように報酬を定義するか？は非常に重要です。強化学習はこいつを最大化する方向にしか動いてくれないので、見落としの先には、、、

環境の変化

実際には環境はアクションを受けて変化するのが常。
今回はシンプルに、アクションでは無くてアクション叩くたびになんかしらのノイズを加えてこの状況を再現します

 self.state += random.randint(-1,1)

1	self.state += random.randint(-1,1)

一応これで環境はできました。自作には上コードの４つの関数は必須だそうです。次に中身を見てみます。
https://github.com/openai/gym/blob/master/gym/core.py

# Get first env
env = ShowEnv()
# temperature example　
print(env.observation_space.sample())
# ==>  [0.12433]
# Action example
print(env.action_space.sample())
# ==> 0

# Get first env

env = ShowEnv()

# temperature example　

print(env.observation_space.sample())

# ==> [0.12433]

# Action example

print(env.action_space.sample())

# ==> 0

観測のサンプルとアクションのSAMPLEを取得しました。実行毎にことなるはずです.
次に、試しに適当に２５ループ回してみた。

episodes =  25
for episode in range(1, episodes+1):
    state = env.reset()
    done = False
    score = 0 
    while not done:
        #env.render()
        action = env.action_space.sample()
        n_state, reward, done, info = env.step(action)
        score+=reward
    print('Episode:{} Score:{}'.format(episode, score))

episodes = 25

for episode in range(1, episodes+1):

state = env.reset()

done = False

score = 0

while not done:

#env.render()

action = env.action_space.sample()

n_state, reward, done, info = env.step(action)

score+=reward

print('Episode:{} Score:{}'.format(episode, score))

ここでは、ひたすら
(1)行動を実行し、
(2)observation, reward, done, infoの4value tupleを受領、
(3)累積報酬計算
なので、とくに環境がどうだ！と言って省みることはしませんので25回実行しても、良くなりません。

ちゃんとエージェントを考える。

環境と対にあるのがエージェント。
エージェントの脳みそであるモデルをつくる。一昔前だとこの脳味噌がDNNではなくて、正解データに基づいた勾配法とかを使っている

隠れ１層の単純モデルにしよう

import numpy as np
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Flatten
from tensorflow.keras.optimizers import Adam

import numpy as np

from tensorflow.keras.models import Sequential

from tensorflow.keras.layers import Dense, Flatten

from tensorflow.keras.optimizers import Adam

# Create Simple DNN
def build_model(states, actions):
  model = Sequential()
  model.add(Dense(24,activation = 'relu', input_shape=states))
  model.add(Dense(24, activation='relu'))
  model.add(Dense(actions, activation='linear')) #return action
  return model

# create
model = build_model(
    states = env.observation_space.shape,
    actions = env.action_space.n
)

#show summary
model.summary()
Model: "sequential_1"
_________________________________________________________________
Layer (type)                 Output Shape              Param #   
=================================================================
dense_3 (Dense)              (None, 24)                48        
_________________________________________________________________
dense_4 (Dense)              (None, 24)                600       
_________________________________________________________________
dense_5 (Dense)              (None, 3)                 75        
=================================================================
Total params: 723
Trainable params: 723
Non-trainable params: 0
_________________________________________________________________

# Create Simple DNN

def build_model(states, actions):

model = Sequential()

model.add(Dense(24,activation = 'relu', input_shape=states))

model.add(Dense(24, activation='relu'))

model.add(Dense(actions, activation='linear')) #return action

return model

# create

model = build_model(

states = env.observation_space.shape,

actions = env.action_space.n

)

#show summary

model.summary()

Model: "sequential_1"

_________________________________________________________________

Layer (type) Output Shape Param #

=================================================================

dense_3 (Dense) (None, 24) 48

_________________________________________________________________

dense_4 (Dense) (None, 24) 600

_________________________________________________________________

dense_5 (Dense) (None, 3) 75

=================================================================

Total params: 723

Trainable params: 723

Non-trainable params: 0

_________________________________________________________________

モデルをエージェント本体に入れる

rlライブラリから便利なモジュールを入れ込む

#便利なエージェントモジュール
from rl.agents import DQNAgent
from rl.policy import BoltzmannQPolicy
from rl.memory import SequentialMemory

#便利なエージェントモジュール

from rl.agents import DQNAgent

from rl.policy import BoltzmannQPolicy

from rl.memory import SequentialMemory

方策としてBoltzmannQPolicyを用いる。
BoltzmannQPolicyは行動のQ値をソフトマックスにかけて選択する手法。つまりは、-1/0/1のどれが良いのかをQ値をもとに選択してくれる。他にはGreedyQPolicyとか色々あるってのが、実際にコードを見るとわかる

https://github.com/keras-rl/keras-rl/blob/master/rl/policy.py
これによると、自作なら(1)Policyクラスを継承の上で(2)select_acitonメソッドを継承しろと言っている

Experience replay

初めて知ったのだが、強化学習にはExperience replayという最適化手法がある。
過去の経験(=行動とその結果)をサンプリングして学習データとして再利用したりする手法で、時間的に近いと「多分」似た経験だから局所最適解に陥っちゃう。
ってことで、ある程度溜め込んで経験をランダムにピックアップしてミニバッチとして入力させることでうまく行くらしい
今回はSequentialMemoryを使用する
https://github.com/keras-rl/keras-rl/blob/master/rl/memory.py
もう一つクラスが見えますが、これはまたの機会に

def build_agent(model, actions):
  policy = BoltzmannQPolicy()
  memory = SequentialMemory(limit=50000, window_length=1)
  dqn = DQNAgent(model=model, memory=memory, policy=policy, nb_actions=actions, nb_steps_warmup=10, target_model_update=1e-2)
  return dqn

dqn = build_agent(
    model = model,
    actions = env.action_space.n
)
dqn.compile(Adam(lr=1e-3),  metrics=['mae'])#lossは設定不可
dqn.fit(env, nb_steps=500, visualize=False, verbose=1)

def build_agent(model, actions):

policy = BoltzmannQPolicy()

memory = SequentialMemory(limit=50000, window_length=1)

dqn = DQNAgent(model=model, memory=memory, policy=policy, nb_actions=actions, nb_steps_warmup=10, target_model_update=1e-2)

return dqn

dqn = build_agent(

model = model,

actions = env.action_space.n

)

dqn.compile(Adam(lr=1e-3), metrics=['mae'])#lossは設定不可

dqn.fit(env, nb_steps=500, visualize=False, verbose=1)

一気に書いてみた。
compileの部分は最適化手法がいくつか使えるのと、metricがtfと同じように使える。
残念なのはloss関数を定義できない点。
今回はすぐ学習終わらせたいのでepisode=500に設定

Testしてみる

import rl.callbacks
# call back も設定できる
class EpisodeLogger(rl.callbacks.Callback):
    def __init__(self):
        self.observations = {}
        self.rewards = {}
        self.actions = {}

    def on_episode_begin(self, episode, logs):
        self.observations[episode] = []
        self.rewards[episode] = []
        self.actions[episode] = []

    def on_step_end(self, step, logs):
        episode = logs['episode']
        self.observations[episode].append(logs['observation'])
        self.rewards[episode].append(logs['reward'])
        self.actions[episode].append(logs['action'])
cb_ep = EpisodeLogger()

scores = dqn.test(env, nb_episodes=10, visualize=False, callbacks=[cb_ep])
print(np.mean(scores.history['episode_reward']))

import rl.callbacks

# call back も設定できる

class EpisodeLogger(rl.callbacks.Callback):

def __init__(self):

self.observations = {}

self.rewards = {}

self.actions = {}

def on_episode_begin(self, episode, logs):

self.observations[episode] = []

self.rewards[episode] = []

self.actions[episode] = []

def on_step_end(self, step, logs):

episode = logs['episode']

self.observations[episode].append(logs['observation'])

self.rewards[episode].append(logs['reward'])

self.actions[episode].append(logs['action'])

cb_ep = EpisodeLogger()

scores = dqn.test(env, nb_episodes=10, visualize=False, callbacks=[cb_ep])

print(np.mean(scores.history['episode_reward']))

ここにはtensor_callbackみたいなノリでコールバック関数を定義できる。
ちなみに、visualizeはちゃんと環境の法でrenderメソッド定義していないとエラーゲロるので今回はずっとfalse

%matplotlib inline
import matplotlib.pyplot as plt

for obs in cb_ep.observations.values():
  plt.plot(obs)
plt.xlabel("step")
plt.ylabel("pos")

%matplotlib inline

import matplotlib.pyplot as plt

for obs in cb_ep.observations.values():

plt.plot(obs)

plt.xlabel("step")

plt.ylabel("pos")

凡例つければよかったが、10エピソードの中ではいい感じになっている印象はないですね
まぁちゃんと学習すればよいですが

強化学習ではエージェントの脳味噌はモデル（＝DNN）なのんで、
このときのDNNの重みさえ保存してしまえば後で再利用OK

dqn.save_weights('dqn_weights.h5f', overwrite=True)
# dqn.load_weights('dqn_weights.h5f')

1 2	dqn.save_weights('dqn_weights.h5f', overwrite=True) # dqn.load_weights('dqn_weights.h5f')

できた。

こんかい参考にしたsiteたち

えーなんか作りたいねーこれつかって。
その前にやっぱり理論をしりたくなる

機械学習learning, rainforest, tensorflow

Posted by tanico.rikudo

TPUを仕組み理解からTensorFlowで実装まで

最近の気になったものブックマーク(2021-03-28)