953 views
この記事は最終更新から 372日 が経過しています。
【1】やりたいこと
前回の投稿で案出しした内容で、実装を進めたい。
(53)【Othello AI】モンテカルロ法で DQNの教師データを作ろうか？
DQN方式オセロAIの訓練データとして、
モンテカルロ探索方式オセロAIの指し手履歴データを使いたい。
まずは収集する！
【2】やってみる
1) 収集するデータ
MCTS vs MCTSの対局データを一手ごとに保存する。
保存対象のデータは下記の通り。
| # | パラメータ名 | 説明 |
|---|---|---|
| 1 | state | オセロ盤上の石の配置、[6][6]の配列 |
| 2 | action | 指し手 (r,c) |
| 3 | reward | (state, action) の結果、指し手(r,c)により得られた報酬 |
| 4 | next_state | action実行後のオセロ盤上の石の配置、[6][6]の配列 |
| 5 | done | ゲーム終了 or not フラグ |
なぜ next_state が必要なの？
は、こちら（↓）の過去記事に記載している。DQNの戦術で使うのだ。
(45)【Othello AI】強化学習でオセロAIを作る。
※今回は、自分自身の次の手番直前の stateではなく、after-step（自分の指し手直後の相手視点）の stateを採用した。
対戦する MCTS agentの探索回数は、以下の値から選択する。値が大きいほど探索範囲が深くなり、一般的には強くなる。
100, 200, 300, 400, 500, 600, 700, 800, 900, 1000
※ 100～1000の範囲で 100stepの 10通り
形式仕様言語だとこんな風に書ける。
Numbers == { n : ℕ | 100 ≤ n ∧ n ≤ 1000 ∧ (n mod 100 = 0) }
2) プログラム
(1) main_CvC_MCTS.py
汎用の COM vs COM対戦プログラムとして作成済みの play_2_agent() を実行する。
from play import play_2_agent
import argparse
import pickle
from datetime import datetime
#///////////////////////////////////////////////////////////////////////////////
def main():
parser = argparse.ArgumentParser(description="COM vs COM MCTS版")
parser.add_argument("-n", type=int, help="対戦回数", default=1)
parser.add_argument("--miA",type=int, help="MCTS(A)プレイアウト数", default=100 )
parser.add_argument("--miB",type=int, help="MCTS(B)プレイアウト数", default=100 )
parser.add_argument("--rh", type=str, help="指し手履歴ファイル名", default='')
args = parser.parse_args()
#---------------------------------------------------------------------------
recordHistory = True if args.rh != '' else False
#---------------------------------------------------------------------------
cOneGameResults = play_2_agent(
model_pathA = 'MCTS',
model_pathB = 'MCTS',
n_games = args.n,
iterationsA = args.miA,
iterationsB = args.miB,
recordHistory = recordHistory
)
#---------------------------------------------------------------------------
if recordHistory is True: # 指し手履歴の記録指定あり？
saveFiles_per_Player(cOneGameResults, args.rh)
#///////////////////////////////////////////////////////////////////////////////
def saveFiles_per_Player( cOneGameResults, args_rh ):
transitions = [[], []] # 空リストで初期化
for cOneGameResult in cOneGameResults:
transitions[0] += cOneGameResult.transitions[0]
transitions[1] += cOneGameResult.transitions[1]
# --rh で指定されたファイル名で pickle にシリアライズ
timestamp = datetime.now().strftime("%Y_%m%d_%H%M%S")
for i, data in enumerate(transitions, start=1):
fname = f"{args_rh}_P{i}_{timestamp}.pth"
with open(fname, 'wb') as f:
pickle.dump(data, f)
print(f"指し手履歴を {fname} に保存しました。")
#///////////////////////////////////////////////////////////////////////////////
if __name__ == "__main__":
main()
(2) main_Collect_MCTS_training_data.py
100～1000（100step）の MCTS探索上限値の組合せで、MCTS-agent(A) vs MCTS-agent(B) の対戦を実行する。
コマンドライン引数（＝入力パラメータ）を変えて、上記の main_CvC_MCTS.py を CPU 10コア並列 で繰り返し実行する。
モンテカルロ探索では GPUによるデータ並列処理の恩恵が受けられないので、処理並列が可能且つ単体で高速な CPU実行とする。 → 今回はグラボを使わない。
#///////////////////////////////////////////////////////////////////////////////
# MCTS vs MCTSで対戦を繰り返し、DQN学習で使用する指し手履歴データを収集する。
#///////////////////////////////////////////////////////////////////////////////
import subprocess # 外部プログラムを呼び出すためのライブラリ
import sys # Python実行環境に関する情報を扱うライブラリ
from multiprocessing import Pool # 複数のプロセスで並列処理を行うためのライブラリ
from itertools import product # 複数リストの全組合せを生成するためのライブラリ
from tqdm import tqdm # 進捗バー（処理の進行度）を表示するライブラリ
from datetime import datetime # 現在日時を取得するためのライブラリ
#///////////////////////////////////////////////////////////////////////////////
# 各プロセス（並列処理）で実行する関数
def run_main_CvC(params):
#---------------------------------------------------------------------------
uid, MIA, MIB = params # 引数から uid, MIA, MIBの値を取り出す
nGame = 100 # 対戦数
#---------------------------------------------------------------------------
# 外部プログラム main_CvC_MCTS.py を呼び出す
subprocess.run([
sys.executable, "main_CvC_MCTS.py", # sys.executable で実行中のPythonを使用
"-n", "100", # 対戦回数を100回に設定
"--miA", str(MIA), # MCTS(A)の iteration回数（大きいほど探索が深い）
"--miB", str(MIB), # MCTS(B)の iteration回数（大きいほど探索が深い）
"--rh", f"MCTS_{MIA}_vs_{MIB}_x{nGame}_{uid:010}_History" # 出力する履歴ファイル名を指定
])
#///////////////////////////////////////////////////////////////////////////////
# メイン処理部分
if __name__ == "__main__":
process_count = 10 # 並列で動かすプロセスの数（CPUコア数に合わせて調整可）
#---------------------------------------------------------------------------
# MIAとMIBの値を100から1000まで、100ごとに増加させたリストを作る
MIA_values = range(100, 1001, 100)
MIB_values = range(100, 1001, 100)
#---------------------------------------------------------------------------
# MIAとMIBの全ての組合せ（ペア）を作成（例: (100,100), (100,200), ...）
param_combinations = list(product(MIA_values, MIB_values))
total_combinations = len(param_combinations) # 組合せ総数（10×10=100）
indexed_combinations = [(i, mia, mib) for i, (mia, mib) in enumerate(param_combinations)]
#---------------------------------------------------------------------------
# 開始時刻を表示
start_time = datetime.now()
print(f"実行開始時刻: {start_time.strftime('%Y-%m-%d %H:%M:%S')}")
#---------------------------------------------------------------------------
# プロセスプール（複数プロセス）を作成
with Pool(process_count) as pool:
#-----------------------------------------------------------------------
# imap_unorderedを使い、param_combinationsの各要素を並列実行
# 並行して処理が進み、処理結果を順不同で取得できるイテレータが返る
results_iterator = pool.imap_unordered(run_main_CvC, indexed_combinations)
#-----------------------------------------------------------------------
# tqdmで進捗バーを表示（イテレータを進めるたびに進行度が更新される）
for _ in tqdm(results_iterator, total=total_combinations):
pass
#---------------------------------------------------------------------------
# 終了時刻を表示
end_time = datetime.now()
print(f"実行終了時刻: {end_time.strftime('%Y-%m-%d %H:%M:%S')}")
#---------------------------------------------------------------------------
# 開始から終了までの処理時間を計算、表示する。
duration = (end_time - start_time).total_seconds()
print(f"実行時間: {duration:.1f} [sec]")
3) いざ実行
100セット x 100試合の対局データの収集に、我が家の最新マシンで約 3.7時間 を要した。
実行環境のマシンスペックは下記の通り。
| Parts | Name/Spec |
|---|---|
| CPU | Intel Core Ultra7 265KF 20cores |
| Memory | DDR5-5600 96GB |
| Motherboard | Z890 |
| SSD | M.2 NVMe PCIe4.0 2TB |
| グラボ | 今回は CPU実行なので使用せず |
Bash shell上でプログラムを実行する。
ASUS製マザーボードの CPU警告 LEDが赤く点灯するほど、高温になった・・・
$ python main_Collect_MCTS_training_data.py
実行開始時刻: 2025-06-04 11:38:15
100%|████████████████████████████████████████████████████████████████| 100/100 [3:40:04<00:00, 132.05s/it]
実行終了時刻: 2025-06-04 15:18:19
実行時間: 13204 [sec]
収集できた訓練データは全200ファイルだ。
探索回数の組合せ 100セット x 2（先手・後手）= 200ファイル
4) データの中身を確認
無作為に 1試合のデータを選択し、中身を表示してみる。
今回たまたま選んだのは、
MCTS（探索回数900回）vs MCTS（探索回数900回）
の結果だ。
先手、後手のデータファイルをロードする。
各ファイルの中には 100試合分の指し手データ（各約1600手分）が直列に格納されている。
>>> import pickle >>> >>> filename1 = "MCTS_900_vs_900_x100_0000000088_History_P1_2025_0604_144656.pth" >>> with open(filename1, "rb") as f: ... data1 = pickle.load(f) >>> >>> filename2 = "MCTS_900_vs_900_x100_0000000088_History_P2_2025_0604_144656.pth" >>> with open(filename2, "rb") as f: ... data2 = pickle.load(f) >>> >>> len(data1) 1617 >>> len(data2) 1583
先手の指し手データは 1ファイル内に 100試合で全 1617手分、
後手の指し手データは 1ファイル内に 100試合で全 1583手分が存在する。
両者の手数が違うのは、パスにより手番をスキップされる場合があるため。
先手の 1試合目の 第一手を表示してみる。
>>> from pprint import pprint
>>> pprint(data1[0])
(array([[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 0, 0, 0, 0],
[ 0, 0, -1, 1, 0, 0],
[ 0, 0, 1, -1, 0, 0],
[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 0, 0, 0, 0]], dtype=int8),
22,
array([[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 0, 0, 0, 0],
[ 0, 0, -1, 1, 0, 0],
[ 0, 0, 1, 1, 1, 0],
[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 0, 0, 0, 0]], dtype=int8),
np.float64(0.35000000000000003),
False)
後手の 1試合目の 第一手を表示してみる。
石を表す (1, -1) は以下の通り。
| 1 | 自分の石 |
| -1 | 相手の石 |
このため、手番が変われば石の表現も正負が反転 することに注意する。
>>> pprint(data2[0])
(array([[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 1, -1, 0, 0],
[ 0, 0, -1, -1, -1, 0],
[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 0, 0, 0, 0]], dtype=int8),
26,
array([[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 0, 0, 0, 0],
[ 0, 0, 1, -1, 0, 0],
[ 0, 0, 1, -1, -1, 0],
[ 0, 0, 1, 0, 0, 0],
[ 0, 0, 0, 0, 0, 0]], dtype=int8),
np.float64(0.2),
False)
盤上の石の配置が 1, -1 だと見づらいので、o x に変えてみる。
>>> def convert_and_print(board):
... symbol_map = {
... -1: 'x',
... 0: '.',
... 1: 'o'
... }
... for row in board:
... print(' '.join(symbol_map[int(cell)] for cell in row))
...
>>>
>>> convert_and_print(data1[0][0]) # 先手1手目、打つ前
. . . . . .
. . . . . .
. . x o . .
. . o x . .
. . . . . .
. . . . . .
>>> convert_and_print(data1[0][2]) # 先手1手目、打った後
. . . . . .
. . . . . .
. . x o . .
. . o o o .
. . . . . .
. . . . . .
>>>
>>> convert_and_print(data2[0][0]) # 後手1手目、打つ前 ※自分視点に正負反転済み
. . . . . .
. . . . . .
. . o x . .
. . x x x .
. . . . . .
. . . . . .
>>> convert_and_print(data2[0][2]) # 後手1手目、打った後
. . . . . .
. . . . . .
. . o x . .
. . o x x .
. . o . . .
. . . . . .
次に、勝敗が決した場面を見てみる。
第一試合は先手・後手それぞれ 15手目で試合終了したので、この場面を見てみる。
>>> pprint(data1[15]) # 先手の 15手目 (array([[ 1, 1, 1, 1, 1, 1], [-1, -1, -1, -1, -1, 0], [-1, -1, -1, -1, -1, 1], [ 1, 1, -1, 1, -1, 1], [ 0, 1, -1, -1, 1, 1], [-1, 1, 1, 1, 1, 1]], dtype=int8), 11, array([[ 1, 1, 1, 1, 1, 1], [-1, -1, -1, -1, -1, 1], [-1, -1, -1, -1, 1, 1], [ 1, 1, -1, 1, -1, 1], [ 0, 1, -1, -1, 1, 1], [-1, 1, 1, 1, 1, 1]], dtype=int8), np.float64(1.05), True) >>> pprint(data2[15]) # 後手の 15手目 (array([[-1, -1, -1, -1, -1, -1], [ 1, 1, 1, 1, 1, -1], [ 1, 1, 1, 1, -1, -1], [-1, -1, 1, -1, 1, -1], [ 0, -1, 1, 1, -1, -1], [ 1, -1, -1, -1, -1, -1]], dtype=int8), 24, array([[-1, -1, -1, -1, -1, -1], [ 1, 1, 1, 1, 1, -1], [ 1, 1, 1, 1, -1, -1], [ 1, 1, 1, -1, 1, -1], [ 1, 1, 1, 1, -1, -1], [ 1, -1, -1, -1, -1, -1]], dtype=int8), np.float64(1.6), True) >>> >>> data1[15][4] # 先手の終了フラグが Trueになっている。 True >>> data2[15][4] # 後手の終了フラグが Trueになっている。 True >>> >>> data1[15][3] # 先手のこの一手の報酬は 1.05 ※この試合は引き分けだったようだ。 np.float64(1.05) >>> data2[15][3] # 後手のこの一手の報酬は 1.6 np.float64(1.6)
この場面もわかりやすく表示してみる。
確かに 18 - 18 の引き分けで終わっている。
>>> convert_and_print(data1[15][2]) # 先手が最後の一手を打った直後 o o o o o o x x x x x o x x x x o o o o x o x o . o x x o o x o o o o o >>> convert_and_print(data2[15][0]) # 後手が最後の一手を打つ直前 ※上記盤面を自分視点に反転済み x x x x x x o o o o o x o o o o x x x x o x o x . x o o x x o x x x x x >>> convert_and_print(data2[15][2]) # 後手が最後の一手を打った直後 → 最終的な盤面の状態 x x x x x x o o o o o x o o o o x x o o o x o x o o o o x x o x x x x x
【3】今後の予定
現時点までに出来たこと：
MCTS vs MCTSの強弱100通りの組み合わせで 100セット x 100試合分の指し手履歴データが収集できた。
1試合あたりの指し手数が、先手・後手合わせて約 32なので、
100セット x 100試合 x 32手 = 320,000 手分の訓練データを作成できた。
紛失すると困るので、ここにも置いておこう。
training_data_20250604.zip
次にやること：
このデータを DQNの教師データに使い、DQNオセロ AIを訓練する。
果たして MCTSレベルの強いオセロAIに育つのか？
を実験していきたい。
作ったデータのクレンジングやデータ拡張による増殖についても要検討だが、最初は人間判断の最適化をせずにやってみる予定だ。
回転によるデータ 4倍拡張ぐらいはやろうかな・・・
アクセス数（直近7日）: ※試験運用中、BOT除外簡易実装済2026-06-19: 0回 2026-06-18: 0回 2026-06-17: 0回 2026-06-16: 0回 2026-06-15: 0回 2026-06-14: 1回 2026-06-13: 0回