麻雀強化学習環境「Mjx」の覚書

はじめに

目的

このページは、麻雀強化学習環境「Mjx」の導入手順や実行手順についてまとめたページです。
私自身が認識不足だった点をまとめています。

Mjxについて

麻雀強化学習環境「Mjx」は、AIの研究・開発を目的に作られた、リーチ麻雀のシミュレータ兼フレームワークです。
強化学習を前提とした設計としており、Pythonのインターフェースを提供しています。

---

Mjxインストール

mjxインストールの注意

Ubuntu 22.04以降でMjxを導入する場合には、gRPCパッケージが原因でmjxのpipインストールに失敗する事象が確認されています。
Docker環境等で、PythonやgRPCパッケージが古いバージョンの環境で、mjxをインストールした方が推奨されます。

• 不具合情報
  • Ubuntu 22.04でpip installできない #1139
• protobuf
  • Ubuntu に最新版のprotobufをインストール
  • Ubuntu18.04にProtobufとgRPC C++の環境構築

(参考)mjxインストール手順

• mjxインストール手順
  • Docker環境でmjxをインストール

---

Mjx実行

動作概要

MjxをPythonで実行する場合は、以下のような処理フローとなります。

• 処理フロー
  1. mjx.MjxEnv()で卓ゲームを定義、卓ゲームを開始する
     • 関数："env = mjx.MjxEnv()"
  2. agents変数にプレーヤーとなるAgentを定義する
     • 関数："agent = ShantenAgent()"またはagents配列を定義する
  3. env.reset関数で卓ゲームの初期化を行い、obs_dict値を定義する
     • 関数："obs_dict = env.reset()"
  4. 以下でループを行う
     1. obs_dictからObservationを抽出　　
        • 関数："for player_id, obs in obs_dict.items()"
     2. agentにObservationオブジェクトを引き渡し
        • 関数："agent.act(obs)"
     3. Agent内処理を行いAction決定
     4. Actionの値を設定
        • 関数："actions[player_id] = agent.act(obs)"
     5. ステップ実行とobs_dict値を設定
        • 関数："obs_dict = env.step(actions)"
     6. env.endがtrueなら処理終了、falseならループの最初に戻る
  5. env.state関数で卓ゲームの結果情報を定義する
     • 関数："state = env.state()"
  6. いくつかの方法で、State情報を出力する
     1. 画面表示
        • 関数："state"
     2. protobuf形式に変換
        • 関数："state.to_proto()"
     3. JSON形式に変換
        • 関数："state_json = json.loads(state.to_json())"

公式ドキュメント内の実行手順

• Run 100 games

---

Mjx実行\サンプルプログラム

サンプルプログラム

以下を実行することで、局ごとのJSON形式のState情報を出力することができます。

import json
import mjx
from mjx.agents import ShantenAgent

# 卓ゲーム定義
env = mjx.MjxEnv()

# エージェントの準備
agent = ShantenAgent()

# 卓ゲーム初期化
obs_dict = env.reset()

# actions を空の辞書として初期化
actions = {}

# 行動が求められているプレイヤー（obs_dictのキー）ごとにアクションを決定
while not env.done():

    # アクション定義
    actions = {player_id: agent.act(obs) for player_id, obs in obs_dict.items()}

    # ステップ実行
    obs_dict = env.step(actions)

    # state_json定義
    state = env.state()

    # 局が終了しているなら、state_jsonを表示
    state_json = json.loads(state.to_json())
    if "roundTerminal" in state_json:

        # JSON表示
        print (state_json)

(処理を一部で分割して実行)

import json
import mjx
from mjx.agents import ShantenAgent

# 卓ゲーム定義
env = mjx.MjxEnv()

agents = {
    "player_0": ShantenAgent(),
    "player_1": ShantenAgent(),
    "player_2": ShantenAgent(),
    "player_3": ShantenAgent()
}

# 卓ゲーム初期化
obs_dict = env.reset()

# 行動が求められているプレイヤー（obs_dictのキー）ごとにアクションを決定
while not env.done():

    # actions を空の辞書として初期化
    actions = {}
    
    # 行動が求められているプレイヤー（obs_dictのキー）ごとにアクションを決定
    for player_id, obs in obs_dict.items():
        agent = agents[player_id]
        
        # agent.act(obs) は mjx.Action オブジェクトを返す
        action = agent.act(obs) 
        
        # 辞書にプレイヤーIDをキーとして格納
        actions[player_id] = action 
    
    # 作成した actions を環境に渡して一歩進める
    obs_dict = env.step(actions)

    # state_json定義
    state = env.state()

    # 局が終了しているなら、state_jsonを表示
    state_json = json.loads(state.to_json())
    if "roundTerminal" in state_json:

        # JSON表示
        print (state_json)

---

Mjxの定義情報

Mjxの定義情報_概要

Mjx内で実行されている卓ゲームの情報として以下情報があります。

• Envオブジェクト
  • 卓ゲームを処理するためのオブジェクトです
  • 関数："env = mjx.MjxEnv()"で値を取得します
• Stateオブジェクト
  • 卓ゲームの全ての情報を定義しています
  • 関数："state = env.state()"を実行して値を取得します
• Obs_dictオブジェクト
  • 全プレーヤのmjx.Observationを保持するオブジェクト
  • 関数："obs_dict = env.reset()"又は"obs_dict = env.step(actions)"を実行して値を取得します
  • 型は Dict[player_x, mjx.Observation]
• Observationオブジェクト
  • 卓ゲームで公開されている情報を定義しています
  • 関数：obs = obs_dict["player_0"]"でプレーヤごとの情報を取得
  • ループ文"for player_id, obs in obs_dict.items():"で全プレーヤの情報を取得するができます。
• Agentオブジェクト
  • Mjxのプレーヤを定義するオブジェクト
• Actionオブジェクト
  • Mjxのプレーヤのアクションを定義するオブジェクト
  • 全プレーヤのアクションを定義した"actions"変数を渡して、卓ゲームを継続する
  • 関数："obs_dict = env.step(actions)"
• Rewardオブジェクト
  • ゲーム結果から算出された報酬の値
  • Envオブジェクトから取得することができます
• 麻雀の強化学習環境Mjx(v0.1.0)を触る（1/3）ライブラリ機能の確認編

情報取得手順

• protobuf形式に変換
  • 関数："state.to_proto()"
  • 関数："obs.to_proto()"
• JSON形式に変換
  • 関数："state_json = json.loads(state.to_json())"
  • 関数："obs_json = json.loads(obs.to_json())"
  • 関数："action_json = json.loads(action.to_json())"
• Data structure

Observationの詳細情報

• Observationの詳細情報
  1. いま選べる有効な行動をすべてリストアップ
     • 関数："legal_actions = obs.legal_actions()"
  2. AI（ディープラーニングモデル）の入力用に、行列データへ変換
     • 関数："matrix_data = obs.to_features(feature_name="mjx-small-v0")"

---

Mjxの牌定義

Mjxの牌定義

Mjxで利用されている牌定義は、「天鳳牌譜 mjlog形式」をそのまま踏襲しています。
プレーヤの鳴き情報も「天鳳牌譜 mjlog形式」と同一の値になります。

• 参考/天鳳牌譜 mjlog形式
  • 天鳳牌譜 mjlog形式の牌種について
  • 天鳳牌譜 mjlog形式について
• 参考/天鳳牌譜 mjlogソースファイル
  • tehai.js
  • entsu136.txt

---

Actionオブジェクトの詳細

Actionオブジェクトの詳細

AgentオブジェクがActionオブジェクトを定義して、プレーヤのアクションを決定できます。

Observationのlegal_actionsの詳細

Observation(obs)内にあるlegal_actionsは、いま選べる有効な行動がリスト化されている。legal_actionsからactionに分割して有効なアクションを確認することができます。

• 関数："legal_actions = obs.legal_actions()"

legal_actionsの表示

以下でlegal_actions内legal_actionsを確認することができます。

import json
import mjx
from mjx.agents import ShantenAgent

# 卓ゲーム定義
env = mjx.MjxEnv()

# エージェントの準備
agents = {
    "player_0": ShantenAgent(),
    "player_1": ShantenAgent(),
    "player_2": ShantenAgent(),
    "player_3": ShantenAgent()
}

# 卓ゲーム初期化
obs_dict = env.reset()

# actions を空の辞書として初期化
actions = {}

# プレーヤごとの処理
for player_id, obs in obs_dict.items():

    # オブジェクト情報
    agent = agents[player_id]
    
    # legal_actionsの取得
    legal_actions = obs.legal_actions()

    # legal_actionsを各actionに分割
    for action in obs.legal_actions():

        # actionをJSONに変換
        action_json = json.loads(action.to_json())

        # action情報詳細
        print(action_json, action.type(), action.tile().id(), action.tile().type(), action.tile().is_red(), action.to_idx())

RiichiLab(riichi.dev)での応用

RiichiLabではサーバからObservationオブジェクトが送信されます。
Observationオブジェクトから現時点のlegal_actionsのアクション一覧が可能であり、legal_actionsからアクションを選択してmjai形式のJSONに変換することもできます。
(ただし、request_actionのpossible_actionsキー内に同様の可能アクション一覧も存在します)

(RiichiLabのサンプルWebSocketBotからの抜粋)

case "request_action":

  # Deserialize the observation from the server
  obs = Observation.deserialize_from_base64(msg["observation"])
  ⇒ Observationオブジェクトの情報がサーバから送信される
  ⇒ 関数"legal_actions = obs.legal_actions()"から、現時点の実行可能アクションを確認することができる

  action = agent.act(obs)
  ⇒ agent内でも、関数"legal_actions = obs.legal_actions()"を実行して、legal_actionsオブジェクトの情報を取得することができる。
  ⇒ "for action in obs.legal_actions():"と"action_json = json.loads(action.to_json())"を組み合わせて、json形式のlegal_actionsの情報を取得することができる。

  resp = json.loads(action.to_mjai())
  ⇒ legal_actionsから選択し定義したactionオブジェクトからmjai形式のJSONに変換が可能

• 参考
  • RiichiLab/ローカルテスト

---

参考情報

公式資料リンク

• github
  • github
• Googleドキュメント
  • Mjx
• IEEE

参考資料

• 麻雀AIプログラミングを行う際に有用となるサイト

---