強化学習実装の準備として、牌譜の記録/再生を実装した。
PPOでは、エピソードの終端の報酬を、GAEで各ステップの状態に割り当てるため、エピソードの各ステップの状態を再生できるように記録しておく必要がある。
また、ランダムな初期局面からの対局を再生できるように、対局の開始状態（手牌、河、牌山）を記録しておく必要がある。通常の麻雀の牌譜で使われるフォーマットでは対応できないため、独自のフォーマットで記録する。
フォーマット
記録が必要な情報
- ゲームルール（赤牌有無やクイタン有無など）
- 起家
- 各局の初期局面（手牌、河、牌山）
- 各局のアクション（打牌、副露、槓、和了）
打牌後に副露できる状態で誰も副露しなかった場合、副露なしのアクションは冗長なため記録しない。
和了した役や最終順位・得点は、再計算可能で冗長なため記録しない。
文字列フォーマット
cmajiangは、内部的に牌を文字列で表現しているため、牌譜でもそのまま文字列を使うことにする。
1ゲーム1行、各情報をスペース区切り
各情報をスペースで区切って出力し、1ゲームを改行で区切る。
可変個の情報は、牌山などそのまま連結しても再生時にロジックで分解できるものは区切り文字なしで連結し、アクションなど区切り文字が必要な情報はカンマ区切って連結する。
牌譜の例
25000 20.0 10.0 -10.0 -20.0 0 1 1 1 1 0 2 1 1 0 1 2 2 1 1 1 1 1 1 1 1 0 2 1 1 1 1 0 1 0 0 0 0 25000 25000 25000 25000 p1s4m7z1m5s9m5m8p3m4m6p2m6p4m2s8z5z3p8s8s3z7m4s6s3m0z1p5m4p5p6m8s6s5m2s9m1p6s1s3p8s2p0s4s6 s78z11222444577 p36677s12244z667 m1148999p999s123 p2457z233345566 m1m3p9s1m2s0m3p4m6p8 p1m7s9p7m5p4m6m8m2p3 s7s5s8p2s9s7p3p1m3s6 m7p2m3m7s5m9p8s7p1 1 2 3 0 2 1z2,1z4,1p0_,1m4,1p4,1s3_,2s123-,1p7,1m8*,1p5,1z4,1s2,1m2_,1z6,1z7,1s4,1p6_,2p666+,1p3,1p5_,2p5-67,1z5,1s4,1p5_,1z1_,2z111=,1z2,1m0_,4
実装
牌譜の入出力は実装が容易なため、C++のストリームを使用して実装する。
牌譜の再生は、PaipuReplayクラスに牌譜を与えて、next()を呼び出すたびに、次のアクション実行前の状態まで遷移するようにする。
打牌後に誰も副露しなかった場合も、エージェントが副露を選択しなかったことを学習する必要があるため、その時点で止めるようにする。
cmajiang/src_cpp/paipu.cpp at main · TadaoYamaoka/cmajiang · GitHub
実行例
以下は、1局だけ対局を行い、記録された牌譜を再生する例。
from cmajiang import Game, random_game_state, Status, Message, xiangting, set_seed, PaipuReplay import random set_seed(8) random.seed(0) n_xiangting = 3 # 初期局面生成 game = Game() game.kaiju() random_game_state(game, n_xiangting) print([str(shoupai) for shoupai in game.shoupai]) print("xiangting", [xiangting(shoupai) for shoupai in game.shoupai]) print("lunban", game.lunban) # 和了/流局まで繰り返す while game.status not in (Status.HULE, Status.PINGJU): match game.status: case Status.ZIMO | Status.GANGZIMO: if game.allow_hule(): # 和了 game.reply(game.lunban_player_id, Message.HULE) else: dapai = game.get_dapai() # 立直 dapai.extend(p + '*' for p in game.allow_lizhi()[1]) # 暗槓もしくは加槓 dapai.extend(m for m in game.get_gang_mianzi()) p = random.choice(dapai) if len(p) > 4: game.reply(game.lunban_player_id, Message.GANG, p) else: game.reply(game.lunban_player_id, Message.DAPAI, p) case Status.DAPAI: # 他家の応答 ロン、副露 for player_id in range(4): if player_id == game.lunban_player_id: continue player_lunban = game.player_lunban(player_id) if game.allow_hule(player_lunban): # ロン game.reply(player_id, Message.HULE) else: # 副露 mianzi = [None] mianzi.extend(game.get_chi_mianzi(player_lunban)) mianzi.extend(game.get_peng_mianzi(player_lunban)) mianzi.extend(game.get_gang_mianzi(player_lunban)) m = random.choice(mianzi) if m: if len(m) > 5: game.reply(player_id, Message.GANG, m) else: game.reply(player_id, Message.FULOU, m) case Status.FULOU: dapai = game.get_dapai() p = random.choice(dapai) game.reply(game.lunban_player_id, Message.DAPAI, p) case Status.GANG: for player_id in range(4): if player_id == game.lunban_player_id: continue player_lunban = game.player_lunban(player_id) if game.allow_hule(player_lunban): # ロン(槍槓) game.reply(player_id, Message.HULE) for l in range(4): reply = game.get_reply(l) if reply.msg != Message.NONE: print(l, reply.msg, reply.arg) game.next() print(game.status) if game.status == Status.HULE: print(game._defen) print(game.fenpei) # 棋譜再生 print("------") print("replay") paipu = game.paipu replay = PaipuReplay(paipu) while replay.status not in (Status.HULE, Status.PINGJU): replay.next() print(replay.game.status) if replay.game.status == Status.HULE: print(replay.game._defen) print(replay.game.fenpei)
実行結果
['s78z11222444577', 'p36677s12244z667', 'm1148999p999s123', 'p2457z233345566'] xiangting [1, 1, 1, 2] lunban 2 3 Message.DAPAI z2 (略) 2 Message.HULE Status.HULE 立直 1翻 門前清自摸和 1翻 一盃口 1翻 純全帯幺九 3翻 40符 6翻 跳満 12000点 [-6000, -3000, 13000, -3000] ------ replay Status.HULE 立直 1翻 門前清自摸和 1翻 一盃口 1翻 純全帯幺九 3翻 40符 6翻 跳満 12000点 [-6000, -3000, 13000, -3000]
実際の対局の結果と、牌譜を再生した結果が一致することが確認できる。
まとめ
強化学習実装の準備として、牌譜の記録/再生機能をcmajiangに実装した。
次は、自己対局でエピソードを収集する処理を実装したい。
