麻雀AI実装とオープンアリーナの公開

麻雀AIのシミュレーション実装をオープンソースとして公開しました。また麻雀AIを対戦させるためのオープンアリーナ環境として「RiichiLab」をリニューアルオープンしました。WebSocket接続して24時間いつでも他人の麻雀AIと対戦させてレーティングを付けることができます。

シミュレーション実装 RiichiEnv とオープンアリーナ

麻雀AIは不完全情報ゲームであるため探索アルゴリズムが活用しづらい非常に面白い研究テーマです。しかし将棋や囲碁と比較して開発者が少ないように感じます。その原因にはルールの複雑さ、ランダム性の高さによる評価の難しさ、再利用可能なライブラリの不足などにあると考えています。喰い替え禁止や槍槓のようなイレギュラーなイベントの正確なハンドリングなど、すべてを正確な状態に保つのは大変です。打牌だけでなくカンした牌に対してロンできるなんてルール、シンプルにモジュール化して実装したいと考える側からすると面倒この上ない。

これらの参入障壁を取り除くべく、再利用可能なシミュレーション実装として Rust のコア実装である riichienv-core クレート、それを用いた Gym-like インターフェースを持つ Python パッケージとして RiichiEnv を新たに実装してOSSとして公開しました。wasm ビルトインの ES Modules (ESM) としてビルドしてブラウザからの可視化に役立てるモジュールも提供していますが、こちらはまだ整備中です。

またこれをベースとして麻雀AIを対戦させるオープンアリーナである RiichiLab をリニューアルオープンしました。開発した麻雀AI を他人の麻雀AIと対戦させて OpenSkill レーティング によって評価することが可能です。

RiichiLab

是非これらを活用して面白い成果物や最強麻雀AIの開発に役立てていただければ幸いです。

RiichiEnv の使い方

RiichiEnv は Python パッケージとしてシミュレーション実装を提供します。以下のような Gym-like な API を提供します。

from riichienv import RiichiEnv
from riichienv.agents import RandomAgent

agent = RandomAgent()
env = RiichiEnv()
obs_dict = env.reset()
while not env.done():
    actions = {player_id: agent.act(obs)
               for player_id, obs in obs_dict.items()}
    obs_dict = env.step(actions)

scores, ranks = env.scores(), env.ranks()
print(scores, ranks)

このコードでは以下のような流れで麻雀AIのシミュレーションを行っています。

  • obs_dict = env.reset() で環境を初期化して、行動可能なプレイヤーごとの観測情報を dict[int, Observation] 形式で受け取ります。
  • ゲームが終了したかどうかの判定は env.done() で行います。
  • 行動可能なプレイヤーの観測情報を agent.act(obs) に渡して Action を受け取ります。
  • env.step(actions) で行動可能なプレイヤーごとの行動を dict[int, Action] で環境に渡し、次の観測情報を得ます。

麻雀AI の開発者はこのように観測情報 Observation を受け取り、行動 Action を返却する必要があります。 Observation オブジェクトはプレイヤーが観測するすべての情報にアクセスする手段を提供します。これには行動可能な Action のリストを含みます。

>>> obs.legal_actions()
[Action(action_type=Discard, tile=Some(1), ...), ...]

麻雀AIの開発者は act(obs) メソッドによって Action を返すクラスを実装する必要があります。 例えばランダムな行動を行う麻雀AI は以下のように実装できます。

import random

from riichienv import Action, Observation

class RandomAgent:
    def __init__(self, seed: int | None = None):
        self._rng = random.Random(seed)

    def act(self, obs: Observation) -> Action:
        """
        Returns a valid Action object (DISCARD, RON, etc).
        For now, mostly discards randomly from legal moves.
        """
        return self._rng.choice(obs.legal_actions())

Mortal などの既存実装をこのインターフェースに適合させる方法や Observation の他の機能など、詳しくは README.md を参照ください。

RiichiLab の基本的な接続方法

RiichiLab にログイン(GitHub OAuth 認証)した後、Bots ページから麻雀AIの Bot 名を設定してトークンを取得します。このトークンによって認証を行い、WebSocket 接続します。

以下はツモ切りをする Bot の例です。

import asyncio
import json
import websockets

# 取得したトークンを設定
TOKEN = "your-bot-token-here"
# 認証用のヘッダー
HEADERS = {"Authorization": f"Bearer {TOKEN}"}

async def bot(url: str):
    # 認証用ヘッダーを設定して WebSocket に接続
    async with websockets.connect(url, additional_headers=HEADERS) as ws:
        my_seat = None
        last_tsumo = None
        while True:
            # イベントメッセージを受け取り
            msg = json.loads(await ws.recv())
            match msg["type"]:
                # ゲームの開始
                case "start_game":
                    my_seat = msg.get("id", 0)
                # ツモイベント
                case "tsumo":
                    pai = msg.get("pai")
                    # 自分のツモ牌以外は "?" が得られる
                    # ツモ牌が "?" でない場合は最後のツモ牌を更新する
                    if pai and pai != "?":
                        # ツモ牌を更新
                        last_tsumo = pai
                # 行動要求イベント
                case "request_action":
                    if last_tsumo:
                        resp = {
                            "type": "dahai",
                            "actor": my_seat,
                            "pai": last_tsumo,
                            "tsumogiri": True,
                        }
                    else:
                        resp = {"type": "none"}
                    # サーバーに行動を送信する
                    await ws.send(json.dumps(resp))
                    last_tsumo = None
                case "end_game":
                    # ゲームが終了したらループを抜ける
                    break

# 検証用の WebSocket URL に接続
asyncio.run(bot("wss://game.riichi.dev/ws/validate"))

少しコードが煩雑になっていますが、これは例としてイベントごとに処理を分けているためです。 実際には RiichiLab では Observation のシリアライズしたデータを提供するため、実装をもっと簡略化できます。

request_action メッセージには Observation をシリアライズしたデータが含まれています。Observation.deserialize_from_base64() によって riichienvObservation オブジェクトをデシリアライズすることができます。これを使うことで、tsumo, start_game などの個々のメッセージを解釈する必要は無くなります。request_actionend_game を受け取るまですべて無視できます。

import asyncio
import json
import websockets

from riichienv import Observation

TOKEN = "your-bot-token-here"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}

async def bot(url: str):
    async with websockets.connect(url, additional_headers=HEADERS) as ws:
        # MyAgent は Observation を受け取って act() メソッドで Action を返すクラス
        agent = MyAgent()
        while True:
            msg = json.loads(await ws.recv())
            match msg["type"]:
                case "request_action":
                    # Deserialize the observation from the server
                    obs = Observation.deserialize_from_base64(
                        msg["observation"]
                    )
                    action = agent.act(obs)
                    await ws.send(action.to_mjai())
                case "end_game":
                    break

asyncio.run(bot("wss://game.riichi.dev/ws/validate"))

とてもシンプルになりました。MyAgent は RiichiEnv のインターフェースと同じです。 これによって通信とAIの実装を分離することができます。

from riichienv import Action, ActionType, Observation

class MyAgent:
    """Your bot agent. Implement act() to return an Action."""

    def act(self, obs: Observation) -> Action:
        actions = obs.legal_actions()

        # Win if possible
        for a in actions:
            if a.action_type == ActionType.Tsumo or a.action_type == ActionType.Ron:
                return a

        # Declare riichi if possible
        for a in actions:
            if a.action_type == ActionType.Riichi:
                return a

        # Default: discard the drawn tile (tsumogiri)
        for a in actions:
            if a.action_type == ActionType.Discard:
                return a

        # Pass on claims
        return Action(ActionType.Pass)

RiichiEnv のゲームサーバーの仕様、レーティングの計算方法やマッチングの仕様など、詳細については RiichiLab のドキュメントを参照してください。

ゲームの可視化:キーイベントへのショートカット

ゲームの可視化ではキーイベントへのショートカットを提供しています。個人的にはこれがとても気に入っています。「オーラスで対面がテンパイしたタイミングが知りたい」といった要求に対して、2クリックで到達可能になります。待ちの変更やテンパイ崩しも瞬時に把握することができ、定性評価やデバッグに役立ちます。

alt text

画面右下には、現在の表示画面の固有リンク(permalink)をコピーするボタンがあります。バグレポートやデバッグのためのメモなどに活用ください。

Kaggle Notebook

学習済みモデルを非公開の Kaggle Dataset としてアップロードして、Kaggle の GPU リソースを使ってゲーム実行を行い、そのリプレイを Kaggle Notebook 上で可視化してブログに埋め込むことができます。おもしろ!

おわりに

麻雀AIを強くするための改善余地はたくさん残っています。特に特徴量。和了率や放銃率などの特徴量が CNN backbone のモデルでは重要であることが私の中でよく知られています。

よく知られているものはオープン実装として開発しつつ、特別なものはプライベートで開発しつつ、引き続き開発を進めていきます。