強化学習 AlphaZero 6 (三目並べ アルファベータ法)

三目並べをアルファベータ法で解いていきます。

アルファベータ法は、ミニマックス法から計算しなくても同じ計算結果になる部分を読まない処理(枝刈り)を実装し、高速化したものです。探索手順はミニマックス法と同じです。

まず、三目並べを実装します。(前回のミニマックス法で定義したものと同じです。)

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
# 三目並べの作成
import random

# ゲームの状態
class State:
# 初期化
def __init__(self, pieces=None, enemy_pieces=None):
# 石の配置
self.pieces = pieces if pieces != None else [0] * 9
self.enemy_pieces = enemy_pieces if enemy_pieces != None else [0] * 9

# 石の数の取得
def piece_count(self, pieces):
count = 0
for i in pieces:
if i == 1:
count += 1
return count

# 負けかどうか
def is_lose(self):
# 3並びかどうか
def is_comp(x, y, dx, dy):
for k in range(3):
if y < 0 or 2 < y or x < 0 or 2 < x or \
self.enemy_pieces[x+y*3] == 0:
return False
x, y = x+dx, y+dy
return True

# 負けかどうか
if is_comp(0, 0, 1, 1) or is_comp(0, 2, 1, -1):
return True
for i in range(3):
if is_comp(0, i, 1, 0) or is_comp(i, 0, 0, 1):
return True
return False

# 引き分けかどうか
def is_draw(self):
return self.piece_count(self.pieces) + self.piece_count(self.enemy_pieces) == 9

# ゲーム終了かどうか
def is_done(self):
return self.is_lose() or self.is_draw()

# 次の状態の取得
def next(self, action):
pieces = self.pieces.copy()
pieces[action] = 1
return State(self.enemy_pieces, pieces)

# 合法手のリストの取得
def legal_actions(self):
actions = []
for i in range(9):
if self.pieces[i] == 0 and self.enemy_pieces[i] == 0:
actions.append(i)
return actions

# 先手かどうか
def is_first_player(self):
return self.piece_count(self.pieces) == self.piece_count(self.enemy_pieces)

# 文字列表示
def __str__(self):
ox = ('o', 'x') if self.is_first_player() else ('x', 'o')
str = ''
for i in range(9):
if self.pieces[i] == 1:
str += ox[0]
elif self.enemy_pieces[i] == 1:
str += ox[1]
else:
str += '-'
if i % 3 == 2:
str += '\n'
return str

ミニマックス法で状態価値を計算します。(前回と同じです。)

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
# ミニマックス法で状態価値計算
def mini_max(state):
# 負けは状態価値-1
if state.is_lose():
return -1

# 引き分けは状態価値0
if state.is_draw():
return 0

# 合法手の状態価値の計算
best_score = -float('inf')
for action in state.legal_actions():
score = -mini_max(state.next(action))
if score > best_score:
best_score = score

# 合法手の状態価値の最大値を返す
return best_score

# ミニマックス法で行動選択
def mini_max_action(state):
# 合法手の状態価値の計算
best_action = 0
best_score = -float('inf')
for action in state.legal_actions():
score = -mini_max(state.next(action))
if score > best_score:
best_action = action
best_score = score

# 合法手の状態価値の最大値を持つ行動を返す
return best_action

アルファベータ法で状態価値を計算計算します。

引数のalphaは親ノードの親ノード(つまり自分)のベストスコアで、betaは親ノードのベストスコアです。

  • alphaは自分のベストスコアなので、alpha以上のスコアを取得したらalphaを更新します。
  • betaは親ノードのベストスコアなのでalpha以下だったら探索を終了します。

メソッドを再帰的に呼び出すとき、alphaとbetaを左右入れ替えてるのがポイントです。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# アルファベータ法で状態価値計算
def alpha_beta(state, alpha, beta):
# 負けは状態価値-1
if state.is_lose():
return -1

# 引き分けは状態価値0
if state.is_draw():
return 0

# 合法手の状態価値の計算
for action in state.legal_actions():
score = -alpha_beta(state.next(action), -beta, -alpha)
if score > alpha:
alpha = score

# 現ノードのベストスコアが親ノードを超えたら探索終了
if alpha >= beta:
return alpha

# 合法手の状態価値の最大値を返す
return alpha

アルファベータ法で状態に応じて行動を返す関数を実装します。
ミニマックス法との違いは次の2点です

  • best_scoreをalphaとする。
  • betaの初期値に-float(‘inf’)を設定する。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# アルファベータ法で行動選択
def alpha_beta_action(state):
# 合法手の状態価値の計算
best_action = 0
alpha = -float('inf')
str = ['','']
for action in state.legal_actions():
score = -alpha_beta(state.next(action), -float('inf'), -alpha)
if score > alpha:
best_action = action
alpha = score

str[0] = '{}{:2d},'.format(str[0], action)
str[1] = '{}{:2d},'.format(str[1], score)
print('action:', str[0], '\nscore: ', str[1], '\n')

# 合法手の状態価値の最大値を持つ行動を返す
return best_action

最後に、アルファベータ法とミニマックス法で対戦します。
先手はアルファベータ法、後手はミニマックス法を使います。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# アルファベータ法とミニマックス法の対戦

# 状態の生成
state = State()

# ゲーム終了までのループ
while True:
# ゲーム終了時
if state.is_done():
break

# 行動の取得
if state.is_first_player():
print('----- アルファベータ法で行動選択 -----')
action = alpha_beta_action(state)
else:
print('----- ミニマックス法で行動選択 -----')
action = mini_max_action(state)

# 次の状態の取得
state = state.next(action)

# 文字列表示
print(state)

結果は次の通りです。

結果
アルファベータ法の方が実行速度が上がりましたが、強さは互角になりました。

(Google Colaboratoryで動作確認しています。)

参考

AlphaZero 深層学習・強化学習・探索 人工知能プログラミング実践入門 サポートページ

強化学習 AlphaZero 5 (三目並べ ミニマックス法)

三目並べをミニマックス法で解いていきます。

ミニマックス法は、自分は自分にとって最善手を選び、相手は自分にとって最悪手を選ぶと仮定して最善手を探すアルゴリズムです。

ポイントは次の通りです。

  • next()は行動に応じて次の状態を取得する。
  • 行動は石を配置するマスの位置を0~8で指定する。
  • legal_actions()は選択可能な行動を取得します。
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
# 三目並べの作成
import random

# ゲームの状態
class State:
# 初期化
def __init__(self, pieces=None, enemy_pieces=None):
# 石の配置
self.pieces = pieces if pieces != None else [0] * 9
self.enemy_pieces = enemy_pieces if enemy_pieces != None else [0] * 9

# 石の数の取得
def piece_count(self, pieces):
count = 0
for i in pieces:
if i == 1:
count += 1
return count

# 負けかどうか
def is_lose(self):
# 3並びかどうか
def is_comp(x, y, dx, dy):
for k in range(3):
if y < 0 or 2 < y or x < 0 or 2 < x or \
self.enemy_pieces[x+y*3] == 0:
return False
x, y = x+dx, y+dy
return True

# 負けかどうか
if is_comp(0, 0, 1, 1) or is_comp(0, 2, 1, -1):
return True
for i in range(3):
if is_comp(0, i, 1, 0) or is_comp(i, 0, 0, 1):
return True
return False

# 引き分けかどうか
def is_draw(self):
return self.piece_count(self.pieces) + self.piece_count(self.enemy_pieces) == 9

# ゲーム終了かどうか
def is_done(self):
return self.is_lose() or self.is_draw()

# 次の状態の取得
def next(self, action):
pieces = self.pieces.copy()
pieces[action] = 1
return State(self.enemy_pieces, pieces)

# 合法手のリストの取得
def legal_actions(self):
actions = []
for i in range(9):
if self.pieces[i] == 0 and self.enemy_pieces[i] == 0:
actions.append(i)
return actions

# 先手かどうか
def is_first_player(self):
return self.piece_count(self.pieces) == self.piece_count(self.enemy_pieces)

# 文字列表示
def __str__(self):
ox = ('o', 'x') if self.is_first_player() else ('x', 'o')
str = ''
for i in range(9):
if self.pieces[i] == 1:
str += ox[0]
elif self.enemy_pieces[i] == 1:
str += ox[1]
else:
str += '-'
if i % 3 == 2:
str += '\n'
return str

ランダムで行動選択する関数を実装します。
leagl_actions()で合法手を取得し、その中からランダムに手を選びます。

1
2
3
4
# ランダムで行動選択
def random_action(state):
legal_actions = state.legal_actions()
return legal_actions[random.randint(0, len(legal_actions)-1)]

ランダムとランダムで三目並べを対戦させます。
ゲーム終了まで行動の取得と次の状態の取得を繰り返します。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# ランダムとランダムで対戦

# 状態の生成
state = State()

# ゲーム終了までのループ
while True:
# ゲーム終了時
if state.is_done():
break;

# 行動の取得
action = random_action(state)

# 次の状態の取得
state = state.next(action)

# 文字列表示
print(state)
print()


結果

ミニマックス法で状態価値の計算を行います。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# ミニマックス法で状態価値計算
def mini_max(state):
# 負けは状態価値-1
if state.is_lose():
return -1

# 引き分けは状態価値0
if state.is_draw():
return 0

# 合法手の状態価値の計算
best_score = -float('inf')
for action in state.legal_actions():
score = -mini_max(state.next(action))
if score > best_score:
best_score = score

# 合法手の状態価値の最大値を返す
return best_score

ミニマックス法で状態に応じて行動を返す関数を実装します。
Stateを渡すと行動(石を置くマス:0~8)を返します。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# ミニマックス法で行動選択
def mini_max_action(state):
# 合法手の状態価値の計算
best_action = 0
best_score = -float('inf')
str = ['','']
for action in state.legal_actions():
score = -mini_max(state.next(action))
if score > best_score:
best_action = action
best_score = score

str[0] = '{}{:2d},'.format(str[0], action)
str[1] = '{}{:2d},'.format(str[1], score)
print('action:', str[0], '\nscore: ', str[1], '\n')

# 合法手の状態価値の最大値を持つ行動を返す
return best_action

ミニマックス法とランダムで対戦します。
先手はmini_max_action()、後手はrandmo_action()を使います。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# ミニマックス法とランダムで対戦

# 状態の生成
state = State()

# ゲーム終了までのループ
while True:
# ゲーム終了時
if state.is_done():
break

# 行動の取得
if state.is_first_player():
action = mini_max_action(state)
else:
action = random_action(state)

# 次の状態の取得
state = state.next(action)

# 文字列表示
print(state)
print()

実行ごとに結果がかわるので複数回実行してみてください。
ただランダムが勝つことはありません。

結果

(Google Colaboratoryで動作確認しています。)

参考

AlphaZero 深層学習・強化学習・探索 人工知能プログラミング実践入門 サポートページ

強化学習 AlphaZero 4 (カートポール Deep Q-network)

カートポールをDQN(Deep Q-network)で解いていきます。

カートポールは棒を倒さないようにバランスをとるゲームです。

今回の強化学習のポイントは下記です。

  • 目的は棒を倒さないようにバランスをとること。
  • 1エピソードは棒を倒すまで。
  • 状態は「カートの位置」「カートの速度」「棒の角度」「棒の角速度」の4種類。
  • 行動は「カートの左移動」「カートの右移動」の2種類。
  • 報酬はエピソード完了時に190ステップ以上で+1.
  • パラメータ更新間隔は行動1回ごと。

DQNでは行動価値関数を表形式ではなく、ニューラルネットワークで表現します。
またそれ以外にもDQNでは次の4つの工夫がされています。

  • Experience Replayd(経験を貯めておきあとでランダムに学習する)
  • Fixed Target Q-Network(更新計算のためにニューラルネットワークをもう1つ作る)
  • Reward Clipping(報酬のスケールを-1,0,1に統一する)
  • Huber Loss(誤差が大きい場合でも安定している関数huberを使う)

まずはGoogle Colabaratoryで動作確認するために環境を構築します。
tensorflowのバージョンは1.13.1が必須のため、次のコマンドを実行します。

1
2
!pip uninstall tensorflow -y
!pip install tensorflow==1.13.1

カートポールの様子をアニメーションで確認するために、ディスプレイの設定を行います。

1
2
3
4
5
6
7
8
9
10
11
12
# ディスプレイ設定のインストール
!apt-get -qq -y install xvfb freeglut3-dev ffmpeg> /dev/null
!pip install pyglet==1.3.2
!pip install pyopengl
!pip install pyvirtualdisplay

# ディスプレイ設定の適用
from pyvirtualdisplay import Display
import os
disp = Display(visible=0, size=(1024, 768))
disp.start()
os.environ['DISPLAY'] = ':' + str(disp.display) + '.' + str(disp.screen)

ここでいったんランタイムの再接続が必要になります。

次に必要なパッケージをインポートします。

1
2
3
4
5
6
7
8
# パッケージのインポート
import gym
import numpy as np
from keras.models import Sequential
from keras.layers import Dense
from keras.optimizers import Adam
from collections import deque
from tensorflow.losses import huber_loss

パラメータを準備します。パラメータの内容はコメントを参照して下さい。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# パラメータの準備
NUM_EPISODES = 500 # エピソード数
MAX_STEPS = 200 # 最大ステップ数
GAMMA = 0.99 # 時間割引率
WARMUP = 10 # 無操作ステップ数

# 探索パラメータ
E_START = 1.0 # εの初期値
E_STOP = 0.01 # εの最終値
E_DECAY_RATE = 0.001 # εの減衰率

# メモリパラメータ
MEMORY_SIZE = 10000 # 経験メモリのサイズ
BATCH_SIZE = 32 # バッチサイズ

行動評価関数となるニューラルネットワークモデルを作成します。
全結合層を4つ重ねたモデルで、入力は状態数、出力数は行動数になります。

1
2
3
4
5
6
7
8
9
10
11
12
13
# 行動価値関数の定義
class QNetwork:
# 初期化
def __init__(self, state_size, action_size):
# モデルの作成
self.model = Sequential()
self.model.add(Dense(16, activation='relu', input_dim=state_size))
self.model.add(Dense(16, activation='relu'))
self.model.add(Dense(16, activation='relu'))
self.model.add(Dense(action_size, activation='linear'))

# モデルのコンパイル
self.model.compile(loss=huber_loss, optimizer=Adam(lr=0.001))

経験メモリを定義します。メモリの内容は「状態」「行動」「報酬」「次の状態」の4種類となります。
dequeはmaxlen以上の要素を追加すると自動的に先頭の要素が削除される関数です。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 経験メモリの定義
class Memory():
# 初期化
def __init__(self, memory_size):
self.buffer = deque(maxlen=memory_size)

# 経験の追加
def add(self, experience):
self.buffer.append(experience)

# バッチサイズ分の経験をランダムに取得
def sample(self, batch_size):
idx = np.random.choice(np.arange(len(self.buffer)), size=batch_size, replace=False)
return [self.buffer[i] for i in idx]

# 経験メモリのサイズ
def __len__(self):
return len(self.buffer)

環境を作成します。OpenAI Gymのgym.make()を使用します。
さらに、定義したQNetworkクラスとMemoryクラスを利用して、2つのニューラルネットワークと経験メモリを作成します。

1
2
3
4
5
6
7
8
9
10
11
12
13
# 環境の作成
env = gym.make('CartPole-v0')
state_size = env.observation_space.shape[0] # 行動数
action_size = env.action_space.n # 状態数

# main-networkの作成
main_qn = QNetwork(state_size, action_size)

# target-networkの作成
target_qn = QNetwork(state_size, action_size)

# 経験メモリの作成
memory = Memory(MEMORY_SIZE)

学習を開始します。

  1. 環境をリセットする。
  2. 定義したエピソード数だけエピソードを繰り返す。
  3. target-networkを更新する。
  4. 1エピソード分、ゲーム終了まで処理を実行する。
    1. εを減らす。
    2. ランダムまたは行動価値関数に従って、行動を取得する。
    3. 行動に応じて状態を報酬を得る。
    4. エピソード完了時に190ステップ以上で報酬+1、成功回数に1加算する。
    5. エピソード完了でないとき報酬に0を指定し、経験メモリに経験を追加する。
    6. 行動価値を更新する。
      1. ニューラルネットワークの入力と出力を準備する。
      2. バッチサイズ分の経験をランダムに取得する。
      3. ニューラルネットワークの入力と出力を生成する。
      4. 行動価値を更新する。
  5. エピソード完了時のログを表示する。
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
# 学習の開始

# 環境の初期化
state = env.reset()
state = np.reshape(state, [1, state_size])

# エピソード数分のエピソードを繰り返す
total_step = 0 # 総ステップ数
success_count = 0 # 成功数
for episode in range(1, NUM_EPISODES+1):
step = 0 # ステップ数

# target-networkの更新
target_qn.model.set_weights(main_qn.model.get_weights())

# 1エピソードのループ
for _ in range(1, MAX_STEPS+1):
step += 1
total_step += 1

# εを減らす
epsilon = E_STOP + (E_START - E_STOP)*np.exp(-E_DECAY_RATE*total_step)

# ランダムな行動を選択
if epsilon > np.random.rand():
action = env.action_space.sample()
# 行動価値関数で行動を選択
else:
action = np.argmax(main_qn.model.predict(state)[0])

# 行動に応じて状態と報酬を得る
next_state, _, done, _ = env.step(action)
next_state = np.reshape(next_state, [1, state_size])

# エピソード完了時
if done:
# 報酬の指定
if step >= 190:
success_count += 1
reward = 1
else:
success_count = 0
reward = 0

# 次の状態に状態なしを代入
next_state = np.zeros(state.shape)

# 経験の追加
if step > WARMUP:
memory.add((state, action, reward, next_state))
# エピソード完了でない時
else:
# 報酬の指定
reward = 0

# 経験の追加
if step > WARMUP:
memory.add((state, action, reward, next_state))

# 状態に次の状態を代入
state = next_state

# 行動価値関数の更新
if len(memory) >= BATCH_SIZE:
# ニューラルネットワークの入力と出力の準備
inputs = np.zeros((BATCH_SIZE, 4)) # 入力(状態)
targets = np.zeros((BATCH_SIZE, 2)) # 出力(行動ごとの価値)

# バッチサイズ分の経験をランダムに取得
minibatch = memory.sample(BATCH_SIZE)

# ニューラルネットワークの入力と出力の生成
for i, (state_b, action_b, reward_b, next_state_b) in enumerate(minibatch):

# 入力に状態を指定
inputs[i] = state_b

# 採った行動の価値を計算
if not (next_state_b == np.zeros(state_b.shape)).all(axis=1):
target = reward_b + GAMMA * np.amax(target_qn.model.predict(next_state_b)[0])
else:
target = reward_b

# 出力に行動ごとの価値を指定
targets[i] = main_qn.model.predict(state_b)
targets[i][action_b] = target # 採った行動の価値

# 行動価値関数の更新
main_qn.model.fit(inputs, targets, epochs=1, verbose=0)

# エピソード完了時
if done:
# エピソードループを抜ける
break

# エピソード完了時のログ表示
print('エピソード: {}, ステップ数: {}, epsilon: {:.4f}'.format(episode, step, epsilon))

# 5回連続成功で学習終了
if success_count >= 5:
break

# 環境のリセット
state = env.reset()
state = np.reshape(state, [1, state_size])

最大ステップ数の200に少しずつ近づいていきます。
結果(エピソード1から20)
結果(エピソード90から114)

アニメーションフレームを作成します。

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
# 評価
frames = [] # アニメーションフレーム

# 環境のリセット
state = env.reset()
state = np.reshape(state, [1, state_size])

# 1エピソードのループ
step = 0 # ステップ数
for step in range(1, MAX_STEPS+1):
step += 1

# アニメーションフレームの追加
frames.append(env.render(mode='rgb_array'))

# 最適行動を選択
action = np.argmax(main_qn.model.predict(state)[0])

# 行動に応じて状態と報酬を得る
next_state, reward, done, _ = env.step(action)
next_state = np.reshape(next_state, [1, state_size])

# エピソード完了時
if done:
# 次の状態に状態なしを代入
next_state = np.zeros(state.shape)

# エピソードループを抜ける
break
else:
# 状態に次の状態を代入
state = next_state

# エピソード完了時のログ表示
print('ステップ数: {}'.format(step))

結果

アニメーションフレームをアニメーションに変換して最終動作を確認します。

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
# JSAnimationのインストール
!pip install JSAnimation

# パッケージのインポート
import matplotlib.pyplot as plt
from matplotlib import animation
from JSAnimation.IPython_display import display_animation
from IPython.display import display

# アニメーション再生の定義
def display_frames_as_gif(frames):
plt.figure(figsize=(frames[0].shape[1]/72.0, frames[0].shape[0]/72.0), dpi=72)
patch = plt.imshow(frames[0])
plt.axis('off')

# アニメーションの定期処理
def animate(i):
patch.set_data(frames[i])

# アニメーション再生
anim = animation.FuncAnimation(plt.gcf(), animate, frames=len(frames), interval=50)
display(display_animation(anim, default_mode='loop'))

# アニメーション再生
display_frames_as_gif(frames)

最終的にうまくポールのバランスととることができています。

(Google Colaboratoryで動作確認しています。)

参考

AlphaZero 深層学習・強化学習・探索 人工知能プログラミング実践入門 サポートページ

強化学習 AlphaZero 3 (迷路ゲーム Sarsa/Q-learning)

迷路ゲームをSarsaとQ-learningで解いていきます。

SarsaとQ-learningは価値反復法と呼ばれ、ある行動をとるたびに状態価値を増やす手法です。
(Sarsaは収束が遅い一方局所解に陥りにくく、Q-learningは収束が早い一方で局所解に陥りやすいと言われています。)

今回の強化学習のポイントは下記です。

  • 目的はゴールすること。
  • 状態は位置。
  • 行動は上下左右の4種類。
  • 報酬はゴールしたら+1。
  • パラメータl更新間隔は行動1回ごと。

価値反復法の学習手順は次の通りです。

  1. ランダム行動の準備する。
  2. 行動価値関数を準備する。
  3. 行動に従って、次の状態を取得する。
  4. ランダムまたは行動価値関数に従って、行動を取得する。
  5. 行動価値関数を更新する。
  6. ゴールするまで3~5を繰り返す。

まずは必要なパッケージをインポートします。

1
2
3
4
5
6
# パッケージのインポート
import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline
from matplotlib import animation
from IPython.display import HTML

迷路を作成します。

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
# 迷路の作成
fig = plt.figure(figsize=(3, 3))

# 壁
plt.plot([0, 3], [3, 3], color='k')
plt.plot([0, 3], [0, 0], color='k')
plt.plot([0, 0], [0, 2], color='k')
plt.plot([3, 3], [1, 3], color='k')
plt.plot([1, 1], [1, 2], color='k')
plt.plot([2, 3], [2, 2], color='k')
plt.plot([2, 1], [1, 1], color='k')
plt.plot([2, 2], [0, 1], color='k')

# 数字
for i in range(3):
for j in range(3):
plt.text(0.5+i, 2.5-j, str(i+j*3), size=20, ha='center', va='center')

# 円
circle, = plt.plot([0.5], [2.5], marker='o', color='#d3d3d3', markersize=40)

# 目盛りと枠の非表示
plt.tick_params(axis='both', which='both', bottom='off', top= 'off',
labelbottom='off', right='off', left='off', labelleft='off')
plt.box('off')

結果

ランダム行動を準備します。
パラメータθと方策を作成し、np.random.choice()でばらつきを加えて作成します。
前回の方策勾配法とは違い、今回パラメータθと方策の更新はありません。

1
2
3
4
5
6
7
8
9
10
# パラメータθの初期値の準備
theta_0 = np.array([
[np.nan, 1, 1, np.nan], # 0
[np.nan, 1, 1, 1], # 1
[np.nan, np.nan, np.nan, 1], # 2
[1, np.nan, 1, np.nan], # 3
[1, 1, np.nan, np.nan], # 4
[np.nan, np.nan, 1, 1], # 5
[1, 1, np.nan, np.nan], # 6
[np.nan, np.nan, np.nan, 1]]) # 7

パラメータθから方策のへの変換には単純な割合計算を行います。

1
2
3
4
5
6
7
8
9
# パラメータθを方策に変換
def get_pi(theta):
# 割合の計算
[m, n] = theta.shape
pi = np.zeros((m, n))
for i in range(0, m):
pi[i, :] = theta[i, :] / np.nansum(theta[i, :])
pi = np.nan_to_num(pi)
return pi

パラメータθの初期値を変換します。列の合計が1になっています。

1
2
3
# パラメータθの初期値を方策に変換
pi_0 = get_pi(theta_0)
print(pi_0)

結果

行動価値関数を表形式で準備します。
学習前は行動価値関数は不明なので、移動可能が方向は乱数、移動不可の方向は欠損値(np.nan)で初期化します。

1
2
3
4
# 行動価値関数の準備
[a, b] = theta_0.shape
Q = np.random.rand(a, b) * theta_0 * 0.01
print(Q)

結果

行動に従って次の状態を取得する関数を作成します。

1
2
3
4
5
6
7
8
9
10
# 行動に従って次の状態を取得
def get_s_next(s, a):
if a == 0: # 上
return s - 3
elif a == 1: # 右
return s + 1
elif a == 2: # 下
return s + 3
elif a == 3: # 左
return s - 1

確率ε(0以上1以下)でランダムに行動し、確率1-εで行動価値関数の行動を選択します。

1
2
3
4
5
6
7
8
# ランダムまたは行動価値関数に従って行動を取得
def get_a(s, Q, epsilon, pi_0):
if np.random.rand() < epsilon:
# ランダムに行動を選択
return np.random.choice([0, 1, 2, 3], p=pi_0[s])
else:
# 行動価値関数で行動を選択
return np.nanargmax(Q[s])

Sarsaの行動価値関数の更新は、パラメータθに学習係数とTD誤差を掛けた値を加算します。

1
2
3
4
5
6
7
8
9
10
# Sarsaによる行動価値関数の更新
def sarsa(s, a, r, s_next, a_next, Q):
eta = 0.1 # 学習係数
gamma = 0.9 # 時間割引率

if s_next == 8:
Q[s, a] = Q[s, a] + eta * (r - Q[s, a])
else:
Q[s, a] = Q[s, a] + eta * (r + gamma * Q[s_next, a_next] - Q[s, a])
return Q

Q-learningの行動価値関数の更新は、次ステップの価値最大の行動を使用します。
(Sarsaと異なりランダム性を含みません)

1
2
3
4
5
6
7
8
9
10
# Q学習による行動価値関数の更新
def q_learning(s, a, r, s_next, a_next, Q):
eta = 0.1 # 学習係数
gamma = 0.9 # 時間割引率

if s_next == 8:
Q[s, a] = Q[s, a] + eta * (r - Q[s, a])
else:
Q[s, a] = Q[s, a] + eta * (r + gamma * np.nanmax(Q[s_next, :]) - Q[s, a])
return Q

1エピソードを実行して履歴と行動価値を取得します。
履歴は[状態, 行動]のリストです。

Sarsaの場合は行動価値関数としてsarsa()を使い、Q-learnigの場合はq-learning()を使います。

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
# 1エピソードの実行
def play(Q, epsilon, pi, type):
s = 0 # 状態
a = a_next = get_a(s, Q, epsilon, pi) # 行動の初期値
s_a_history = [[0, np.nan]] # 状態と行動の履歴

# エピソード完了までループ
while True:
# 行動に従って次の状態の取得
a = a_next
s_next = get_s_next(s, a)

# 履歴の更新
s_a_history[-1][1] = a
s_a_history.append([s_next, np.nan])

# 終了判定
if s_next == 8:
r = 1
a_next = np.nan
else:
r = 0
# 行動価値関数Qに従って行動の取得
a_next = get_a(s_next, Q, epsilon, pi)

# 行動価値関数の更新
if type == 1:
Q = sarsa(s, a, r, s_next, a_next, Q)
else:
Q = q_learning(s, a, r, s_next, a_next, Q)

# 終了判定
if s_next == 8:
break
else:
s = s_next

# 履歴と行動価値関数を返す
return [s_a_history, Q]

エピソードを繰り返し実行して学習を行います。今回は10回エピソードを実行しています。

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
epsilon = 0.5  # ε-greedy法のεの初期値
# 行動価値関数を初期化
Q = np.random.rand(a, b) * theta_0 * 0.01

# エピソードを繰り返し実行して学習
print('=== Sarsa ===')
for episode in range(10):
# ε-greedyの値を少しずつ小さくする
epsilon = epsilon / 2

# 1エピソード実行して履歴と行動価値関数を取得
[s_a_history, Q] = play(Q, epsilon, pi_0, 1)

# 出力
print('エピソード: {}, ステップ: {}'.format(
episode, len(s_a_history)-1))

epsilon = 0.5 # ε-greedy法のεの初期値
# 行動価値関数を初期化
Q = np.random.rand(a, b) * theta_0 * 0.01

# エピソードを繰り返し実行して学習
print('=== Q-learning ===')
for episode in range(10):
# ε-greedyの値を少しずつ小さくする
epsilon = epsilon / 2

# 1エピソード実行して履歴と行動価値関数を取得
[s_a_history, Q] = play(Q, epsilon, pi_0, 2)

# 出力
print('エピソード: {}, ステップ: {}'.format(
episode, len(s_a_history)-1))

SarsaとQ-learningでそれぞれ最短ステップの4に収束していることがわかります。
またSarsaよりQ-learningの方が収束が早いこともわかります。
結果

最後の履歴をもとにアニメーション表示を行ってみます。
(Sarsa、Q-learningともに最終的には同じ結果になります。)

1
2
3
4
5
6
7
8
9
10
# アニメーションの定期処理を行う関数
def animate(i):
state = s_a_history[i][0]
circle.set_data((state % 3) + 0.5, 2.5 - int(state / 3))
return circle

# アニメーションの表示
anim = animation.FuncAnimation(fig, animate, \
frames=len(s_a_history), interval=1000, repeat=False)
HTML(anim.to_jshtml())

(Google Colaboratoryで動作確認しています。)

参考

AlphaZero 深層学習・強化学習・探索 人工知能プログラミング実践入門 サポートページ

強化学習 AlphaZero 2 (迷路ゲーム 方策勾配法)

迷路ゲームを方策勾配法で解いていきます。

方策勾配法では成功時の行動を重要と考え、その行動を多く取り入れる手法です。

  • 目的はゴールすること。
  • 状態は位置。
  • 行動は上下左右の4種類。
  • 報酬はゴールした行動を重要視。
  • パラメータ更新間隔は1エピソード(ゴールするまで)

学習手順は次の通りです。

  1. パラメータθを準備。
  2. パラメータθを方策に変換。
  3. 方策に従って、行動をゴールするまで繰り返す。
  4. 成功した行動を多く取り入れるようにパラメータθを更新する。
  5. 方策の変化量が閾値以下になるまで2~4を繰り返す。

パラメータθは深層学習の重みパラメータにあたるものです。

まずは必要なパッケージをインポートします。

1
2
3
4
5
6
# パッケージのインポート
import numpy as np
import matplotlib.pyplot as plt
%matplotlib inline
from matplotlib import animation
from IPython.display import HTML

迷路を作成します。

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
# 迷路の作成
fig = plt.figure(figsize=(3, 3))

# 壁
plt.plot([0, 3], [3, 3], color='k')
plt.plot([0, 3], [0, 0], color='k')
plt.plot([0, 0], [0, 2], color='k')
plt.plot([3, 3], [1, 3], color='k')
plt.plot([1, 1], [1, 2], color='k')
plt.plot([2, 3], [2, 2], color='k')
plt.plot([2, 1], [1, 1], color='k')
plt.plot([2, 2], [0, 1], color='k')

# 数字
for i in range(3):
for j in range(3):
plt.text(0.5+i, 2.5-j, str(i+j*3), size=20, ha='center', va='center')

# 円
circle, = plt.plot([0.5], [2.5], marker='o', color='#d3d3d3', markersize=40)

# 目盛りと枠の非表示
plt.tick_params(axis='both', which='both', bottom='off', top= 'off',
labelbottom='off', right='off', left='off', labelleft='off')
plt.box('off')

結果

パラメータθを準備します。
学習前は正しいパラメータθは不明なので、移動可能な方向は1、移動不可な方向はnp.nan(欠損値)で初期化します。
スタートのマスはインデックス0で、ゴールのマス(インデックス8)は存在しません。

1
2
3
4
5
6
7
8
9
10
# パラメータθの初期値の準備
theta_0 = np.array([
[np.nan, 1, 1, np.nan], # 0 上,右,下,左
[np.nan, 1, 1, 1], # 1
[np.nan, np.nan, np.nan, 1], # 2
[1, np.nan, 1, np.nan], # 3
[1, 1, np.nan, np.nan], # 4
[np.nan, np.nan, 1, 1], # 5
[1, 1, np.nan, np.nan], # 6
[np.nan, np.nan, np.nan, 1]]) # 7

パラメータθを方策に変換します。
変換関数にはソフトマックス関数を利用します。マスごとに合計を1になる実数値に落とし込む関数です。
例)[np,nan, 1, 1, np.nan] -> [0, 0.5, 0.5, 0]

1
2
3
4
5
6
7
8
9
10
# パラメータθを方策に変換
def get_pi(theta):
# ソフトマックス関数で変換
[m, n] = theta.shape
pi = np.zeros((m, n))
exp_theta = np.exp(theta)
for i in range(0, m):
pi[i, :] = exp_theta[i, :] / np.nansum(exp_theta[i, :])
pi = np.nan_to_num(pi)
return pi

パラメータθの初期値を方策に変換します。

1
2
3
# パラメータθの初期値を方策に変換
pi_0 = get_pi(theta_0)
print(pi_0)

列の合計が1になっていることが分かります。
結果

方策に従って行動を取得します。
np.random.choice()にて引数pの確率分布に従って、配列の要素をランダムに返します。

1
2
3
4
# 方策に従って行動を取得
def get_a(pi, s):
# 方策の確率に従って行動を返す
return np.random.choice([0, 1, 2, 3], p=pi[s])

行動に従って次の状態を取得する関数を定義します、
3 x 3 の迷路なので、左右移動は±1、上下移動は±3になります。

1
2
3
4
5
6
7
8
9
10
# 行動に従って次の状態を取得
def get_s_next(s, a):
if a == 0: # 上
return s - 3
elif a == 1: # 右
return s + 1
elif a == 2: # 下
return s + 3
elif a == 3: # 左
return s - 1

1エピソードを実行して、履歴を取得します。履歴は、[状態, 行動]のリストです。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 1エピソード実行して履歴取得
def play(pi):
s = 0 # 状態
s_a_history = [[0, np.nan]] # 状態と行動の履歴

# エピソード完了までループ
while True:
# 方策に従って行動を取得
a = get_a(pi, s)

# 行動に従って次の状態を取得
s_next = get_s_next(s, a)

# 履歴の更新
s_a_history[-1][1] = a
s_a_history.append([s_next, np.nan])

# 終了判定
if s_next == 8:
break
else:
s = s_next

return s_a_history

1エピソードの実行と履歴を確認します。
ゴールまでにどのような経路をたどり、何ステップかかったかを把握することができます。

1
2
3
4
# 1エピソードの実行と履歴の確認
s_a_history = play(pi_0)
print(s_a_history)
print('1エピソードのステップ数:{}'.format(len(s_a_history)+1))

結果

パラメータθを更新します。
パラメータθに「学習係数」と「パラメータθの変化量」を掛けた値を加算します。
学習係数は1回の学習で更新される大きさです。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
def update_theta(theta, pi, s_a_history):
eta = 0.1 # 学習係数
total = len(s_a_history) - 1 # ゴールまでにかかった総ステップ数
[s_count, a_count] = theta.shape # 状態数, 行動数

# パラメータθの変化量の計算
delta_theta = theta.copy()
for i in range(0, s_count):
for j in range(0, a_count):
if not(np.isnan(theta[i, j])):
# ある状態である行動を採る回数
sa_ij = [sa for sa in s_a_history if sa == [i, j]]
n_ij = len(sa_ij)

# ある状態でなんらかの行動を採る回数
sa_i = [sa for sa in s_a_history if sa[0] == i]
n_i = len(sa_i)

# パラメータθの変化量
delta_theta[i, j] = (n_ij + pi[i, j] * n_i) / total

# パラメータθの更新
return theta + eta * delta_theta

エピソードを繰り返し実行して学習を行います。

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
stop_epsilon = 10**-3 # しきい値
theta = theta_0 # パラメータθ
pi = pi_0 # 方策

# エピソードを繰り返し実行して学習
for episode in range(1000):
# 1エピソード実行して履歴取得
s_a_history = play(pi)

# パラメータθの更新
theta = update_theta(theta, pi, s_a_history)

# 方策の更新
pi_new = get_pi(theta)

# 方策の変化量
pi_delta = np.sum(np.abs(pi_new-pi))
pi = pi_new

# 出力
print('エピソード: {}, ステップ: {}, 方策変化量: {:.4f}'.format(
episode, len(s_a_history)-1, pi_delta))

# 終了判定
if pi_delta < stop_epsilon: # 方策の変化量がしきい値以下
break

結果(エピソード0から27)
(中略)
結果(エピソード208から231)
ゴールへの最短ステップ数である4に少しずつ近づいているのが分かります。

最後の履歴をもとにアニメーション表示を行ってみます。

1
2
3
4
5
6
7
8
9
10
# アニメーションの定期処理を行う関数
def animate(i):
state = s_a_history[i][0]
circle.set_data((state % 3) + 0.5, 2.5 - int(state / 3))
return circle

# アニメーションの表示
anim = animation.FuncAnimation(fig, animate, \
frames=len(s_a_history), interval=1000, repeat=False)
HTML(anim.to_jshtml())

(Google Colaboratoryで動作確認しています。)

参考

AlphaZero 深層学習・強化学習・探索 人工知能プログラミング実践入門 サポートページ

強化学習 AlphaZero 1 (スロットマシーン)

スロットマシーン問題を強化学習で解いていきます。

  • スロットマシーンの目的はコインを多く出すことです。
  • 行動は1回ごとで状態はありません。
  • 行動はどのアームを選択するかであり、報酬はコインが出れば+1です。
  • 学習方法はε-greedyとUCB1を使います。
  • パラメータの更新間隔は行動1回ごとです。

まずは必要なパッケージをインポートします。

1
2
3
4
5
6
7
# パッケージのインポート
import numpy as np
import random
import math
import pandas as pd
import matplotlib.pyplot as plt
%matplotlib inline

スロットのアームを表すクラスを作成します。
コンストラクタには「コインが出る確率」を指定します。
draw()でアームを選択したときの報酬を取得します。

1
2
3
4
5
6
7
8
9
10
11
12
# スロットのアームの作成
class SlotArm():
# スロットのアームの初期化
def __init__(self, p):
self.p = p # コインが出る確率

# アームを選択した時の報酬の取得
def draw(self):
if self.p > random.random() :
return 1.0
else:
return 0.0

ε-greedyの計算処理を実装します。
コンストラクタにアームの数を指定し、select_arm()でポリシーに従ってアームを選択します。
その後に、update()で試行回数と価値を更新します。
報酬は、「前回の平均報酬」と「今回の報酬」から平均報酬を算出しています。

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
# ε-greedyの計算処理の作成
class EpsilonGreedy():
# ε-greedyの計算処理の初期化
def __init__(self, epsilon):
self.epsilon = epsilon # 探索する確率

# 試行回数と価値のリセット
def initialize(self, n_arms):
self.n = np.zeros(n_arms) # 各アームの試行回数
self.v = np.zeros(n_arms) # 各アームの価値

# アームの選択
def select_arm(self):
if self.epsilon > random.random():
# ランダムにアームを選択
return np.random.randint(0, len(self.v))
else:
# 価値が高いアームを選択
return np.argmax(self.v)

# アルゴリズムのパラメータの更新
def update(self, chosen_arm, reward, t):
# 選択したアームの試行回数に1加算
self.n[chosen_arm] += 1

# 選択したアームの価値の更新
n = self.n[chosen_arm]
v = self.v[chosen_arm]
self.v[chosen_arm] = ((n-1) / float(n)) * v + (1 / float(n)) * reward

# 文字列情報の取得
def label(self):
return 'ε-greedy('+str(self.epsilon)+')'

次にUCB1の計算処理を実装します。
初期化(initialize)の引数にアーム数を指定し、select_arm()でポリシーに従ってアームを選択します。
その後に、update()で試行回数と価値を更新します。

UCB1のアルゴリズムのパラメータ更新方法は次の通りです。

  • 成功時は選択したアームの成功回数に1を加算
  • 試行回数が0のアームがあるときは価値を更新しない(0の除算を防ぐため)
  • 各アームの価値の更新(選択されたアームだけではなく全アームの価値が更新される)
    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
    # UCB1アルゴリズム
    class UCB1():
    # 試行回数と成功回数と価値のリセット
    def initialize(self, n_arms):
    self.n = np.zeros(n_arms) # 各アームの試行回数
    self.w = np.zeros(n_arms) # 各アームの成功回数
    self.v = np.zeros(n_arms) # 各アームの価値

    # アームの選択
    def select_arm(self):
    # nが全て1以上になるようにアームを選択
    for i in range(len(self.n)):
    if self.n[i] == 0:
    return i

    # 価値が高いアームを選択
    return np.argmax(self.v)

    # アルゴリズムのパラメータの更新
    def update(self, chosen_arm, reward, t):
    # 選択したアームの試行回数に1加算
    self.n[chosen_arm] += 1

    # 成功時は選択したアームの成功回数に1加算
    if reward == 1.0:
    self.w[chosen_arm] += 1

    # 試行回数が0のアームの存在時は価値を更新しない
    for i in range(len(self.n)):
    if self.n[i] == 0:
    return

    # 各アームの価値の更新
    for i in range(len(self.v)):
    self.v[i] = self.w[i] / self.n[i] + (2 * math.log(t) / self.n[i]) ** 0.5

    # 文字列情報の取得
    def label(self):
    return 'ucb1'

シミュレーションを実行(play)します。

  • アームを準備します。(確率は30%、50%、90%とします)
  • アルゴリズムを準備します。(ε-greedyとUCB1を使います。)
  • シミュレーションを実行します。(250回を1セットとして1000セット実行します)
  • グラフを表示します。(DataFrameのgroupby()でグループ化し、mean()で平均報酬を算出します)
    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
    # シミュレーションの実行
    def play(algo, arms, num_sims, num_time):
    # 履歴の準備
    times = np.zeros(num_sims * num_time) # ゲーム回数の何回目か
    rewards = np.zeros(num_sims * num_time) # 報酬

    # シミュレーション回数分ループ
    for sim in range(num_sims):
    algo.initialize(len(arms)) # アルゴリズム設定の初期化

    # ゲーム回数分ループ
    for time in range(num_time):
    # インデックスの計算
    index = sim * num_time + time

    # 履歴の計算
    times[index] = time+1
    chosen_arm = algo.select_arm()
    reward = arms[chosen_arm].draw()
    rewards[index] = reward

    # アルゴリズムのパラメータの更新
    algo.update(chosen_arm, reward, time+1)

    # [ゲーム回数の何回目か, 報酬]
    return [times, rewards]

    # アームの準備
    arms = (SlotArm(0.3), SlotArm(0.5), SlotArm(0.9))

    # アルゴリズムの準備
    algos = (EpsilonGreedy(0.1), UCB1())

    for algo in algos:
    # シミュレーションの実行
    results = play(algo, arms, 1000, 250)

    # グラフの表示
    df = pd.DataFrame({'times': results[0], 'rewards': results[1]})
    mean = df['rewards'].groupby(df['times']).mean()
    plt.plot(mean, label=algo.label())

    # グラフの表示
    plt.xlabel('Step')
    plt.ylabel('Average Reward')
    plt.legend(loc='best')
    plt.show()

結果は下記の通りです。
結果
UCB1の方がゲーム最初から報酬が高いですが、最終的には両方とも安定した結果になっていることが分かります。

(Google Colaboratoryで動作確認しています。)

参考

AlphaZero 深層学習・強化学習・探索 人工知能プログラミング実践入門 サポートページ

強化学習 x ニューラルネットワーク 8 (A2C)

パラメータを持った関数で戦略を実装します。攻略する環境はCartPoleです。

まずは親クラスとなるフレームワークを作成します。(前回のソースと同じです。)

fn_framework.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
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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
import os
import io
import re
from collections import namedtuple
from collections import deque
import numpy as np
import tensorflow as tf
from tensorflow.python import keras as K
from PIL import Image
import matplotlib.pyplot as plt

# s:状態
# a:行動
# r:報酬
# n_s:遷移先の状態
# d:エピソード終了フラグ
Experience = namedtuple("Experience",
["s", "a", "r", "n_s", "d"])

# ニューラルネットワークを使い状態から評価を行う。
class FNAgent():

def __init__(self, epsilon, actions):
self.epsilon = epsilon
self.actions = actions
self.model = None
self.estimate_probs = False
self.initialized = False

# 学習したエージェントを保存
def save(self, model_path):
self.model.save(model_path, overwrite=True, include_optimizer=False)

# 学習したエージェントを読み込み
@classmethod
def load(cls, env, model_path, epsilon=0.0001):
actions = list(range(env.action_space.n))
agent = cls(epsilon, actions)
agent.model = K.models.load_model(model_path)
agent.initialized = True
return agent

# 初期化
# experiences:エージェントの経験
def initialize(self, experiences):
raise Exception("You have to implements estimate method.")

# 関数による予測
def estimate(self, s):
raise Exception("You have to implements estimate method.")

# パラメータの更新
def update(self, experiences, gamma):
raise Exception("You have to implements update method.")

def policy(self, s):
if np.random.random() < self.epsilon or not self.initialized:
return np.random.randint(len(self.actions))
else:
estimates = self.estimate(s)
if self.estimate_probs:
action = np.random.choice(self.actions,
size=1, p=estimates)[0]
return action
else:
return np.argmax(estimates)

def play(self, env, episode_count=5, render=True):
for e in range(episode_count):
s = env.reset()
done = False
episode_reward = 0
while not done:
if render:
env.render()
a = self.policy(s)
n_state, reward, done, info = env.step(a)
episode_reward += reward
s = n_state
else:
print("Get reward {}.".format(episode_reward))

# エージェントの学習を行う
class Trainer():

def __init__(self, buffer_size=1024, batch_size=32,
gamma=0.9, report_interval=10, log_dir=""):
self.buffer_size = buffer_size
self.batch_size = batch_size
self.gamma = gamma
self.report_interval = report_interval
self.logger = Logger(log_dir, self.trainer_name)
# エージェントの行動履歴(古い行動からすてる)
self.experiences = deque(maxlen=buffer_size)
self.training = False
self.training_count = 0
self.reward_log = []

@property
def trainer_name(self):
class_name = self.__class__.__name__
snaked = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", class_name)
snaked = re.sub("([a-z0-9])([A-Z])", r"\1_\2", snaked).lower()
snaked = snaked.replace("_trainer", "")
return snaked

def train_loop(self, env, agent, episode=200, initial_count=-1,
render=False, observe_interval=0):
self.experiences = deque(maxlen=self.buffer_size)
self.training = False
self.training_count = 0
self.reward_log = []
frames = []

for i in range(episode):
s = env.reset()
done = False
step_count = 0
self.episode_begin(i, agent)
while not done:
if render:
env.render()
if self.training and observe_interval > 0 and\
(self.training_count == 1 or
self.training_count % observe_interval == 0):
frames.append(s)

a = agent.policy(s)
n_state, reward, done, info = env.step(a)
e = Experience(s, a, reward, n_state, done)
self.experiences.append(e)
if not self.training and \
len(self.experiences) == self.buffer_size:
self.begin_train(i, agent)
self.training = True

self.step(i, step_count, agent, e)

s = n_state
step_count += 1
else:
self.episode_end(i, step_count, agent)

if not self.training and \
initial_count > 0 and i >= initial_count:
self.begin_train(i, agent)
self.training = True

if self.training:
if len(frames) > 0:
self.logger.write_image(self.training_count,
frames)
frames = []
self.training_count += 1

def episode_begin(self, episode, agent):
pass

def begin_train(self, episode, agent):
pass

def step(self, episode, step_count, agent, experience):
pass

def episode_end(self, episode, step_count, agent):
pass

def is_event(self, count, interval):
return True if count != 0 and count % interval == 0 else False

def get_recent(self, count):
recent = range(len(self.experiences) - count, len(self.experiences))
return [self.experiences[i] for i in recent]

# 環境から取得される「状態」の前処理を行う
class Observer():

def __init__(self, env):
self._env = env

@property
def action_space(self):
return self._env.action_space

@property
def observation_space(self):
return self._env.observation_space

def reset(self):
return self.transform(self._env.reset())

def render(self):
self._env.render()

def step(self, action):
n_state, reward, done, info = self._env.step(action)
return self.transform(n_state), reward, done, info

def transform(self, state):
raise Exception("You have to implements transform method.")

# 学習経過の記録を行う
class Logger():

def __init__(self, log_dir="", dir_name=""):
self.log_dir = log_dir
if not log_dir:
self.log_dir = os.path.join(os.path.dirname(__file__), "logs")
if not os.path.exists(self.log_dir):
os.mkdir(self.log_dir)

if dir_name:
self.log_dir = os.path.join(self.log_dir, dir_name)
if not os.path.exists(self.log_dir):
os.mkdir(self.log_dir)

self._callback = K.callbacks.TensorBoard(self.log_dir)

@property
def writer(self):
return self._callback.writer

def set_model(self, model):
self._callback.set_model(model)

def path_of(self, file_name):
return os.path.join(self.log_dir, file_name)

def describe(self, name, values, episode=-1, step=-1):
mean = np.round(np.mean(values), 3)
std = np.round(np.std(values), 3)
desc = "{} is {} (+/-{})".format(name, mean, std)
if episode > 0:
print("At episode {}, {}".format(episode, desc))
elif step > 0:
print("At step {}, {}".format(step, desc))

def plot(self, name, values, interval=10):
indices = list(range(0, len(values), interval))
means = []
stds = []
for i in indices:
_values = values[i:(i + interval)]
means.append(np.mean(_values))
stds.append(np.std(_values))
means = np.array(means)
stds = np.array(stds)
plt.figure()
plt.title("{} History".format(name))
plt.grid()
plt.fill_between(indices, means - stds, means + stds,
alpha=0.1, color="g")
plt.plot(indices, means, "o-", color="g",
label="{} per {} episode".format(name.lower(), interval))
plt.legend(loc="best")
plt.show()

def write(self, index, name, value):
summary = tf.Summary()
summary_value = summary.value.add()
summary_value.tag = name
summary_value.simple_value = value
self.writer.add_summary(summary, index)
self.writer.flush()

def write_image(self, index, frames):
# Deal with a 'frames' as a list of sequential gray scaled image.
last_frames = [f[:, :, -1] for f in frames]
if np.min(last_frames[-1]) < 0:
scale = 127 / np.abs(last_frames[-1]).max()
offset = 128
else:
scale = 255 / np.max(last_frames[-1])
offset = 0
channel = 1 # gray scale
tag = "frames_at_training_{}".format(index)
values = []

for f in last_frames:
height, width = f.shape
array = np.asarray(f * scale + offset, dtype=np.uint8)
image = Image.fromarray(array)
output = io.BytesIO()
image.save(output, format="PNG")
image_string = output.getvalue()
output.close()
image = tf.Summary.Image(
height=height, width=width, colorspace=channel,
encoded_image_string=image_string)
value = tf.Summary.Value(tag=tag, image=image)
values.append(value)

summary = tf.Summary(value=values)
self.writer.add_summary(summary, index)
self.writer.flush()

戦略に深層学習を適用したAdvantage Actor Critic(A2C)で実装します。

a2c_agent.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
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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
import random
import argparse
from collections import deque
import numpy as np
import tensorflow as tf
from tensorflow.python import keras as K
from PIL import Image
import gym
import gym_ple
from fn_framework import FNAgent, Trainer, Observer, Experience

class ActorCriticAgent(FNAgent):

def __init__(self, epsilon, actions):
super().__init__(epsilon, actions)
self._updater = None

@classmethod
def load(cls, env, model_path, epsilon=0.0001):
actions = list(range(env.action_space.n))
agent = cls(epsilon, actions)
agent.model = K.models.load_model(model_path, custom_objects={
"SampleLayer": SampleLayer})
agent.initialized = True
return agent

def initialize(self, experiences, optimizer):
feature_shape = experiences[0].s.shape
self.make_model(feature_shape)
self.set_updater(optimizer)
self.initialized = True

def make_model(self, feature_shape):
normal = K.initializers.glorot_normal()
model = K.Sequential()
model.add(K.layers.Conv2D(
32, kernel_size=8, strides=4, padding="same",
input_shape=feature_shape,
kernel_initializer=normal, activation="relu"))
model.add(K.layers.Conv2D(
64, kernel_size=4, strides=2, padding="same",
kernel_initializer=normal, activation="relu"))
model.add(K.layers.Conv2D(
64, kernel_size=3, strides=1, padding="same",
kernel_initializer=normal, activation="relu"))
model.add(K.layers.Flatten())
model.add(K.layers.Dense(256, kernel_initializer=normal,
activation="relu"))

actor_layer = K.layers.Dense(len(self.actions),
kernel_initializer=normal)
action_evals = actor_layer(model.output)
actions = SampleLayer()(action_evals)

critic_layer = K.layers.Dense(1, kernel_initializer=normal)
values = critic_layer(model.output)

self.model = K.Model(inputs=model.input,
outputs=[actions, action_evals, values])

def set_updater(self, optimizer,
value_loss_weight=1.0, entropy_weight=0.1):
actions = tf.placeholder(shape=(None), dtype="int32")
rewards = tf.placeholder(shape=(None), dtype="float32")

_, action_evals, values = self.model.output

neg_logs = tf.nn.sparse_softmax_cross_entropy_with_logits(
logits=action_evals, labels=actions)
advantages = rewards - values

policy_loss = tf.reduce_mean(neg_logs * tf.nn.softplus(advantages))
value_loss = tf.losses.mean_squared_error(rewards, values)
action_entropy = tf.reduce_mean(self.categorical_entropy(action_evals))

loss = policy_loss + value_loss_weight * value_loss
loss -= entropy_weight * action_entropy

updates = optimizer.get_updates(loss=loss,
params=self.model.trainable_weights)

self._updater = K.backend.function(
inputs=[self.model.input,
actions, rewards],
outputs=[loss,
policy_loss,
tf.reduce_mean(neg_logs),
tf.reduce_mean(advantages),
value_loss,
action_entropy],
updates=updates)

def categorical_entropy(self, logits):
"""
From OpenAI baseline implementation
https://github.com/openai/baselines/blob/master/baselines/common/distributions.py#L192
"""
a0 = logits - tf.reduce_max(logits, axis=-1, keepdims=True)
ea0 = tf.exp(a0)
z0 = tf.reduce_sum(ea0, axis=-1, keepdims=True)
p0 = ea0 / z0
return tf.reduce_sum(p0 * (tf.log(z0) - a0), axis=-1)

def policy(self, s):
if np.random.random() < self.epsilon or not self.initialized:
return np.random.randint(len(self.actions))
else:
action, action_evals, values = self.model.predict(np.array([s]))
return action[0]

def estimate(self, s):
action, action_evals, values = self.model.predict(np.array([s]))
return values[0][0]

def update(self, states, actions, rewards):
return self._updater([states, actions, rewards])

class SampleLayer(K.layers.Layer):

def __init__(self, **kwargs):
self.output_dim = 1 # sample one action from evaluations
super(SampleLayer, self).__init__(**kwargs)

def build(self, input_shape):
super(SampleLayer, self).build(input_shape)

def call(self, x):
noise = tf.random_uniform(tf.shape(x))
return tf.argmax(x - tf.log(-tf.log(noise)), axis=1)

def compute_output_shape(self, input_shape):
return (input_shape[0], self.output_dim)

class ActorCriticAgentTest(ActorCriticAgent):

def make_model(self, feature_shape):
normal = K.initializers.glorot_normal()
model = K.Sequential()
model.add(K.layers.Dense(64, input_shape=feature_shape,
kernel_initializer=normal, activation="relu"))
model.add(K.layers.Dense(64, kernel_initializer=normal,
activation="relu"))

actor_layer = K.layers.Dense(len(self.actions),
kernel_initializer=normal)

action_evals = actor_layer(model.output)
actions = SampleLayer()(action_evals)

critic_layer = K.layers.Dense(1, kernel_initializer=normal)
values = critic_layer(model.output)

self.model = K.Model(inputs=model.input,
outputs=[actions, action_evals, values])

class CatcherObserver(Observer):

def __init__(self, env, width, height, frame_count):
super().__init__(env)
self.width = width
self.height = height
self.frame_count = frame_count
self._frames = deque(maxlen=frame_count)

def transform(self, state):
grayed = Image.fromarray(state).convert("L")
resized = grayed.resize((self.width, self.height))
resized = np.array(resized).astype("float")
normalized = resized / 255.0 # scale to 0~1
if len(self._frames) == 0:
for i in range(self.frame_count):
self._frames.append(normalized)
else:
self._frames.append(normalized)
feature = np.array(self._frames)
# Convert the feature shape (f, w, h) => (w, h, f).
feature = np.transpose(feature, (1, 2, 0))
return feature

class ActorCriticTrainer(Trainer):

def __init__(self, buffer_size=50000, batch_size=32,
gamma=0.99, initial_epsilon=0.1, final_epsilon=1e-3,
learning_rate=1e-3, report_interval=10,
log_dir="", file_name=""):
super().__init__(buffer_size, batch_size, gamma,
report_interval, log_dir)
self.file_name = file_name if file_name else "a2c_agent.h5"
self.initial_epsilon = initial_epsilon
self.final_epsilon = final_epsilon
self.learning_rate = learning_rate
self.d_experiences = deque(maxlen=self.buffer_size)
self.training_episode = 0
self.losses = {}
self._max_reward = -10

def train(self, env, episode_count=900, initial_count=10,
test_mode=False, render=False, observe_interval=100):
actions = list(range(env.action_space.n))
if not test_mode:
agent = ActorCriticAgent(1.0, actions)
else:
agent = ActorCriticAgentTest(1.0, actions)
observe_interval = 0
self.training_episode = episode_count

self.train_loop(env, agent, episode_count, initial_count, render,
observe_interval)
return agent

def episode_begin(self, episode, agent):
self.losses = {}
for key in ["loss", "loss_policy", "loss_action", "loss_advantage",
"loss_value", "entropy"]:
self.losses[key] = []
self.experiences = []

def step(self, episode, step_count, agent, experience):
if self.training:
loss, lp, ac, ad, vl, en = agent.update(*self.make_batch())
self.losses["loss"].append(loss)
self.losses["loss_policy"].append(lp)
self.losses["loss_action"].append(ac)
self.losses["loss_advantage"].append(ad)
self.losses["loss_value"].append(vl)
self.losses["entropy"].append(en)

def make_batch(self):
batch = random.sample(self.d_experiences, self.batch_size)
states = [e.s for e in batch]
actions = [e.a for e in batch]
rewards = [e.r for e in batch]
return states, actions, rewards

def begin_train(self, episode, agent):
self.logger.set_model(agent.model)
agent.epsilon = self.initial_epsilon
self.training_episode -= episode
print("Done initialization. From now, begin training!")

def episode_end(self, episode, step_count, agent):
rewards = [e.r for e in self.experiences]
self.reward_log.append(sum(rewards))

if not agent.initialized:
optimizer = K.optimizers.Adam(lr=self.learning_rate, clipnorm=5.0)
agent.initialize(self.experiences, optimizer)

discounteds = []
for t, r in enumerate(rewards):
future_r = [_r * (self.gamma ** i) for i, _r in
enumerate(rewards[t:])]
_r = sum(future_r)
discounteds.append(_r)

for i, e in enumerate(self.experiences):
s, a, r, n_s, d = e
d_r = discounteds[i]
d_e = Experience(s, a, d_r, n_s, d)
self.d_experiences.append(d_e)

if not self.training and len(self.d_experiences) == self.buffer_size:
self.begin_train(i, agent)
self.training = True

if self.training:
reward = sum(rewards)
self.logger.write(self.training_count, "reward", reward)
self.logger.write(self.training_count, "reward_max", max(rewards))
self.logger.write(self.training_count, "epsilon", agent.epsilon)
for k in self.losses:
loss = sum(self.losses[k]) / step_count
self.logger.write(self.training_count, "loss/" + k, loss)
if reward > self._max_reward:
agent.save(self.logger.path_of(self.file_name))
self._max_reward = reward

diff = (self.initial_epsilon - self.final_epsilon)
decay = diff / self.training_episode
agent.epsilon = max(agent.epsilon - decay, self.final_epsilon)

if self.is_event(episode, self.report_interval):
recent_rewards = self.reward_log[-self.report_interval:]
self.logger.describe("reward", recent_rewards, episode=episode)

def main(play, is_test):
file_name = "a2c_agent.h5" if not is_test else "a2c_agent_test.h5"
trainer = ActorCriticTrainer(file_name=file_name)
path = trainer.logger.path_of(trainer.file_name)
agent_class = ActorCriticAgent

if is_test:
print("Train on test mode")
obs = gym.make("CartPole-v0")
agent_class = ActorCriticAgentTest
else:
env = gym.make("Catcher-v0")
obs = CatcherObserver(env, 80, 80, 4)
trainer.learning_rate = 7e-5

if play:
agent = agent_class.load(obs, path)
agent.play(obs, episode_count=10, render=True)
else:
trainer.train(obs, test_mode=is_test)

if __name__ == "__main__":
parser = argparse.ArgumentParser(description="A2C Agent")
parser.add_argument("--play", action="store_true", help="play with trained model")
parser.add_argument("--test", action="store_true", help="train by test mode")

args = parser.parse_args()
main(args.play, args.test)

テスト用のCartPoleを実行してみます。
まずは学習学習です。

1
python a2c_agent.py --test

結果(後半部)
あまり報酬が増えてないような気もしますが、この学習データを使ってゲームをプレイします。

1
python a2c_agent.py --test --play

結果(コンソール)
プレイしてる様子は下記の動画で確認できます。

・・・もうちょっと頑張ってほしいところです。

強化学習 x ニューラルネットワーク 7 (Policy Gradient)

パラメータを持った関数で戦略を実装します。攻略する環境はCartPoleです。

まずは親クラスとなるフレームワークを作成します。(前回のCatcherと同じです。)

fn_framework.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
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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
import os
import io
import re
from collections import namedtuple
from collections import deque
import numpy as np
import tensorflow as tf
from tensorflow.python import keras as K
from PIL import Image
import matplotlib.pyplot as plt

# s:状態
# a:行動
# r:報酬
# n_s:遷移先の状態
# d:エピソード終了フラグ
Experience = namedtuple("Experience",
["s", "a", "r", "n_s", "d"])

# ニューラルネットワークを使い状態から評価を行う。
class FNAgent():

def __init__(self, epsilon, actions):
self.epsilon = epsilon
self.actions = actions
self.model = None
self.estimate_probs = False
self.initialized = False

# 学習したエージェントを保存
def save(self, model_path):
self.model.save(model_path, overwrite=True, include_optimizer=False)

# 学習したエージェントを読み込み
@classmethod
def load(cls, env, model_path, epsilon=0.0001):
actions = list(range(env.action_space.n))
agent = cls(epsilon, actions)
agent.model = K.models.load_model(model_path)
agent.initialized = True
return agent

# 初期化
# experiences:エージェントの経験
def initialize(self, experiences):
raise Exception("You have to implements estimate method.")

# 関数による予測
def estimate(self, s):
raise Exception("You have to implements estimate method.")

# パラメータの更新
def update(self, experiences, gamma):
raise Exception("You have to implements update method.")

def policy(self, s):
if np.random.random() < self.epsilon or not self.initialized:
return np.random.randint(len(self.actions))
else:
estimates = self.estimate(s)
if self.estimate_probs:
action = np.random.choice(self.actions,
size=1, p=estimates)[0]
return action
else:
return np.argmax(estimates)

def play(self, env, episode_count=5, render=True):
for e in range(episode_count):
s = env.reset()
done = False
episode_reward = 0
while not done:
if render:
env.render()
a = self.policy(s)
n_state, reward, done, info = env.step(a)
episode_reward += reward
s = n_state
else:
print("Get reward {}.".format(episode_reward))

# エージェントの学習を行う
class Trainer():

def __init__(self, buffer_size=1024, batch_size=32,
gamma=0.9, report_interval=10, log_dir=""):
self.buffer_size = buffer_size
self.batch_size = batch_size
self.gamma = gamma
self.report_interval = report_interval
self.logger = Logger(log_dir, self.trainer_name)
# エージェントの行動履歴(古い行動からすてる)
self.experiences = deque(maxlen=buffer_size)
self.training = False
self.training_count = 0
self.reward_log = []

@property
def trainer_name(self):
class_name = self.__class__.__name__
snaked = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", class_name)
snaked = re.sub("([a-z0-9])([A-Z])", r"\1_\2", snaked).lower()
snaked = snaked.replace("_trainer", "")
return snaked

def train_loop(self, env, agent, episode=200, initial_count=-1,
render=False, observe_interval=0):
self.experiences = deque(maxlen=self.buffer_size)
self.training = False
self.training_count = 0
self.reward_log = []
frames = []

for i in range(episode):
s = env.reset()
done = False
step_count = 0
self.episode_begin(i, agent)
while not done:
if render:
env.render()
if self.training and observe_interval > 0 and\
(self.training_count == 1 or
self.training_count % observe_interval == 0):
frames.append(s)

a = agent.policy(s)
n_state, reward, done, info = env.step(a)
e = Experience(s, a, reward, n_state, done)
self.experiences.append(e)
if not self.training and \
len(self.experiences) == self.buffer_size:
self.begin_train(i, agent)
self.training = True

self.step(i, step_count, agent, e)

s = n_state
step_count += 1
else:
self.episode_end(i, step_count, agent)

if not self.training and \
initial_count > 0 and i >= initial_count:
self.begin_train(i, agent)
self.training = True

if self.training:
if len(frames) > 0:
self.logger.write_image(self.training_count,
frames)
frames = []
self.training_count += 1

def episode_begin(self, episode, agent):
pass

def begin_train(self, episode, agent):
pass

def step(self, episode, step_count, agent, experience):
pass

def episode_end(self, episode, step_count, agent):
pass

def is_event(self, count, interval):
return True if count != 0 and count % interval == 0 else False

def get_recent(self, count):
recent = range(len(self.experiences) - count, len(self.experiences))
return [self.experiences[i] for i in recent]

# 環境から取得される「状態」の前処理を行う
class Observer():

def __init__(self, env):
self._env = env

@property
def action_space(self):
return self._env.action_space

@property
def observation_space(self):
return self._env.observation_space

def reset(self):
return self.transform(self._env.reset())

def render(self):
self._env.render()

def step(self, action):
n_state, reward, done, info = self._env.step(action)
return self.transform(n_state), reward, done, info

def transform(self, state):
raise Exception("You have to implements transform method.")

# 学習経過の記録を行う
class Logger():

def __init__(self, log_dir="", dir_name=""):
self.log_dir = log_dir
if not log_dir:
self.log_dir = os.path.join(os.path.dirname(__file__), "logs")
if not os.path.exists(self.log_dir):
os.mkdir(self.log_dir)

if dir_name:
self.log_dir = os.path.join(self.log_dir, dir_name)
if not os.path.exists(self.log_dir):
os.mkdir(self.log_dir)

self._callback = K.callbacks.TensorBoard(self.log_dir)

@property
def writer(self):
return self._callback.writer

def set_model(self, model):
self._callback.set_model(model)

def path_of(self, file_name):
return os.path.join(self.log_dir, file_name)

def describe(self, name, values, episode=-1, step=-1):
mean = np.round(np.mean(values), 3)
std = np.round(np.std(values), 3)
desc = "{} is {} (+/-{})".format(name, mean, std)
if episode > 0:
print("At episode {}, {}".format(episode, desc))
elif step > 0:
print("At step {}, {}".format(step, desc))

def plot(self, name, values, interval=10):
indices = list(range(0, len(values), interval))
means = []
stds = []
for i in indices:
_values = values[i:(i + interval)]
means.append(np.mean(_values))
stds.append(np.std(_values))
means = np.array(means)
stds = np.array(stds)
plt.figure()
plt.title("{} History".format(name))
plt.grid()
plt.fill_between(indices, means - stds, means + stds,
alpha=0.1, color="g")
plt.plot(indices, means, "o-", color="g",
label="{} per {} episode".format(name.lower(), interval))
plt.legend(loc="best")
plt.show()

def write(self, index, name, value):
summary = tf.Summary()
summary_value = summary.value.add()
summary_value.tag = name
summary_value.simple_value = value
self.writer.add_summary(summary, index)
self.writer.flush()

def write_image(self, index, frames):
# Deal with a 'frames' as a list of sequential gray scaled image.
last_frames = [f[:, :, -1] for f in frames]
if np.min(last_frames[-1]) < 0:
scale = 127 / np.abs(last_frames[-1]).max()
offset = 128
else:
scale = 255 / np.max(last_frames[-1])
offset = 0
channel = 1 # gray scale
tag = "frames_at_training_{}".format(index)
values = []

for f in last_frames:
height, width = f.shape
array = np.asarray(f * scale + offset, dtype=np.uint8)
image = Image.fromarray(array)
output = io.BytesIO()
image.save(output, format="PNG")
image_string = output.getvalue()
output.close()
image = tf.Summary.Image(
height=height, width=width, colorspace=channel,
encoded_image_string=image_string)
value = tf.Summary.Value(tag=tag, image=image)
values.append(value)

summary = tf.Summary(value=values)
self.writer.add_summary(summary, index)
self.writer.flush()

上記の親クラスを継承し、パラメータを持った関数として戦略を実装していきます。

policy_gradient_agent.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
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
import os
import argparse
import random
from collections import deque
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.externals import joblib
import tensorflow as tf
from tensorflow.python import keras as K
import gym
from fn_framework import FNAgent, Trainer, Observer, Experience

class PolicyGradientAgent(FNAgent):

def __init__(self, epsilon, actions):
super().__init__(epsilon, actions)
self.estimate_probs = True
self.scaler = None
self._updater = None

def save(self, model_path):
super().save(model_path)
joblib.dump(self.scaler, self.scaler_path(model_path))

@classmethod
def load(cls, env, model_path, epsilon=0.0001):
agent = super().load(env, model_path, epsilon)
agent.scaler = joblib.load(agent.scaler_path(model_path))
return agent

def scaler_path(self, model_path):
fname, _ = os.path.splitext(model_path)
fname += "_scaler.pkl"
return fname

def initialize(self, experiences, optimizer):
self.scaler = StandardScaler()
states = np.vstack([e.s for e in experiences])
self.scaler.fit(states)

feature_size = states.shape[1]
self.model = K.models.Sequential([
K.layers.Dense(10, activation="relu", input_shape=(feature_size,)),
K.layers.Dense(10, activation="relu"),
K.layers.Dense(len(self.actions), activation="softmax")
])
self.set_updater(optimizer)
self.initialized = True
print("Done initialization. From now, begin training!")

def set_updater(self, optimizer):
actions = tf.placeholder(shape=(None), dtype="int32")
rewards = tf.placeholder(shape=(None), dtype="float32")
one_hot_actions = tf.one_hot(actions, len(self.actions), axis=1)
action_probs = self.model.output
selected_action_probs = tf.reduce_sum(one_hot_actions * action_probs,
axis=1)
clipped = tf.clip_by_value(selected_action_probs, 1e-10, 1.0)
loss = - tf.log(clipped) * rewards
loss = tf.reduce_mean(loss)

updates = optimizer.get_updates(loss=loss,
params=self.model.trainable_weights)
self._updater = K.backend.function(
inputs=[self.model.input,
actions, rewards],
outputs=[loss],
updates=updates)

def estimate(self, s):
normalized = self.scaler.transform(s)
action_probs = self.model.predict(normalized)[0]
return action_probs

def update(self, states, actions, rewards):
normalizeds = self.scaler.transform(states)
actions = np.array(actions)
rewards = np.array(rewards)
self._updater([normalizeds, actions, rewards])

class CartPoleObserver(Observer):

def transform(self, state):
return np.array(state).reshape((1, -1))

class PolicyGradientTrainer(Trainer):

def __init__(self, buffer_size=1024, batch_size=32,
gamma=0.9, report_interval=10, log_dir=""):
super().__init__(buffer_size, batch_size, gamma,
report_interval, log_dir)
self._reward_scaler = None
self.d_experiences = deque(maxlen=buffer_size)

def train(self, env, episode_count=220, epsilon=0.1, initial_count=-1,
render=False):
actions = list(range(env.action_space.n))
agent = PolicyGradientAgent(epsilon, actions)

self.train_loop(env, agent, episode_count, initial_count, render)
return agent

def episode_begin(self, episode, agent):
self.experiences = []

def step(self, episode, step_count, agent, experience):
if agent.initialized:
agent.update(*self.make_batch())

def make_batch(self):
batch = random.sample(self.d_experiences, self.batch_size)
states = np.vstack([e.s for e in batch])
actions = [e.a for e in batch]
rewards = [e.r for e in batch]
rewards = np.array(rewards).reshape((-1, 1))
rewards = self._reward_scaler.transform(rewards).flatten()
return states, actions, rewards

def begin_train(self, episode, agent):
optimizer = K.optimizers.Adam(clipnorm=1.0)
agent.initialize(self.d_experiences, optimizer)
self._reward_scaler = StandardScaler(with_mean=False)
rewards = np.array([[e.r] for e in self.d_experiences])
self._reward_scaler.fit(rewards)

def episode_end(self, episode, step_count, agent):
rewards = [e.r for e in self.experiences]
self.reward_log.append(sum(rewards))

discounteds = []
for t, r in enumerate(rewards):
d_r = [_r * (self.gamma ** i) for i, _r in
enumerate(rewards[t:])]
d_r = sum(d_r)
discounteds.append(d_r)

for i, e in enumerate(self.experiences):
s, a, r, n_s, d = e
d_r = discounteds[i]
d_e = Experience(s, a, d_r, n_s, d)
self.d_experiences.append(d_e)

if not self.training and len(self.d_experiences) == self.buffer_size:
self.begin_train(i, agent)
self.training = True

if self.is_event(episode, self.report_interval):
recent_rewards = self.reward_log[-self.report_interval:]
self.logger.describe("reward", recent_rewards, episode=episode)

def main(play):
env = CartPoleObserver(gym.make("CartPole-v0"))
trainer = PolicyGradientTrainer()
path = trainer.logger.path_of("policy_gradient_agent.h5")

if play:
agent = PolicyGradientAgent.load(env, path)
agent.play(env)
else:
trained = trainer.train(env, episode_count=250)
trainer.logger.plot("Rewards", trainer.reward_log,
trainer.report_interval)
trained.save(path)

if __name__ == "__main__":
parser = argparse.ArgumentParser(description="PG Agent")
parser.add_argument("--play", action="store_true", help="play with trained model")

args = parser.parse_args()
main(args.play)

結果は次の通りです。
結果(コンソール)

結果(グラフ)
前回実装した価値関数の場合より、報酬が増えるようになるには時間がかかるようです。

参考

Pythonで学ぶ強化学習 -入門から実践まで- サンプルコード

強化学習 x ニューラルネットワーク 6 (Catcher)

ニューラルネットワークでCatcherという環境を攻略してみます。
Catcherは、ボールキャッチを行うゲームです。

前回のCartPoleでは4つの入力パラメータを使って学習を行いましたが、今回は画面を入力パラメータとしています。

まずは親クラスとなるフレームワークを作成します。(前回のCartPoleと同じです。)

fn_framework.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
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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
import os
import io
import re
from collections import namedtuple
from collections import deque
import numpy as np
import tensorflow as tf
from tensorflow.python import keras as K
from PIL import Image
import matplotlib.pyplot as plt

# s:状態
# a:行動
# r:報酬
# n_s:遷移先の状態
# d:エピソード終了フラグ
Experience = namedtuple("Experience",
["s", "a", "r", "n_s", "d"])

# ニューラルネットワークを使い状態から評価を行う。
class FNAgent():

def __init__(self, epsilon, actions):
self.epsilon = epsilon
self.actions = actions
self.model = None
self.estimate_probs = False
self.initialized = False

# 学習したエージェントを保存
def save(self, model_path):
self.model.save(model_path, overwrite=True, include_optimizer=False)

# 学習したエージェントを読み込み
@classmethod
def load(cls, env, model_path, epsilon=0.0001):
actions = list(range(env.action_space.n))
agent = cls(epsilon, actions)
agent.model = K.models.load_model(model_path)
agent.initialized = True
return agent

# 初期化
# experiences:エージェントの経験
def initialize(self, experiences):
raise Exception("You have to implements estimate method.")

# 関数による予測
def estimate(self, s):
raise Exception("You have to implements estimate method.")

# パラメータの更新
def update(self, experiences, gamma):
raise Exception("You have to implements update method.")

def policy(self, s):
if np.random.random() < self.epsilon or not self.initialized:
return np.random.randint(len(self.actions))
else:
estimates = self.estimate(s)
if self.estimate_probs:
action = np.random.choice(self.actions,
size=1, p=estimates)[0]
return action
else:
return np.argmax(estimates)

def play(self, env, episode_count=5, render=True):
for e in range(episode_count):
s = env.reset()
done = False
episode_reward = 0
while not done:
if render:
env.render()
a = self.policy(s)
n_state, reward, done, info = env.step(a)
episode_reward += reward
s = n_state
else:
print("Get reward {}.".format(episode_reward))

# エージェントの学習を行う
class Trainer():

def __init__(self, buffer_size=1024, batch_size=32,
gamma=0.9, report_interval=10, log_dir=""):
self.buffer_size = buffer_size
self.batch_size = batch_size
self.gamma = gamma
self.report_interval = report_interval
self.logger = Logger(log_dir, self.trainer_name)
# エージェントの行動履歴(古い行動からすてる)
self.experiences = deque(maxlen=buffer_size)
self.training = False
self.training_count = 0
self.reward_log = []

@property
def trainer_name(self):
class_name = self.__class__.__name__
snaked = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", class_name)
snaked = re.sub("([a-z0-9])([A-Z])", r"\1_\2", snaked).lower()
snaked = snaked.replace("_trainer", "")
return snaked

def train_loop(self, env, agent, episode=200, initial_count=-1,
render=False, observe_interval=0):
self.experiences = deque(maxlen=self.buffer_size)
self.training = False
self.training_count = 0
self.reward_log = []
frames = []

for i in range(episode):
s = env.reset()
done = False
step_count = 0
self.episode_begin(i, agent)
while not done:
if render:
env.render()
if self.training and observe_interval > 0 and\
(self.training_count == 1 or
self.training_count % observe_interval == 0):
frames.append(s)

a = agent.policy(s)
n_state, reward, done, info = env.step(a)
e = Experience(s, a, reward, n_state, done)
self.experiences.append(e)
if not self.training and \
len(self.experiences) == self.buffer_size:
self.begin_train(i, agent)
self.training = True

self.step(i, step_count, agent, e)

s = n_state
step_count += 1
else:
self.episode_end(i, step_count, agent)

if not self.training and \
initial_count > 0 and i >= initial_count:
self.begin_train(i, agent)
self.training = True

if self.training:
if len(frames) > 0:
self.logger.write_image(self.training_count,
frames)
frames = []
self.training_count += 1

def episode_begin(self, episode, agent):
pass

def begin_train(self, episode, agent):
pass

def step(self, episode, step_count, agent, experience):
pass

def episode_end(self, episode, step_count, agent):
pass

def is_event(self, count, interval):
return True if count != 0 and count % interval == 0 else False

def get_recent(self, count):
recent = range(len(self.experiences) - count, len(self.experiences))
return [self.experiences[i] for i in recent]

# 環境から取得される「状態」の前処理を行う
class Observer():

def __init__(self, env):
self._env = env

@property
def action_space(self):
return self._env.action_space

@property
def observation_space(self):
return self._env.observation_space

def reset(self):
return self.transform(self._env.reset())

def render(self):
self._env.render()

def step(self, action):
n_state, reward, done, info = self._env.step(action)
return self.transform(n_state), reward, done, info

def transform(self, state):
raise Exception("You have to implements transform method.")

# 学習経過の記録を行う
class Logger():

def __init__(self, log_dir="", dir_name=""):
self.log_dir = log_dir
if not log_dir:
self.log_dir = os.path.join(os.path.dirname(__file__), "logs")
if not os.path.exists(self.log_dir):
os.mkdir(self.log_dir)

if dir_name:
self.log_dir = os.path.join(self.log_dir, dir_name)
if not os.path.exists(self.log_dir):
os.mkdir(self.log_dir)

self._callback = K.callbacks.TensorBoard(self.log_dir)

@property
def writer(self):
return self._callback.writer

def set_model(self, model):
self._callback.set_model(model)

def path_of(self, file_name):
return os.path.join(self.log_dir, file_name)

def describe(self, name, values, episode=-1, step=-1):
mean = np.round(np.mean(values), 3)
std = np.round(np.std(values), 3)
desc = "{} is {} (+/-{})".format(name, mean, std)
if episode > 0:
print("At episode {}, {}".format(episode, desc))
elif step > 0:
print("At step {}, {}".format(step, desc))

def plot(self, name, values, interval=10):
indices = list(range(0, len(values), interval))
means = []
stds = []
for i in indices:
_values = values[i:(i + interval)]
means.append(np.mean(_values))
stds.append(np.std(_values))
means = np.array(means)
stds = np.array(stds)
plt.figure()
plt.title("{} History".format(name))
plt.grid()
plt.fill_between(indices, means - stds, means + stds,
alpha=0.1, color="g")
plt.plot(indices, means, "o-", color="g",
label="{} per {} episode".format(name.lower(), interval))
plt.legend(loc="best")
plt.show()

def write(self, index, name, value):
summary = tf.Summary()
summary_value = summary.value.add()
summary_value.tag = name
summary_value.simple_value = value
self.writer.add_summary(summary, index)
self.writer.flush()

def write_image(self, index, frames):
# Deal with a 'frames' as a list of sequential gray scaled image.
last_frames = [f[:, :, -1] for f in frames]
if np.min(last_frames[-1]) < 0:
scale = 127 / np.abs(last_frames[-1]).max()
offset = 128
else:
scale = 255 / np.max(last_frames[-1])
offset = 0
channel = 1 # gray scale
tag = "frames_at_training_{}".format(index)
values = []

for f in last_frames:
height, width = f.shape
array = np.asarray(f * scale + offset, dtype=np.uint8)
image = Image.fromarray(array)
output = io.BytesIO()
image.save(output, format="PNG")
image_string = output.getvalue()
output.close()
image = tf.Summary.Image(
height=height, width=width, colorspace=channel,
encoded_image_string=image_string)
value = tf.Summary.Value(tag=tag, image=image)
values.append(value)

summary = tf.Summary(value=values)
self.writer.add_summary(summary, index)
self.writer.flush()

Catcherの画面を入力とし、各行動の価値を出力します。
このネットワークをCNNで構築し、学習していきます。

Catcherの行動は次の3つです。

  • 左に移動
  • 右に移動
  • 停止

ボールをキャッチできれば報酬1、キャッチできなければ報酬-1となります。

dqn_agent.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
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
import random
import argparse
from collections import deque
import numpy as np
from tensorflow.python import keras as K
from PIL import Image
import gym
import gym_ple
from fn_framework import FNAgent, Trainer, Observer

class DeepQNetworkAgent(FNAgent):

def __init__(self, epsilon, actions):
super().__init__(epsilon, actions)
self._scaler = None
self._teacher_model = None

def initialize(self, experiences, optimizer):
feature_shape = experiences[0].s.shape
self.make_model(feature_shape)
self.model.compile(optimizer, loss="mse")
self.initialized = True
print("Done initialization. From now, begin training!")

def make_model(self, feature_shape):
normal = K.initializers.glorot_normal()
model = K.Sequential()
model.add(K.layers.Conv2D(
32, kernel_size=8, strides=4, padding="same",
input_shape=feature_shape, kernel_initializer=normal,
activation="relu"))
model.add(K.layers.Conv2D(
64, kernel_size=4, strides=2, padding="same",
kernel_initializer=normal,
activation="relu"))
model.add(K.layers.Conv2D(
64, kernel_size=3, strides=1, padding="same",
kernel_initializer=normal,
activation="relu"))
model.add(K.layers.Flatten())
model.add(K.layers.Dense(256, kernel_initializer=normal,
activation="relu"))
model.add(K.layers.Dense(len(self.actions),
kernel_initializer=normal))
self.model = model
self._teacher_model = K.models.clone_model(self.model)

def estimate(self, state):
return self.model.predict(np.array([state]))[0]

def update(self, experiences, gamma):
states = np.array([e.s for e in experiences])
n_states = np.array([e.n_s for e in experiences])

estimateds = self.model.predict(states)
future = self._teacher_model.predict(n_states)

for i, e in enumerate(experiences):
reward = e.r
if not e.d:
reward += gamma * np.max(future[i])
estimateds[i][e.a] = reward

loss = self.model.train_on_batch(states, estimateds)
return loss

def update_teacher(self):
self._teacher_model.set_weights(self.model.get_weights())

class DeepQNetworkAgentTest(DeepQNetworkAgent):

def __init__(self, epsilon, actions):
super().__init__(epsilon, actions)

def make_model(self, feature_shape):
normal = K.initializers.glorot_normal()
model = K.Sequential()
model.add(K.layers.Dense(64, input_shape=feature_shape,
kernel_initializer=normal, activation="relu"))
model.add(K.layers.Dense(len(self.actions), kernel_initializer=normal,
activation="relu"))
self.model = model
self._teacher_model = K.models.clone_model(self.model)

class CatcherObserver(Observer):

def __init__(self, env, width, height, frame_count):
super().__init__(env)
self.width = width
self.height = height
self.frame_count = frame_count
self._frames = deque(maxlen=frame_count)

def transform(self, state):
grayed = Image.fromarray(state).convert("L")
resized = grayed.resize((self.width, self.height))
resized = np.array(resized).astype("float")
normalized = resized / 255.0 # scale to 0~1
if len(self._frames) == 0:
for i in range(self.frame_count):
self._frames.append(normalized)
else:
self._frames.append(normalized)
feature = np.array(self._frames)
# Convert the feature shape (f, w, h) => (w, h, f).
feature = np.transpose(feature, (1, 2, 0))

return feature

class DeepQNetworkTrainer(Trainer):

def __init__(self, buffer_size=50000, batch_size=32,
gamma=0.99, initial_epsilon=0.5, final_epsilon=1e-3,
learning_rate=1e-3, teacher_update_freq=3, report_interval=10,
log_dir="", file_name=""):
super().__init__(buffer_size, batch_size, gamma,
report_interval, log_dir)
self.file_name = file_name if file_name else "dqn_agent.h5"
self.initial_epsilon = initial_epsilon
self.final_epsilon = final_epsilon
self.learning_rate = learning_rate
self.teacher_update_freq = teacher_update_freq
self.loss = 0
self.training_episode = 0
self._max_reward = -10

def train(self, env, episode_count=1200, initial_count=200,
test_mode=False, render=False, observe_interval=100):
actions = list(range(env.action_space.n))
if not test_mode:
agent = DeepQNetworkAgent(1.0, actions)
else:
agent = DeepQNetworkAgentTest(1.0, actions)
observe_interval = 0
self.training_episode = episode_count

self.train_loop(env, agent, episode_count, initial_count, render,
observe_interval)
return agent

def episode_begin(self, episode, agent):
self.loss = 0

def begin_train(self, episode, agent):
optimizer = K.optimizers.Adam(lr=self.learning_rate, clipvalue=1.0)
agent.initialize(self.experiences, optimizer)
self.logger.set_model(agent.model)
agent.epsilon = self.initial_epsilon
self.training_episode -= episode

def step(self, episode, step_count, agent, experience):
if self.training:
batch = random.sample(self.experiences, self.batch_size)
self.loss += agent.update(batch, self.gamma)

def episode_end(self, episode, step_count, agent):
reward = sum([e.r for e in self.get_recent(step_count)])
self.loss = self.loss / step_count
self.reward_log.append(reward)
if self.training:
self.logger.write(self.training_count, "loss", self.loss)
self.logger.write(self.training_count, "reward", reward)
self.logger.write(self.training_count, "epsilon", agent.epsilon)
if reward > self._max_reward:
agent.save(self.logger.path_of(self.file_name))
self._max_reward = reward
if self.is_event(self.training_count, self.teacher_update_freq):
agent.update_teacher()

diff = (self.initial_epsilon - self.final_epsilon)
decay = diff / self.training_episode
agent.epsilon = max(agent.epsilon - decay, self.final_epsilon)

if self.is_event(episode, self.report_interval):
recent_rewards = self.reward_log[-self.report_interval:]
self.logger.describe("reward", recent_rewards, episode=episode)

def main(play, is_test):
file_name = "dqn_agent.h5" if not is_test else "dqn_agent_test.h5"
trainer = DeepQNetworkTrainer(file_name=file_name)
path = trainer.logger.path_of(trainer.file_name)
print('path', path)
agent_class = DeepQNetworkAgent

if is_test:
print("Train on test mode")
obs = gym.make("CartPole-v0")
agent_class = DeepQNetworkAgentTest
else:
env = gym.make("Catcher-v0")
obs = CatcherObserver(env, 80, 80, 4)
trainer.learning_rate = 1e-4

if play:
agent = agent_class.load(obs, path)
agent.play(obs, render=True)
else:
trainer.train(obs, test_mode=is_test)

if __name__ == "__main__":
parser = argparse.ArgumentParser(description="DQN Agent")
parser.add_argument("--play", action="store_true", help="play with trained model")
parser.add_argument("--test", action="store_true", help="train by test mode")

args = parser.parse_args()
main(args.play, args.test)

実行するために下記のコマンドを実行し、必要な環境をインストールしておきます。

1
2
3
4
pip install gym_ple
git clone https://github.com/ntasfi/PyGame-Learning-Environment.git
cd PyGame-Learning-Environment/
pip install -e .

学習を場合は下記のコマンドを実行します。

1
python dqn_agent.py

学習データを使ってゲームを攻略するには下記コマンドを実行します。

1
python dqn_agent.py --play

上記コマンドでCatcherを実行している様子は下記の動画で確認できます。

なかなかのスピードでボールが落ちてきますが正確にキャッチできていることがわかります。
簡単なゲームながら画面から学習して、攻略している様子をみると応用範囲も期待もふくらみます。

参考

Pythonで学ぶ強化学習 -入門から実践まで- サンプルコード

強化学習 x ニューラルネットワーク 5 (CartPole)

ニューラルネットワークでCartPoleという環境を攻略してみます。
CartPoleは、棒が倒れないようにカートの位置を調整する環境です。

まずは親クラスとなるフレームワークを作成します。
フレームワークは下記の4種類のクラスで構成されています。

  • Agent:パラメータを持った関数(ニューラルネットワーク)で実装されたエージェント。
  • Trainer:エージェントの学習を行う。
  • Observer:環境から取得される「状態」の前処理を行う。
  • Logger:学習経過の記録を行う。
fn_framework.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
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
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
import os
import io
import re
from collections import namedtuple
from collections import deque
import numpy as np
import tensorflow as tf
from tensorflow.python import keras as K
from PIL import Image
import matplotlib.pyplot as plt

# s:状態
# a:行動
# r:報酬
# n_s:遷移先の状態
# d:エピソード終了フラグ
Experience = namedtuple("Experience",
["s", "a", "r", "n_s", "d"])

# ニューラルネットワークを使い状態から評価を行う。
class FNAgent():

def __init__(self, epsilon, actions):
self.epsilon = epsilon
self.actions = actions
self.model = None
self.estimate_probs = False
self.initialized = False

# 学習したエージェントを保存
def save(self, model_path):
self.model.save(model_path, overwrite=True, include_optimizer=False)

# 学習したエージェントを読み込み
@classmethod
def load(cls, env, model_path, epsilon=0.0001):
actions = list(range(env.action_space.n))
agent = cls(epsilon, actions)
agent.model = K.models.load_model(model_path)
agent.initialized = True
return agent

# 初期化
# experiences:エージェントの経験
def initialize(self, experiences):
raise Exception("You have to implements estimate method.")

# 関数による予測
def estimate(self, s):
raise Exception("You have to implements estimate method.")

# パラメータの更新
def update(self, experiences, gamma):
raise Exception("You have to implements update method.")

def policy(self, s):
if np.random.random() < self.epsilon or not self.initialized:
return np.random.randint(len(self.actions))
else:
estimates = self.estimate(s)
if self.estimate_probs:
action = np.random.choice(self.actions,
size=1, p=estimates)[0]
return action
else:
return np.argmax(estimates)

def play(self, env, episode_count=5, render=True):
for e in range(episode_count):
s = env.reset()
done = False
episode_reward = 0
while not done:
if render:
env.render()
a = self.policy(s)
n_state, reward, done, info = env.step(a)
episode_reward += reward
s = n_state
else:
print("Get reward {}.".format(episode_reward))

# エージェントの学習を行う
class Trainer():

def __init__(self, buffer_size=1024, batch_size=32,
gamma=0.9, report_interval=10, log_dir=""):
self.buffer_size = buffer_size
self.batch_size = batch_size
self.gamma = gamma
self.report_interval = report_interval
self.logger = Logger(log_dir, self.trainer_name)
# エージェントの行動履歴(古い行動からすてる)
self.experiences = deque(maxlen=buffer_size)
self.training = False
self.training_count = 0
self.reward_log = []

@property
def trainer_name(self):
class_name = self.__class__.__name__
snaked = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", class_name)
snaked = re.sub("([a-z0-9])([A-Z])", r"\1_\2", snaked).lower()
snaked = snaked.replace("_trainer", "")
return snaked

def train_loop(self, env, agent, episode=200, initial_count=-1,
render=False, observe_interval=0):
self.experiences = deque(maxlen=self.buffer_size)
self.training = False
self.training_count = 0
self.reward_log = []
frames = []

for i in range(episode):
s = env.reset()
done = False
step_count = 0
self.episode_begin(i, agent)
while not done:
if render:
env.render()
if self.training and observe_interval > 0 and\
(self.training_count == 1 or
self.training_count % observe_interval == 0):
frames.append(s)

a = agent.policy(s)
n_state, reward, done, info = env.step(a)
e = Experience(s, a, reward, n_state, done)
self.experiences.append(e)
if not self.training and \
len(self.experiences) == self.buffer_size:
self.begin_train(i, agent)
self.training = True

self.step(i, step_count, agent, e)

s = n_state
step_count += 1
else:
self.episode_end(i, step_count, agent)

if not self.training and \
initial_count > 0 and i >= initial_count:
self.begin_train(i, agent)
self.training = True

if self.training:
if len(frames) > 0:
self.logger.write_image(self.training_count,
frames)
frames = []
self.training_count += 1

def episode_begin(self, episode, agent):
pass

def begin_train(self, episode, agent):
pass

def step(self, episode, step_count, agent, experience):
pass

def episode_end(self, episode, step_count, agent):
pass

def is_event(self, count, interval):
return True if count != 0 and count % interval == 0 else False

def get_recent(self, count):
recent = range(len(self.experiences) - count, len(self.experiences))
return [self.experiences[i] for i in recent]

# 環境から取得される「状態」の前処理を行う
class Observer():

def __init__(self, env):
self._env = env

@property
def action_space(self):
return self._env.action_space

@property
def observation_space(self):
return self._env.observation_space

def reset(self):
return self.transform(self._env.reset())

def render(self):
self._env.render()

def step(self, action):
n_state, reward, done, info = self._env.step(action)
return self.transform(n_state), reward, done, info

def transform(self, state):
raise Exception("You have to implements transform method.")

# 学習経過の記録を行う
class Logger():

def __init__(self, log_dir="", dir_name=""):
self.log_dir = log_dir
if not log_dir:
self.log_dir = os.path.join(os.path.dirname(__file__), "logs")
if not os.path.exists(self.log_dir):
os.mkdir(self.log_dir)

if dir_name:
self.log_dir = os.path.join(self.log_dir, dir_name)
if not os.path.exists(self.log_dir):
os.mkdir(self.log_dir)

self._callback = K.callbacks.TensorBoard(self.log_dir)

@property
def writer(self):
return self._callback.writer

def set_model(self, model):
self._callback.set_model(model)

def path_of(self, file_name):
return os.path.join(self.log_dir, file_name)

def describe(self, name, values, episode=-1, step=-1):
mean = np.round(np.mean(values), 3)
std = np.round(np.std(values), 3)
desc = "{} is {} (+/-{})".format(name, mean, std)
if episode > 0:
print("At episode {}, {}".format(episode, desc))
elif step > 0:
print("At step {}, {}".format(step, desc))

def plot(self, name, values, interval=10):
indices = list(range(0, len(values), interval))
means = []
stds = []
for i in indices:
_values = values[i:(i + interval)]
means.append(np.mean(_values))
stds.append(np.std(_values))
means = np.array(means)
stds = np.array(stds)
plt.figure()
plt.title("{} History".format(name))
plt.grid()
plt.fill_between(indices, means - stds, means + stds,
alpha=0.1, color="g")
plt.plot(indices, means, "o-", color="g",
label="{} per {} episode".format(name.lower(), interval))
plt.legend(loc="best")
plt.show()

def write(self, index, name, value):
summary = tf.Summary()
summary_value = summary.value.add()
summary_value.tag = name
summary_value.simple_value = value
self.writer.add_summary(summary, index)
self.writer.flush()

def write_image(self, index, frames):
# Deal with a 'frames' as a list of sequential gray scaled image.
last_frames = [f[:, :, -1] for f in frames]
if np.min(last_frames[-1]) < 0:
scale = 127 / np.abs(last_frames[-1]).max()
offset = 128
else:
scale = 255 / np.max(last_frames[-1])
offset = 0
channel = 1 # gray scale
tag = "frames_at_training_{}".format(index)
values = []

for f in last_frames:
height, width = f.shape
array = np.asarray(f * scale + offset, dtype=np.uint8)
image = Image.fromarray(array)
output = io.BytesIO()
image.save(output, format="PNG")
image_string = output.getvalue()
output.close()
image = tf.Summary.Image(
height=height, width=width, colorspace=channel,
encoded_image_string=image_string)
value = tf.Summary.Value(tag=tag, image=image)
values.append(value)

summary = tf.Summary(value=values)
self.writer.add_summary(summary, index)
self.writer.flush()

次に価値関数により行動を決定するエージェントを作成し、実際にCartPoleを攻略してみます。
CartPoleにおける「状態」は次の4つです。

  • 位置
  • 加速度
  • ポールの角度
  • ポールの倒れる速度(角速度)

「行動」はカートの左右への移動です。

「報酬」は常に1で、ポールが倒れたらエピソード終了になります。つまりポールが倒れない時間が長いほど報酬が手に入ります。

value_function_agent.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
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
import random
import argparse
import numpy as np
from sklearn.neural_network import MLPRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.externals import joblib
import gym
from fn_framework import FNAgent, Trainer, Observer

class ValueFunctionAgent(FNAgent): # 親クラス(フレームワーク)を継承

def save(self, model_path):
joblib.dump(self.model, model_path)

@classmethod
def load(cls, env, model_path, epsilon=0.0001):
actions = list(range(env.action_space.n))
agent = cls(epsilon, actions)
agent.model = joblib.load(model_path)
agent.initialized = True
return agent

def initialize(self, experiences):
scaler = StandardScaler()
estimator = MLPRegressor(hidden_layer_sizes=(10, 10), max_iter=1)
self.model = Pipeline([("scaler", scaler), ("estimator", estimator)])

states = np.vstack([e.s for e in experiences])
self.model.named_steps["scaler"].fit(states)

# Avoid the predict before fit.
self.update([experiences[0]], gamma=0)
self.initialized = True
print("Done initialization. From now, begin training!")

def estimate(self, s):
estimated = self.model.predict(s)[0]
return estimated

def _predict(self, states):
if self.initialized:
predicteds = self.model.predict(states)
else:
size = len(self.actions) * len(states)
predicteds = np.random.uniform(size=size)
predicteds = predicteds.reshape((-1, len(self.actions)))
return predicteds

def update(self, experiences, gamma):
states = np.vstack([e.s for e in experiences])
n_states = np.vstack([e.n_s for e in experiences])

estimateds = self._predict(states)
future = self._predict(n_states)

for i, e in enumerate(experiences):
reward = e.r
if not e.d:
reward += gamma * np.max(future[i])
estimateds[i][e.a] = reward

estimateds = np.array(estimateds)
states = self.model.named_steps["scaler"].transform(states)
self.model.named_steps["estimator"].partial_fit(states, estimateds)

class CartPoleObserver(Observer):

def transform(self, state):
return np.array(state).reshape((1, -1))

class ValueFunctionTrainer(Trainer):

def train(self, env, episode_count=220, epsilon=0.1, initial_count=-1,
render=False):
actions = list(range(env.action_space.n))
agent = ValueFunctionAgent(epsilon, actions)
self.train_loop(env, agent, episode_count, initial_count, render)
return agent

def begin_train(self, episode, agent):
agent.initialize(self.experiences)

def step(self, episode, step_count, agent, experience):
if self.training:
batch = random.sample(self.experiences, self.batch_size)
# 学習を行う。
agent.update(batch, self.gamma)

def episode_end(self, episode, step_count, agent):
rewards = [e.r for e in self.get_recent(step_count)]
self.reward_log.append(sum(rewards))

if self.is_event(episode, self.report_interval):
recent_rewards = self.reward_log[-self.report_interval:]
self.logger.describe("reward", recent_rewards, episode=episode)

def main(play):
env = CartPoleObserver(gym.make("CartPole-v0"))
trainer = ValueFunctionTrainer()
path = trainer.logger.path_of("value_function_agent.pkl")

if play:
agent = ValueFunctionAgent.load(env, path)
agent.play(env)
else:
trained = trainer.train(env)
trainer.logger.plot("Rewards", trainer.reward_log, trainer.report_interval)
trained.save(path)

if __name__ == "__main__":
parser = argparse.ArgumentParser(description="VF Agent")
parser.add_argument("--play", action="store_true", help="play with trained model")

args = parser.parse_args()
main(args.play)

結果は下記のとおりです。
結果(コンソール)

結果(グラフ)

エピソードをこなすほど獲得報酬が増加していて、うまくカートを動かす方法を学習していることがわかります。

参考

Pythonで学ぶ強化学習 -入門から実践まで- サンプルコード


Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×