強化学習 AlphaZero 18 (コネクトフォー)

AlphaZeroのアルゴリズムを使ってコネクトフォーの対戦ゲームを作ります。

コネクトフォーは石を下から積み重ねていき、縦・横・斜めのいずれかに直線状に石を４つ並べた方が勝ちとなります。

ソースコード一覧は下記のようになります。

ソースコード 説明 ３目並べとの相違
game.py ゲーム状態 全て
dual_network.py デュアルネットワーク パラメータのみ
pv_mcts.py モンテカルロ木探索 なし
self_play.py セルフプレイ部 なし
train_network.py パラメータ更新部 なし
evaluate_network.py 新パラメータ評価部 なし
train_cycle.py 学習サイクルの実行 ベストプレイヤーの評価を削除
human_play.py ゲームUI 全て

まずコネクトフォーのゲーム状態を作成します。

game.py
1
2
3
4
5
6
7
8
9
10
# ====================
# コネクトフォー
# ====================

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

# ゲーム状態
class State:

初期化処理では「自分の石の配置」「相手の石の配置」を長さ４２（６行×７列）の１次元配列で保持します。

game.py
1
2
3
4
5
# 初期化
def __init__(self, pieces=None, enemy_pieces=None):
    # 石の配置
    self.pieces = pieces if pieces != None else [0] * 42
    self.enemy_pieces = enemy_pieces if enemy_pieces != None else [0] * 42

石の数を取得します。

game.py
1
2
3
4
5
6
7
# 石の数の取得
def piece_count(self, pieces):
    count = 0
    for i in pieces:
        if i == 1:
            count +=  1
    return count

負けかどうかを判定します。

game.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 負けかどうか
def is_lose(self):
    # 4並びかどうか
    def is_comp(x, y, dx, dy):
        for k in range(4):
            if y < 0 or 5 < y or x < 0 or 6 < x or \
                self.enemy_pieces[x+y*7] == 0:
                return False
            x, y = x+dx, y+dy
        return True

    # 負けかどうか
    for j in range(6):
        for i in range(7):
            if is_comp(i, j, 1, 0) or is_comp(i, j, 0, 1) or \
                is_comp(i, j, 1, -1) or is_comp(i, j, 1, 1):
                return True
    return False

引き分け化どうかを判定します。

game.py
1
2
3
# 引き分けかどうか
def is_draw(self):
    return self.piece_count(self.pieces) + self.piece_count(self.enemy_pieces) == 42

ゲーム終了かどうかを判定します。

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

行動に応じて次の状態を取得します。
石の配置をコピーした後、指定した列で空いているマスの中で一番下の場所に位置を配置します。

game.py
1
2
3
4
5
6
7
8
# 次の状態の取得
def next(self, action):
    pieces = self.pieces.copy()
    for j in range(5,-1,-1):
        if self.pieces[action+j*7] == 0 and self.enemy_pieces[action+j*7] == 0:
            pieces[action+j*7] = 1
            break
    return State(self.enemy_pieces, pieces)

合法手のリストを取得します。

game.py
1
2
3
4
5
6
7
# 合法手のリストの取得
def legal_actions(self):
    actions = []
    for i in range(7):
        if self.pieces[i] == 0 and self.enemy_pieces[i] == 0:
            actions.append(i)
    return actions

先手かどうかを取得します。

game.py
1
2
3
# 先手かどうか
def is_first_player(self):
    return self.piece_count(self.pieces) == self.piece_count(self.enemy_pieces)

ゲーム状態の文字列表示を行います。

game.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 文字列表示
def __str__(self):
    ox = ('o', 'x') if self.is_first_player() else ('x', 'o')
    str = ''
    for i in range(42):
        if self.pieces[i] == 1:
            str += ox[0]
        elif self.enemy_pieces[i] == 1:
            str += ox[1]
        else:
            str += '-'
        if i % 7 == 6:
            str += '\n'
    return str

動作確認用にランダム対ランダムで対戦するコードを追加します。

game.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# ランダムで行動選択
def random_action(state):
    legal_actions = state.legal_actions()
    return legal_actions[random.randint(0, len(legal_actions)-1)]

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

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

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

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

試しに実行したところ下記のような結果となりました。

最終結果（途中略）

次にデュアルネットワークを実装します。
入力シェイプと行動数のパラメータのみ変更しています。
入力シェイプは、自分の石の配置(7x6)と相手の石の配置(7x6)となります。

dual_network.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
# ====================
# デュアルネットワークの作成
# ====================

# パッケージのインポート
from tensorflow.keras.layers import Activation, Add, BatchNormalization, Conv2D, Dense, GlobalAveragePooling2D, Input
from tensorflow.keras.models import Model
from tensorflow.keras.regularizers import l2
from tensorflow.keras import backend as K
import os

# パラメータの準備
DN_FILTERS  = 128 # 畳み込み層のカーネル数（本家は256）
DN_RESIDUAL_NUM =  16 # 残差ブロックの数（本家は19）
DN_INPUT_SHAPE = (7, 6, 2) # 入力シェイプ 自分の石の配置(7x6)と相手の石の配置(7x6)
DN_OUTPUT_SIZE = 7 # 行動数(配置先(7))

# 畳み込み層の作成
def conv(filters):
    return Conv2D(filters, 3, padding='same', use_bias=False,
        kernel_initializer='he_normal', kernel_regularizer=l2(0.0005))

# 残差ブロックの作成
def residual_block():
    def f(x):
        sc = x
        x = conv(DN_FILTERS)(x)
        x = BatchNormalization()(x)
        x = Activation('relu')(x)
        x = conv(DN_FILTERS)(x)
        x = BatchNormalization()(x)
        x = Add()([x, sc])
        x = Activation('relu')(x)
        return x
    return f

# デュアルネットワークの作成
def dual_network():
    # モデル作成済みの場合は無処理
    if os.path.exists('./model/best.h5'):
        return

    # 入力層
    input = Input(shape=DN_INPUT_SHAPE)

    # 畳み込み層
    x = conv(DN_FILTERS)(input)
    x = BatchNormalization()(x)
    x = Activation('relu')(x)

    # 残差ブロック x 16
    for i in range(DN_RESIDUAL_NUM):
        x = residual_block()(x)

    # プーリング層
    x = GlobalAveragePooling2D()(x)

    # ポリシー出力
    p = Dense(DN_OUTPUT_SIZE, kernel_regularizer=l2(0.0005),
                activation='softmax', name='pi')(x)

    # バリュー出力
    v = Dense(1, kernel_regularizer=l2(0.0005))(x)
    v = Activation('tanh', name='v')(v)

    # モデルの作成
    model = Model(inputs=input, outputs=[p,v])

    # モデルの保存
    os.makedirs('./model/', exist_ok=True) # フォルダがない時は生成
    model.save('./model/best.h5') # ベストプレイヤーのモデル

    # モデルの破棄
    K.clear_session()
    del model

モンテカルロ木探索を実装します。変更はありません。

pv_mcts.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
# ====================
# モンテカルロ木探索の作成
# ====================

# パッケージのインポート
from game import State
from dual_network import DN_INPUT_SHAPE
from math import sqrt
from tensorflow.keras.models import load_model
from pathlib import Path
import numpy as np

# パラメータの準備
PV_EVALUATE_COUNT = 50 # 1推論あたりのシミュレーション回数（本家は1600）

# 推論
def predict(model, state):
    # 推論のための入力テ゛ータのシェイフ゜の変換
    a, b, c = DN_INPUT_SHAPE
    x = np.array([state.pieces, state.enemy_pieces])
    x = x.reshape(c, a, b).transpose(1, 2, 0).reshape(1, a, b, c)

    # 推論
    y = model.predict(x, batch_size=1)

    # 方策の取得
    policies = y[0][0][list(state.legal_actions())] # 合法手のみ
    policies /= sum(policies) if sum(policies) else 1 # 合計1の確率分布に変換

    # 価値の取得
    value = y[1][0][0]
    return policies, value

# ノードのリストを試行回数のリストに変換
def nodes_to_scores(nodes):
    scores = []
    for c in nodes:
        scores.append(c.n)
    return scores

# モンテカルロ木探索のスコアの取得
def pv_mcts_scores(model, state, temperature):
    # モンテカルロ木探索のノードの定義
    class Node:
        # ノードの初期化
        def __init__(self, state, p):
            self.state = state # 状態
            self.p = p # 方策
            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

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

            # 子ノードが存在しない時
            if not self.child_nodes:
                # ニューラルネットワークの推論で方策と価値を取得
                policies, value = predict(model, self.state)

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

                # 子ノードの展開
                self.child_nodes = []
                for action, policy in zip(self.state.legal_actions(), policies):
                    self.child_nodes.append(Node(self.state.next(action), policy))
                return value

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

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

        # アーク評価値が最大の子ノードを取得
        def next_child_node(self):
            # アーク評価値の計算
            C_PUCT = 1.0
            t = sum(nodes_to_scores(self.child_nodes))
            pucb_values = []
            for child_node in self.child_nodes:
                pucb_values.append((-child_node.w / child_node.n if child_node.n else 0.0) +
                    C_PUCT * child_node.p * sqrt(t) / (1 + child_node.n))

            # アーク評価値が最大の子ノードを返す
            return self.child_nodes[np.argmax(pucb_values)]

    # 現在の局面のノードの作成
    root_node = Node(state, 0)

    # 複数回の評価の実行
    for _ in range(PV_EVALUATE_COUNT):
        root_node.evaluate()

    # 合法手の確率分布
    scores = nodes_to_scores(root_node.child_nodes)
    if temperature == 0: # 最大値のみ1
        action = np.argmax(scores)
        scores = np.zeros(len(scores))
        scores[action] = 1
    else: # ボルツマン分布でバラつき付加
        scores = boltzman(scores, temperature)
    return scores

# モンテカルロ木探索で行動選択
def pv_mcts_action(model, temperature=0):
    def pv_mcts_action(state):
        scores = pv_mcts_scores(model, state, temperature)
        return np.random.choice(state.legal_actions(), p=scores)
    return pv_mcts_action

# ボルツマン分布
def boltzman(xs, temperature):
    xs = [x ** (1 / temperature) for x in xs]
    return [x / sum(xs) for x in xs]

セルフプレイを実装します。変更はありません。

self_play.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
# ====================
# セルフプレイ部
# ====================

# パッケージのインポート
from game import State
from pv_mcts import pv_mcts_scores
from dual_network import DN_OUTPUT_SIZE
from datetime import datetime
from tensorflow.keras.models import load_model
from tensorflow.keras import backend as K
from pathlib import Path
import numpy as np
import pickle
import os

# パラメータの準備
SP_GAME_COUNT = 500 # セルフプレイを行うゲーム数（本家は25000）
SP_TEMPERATURE = 1.0 # ボルツマン分布の温度パラメータ

# 先手プレイヤーの価値
def first_player_value(ended_state):
    # 1:先手勝利, -1:先手敗北, 0:引き分け
    if ended_state.is_lose():
        return -1 if ended_state.is_first_player() else 1
    return 0

# 学習データの保存
def write_data(history):
    now = datetime.now()
    os.makedirs('./data/', exist_ok=True) # フォルダがない時は生成
    path = './data/{:04}{:02}{:02}{:02}{:02}{:02}.history'.format(
        now.year, now.month, now.day, now.hour, now.minute, now.second)
    with open(path, mode='wb') as f:
        pickle.dump(history, f)

# 1ゲームの実行
def play(model):
    # 学習データ
    history = []

    # 状態の生成
    state = State()

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

        # 合法手の確率分布の取得
        scores = pv_mcts_scores(model, state, SP_TEMPERATURE)

        # 学習データに状態と方策を追加
        policies = [0] * DN_OUTPUT_SIZE
        for action, policy in zip(state.legal_actions(), scores):
            policies[action] = policy
        history.append([[state.pieces, state.enemy_pieces], policies, None])

        # 行動の取得
        action = np.random.choice(state.legal_actions(), p=scores)

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

    # 学習データに価値を追加
    value = first_player_value(state)
    for i in range(len(history)):
        history[i][2] = value
        value = -value
    return history

# セルフプレイ
def self_play():
    # 学習データ
    history = []

    # ベストプレイヤーのモデルの読み込み
    model = load_model('./model/best.h5')

    # 複数回のゲームの実行
    for i in range(SP_GAME_COUNT):
        # 1ゲームの実行
        h = play(model)
        history.extend(h)

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

    # 学習データの保存
    write_data(history)

    # モデルの破棄
    K.clear_session()
    del model

パラメータ更新部を実装します。変更はありません。

train_network.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
# ====================
# パラメータ更新部
# ====================

# パッケージのインポート
from dual_network import DN_INPUT_SHAPE
from tensorflow.keras.callbacks import LearningRateScheduler, LambdaCallback
from tensorflow.keras.models import load_model
from tensorflow.keras import backend as K
from pathlib import Path
import numpy as np
import pickle

# パラメータの準備
RN_EPOCHS = 100 # 学習回数

# 学習データの読み込み
def load_data():
    history_path = sorted(Path('./data').glob('*.history'))[-1]
    with history_path.open(mode='rb') as f:
        return pickle.load(f)

# デュアルネットワークの学習
def train_network():
    # 学習データの読み込み
    history = load_data()
    xs, y_policies, y_values = zip(*history)

    # 学習のための入力テ゛ータのシェイフ゜の変換
    a, b, c = DN_INPUT_SHAPE
    xs = np.array(xs)
    xs = xs.reshape(len(xs), c, a, b).transpose(0, 2, 3, 1)
    y_policies = np.array(y_policies)
    y_values = np.array(y_values)

    # ベストプレイヤーのモデルの読み込み
    model = load_model('./model/best.h5')

    # モデルのコンパイル
    model.compile(loss=['categorical_crossentropy', 'mse'], optimizer='adam')

    # 学習率
    def step_decay(epoch):
        x = 0.001
        if epoch >= 50: x = 0.0005
        if epoch >= 80: x = 0.00025
        return x
    lr_decay = LearningRateScheduler(step_decay)

    # 出力
    print_callback = LambdaCallback(
        on_epoch_begin=lambda epoch,logs:
                print('\rTrain {}/{}'.format(epoch + 1,RN_EPOCHS), end=''))

    # 学習の実行
    model.fit(xs, [y_policies, y_values], batch_size=128, epochs=RN_EPOCHS,
            verbose=0, callbacks=[lr_decay, print_callback])
    print('')

    # 最新プレイヤーのモデルの保存
    model.save('./model/latest.h5')

    # モデルの破棄
    K.clear_session()
    del model

新パラメータ評価部を実装します。変更はありません。

evaluate_network.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
# ====================
# 新パラメータ評価部
# ====================

# パッケージのインポート
from game import State
from pv_mcts import pv_mcts_action
from tensorflow.keras.models import load_model
from tensorflow.keras import backend as K
from pathlib import Path
from shutil import copy
import numpy as np

# パラメータの準備
EN_GAME_COUNT = 10 # 1評価あたりのゲーム数（本家は400）
EN_TEMPERATURE = 1.0 # ボルツマン分布の温度

# 先手プレイヤーのポイント
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 update_best_player():
    copy('./model/latest.h5', './model/best.h5')
    print('Change BestPlayer')

# ネットワークの評価
def evaluate_network():
    # 最新プレイヤーのモデルの読み込み
    model0 = load_model('./model/latest.h5')

    # ベストプレイヤーのモデルの読み込み
    model1 = load_model('./model/best.h5')

    # PV MCTSで行動選択を行う関数の生成
    next_action0 = pv_mcts_action(model0, EN_TEMPERATURE)
    next_action1 = pv_mcts_action(model1, EN_TEMPERATURE)
    next_actions = (next_action0, next_action1)

    # 複数回の対戦を繰り返す
    total_point = 0
    for i in range(EN_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, EN_GAME_COUNT), end='')
    print('')

    # 平均ポイントの計算
    average_point = total_point / EN_GAME_COUNT
    print('AveragePoint', average_point)

    # モデルの破棄
    K.clear_session()
    del model0
    del model1

    # ベストプレイヤーの交代
    if average_point > 0.5:
        update_best_player()
        return True
    else:
        return False

学習サイクルを実行します。ベストプレイヤーの評価は削除しています。
この処理を実行するとベストプレイヤーのモデルが作成されます。(model/best.h5)

学習完了までにCorei5、メモリ4G、GPUなしのPCでまる２日ほどかかりました。

train_cycle.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# ====================
# 学習サイクルの実行
# ====================

# パッケージのインポート
from dual_network import dual_network
from self_play import self_play
from train_network import train_network
from evaluate_network import evaluate_network

# デュアルネットワークの作成
dual_network()

for i in range(10):
    print('Train',i,'====================')
    # セルフプレイ部
    self_play()

    # パラメータ更新部
    train_network()

    # 新パラメータ評価部
    evaluate_network()

ゲームUIを実装します。

human_play.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# ====================
# 人とAIの対戦
# ====================

# パッケージのインポート
from game import State
from pv_mcts import pv_mcts_action
from tensorflow.keras.models import load_model
from pathlib import Path
from threading import Thread
import tkinter as tk

# ベストプレイヤーのモデルの読み込み
model = load_model('./model/best.h5')

# ゲームUIの定義
class GameUI(tk.Frame):
    # 初期化

ゲームの初期化を行います。
次の３つを準備します。

  • ゲーム状態
  • モンテカルロ木探索で行動選択を行う関数
  • キャンバス
human_play.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
def __init__(self, master=None, model=None):
    tk.Frame.__init__(self, master)
    self.master.title('コネクトフォー')

    # ゲーム状態の生成
    self.state = State()

    # PV MCTSで行動選択を行う関数の生成
    self.next_action = pv_mcts_action(model, 0.0)

    # キャンバスの生成
    self.c = tk.Canvas(self, width = 280, height = 240, highlightthickness = 0)
    self.c.bind('<Button-1>', self.turn_of_human)
    self.c.pack()

    # 描画の更新
    self.on_draw()

人間のターンの処理を行います。

  1. ゲーム終了時はゲーム状態を初期化します。
  2. 先手でない時は操作不可とします。
  3. クリック位置を行動に変換します。
  4. 合法手でない時、無処理とします。
  5. 合法手の場合、次の状態を取得して描画の更新を行います。
  6. AIのターンへ遷移します。
human_play.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
# 人間のターン
def turn_of_human(self, event):
    # ゲーム終了時
    if self.state.is_done():
        self.state = State()
        self.on_draw()
        return

    # 先手でない時
    if not self.state.is_first_player():
        return

    # クリック位置を行動に変換
    x = int(event.x/40)
    if x < 0 or 6 < x: # 範囲外
        return
    action = x

    # 合法手でない時
    if not (action in self.state.legal_actions()):
        return

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

    # AIのターン
    self.master.after(1, self.turn_of_ai)

AIのターンの処理を行います。

  1. ゲーム終了時は無処理とします。
  2. デュアルネットワークで行動を取得します。
  3. 取得した行動に応じて次の状態を取得し、描画の更新を行います。
human_play.py
1
2
3
4
5
6
7
8
9
10
11
12
# AIのターン
def turn_of_ai(self):
    # ゲーム終了時
    if self.state.is_done():
        return

    # 行動の取得
    action = self.next_action(self.state)

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

石の描画を行います。
indexはマス番号、first_playerは先手かどうかを表します。
先手は赤色のマル、後手は黄色マルで描画します。

human_play.py
1
2
3
4
5
6
7
8
# 石の描画
def draw_piece(self, index, first_player):
    x = (index%7)*40+5
    y = int(index/7)*40+5
    if first_player:
        self.c.create_oval(x, y, x+30, y+30, width = 1.0, fill = '#FF0000')
    else:
        self.c.create_oval(x, y, x+30, y+30, width = 1.0, fill = '#FFFF00')

描画の更新を行います。すべてのマス目と石を描画します。

human_play.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 描画の更新
def on_draw(self):
    self.c.delete('all')
    self.c.create_rectangle(0, 0, 280, 240, width = 0.0, fill = '#00A0FF')
    for i in range(42):
        x = (i%7)*40+5
        y = int(i/7)*40+5
        self.c.create_oval(x, y, x+30, y+30, width = 1.0, fill = '#FFFFFF')

    for i in range(42):
        if self.state.pieces[i] == 1:
            self.draw_piece(i, self.state.is_first_player())
        if self.state.enemy_pieces[i] == 1:
            self.draw_piece(i, not self.state.is_first_player())

ゲームUIを実行します。

human_play.py
1
2
3
4
# ゲームUIの実行
f = GameUI(model=model)
f.pack()
f.mainloop()

human_play.pyを実行するとコネクトフォーが始まりAIと対戦することができます。

対戦中

けっこう本気でやっても勝てないくらい強かったです。

参考

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

強化学習 AlphaZero 17 (三目並べ AlphaZero9)

AlphaZeroで学習したモデルと人間で対戦するゲームを作成します。

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

human_play.py
1
2
3
4
5
6
7
8
# パッケージのインポート
from game import State
from pv_mcts import pv_mcts_action
from game import State, random_action, alpha_beta_action, mcts_action
from tensorflow.keras.models import load_model
from pathlib import Path
from threading import Thread
import tkinter as tk

学習したベストプレイヤーのモデルを読み込みます。

human_play.py
1
2
# ベストプレイヤーのモデルの読み込み
model = load_model('./model/best.h5')

ゲームUIのクラスを定義します。
メソッドは下記のとおりです。

  • init ゲームUIを初期化します。
  • turn_of_human 人間のターンを実行します。
  • turn_of _ai AIのターンを実行します。
  • draw_piece 石の描画を行います。
  • on_draw 描画の更新を行います。

初期化処理ではゲーム状態と行動選択を行う関数を生成し、キャンバスを用意します。

human_play.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# ゲームUIの定義
class GameUI(tk.Frame):
    # 初期化
    def __init__(self, master=None, model=None):
        tk.Frame.__init__(self, master)
        self.master.title('三目並べ')

        # ゲーム状態の生成
        self.state = State()

        # PV MCTSで行動選択を行う関数の生成
        self.next_action = pv_mcts_action(model, 0.0)

        # キャンバスの生成
        self.c = tk.Canvas(self, width = 240, height = 240, highlightthickness = 0)
        self.c.bind('<Button-1>', self.turn_of_human)
        #self.c.bind('<Button-1>', self.turn_of_ai)
        self.c.pack()

        # 描画の更新
        self.on_draw()

人間のターンの処理は下記のようになります。

  1. ゲーム終了時はゲームを初期化状態に戻します。
  2. 先手でないときは操作不可とします。
  3. クリック位置を行動に変換します。
  4. 合法手でないとき無処理とします。
  5. 次の状態を取得します。
  6. AIのターンへ遷移します。
human_play.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
# 人間のターン
def turn_of_human(self, event):
    # ゲーム終了時
    if self.state.is_done():
        self.state = State()
        self.on_draw()
        return

    # 先手でない時
    if not self.state.is_first_player():
        return

    # クリック位置を行動に変換
    x = int(event.x/80)
    y = int(event.y/80)
    if x < 0 or 2 < x or y < 0 or 2 < y: # 範囲外
        return
    action = x + y * 3

    # 合法手でない時
    if not (action in self.state.legal_actions()):
        return

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

    # AIのターン
    self.master.after(1, self.turn_of_ai)

AIのターンは下記のようになります。

  1. ゲーム終了時は無処理とします。
  2. ニューラルネットワークで行動を取得します。
  3. 次の状態を取得します。
human_play.py
1
2
3
4
5
6
7
8
9
10
11
12
13
# AIのターン
def turn_of_ai(self):
    # ゲーム終了時
    if self.state.is_done():
        return

    # 先手でない時
    # 行動の取得
    action = self.next_action(self.state)

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

石の描画を行います。

human_play.py
1
2
3
4
5
6
7
8
9
# 石の描画
def draw_piece(self, index, first_player):
    x = (index%3)*80+10
    y = int(index/3)*80+10
    if first_player:
        self.c.create_oval(x, y, x+60, y+60, width = 2.0, outline = '#FFFFFF')
    else:
        self.c.create_line(x, y, x+60, y+60, width = 2.0, fill = '#5D5D5D')
        self.c.create_line(x+60, y, x, y+60, width = 2.0, fill = '#5D5D5D')

描画を更新します。全てのマス目と石を描画します。

human_play.py
1
2
3
4
5
6
7
8
9
10
11
12
13
# 描画の更新
def on_draw(self):
    self.c.delete('all')
    self.c.create_rectangle(0, 0, 240, 240, width = 0.0, fill = '#00A0FF')
    self.c.create_line(80, 0, 80, 240, width = 2.0, fill = '#0077BB')
    self.c.create_line(160, 0, 160, 240, width = 2.0, fill = '#0077BB')
    self.c.create_line(0, 80, 240, 80,  width = 2.0, fill = '#0077BB')
    self.c.create_line(0, 160, 240, 160, width = 2.0, fill = '#0077BB')
    for i in range(9):
        if self.state.pieces[i] == 1:
            self.draw_piece(i, self.state.is_first_player())
        if self.state.enemy_pieces[i] == 1:
            self.draw_piece(i, not self.state.is_first_player())

ゲームUIを実行します。

human_play.py
1
2
3
4
# ゲームUIの実行
f = GameUI(model=model)
f.pack()
f.mainloop()

対戦状況は下記の通りです。

対戦中

何回か対戦してみましたが、それほど強いイメージはありませんでした。

参考

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

強化学習 AlphaZero 16 (三目並べ AlphaZero8)

AlphaZeroの学習サイクルを実装します。

これまで実装したコードを次の手順で実行していきます。

  1. デュアルネットワークを作成します。
  2. セルフプレイを実行します。
  3. パラメータ更新部を実行します。
  4. 新パラメータ評価部を実行します。
  5. ベストプレイヤーを評価します。

2～5を10回(10サイクル)実行します。

では実装に移ります。
まずは必要なパッケージをインポートします。

train_cycle.py
1
2
3
4
5
6
# パッケージのインポート
from dual_network import dual_network
from self_play import self_play
from train_network import train_network
from evaluate_network import evaluate_network
from evaluate_best_player import evaluate_best_player

デュアルネットワークを作成します。
すでにベストプレイヤーのモデル(model/best.h5)がある場合は無処理となります。

train_cycle.py
1
2
# デュアルネットワークの作成
dual_network()

学習サイクルを10回(10サイクル)繰り返し実行します。

train_cycle.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
for i in range(10):
    print('Train',i,'====================')
    # セルフプレイ部
    self_play()

    # パラメータ更新部
    train_network()

    # 新パラメータ評価部
    update_best_player = evaluate_network()

    # ベストプレイヤーの評価
    if update_best_player:
        evaluate_best_player()

10サイクルの学習を完了するまでは、Corei5・メモリ8GB・GPUなしのPCで10時間ほどかかりました。

何回か学習サイクルを実行し、ベストプレイヤーの評価を行った結果は下記の通りです。

10サイクル 20サイクル 30サイクル

学習サイクルを増やすほど各勝率が増えていくかと予想していたのですがそうでもないようです。
ただ最強ロジックのアルファベータ法と互角に勝負できているのでAlphaZeroという手法は秀逸だと思います。

参考

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

強化学習 AlphaZero 15 (三目並べ AlphaZero7)

AlphaZeroのベストプレイヤーの評価を行います。

評価方法としては、下記の３手法と対戦させて勝率を表示します。

  • ランダム
  • アルファベータ法 <= 最強
  • モンテカルロ木探索 <= AlphaZeroのベース

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

evaluate_best_player.py
1
2
3
4
5
6
7
# パッケージのインポート
from game import State, random_action, alpha_beta_action, mcts_action
from pv_mcts import pv_mcts_action
from tensorflow.keras.models import load_model
from tensorflow.keras import backend as K
from pathlib import Path
import numpy as np

パラメータを準備します。
EP_GAME_COUNTは勝率を計算するために行うゲーム数になります。

evaluate_best_player.py
1
2
# パラメータの準備
EP_GAME_COUNT = 10  # 1評価あたりのゲーム数

最終局面から先手プレイヤーのポイントを返す関数を定義します。

evaluate_best_player.py
1
2
3
4
5
6
# 先手プレイヤーのポイント
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

１ゲームを最後まで実行して、先手プレイヤーのポイントを返す関数を定義します。

evaluate_best_player.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 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)

任意のアルゴリズムに対して10ゲーム行い勝率を算出します。

evaluate_best_player.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 任意のアルゴリズムの評価
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('\r Evaluate {}/{}'.format(i + 1, EP_GAME_COUNT), end='')
    print('')

    # 平均ポイントの計算
    average_point = total_point / EP_GAME_COUNT
    print(label, '{:.1f}%'.format(average_point * 100))

ベストプレイヤーの評価を行います。
ベストプレイヤーとランダム、アルファベータ法、モンテカルロ木探索との対戦を行いそれぞれの勝率を表示します。

evaluate_best_player.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
# ベストプレイヤーの評価
def evaluate_best_player():
    # ベストプレイヤーのモデルの読み込み
    model = load_model('./model/best.h5')

    # PV MCTSで行動選択を行う関数の生成
    next_pv_mcts_action = pv_mcts_action(model, 0.0)

    print()
    # VSランダム
    next_actions = (next_pv_mcts_action, random_action)
    evaluate_algorithm_of(' 対 ランダム', next_actions)
    print()

    # VSアルファベータ法
    next_actions = (next_pv_mcts_action, alpha_beta_action)
    evaluate_algorithm_of(' 対 アルファベータ法', next_actions)
    print()

    # VSモンテカルロ木探索
    next_actions = (next_pv_mcts_action, mcts_action)
    evaluate_algorithm_of(' 対 モンテカルロ木探索', next_actions)
    print()

    # モデルの破棄
    K.clear_session()
    del model

動作確認のための実行コードを実装します。

evaluate_best_player.py
1
2
3
# 動作確認
if __name__ == '__main__':
    evaluate_best_player()

実行結果は下記の通りです。

結果

現在のベストプレイヤーは自己対戦500回、学習回数100回ですが最強のアルファベータ法に対しては勝率25%です。
もうすこし学習サイクルを増やして、強いモデルにしたいところです。

参考

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

強化学習 AlphaZero 14 (三目並べ AlphaZero6)

AlphaZeroの新パラメータ評価部を作成します。

ベストプレイヤー(model/best.h5)と最新プレイヤー(model/latest.h5)を対戦させて、より強い方をベストプレイヤーとして残します。

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

evaluate_network.py
1
2
3
4
5
6
7
8
9
10
11
12
# ====================
# 新パラメータ評価部
# ====================

# パッケージのインポート
from game import State
from pv_mcts import pv_mcts_action
from tensorflow.keras.models import load_model
from tensorflow.keras import backend as K
from pathlib import Path
from shutil import copy
import numpy as np

パラメータを準備します。
ボルツマン分布の温度パラメータはバラつきに関するパラメータになります。

evaluate_network.py
1
2
3
# パラメータの準備
EN_GAME_COUNT = 10 # 1評価あたりのゲーム数（本家は400）
EN_TEMPERATURE = 1.0 # ボルツマン分布の温度パラメータ

最終局面から先手プレイヤーのポイントを返す関数を定義します。

evaluate_network.py
1
2
3
4
5
6
# 先手プレイヤーのポイント
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

１ゲームを最後まで実行して、先手プレイヤーのポイントを返す関数を定義します。

evaluate_network.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 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)

最新プレイヤーをベストプレイヤーに置き換える関数を定義します。

evaluate_network.py
1
2
3
4
# ベストプレイヤーの交代
def update_best_player():
    copy('./model/latest.h5', './model/best.h5')
    print('Change BestPlayer')

ネットワークの評価を行う関数を定義します。

最新プレイヤーとベストプレイヤーのモデルを読み込み、１０回対戦させます。
勝率が高い方をベストプレイヤーとして保存します。

evaluate_network.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
# ネットワークの評価
def evaluate_network():
    # 最新プレイヤーのモデルの読み込み
    model0 = load_model('./model/latest.h5')

    # ベストプレイヤーのモデルの読み込み
    model1 = load_model('./model/best.h5')

    # PV MCTSで行動選択を行う関数の生成
    next_action0 = pv_mcts_action(model0, EN_TEMPERATURE)
    next_action1 = pv_mcts_action(model1, EN_TEMPERATURE)
    next_actions = (next_action0, next_action1)

    # 複数回の対戦を繰り返す
    total_point = 0
    for i in range(EN_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, EN_GAME_COUNT), end='')
    print('')

    # 平均ポイントの計算
    average_point = total_point / EN_GAME_COUNT
    print('AveragePoint', average_point)

    # モデルの破棄
    K.clear_session()
    del model0
    del model1

    # ベストプレイヤーの交代
    if average_point > 0.5:
        update_best_player()
        return True
    else:
        return False

動作確認のための実行コードを実装します。

evaluate_network.py
1
2
3
# 動作確認
if __name__ == '__main__':
    evaluate_network()

実行結果は下記の通りです。

結果

今回は最新プレイヤーの勝率が70%だったので、新たなベストプレイヤーとして更新しています。

参考

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

強化学習 AlphaZero 13 (三目並べ AlphaZero5)

AlphaZeroのパラメータ更新部を作成します。

前回行ったセルフプレイで作成した学習データ(data/*.history)とベストプレイヤーのモデル(model/best.h5)が入力となります。
学習を行ったモデルを最新プレイヤー(model/latest.h5)として出力します。

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

train_network.py
1
2
3
4
5
6
7
8
9
10
11
12
# ====================
# パラメータ更新部
# ====================

# パッケージのインポート
from dual_network import DN_INPUT_SHAPE
from tensorflow.keras.callbacks import LearningRateScheduler, LambdaCallback
from tensorflow.keras.models import load_model
from tensorflow.keras import backend as K
from pathlib import Path
import numpy as np
import pickle

パラメータを定義します。RN_EPOCHSは学習回数を表します。

train_network.py
1
2
# パラメータの準備
RN_EPOCHS = 100 # 学習回数

セルフデータ部で保存した学習データ（data/*.history）を読み込む関数を定義します。

train_network.py
1
2
3
4
5
# 学習データの読み込み
def load_data():
    history_path = sorted(Path('./data').glob('*.history'))[-1]
    with history_path.open(mode='rb') as f:
        return pickle.load(f)

デュアルネットワークの学習を行います。学習手順は以下の通りです。

  1. 学習データの読み込みを行います。
  2. 学習のために入力データのシェイプを変換します。
  3. ベストプレイヤーのモデルを読み込みます。
  4. モデルをコンパイルします。
  5. 学習率をステップ数に応じて減らしていきます。
  6. １ゲームごとに経過を出力します。
  7. 学習を実行します。
  8. 最新プレイヤーのモデルを保存します。
  9. モデルを破棄します。
train_network.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
# デュアルネットワークの学習
def train_network():
    # 学習データの読み込み
    history = load_data()
    xs, y_policies, y_values = zip(*history)

    # 学習のための入力テ゛ータのシェイフ゜の変換
    a, b, c = DN_INPUT_SHAPE
    xs = np.array(xs)
    xs = xs.reshape(len(xs), c, a, b).transpose(0, 2, 3, 1)
    y_policies = np.array(y_policies)
    y_values = np.array(y_values)

    # ベストプレイヤーのモデルの読み込み
    model = load_model('./model/best.h5')

    # モデルのコンパイル
    model.compile(loss=['categorical_crossentropy', 'mse'], optimizer='adam')

    # 学習率
    def step_decay(epoch):
        x = 0.001
        if epoch >= 50: x = 0.0005
        if epoch >= 80: x = 0.00025
        return x
    lr_decay = LearningRateScheduler(step_decay)

    # 出力
    print_callback = LambdaCallback(
        on_epoch_begin=lambda epoch,logs:
                print('\rTrain {}/{}'.format(epoch + 1,RN_EPOCHS), end=''))

    # 学習の実行
    model.fit(xs, [y_policies, y_values], batch_size=128, epochs=RN_EPOCHS,
            verbose=0, callbacks=[lr_decay, print_callback])
    print('')

    # 最新プレイヤーのモデルの保存
    model.save('./model/latest.h5')

    # モデルの破棄
    K.clear_session()
    del model

動作確認のための実行コードを実装します。

train_network.py
1
2
3
# 動作確認
if __name__ == '__main__':
    train_network()

実行結果は下記の通りです。

結果

modelフォルダにlatest.h5ファイル(最新プレイヤー)が作成されます。
最新プレイヤーはベストプレイヤーをコピーし、学習を行ったモデルとなります。

参考

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

強化学習 AlphaZero 12 (三目並べ AlphaZero4)

AlphaZeroのセルフプレイ部を作成します。
セルフプレイ（自己対戦）では、デュアルネットワークの学習に利用する学習データを作成します。

まずセルフプレイに必要なパッケージをインポートします。

self_play.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# ====================
# セルフプレイ部
# ====================

# パッケージのインポート
from game import State
from pv_mcts import pv_mcts_scores
from dual_network import DN_OUTPUT_SIZE
from datetime import datetime
from tensorflow.keras.models import load_model
from tensorflow.keras import backend as K
from pathlib import Path
import numpy as np
import pickle
import os

セルフプレイのパラメータを定義します。

self_play.py
1
2
3
# パラメータの準備
SP_GAME_COUNT = 500 # セルフプレイを行うゲーム数（本家は25000）
SP_TEMPERATURE = 1.0 # ボルツマン分布の温度パラメータ

最終局面から先手プレイヤーの価値を計算します。
先手が勝利の時は1、敗北の時は-1、引き分けの時は0を返します。

self_play.py
1
2
3
4
5
6
# 先手プレイヤーの価値
def first_player_value(ended_state):
    # 1:先手勝利, -1:先手敗北, 0:引き分け
    if ended_state.is_lose():
        return -1 if ended_state.is_first_player() else 1
    return 0

セルフプレイを実行して収集した学習データ（状態と方策、価値のセット）をファイル出力します。

self_play.py
1
2
3
4
5
6
7
8
9
# 学習データの保存
def write_data(history):
    now = datetime.now()
    os.makedirs('./data/', exist_ok=True) # フォルダがない時は生成
    path = './data/{:04}{:02}{:02}{:02}{:02}{:02}.history'.format(
        now.year, now.month, now.day, now.hour, now.minute, now.second)
    with open(path, mode='wb') as f:
        pickle.dump(history, f)

１ゲーム分実行して、学習データを収集します。

ステップ毎に、pv_mcts_scores()で「方策」を取得します。
これは合法手の確率分布なので、すべての手の確率分布に変換して保持します。

「価値」は１ゲーム終了したときにfirst_player_value()で計算します。

self_play.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
# 1ゲームの実行
def play(model):
    # 学習データ
    history = []

    # 状態の生成
    state = State()

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

        # 合法手の確率分布の取得
        scores = pv_mcts_scores(model, state, SP_TEMPERATURE)

        # 学習データに状態と方策を追加
        policies = [0] * DN_OUTPUT_SIZE
        for action, policy in zip(state.legal_actions(), scores):
            policies[action] = policy
        history.append([[state.pieces, state.enemy_pieces], policies, None])

        # 行動の取得
        action = np.random.choice(state.legal_actions(), p=scores)

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

    # 学習データに価値を追加
    value = first_player_value(state)
    for i in range(len(history)):
        history[i][2] = value
        value = -value
    return history

セルフプレイを行います。
ベストプレイヤーのモデルを読み込み、指定回数分のゲームを行います。
最後に、収集した学習データをファイル出力します。

self_play.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
# セルフプレイ
def self_play():
    # 学習データ
    history = []

    # ベストプレイヤーのモデルの読み込み
    model = load_model('./model/best.h5')

    # 複数回のゲームの実行
    for i in range(SP_GAME_COUNT):
        # 1ゲームの実行
        h = play(model)
        history.extend(h)

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

    # 学習データの保存
    write_data(history)

    # モデルの破棄
    K.clear_session()
    del model

動作確認のための実行コードを実装します。

self_play.py
1
2
3
# 動作確認
if __name__ == '__main__':
    self_play()

実行結果は下記の通りです。

結果

dataフォルダ配下に[タイムスタンプ].historyファイルが作成されます。
次回はこの学習データを使ってデュアルネットワークの学習を行います。

参考

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

強化学習 AlphaZero 11 (三目並べ AlphaZero3)

AlphaZeroのベースとなるモンテカルロ木探索を作成します。

ニューラルネットワークで「方策」と「価値」を取得します。
「方策」はアーク評価値の計算、「価値」は累計価値の更新に利用します。

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

pv_mcts.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# ====================
# モンテカルロ木探索の作成
# ====================

# パッケージのインポート
from game import State
from dual_network import DN_INPUT_SHAPE
from math import sqrt
from tensorflow.keras.models import load_model
from pathlib import Path
import numpy as np

# パラメータの準備
PV_EVALUATE_COUNT = 50 # 1推論あたりのシミュレーション回数（本家は1600）

ニューラルネットワークの推論を行います。

  1. 推論のために入力データのシェイプを変換します。
    自分の石の配置と相手の石の配置を(1, 3, 3, 2)に変換します。
  2. 推論を行います。
    model.predict()を使います。
  3. 方策を取得します。
    合法手のみ抽出し、合計で割って合法手のみの確率分布に変換します。
  4. 価値を取得します。
    価値の配列から値のみを抽出します。
pv_mcts.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 推論
def predict(model, state):
    # 推論のための入力テ゛ータのシェイフ゜の変換
    a, b, c = DN_INPUT_SHAPE
    x = np.array([state.pieces, state.enemy_pieces])
    x = x.reshape(c, a, b).transpose(1, 2, 0).reshape(1, a, b, c)

    # 推論
    y = model.predict(x, batch_size=1)

    # 方策の取得
    policies = y[0][0][list(state.legal_actions())] # 合法手のみ
    policies /= sum(policies) if sum(policies) else 1 # 合計1の確率分布に変換

    # 価値の取得
    value = y[1][0][0]
    return policies, value

ノードのリストを試行回数のリストに変換します。

pv_mcts.py
1
2
3
4
5
6
# ノードのリストを試行回数のリストに変換
def nodes_to_scores(nodes):
    scores = []
    for c in nodes:
        scores.append(c.n)
    return scores

モンテカルロ木探索のスコアを取得します。

  1. モンテカルロ木探索のノードを定義します。
  2. 現在の局面のノードを作成します。
  3. 複数回の評価を実行します。

4.合法手の確率分布を返します。

pv_mcts.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
# モンテカルロ木探索のスコアの取得
def pv_mcts_scores(model, state, temperature):
    # モンテカルロ木探索のノードの定義
    class Node:
        # ノードの初期化
        def __init__(self, state, p):
            self.state = state # 状態
            self.p = p # 方策
            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

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

            # 子ノードが存在しない時
            if not self.child_nodes:
                # ニューラルネットワークの推論で方策と価値を取得
                policies, value = predict(model, self.state)

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

                # 子ノードの展開
                self.child_nodes = []
                for action, policy in zip(self.state.legal_actions(), policies):
                    self.child_nodes.append(Node(self.state.next(action), policy))
                return value

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

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

        # アーク評価値が最大の子ノードを取得
        def next_child_node(self):
            # アーク評価値の計算
            C_PUCT = 1.0
            t = sum(nodes_to_scores(self.child_nodes))
            pucb_values = []
            for child_node in self.child_nodes:
                pucb_values.append((-child_node.w / child_node.n if child_node.n else 0.0) +
                    C_PUCT * child_node.p * sqrt(t) / (1 + child_node.n))

            # アーク評価値が最大の子ノードを返す
            return self.child_nodes[np.argmax(pucb_values)]

    # 現在の局面のノードの作成
    root_node = Node(state, 0)

    # 複数回の評価の実行
    for _ in range(PV_EVALUATE_COUNT):
        root_node.evaluate()

    # 合法手の確率分布
    scores = nodes_to_scores(root_node.child_nodes)
    if temperature == 0: # 最大値のみ1
        action = np.argmax(scores)
        scores = np.zeros(len(scores))
        scores[action] = 1
    else: # ボルツマン分布でバラつき付加
        scores = boltzman(scores, temperature)
    return scores

モンテカルロ木探索で行動を選択します。

pv_mcts.py
1
2
3
4
5
6
7
# モンテカルロ木探索で行動選択
def pv_mcts_action(model, temperature=0):
    def pv_mcts_action(state):
        scores = pv_mcts_scores(model, state, temperature)
        return np.random.choice(state.legal_actions(), p=scores)
    return pv_mcts_action

ボルツマン分布によるバラつきを付加します。

pv_mcts.py
1
2
3
4
5
# ボルツマン分布
def boltzman(xs, temperature):
    xs = [x ** (1 / temperature) for x in xs]
    return [x / sum(xs) for x in xs]

動作確認のためにモンテカルロ木探索行動選択「pv_mcts_action」を使ってゲーム終了まで実行します。

pv_mcts.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
# 動作確認
if __name__ == '__main__':
    # モデルの読み込み
    path = sorted(Path('./model').glob('*.h5'))[-1]
    model = load_model(str(path))

    # 状態の生成
    state = State()

    # モンテカルロ木探索で行動取得を行う関数の生成
    next_action = pv_mcts_action(model, 1.0)

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

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

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

        # 文字列表示
        print(state)

実行結果は下記の通りです。

結果

今回は引き分けとなりました。

参考

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

強化学習 AlphaZero 10 (三目並べ AlphaZero2)

AlphaZeroでは局面に応じて「方策」「価値」を出力する「デュアルネットワーク」を使います。
ResNetの残差ブロックでゲームの盤面の特徴を抽出し、「ポリシー出力」と「バリュー出力」の２つに分岐させて、「方策」（次の一手）と「価値」（勝敗予測）の２つを推論します。

dual_network.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
# ====================
# デュアルネットワークの作成
# ====================

# パッケージのインポート
from tensorflow.keras.layers import Activation, Add, BatchNormalization, Conv2D, Dense, GlobalAveragePooling2D, Input
from tensorflow.keras.models import Model
from tensorflow.keras.regularizers import l2
from tensorflow.keras import backend as K
import os

# パラメータの準備
DN_FILTERS  = 128 # 畳み込み層のカーネル数（本家は256）
DN_RESIDUAL_NUM =  16 # 残差ブロックの数（本家は19）
DN_INPUT_SHAPE = (3, 3, 2) # 入力シェイプ
DN_OUTPUT_SIZE = 9 # 行動数(配置先(3*3))

# 畳み込み層の作成
def conv(filters):
    return Conv2D(filters, 3, padding='same', use_bias=False,
        kernel_initializer='he_normal', kernel_regularizer=l2(0.0005))

# 残差ブロックの作成
def residual_block():
    def f(x):
        sc = x
        x = conv(DN_FILTERS)(x)
        x = BatchNormalization()(x)
        x = Activation('relu')(x)
        x = conv(DN_FILTERS)(x)
        x = BatchNormalization()(x)
        x = Add()([x, sc])
        x = Activation('relu')(x)
        return x
    return f

デュアルネットワークの作成手順は次の通りです。

  1. モデル作成済みの場合は何も処理しません。
  2. モデルを作成します。
    入力層、畳み込み層、残差ブロック×１６、プーリング層、ポリシー出力、バリュー出力、モデルと順番に作成します。
  3. モデルを保存します。
  4. モデルを破棄します。
dual_network.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
# デュアルネットワークの作成
def dual_network():
    # モデル作成済みの場合は無処理
    if os.path.exists('./model/best.h5'):
        return

    # 入力層
    input = Input(shape=DN_INPUT_SHAPE)

    # 畳み込み層
    x = conv(DN_FILTERS)(input)
    x = BatchNormalization()(x)
    x = Activation('relu')(x)

    # 残差ブロック x 16
    for i in range(DN_RESIDUAL_NUM):
        x = residual_block()(x)

    # プーリング層
    x = GlobalAveragePooling2D()(x)

    # ポリシー出力
    p = Dense(DN_OUTPUT_SIZE, kernel_regularizer=l2(0.0005),
                activation='softmax', name='pi')(x)

    # バリュー出力
    v = Dense(1, kernel_regularizer=l2(0.0005))(x)
    v = Activation('tanh', name='v')(v)

    # モデルの作成
    model = Model(inputs=input, outputs=[p,v])

    # モデルの保存
    os.makedirs('./model/', exist_ok=True) # フォルダがない時は生成
    model.save('./model/best.h5') # ベストプレイヤーのモデル

    # モデルの破棄
    K.clear_session()
    del model

# 動作確認
if __name__ == '__main__':
    dual_network()

実行すると、「./model.best.h5」ファイルが出力されます。
これはまだ未学習のモデルとなります。

参考

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

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

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

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

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

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

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

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

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

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

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

game.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
# パッケージのインポート
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
1
2
3
4
# ランダムで行動選択
def random_action(state):
    legal_actions = state.legal_actions()
    return legal_actions[random.randint(0, len(legal_actions)-1)]

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

game.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
# アルファベータ法で状態価値計算
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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# プレイアウト
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
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
# モンテカルロ木探索の行動選択
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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 動作確認
if __name__ == '__main__':
    # 状態の生成
    state = State()

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

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

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

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

結果

参考

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