強化学習 AlphaZero 9 (三目並べ AlphaZero1)

10月 3, 2019

三目並べをAlphaZeroで解いていきます。
今回はAlphaZeroを実装する事前準備として、ゲームの全体像と三目並べを実装します。

AlphaZeroは、モンテカルロ木探索をベースに次の２点を組み込んでいます。

深層学習の曲面から最善手を予測する。
強化学習の自己対戦による経験を組み合わせる。

AlphaZeroによる三目並べのポイントは次の通りです。

目的は勝つことです。
状態は局面です。
行動は手を打つことです。
報酬は勝ったら+1、負けたら-1です。
学習方法はモンテカルロ木検索＋ResNet＋セルフプレイです。
パラメータ更新間隔は１エピソードごとです。

AlphaZeroの強化学習は下記のようになります。

デュアルネットワークの作成
ニューラルネットワークを定義し、ベストプレイヤー（過去最強）のモデルを作成します。
セルフプレイ部
AI同士でゲーム終了までプレイします。
パラメータ更新部
セルフプレイで作成した学習データで最新プレイヤーを作成します。
初期状態としてはベストプレイヤーをコピーして作成します。
新パラメータ更新部
最新プレイヤーとベストプレイヤーで対戦し、勝ち越した場合はベストプレイヤーを更新します。

2～4を繰り返して学習を行います。

まず三目並べを実装します。

game.py

# パッケージのインポート
import random
import math

# ゲーム状態
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

ランダムで行動を選択する関数を定義します。

game.py

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

アルファベータ法で行動を選択する関数を定義します。

game.py

# アルファベータ法で状態価値計算
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

# アルファベータ法で行動選択
def alpha_beta_action(state):
    # 合法手の状態価値の計算
    best_action = 0
    alpha = -float('inf')
    for action in state.legal_actions():
        score = -alpha_beta(state.next(action), -float('inf'), -alpha)
        if score > alpha:
            best_action = action
            alpha = score

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

プレイアウト関数を定義します。

game.py

# プレイアウト
def playout(state):
    # 負けは状態価値-1
    if state.is_lose():
        return -1

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

    # 次の状態の状態価値
    return -playout(state.next(random_action(state)))

# 最大値のインデックスを返す
def argmax(collection):
    return collection.index(max(collection))

モンテカルロ木探索で行動を選択する関数を定義します。

game.py

# モンテカルロ木探索の行動選択
def mcts_action(state):
    # モンテカルロ木探索のノード
    class node:
        # 初期化
        def __init__(self, state):
            self.state = state # 状態
            self.w = 0 # 累計価値
            self.n = 0 # 試行回数
            self.child_nodes = None  # 子ノード群

        # 評価
        def evaluate(self):
            # ゲーム終了時
            if self.state.is_done():
                # 勝敗結果で価値を取得
                value = -1 if self.state.is_lose() else 0 # 負けは-1、引き分けは0

                # 累計価値と試行回数の更新
                self.w += value
                self.n += 1
                return value

            # 子ノードが存在しない時
            if not self.child_nodes:
                # プレイアウトで価値を取得
                value = playout(self.state)

                # 累計価値と試行回数の更新
                self.w += value
                self.n += 1

                # 子ノードの展開
                if self.n == 10:
                    self.expand()
                return value

            # 子ノードが存在する時
            else:
                # UCB1が最大の子ノードの評価で価値を取得
                value = -self.next_child_node().evaluate()

                # 累計価値と試行回数の更新
                self.w += value
                self.n += 1
                return value

        # 子ノードの展開
        def expand(self):
            legal_actions = self.state.legal_actions()
            self.child_nodes = []
            for action in legal_actions:
                self.child_nodes.append(node(self.state.next(action)))

        # UCB1が最大の子ノードを取得
        def next_child_node(self):
             # 試行回数nが0の子ノードを返す
            for child_node in self.child_nodes:
                if child_node.n == 0:
                    return child_node

            # UCB1の計算
            t = 0
            for c in self.child_nodes:
                t += c.n
            ucb1_values = []
            for child_node in self.child_nodes:
                ucb1_values.append(-child_node.w/child_node.n+2*(2*math.log(t)/child_node.n)**0.5)

            # UCB1が最大の子ノードを返す
            return self.child_nodes[argmax(ucb1_values)]

    # ルートノードの生成
    root_node = node(state)
    root_node.expand()

    # ルートノードを100回評価
    for _ in range(100):
        root_node.evaluate()

    # 試行回数の最大値を持つ行動を返す
    legal_actions = state.legal_actions()
    n_list = []
    for c in root_node.child_nodes:
        n_list.append(c.n)
    return legal_actions[argmax(n_list)]

動作確認のためのコードを実装します。ランダムとランダムで対戦するコードです。

game.py

# 動作確認
if __name__ == '__main__':
    # 状態の生成
    state = State()

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

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

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

１手ごとの局面が表示され、ゲーム終了まで実行されます。（１エピソードに対応）

参考

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

強化学習 AlphaZero 8 (三目並べモンテカルロ木探索)

10月 2, 2019

三目並べをモンテカルロ木探索で解いていきます。

モンテカルロ木探索は、原始モンテカルロ探索をベースに「有望な手」をより深く調べることで改善を試みます。具体的には、選択・評価・展開・更新の４操作を行います。

まず、三目並べを実装します。（前回の原始モンテカルロ探索で定義したものと同じです。）

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

ランダムで行動を選択する関数を定義します。対戦相手用です。

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

アルファベータ法で行動を選択する関数を定義します。対戦相手用です。

# アルファベータ法で状態価値計算
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

# アルファベータ法で行動選択
def alpha_beta_action(state):
    # 合法手の状態価値の計算
    best_action = 0
    alpha = -float('inf')
    for action in state.legal_actions():
        score = -alpha_beta(state.next(action), -float('inf'), -alpha)
        if score > alpha:
            best_action = action
            alpha = score

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

プレイアウト関数を定義します。

# プレイアウト
def playout(state):
    # 負けは状態価値-1
    if state.is_lose():
        return -1
    
    # 引き分けは状態価値0
    if state.is_draw():
        return  0
    
    # 次の状態の状態価値
    return -playout(state.next(random_action(state)))

原始モンテカルロ探索で行動を選択する関数を定義します。対戦相手用です。

# 原始モンテカルロ探索で行動選択
def mcs_action(state):
    # 合法手ごとに10回プレイアウトした時の状態価値の合計の計算
    legal_actions = state.legal_actions()
    values = [0] * len(legal_actions)
    for i, action in enumerate(legal_actions):
        for _ in range(10):
            values[i] += -playout(state.next(action))

    # 合法手の状態価値の合計の最大値を持つ行動を返す 
    return legal_actions[argmax(values)]

# 最大値のインデックスを返す
def argmax(collection, key=None):
    return collection.index(max(collection))

モンテカルロ木探索で行動を選択する関数を定義します。

import math

# モンテカルロ木探索の行動選択
def mcts_action(state):
    # モンテカルロ木探索のノードの定義
    class Node:
        # ノードの初期化
        def __init__(self, state):
            self.state = state # 状態
            self.w = 0 # 累計価値
            self.n = 0 # 試行回数
            self.child_nodes = None  # 子ノード群

        # 局面の価値の計算
        def evaluate(self):
            # ゲーム終了時
            if self.state.is_done():
                # 勝敗結果で価値を取得
                value = -1 if self.state.is_lose() else 0 # 負けは-1、引き分けは0

                # 累計価値と試行回数の更新
                self.w += value
                self.n += 1
                return value

            # 子ノードが存在しない時
            if not self.child_nodes:
                # プレイアウトで価値を取得
                value = playout(self.state)

                # 累計価値と試行回数の更新
                self.w += value
                self.n += 1

                # 子ノードの展開
                if self.n == 10: 
                    self.expand()
                return value

            # 子ノードが存在する時
            else:
                # UCB1が最大の子ノードの評価で価値を取得
                value = -self.next_child_node().evaluate() 

                # 累計価値と試行回数の更新
                self.w += value
                self.n += 1
                return value

        # 子ノードの展開
        def expand(self):
            legal_actions = self.state.legal_actions()
            self.child_nodes = []
            for action in legal_actions:
                self.child_nodes.append(Node(self.state.next(action)))

        # UCB1が最大の子ノードの取得
        def next_child_node(self):
             # 試行回数が0の子ノードを返す
            for child_node in self.child_nodes:
                if child_node.n == 0:
                    return child_node

            # UCB1の計算
            t = 0
            for c in self.child_nodes:
                t += c.n
            ucb1_values = []
            for child_node in self.child_nodes:
                ucb1_values.append(-child_node.w/child_node.n+(2*math.log(t)/child_node.n)**0.5)

            # UCB1が最大の子ノードを返す
            return self.child_nodes[argmax(ucb1_values)]    
    
    # 現在の局面のノードの作成
    root_node = Node(state)
    root_node.expand()

    # 100回のシミュレーションを実行
    for _ in range(100):
        root_node.evaluate()

    # 試行回数の最大値を持つ行動を返す
    legal_actions = state.legal_actions()
    n_list = []
    for c in root_node.child_nodes:
        n_list.append(c.n)
    return legal_actions[argmax(n_list)]

モンテカルロ木探索とランダムおよびアルファベータ法、原始モンテカルロ探索の対戦を行います。

# モンテカルロ木探索とランダムおよびアルファベータ法、原始モンテカルロ探索の対戦

# パラメータ
EP_GAME_COUNT = 100  # 1評価あたりのゲーム数

# 先手プレイヤーのポイント
def first_player_point(ended_state):
    # 1:先手勝利, 0:先手敗北, 0.5:引き分け
    if ended_state.is_lose():
        return 0 if ended_state.is_first_player() else 1
    return 0.5

# 1ゲームの実行
def play(next_actions):
    # 状態の生成
    state = State()

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

        # 行動の取得
        next_action = next_actions[0] if state.is_first_player() else next_actions[1]
        action = next_action(state)

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

    # 先手プレイヤーのポイントを返す
    return first_player_point(state)

# 任意のアルゴリズムの評価
def evaluate_algorithm_of(label, next_actions):
    # 複数回の対戦を繰り返す
    total_point = 0
    for i in range(EP_GAME_COUNT):
        # 1ゲームの実行
        if i % 2 == 0:
            total_point += play(next_actions)
        else:
            total_point += 1 - play(list(reversed(next_actions)))

        # 出力
        print('\rEvaluate {}/{}'.format(i + 1, EP_GAME_COUNT), end='')
    print('')

    # 平均ポイントの計算
    average_point = total_point / EP_GAME_COUNT
    print(label.format(average_point * 100))

# VSランダム
next_actions = (mcts_action, random_action)
evaluate_algorithm_of('モンテカルロ木探索 VS ランダム {:.2f}%\n', next_actions)

# VSアルファベータ法
next_actions = (mcts_action, alpha_beta_action)
evaluate_algorithm_of('モンテカルロ木探索 VS アルファベータ法 {:.2f}%\n', next_actions)

# VSモンテカルロ法
next_actions = (mcts_action, mcs_action)
evaluate_algorithm_of('モンテカルロ木探索 VS 原始モンテカルロ探索 {:.2f}%\n', next_actions)

結果は次の通りです。

ランダムとの対戦では勝率93%と圧勝ですが、アルファベータ法との対戦では勝率36.5%となり原始モンテカルロ探索よりは勝率が10%以上あがりました。

モンテカルロ木探索と原始モンテカルロ探索の対戦では67.00%と原始モンテカルロ探索よりモンテカルロ木探索がやや強い結果となりました。

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

参考

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

強化学習 AlphaZero 7 (三目並べ原始モンテカルロ探索)

10月 1, 2019

三目並べを原始モンテカルロ探索で解いていきます。

原始モンテカルロ探索はランダムシミュレーションによって状態価値を計算する手法です。
何回もランダムプレイを行い、勝率の高い手を価値が高いと評価します。

まず、三目並べを実装します。（前回のアルファベータ法で定義したものと同じです。）

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

ランダムで行動を選択する関数を定義します。対戦相手用です。

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

アルファベータ法で行動を選択する関数を定義します。対戦相手用です。

# アルファベータ法で状態価値計算
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

# アルファベータ法で行動選択
def alpha_beta_action(state):
    # 合法手の状態価値の計算
    best_action = 0
    alpha = -float('inf')
    for action in state.legal_actions():
        score = -alpha_beta(state.next(action), -float('inf'), -alpha)
        if score > alpha:
            best_action = action
            alpha = score

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

いよいよ原始モンテカルロ探索の実装にはいります。
まずはプレイアウト関数を定義します。プレイアウトとはたんに現在の局面からゲーム終了までプレイすることです。

# プレイアウト
def playout(state):
    # 負けは状態価値-1
    if state.is_lose():
        return -1
    
    # 引き分けは状態価値0
    if state.is_draw():
        return  0
    
    # 次の状態の状態価値
    return -playout(state.next(random_action(state)))

原始モンテカルロ探索で行動選択する関数を定義します。
合法手ごとに10回プレイアウトした時の状態価値の合計の計算し、その合計が最も大きな行動を選択します。

# 原始モンテカルロ探索で行動選択
def mcs_action(state):
    # 合法手ごとに10回プレイアウトした時の状態価値の合計の計算
    legal_actions = state.legal_actions()
    values = [0] * len(legal_actions)
    for i, action in enumerate(legal_actions):
        for _ in range(10):
            values[i] += -playout(state.next(action))

    # 合法手の状態価値の合計の最大値を持つ行動を返す 
    return legal_actions[argmax(values)]

# 最大値のインデックスを返す
def argmax(collection, key=None):
    return collection.index(max(collection))

原始モンテカルロ探索とランダム・アルファベータ法で対戦してみます。

# 原始モンテカルロ探索とランダム・アルファベータ法の対戦

# パラメータ
EP_GAME_COUNT = 100  # 1評価あたりのゲーム数

# 先手プレイヤーのポイント
def first_player_point(ended_state):
    # 1:先手勝利, 0:先手敗北, 0.5:引き分け
    if ended_state.is_lose():
        return 0 if ended_state.is_first_player() else 1
    return 0.5

# 1ゲームの実行
def play(next_actions):
    # 状態の生成
    state = State()

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

        # 行動の取得
        next_action = next_actions[0] if state.is_first_player() else next_actions[1]
        action = next_action(state)

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

    # 先手プレイヤーのポイントを返す
    return first_player_point(state)

# 任意のアルゴリズムの評価
def evaluate_algorithm_of(label, next_actions):
    # 複数回の対戦を繰り返す
    total_point = 0
    for i in range(EP_GAME_COUNT):
        # 1ゲームの実行
        if i % 2 == 0:
            total_point += play(next_actions)
        else:
            total_point += 1 - play(list(reversed(next_actions)))

        # 出力
        print('\rEvaluate {}/{}'.format(i + 1, EP_GAME_COUNT), end='')
    print('')

    # 平均ポイントの計算
    average_point = total_point / EP_GAME_COUNT
    print(label.format(average_point))

# VSランダム
next_actions = (mcs_action, random_action)
evaluate_algorithm_of('VS_Random {:.3f}', next_actions)

# VSアルファベータ法
next_actions = (mcs_action, alpha_beta_action)
evaluate_algorithm_of('VS_AlphaBeta {:.3f}', next_actions)

結果は次の通りです。

ランダムとの対戦では勝率93%と圧勝ですが、アルファベータ法との対戦では勝率23.5%となかなか厳しい結果となりました。

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

参考

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

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

9月 30, 2019

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

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

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

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

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

# ミニマックス法で状態価値計算
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を左右入れ替えてるのがポイントです。

# アルファベータ法で状態価値計算
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

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

best_scoreをalphaとする。
betaの初期値に-float(‘inf’)を設定する。

# アルファベータ法で行動選択
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

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

# アルファベータ法とミニマックス法の対戦

# 状態の生成
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 (三目並べミニマックス法)

9月 29, 2019

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

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

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

next()は行動に応じて次の状態を取得する。
行動は石を配置するマスの位置を０～８で指定する。
legal_actions()は選択可能な行動を取得します。

# 三目並べの作成
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()で合法手を取得し、その中からランダムに手を選びます。

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

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

# ランダムとランダムで対戦

# 状態の生成
state = State()

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

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

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

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

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

# ミニマックス法で状態価値計算
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を渡すと行動（石を置くマス：０～８）を返します。

# ミニマックス法で行動選択
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()を使います。

# ミニマックス法とランダムで対戦

# 状態の生成
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)

9月 28, 2019

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

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

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

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

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

Experience Replayd(経験を貯めておきあとでランダムに学習する)
Fixed Target Q-Network(更新計算のためにニューラルネットワークをもう１つ作る)
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

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

# ディスプレイ設定のインストール
!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)

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

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

# パッケージのインポート
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

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

# パラメータの準備
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 # バッチサイズ

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

# 行動価値関数の定義
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))

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

# 経験メモリの定義
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クラスを利用して、２つのニューラルネットワークと経験メモリを作成します。

# 環境の作成
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)

学習を開始します。

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

# 学習の開始

# 環境の初期化
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])

最大ステップ数の２００に少しずつ近づいていきます。

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

# 評価
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))

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

# 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)

9月 27, 2019

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

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

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

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

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

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

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

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

迷路を作成します。

# 迷路の作成
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()でばらつきを加えて作成します。
前回の方策勾配法とは違い、今回パラメータθと方策の更新はありません。

# パラメータθの初期値の準備
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

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

# パラメータθを方策に変換
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
2
3

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

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

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

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

# 行動に従って次の状態を取得
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-εで行動価値関数の行動を選択します。

# ランダムまたは行動価値関数に従って行動を取得
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誤差を掛けた値を加算します。

# 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と異なりランダム性を含みません）

# 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

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

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

# 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]

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

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でそれぞれ最短ステップの４に収束していることがわかります。
またSarsaよりQ-learningの方が収束が早いこともわかります。

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

# アニメーションの定期処理を行う関数
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 (迷路ゲーム方策勾配法)

9月 26, 2019

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

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

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

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

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

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

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

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

迷路を作成します。

# 迷路の作成
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）は存在しません。

# パラメータθの初期値の準備
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]

# パラメータθを方策に変換
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の確率分布に従って、配列の要素をランダムに返します。

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

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

# 行動に従って次の状態を取得
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エピソード実行して履歴取得
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エピソードの実行と履歴の確認
s_a_history = play(pi_0)
print(s_a_history)
print('1エピソードのステップ数：{}'.format(len(s_a_history)+1))

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

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

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

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

（中略）

ゴールへの最短ステップ数である４に少しずつ近づいているのが分かります。

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

# アニメーションの定期処理を行う関数
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())```

{% youtube BUoy31PYMMY %}


([Google Colaboratory](https://colab.research.google.com/notebooks/welcome.ipynb)で動作確認しています。)

参考
> <cite>AlphaZero 深層学習・強化学習・探索 人工知能プログラミング実践入門 <a href="https://www.borndigital.co.jp/book/14383.html" > サポートページ</a> </cite>

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

9月 25, 2019

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

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

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

# パッケージのインポート
import numpy as np
import random
import math
import pandas as pd
import matplotlib.pyplot as plt
%matplotlib inline

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

# スロットのアームの作成
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()で試行回数と価値を更新します。
報酬は、「前回の平均報酬」と「今回の報酬」から平均報酬を算出しています。

# ε-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のアルゴリズムのパラメータ更新方法は次の通りです。

成功時は選択したアームの成功回数に１を加算
試行回数が０のアームがあるときは価値を更新しない（0の除算を防ぐため）

各アームの価値の更新（選択されたアームだけではなく全アームの価値が更新される）

# 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, ｔ):
        # 選択したアームの試行回数に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()で平均報酬を算出します）

# シミュレーションの実行
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)

9月 24, 2019

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

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

fn_framework.py

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

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

プレイしてる様子は下記の動画で確認できます。

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