1. 始める前に
AlphaGo と AlphaStar の目覚ましい進展は、ML を使用して超人レベルのゲーム エージェントを構築する可能性を示しました。ML を活用した小規模なゲームを作成し、強力なゲーム エージェントの作成に必要なスキルを身に付けるための楽しい演習です。
この Codelab では、以下を使用してボードゲームを作成する方法を学習します。
- 強化学習でゲーム エージェントをトレーニングする TensorFlow エージェント
- モデルを提供する TensorFlow Serving
- Flutter でクロス プラットフォームのボードゲーム アプリを作成する
前提条件
- Dart を使った Flutter の開発に関する基本的な知識
- TensorFlow を使用した機械学習に関する基本的な知識(トレーニングとデプロイなど)
- Python、ターミナル、Docker に関する基本的な知識
学習内容
- TensorFlow エージェントを使用して非プレーヤー キャラクター(NPC)エージェントをトレーニングする方法
- TensorFlow Serving を使用してトレーニング済みモデルを提供する方法
- クロス プラットフォームの Flutter ボードゲームを作成する方法
必要なもの
- Flutter SDK
- Flutter を使用するための Android と iOS の設定
- Flutter を使用するためのパソコンの設定
- Flutter を使用するためのウェブの設定
- Flutter と Dart を使用するための Visual Studio Code(VS Code)の設定
- Docker
- Bash
- Python 3.7+
2. Plane Strike ゲーム
この Codelab で作成するゲームは「プレーン ストライク」と呼ばれ、ボードゲーム「バトルシップ」に似た小さな 2 プレーヤー ボードゲームです。ルールは非常にシンプルです。
- 人間のプレーヤーが、機械学習でトレーニングされた NPC エージェントと対戦します。人間のプレーヤーは、エージェントのボードのセルをタップするとゲームを開始できます。
- ゲーム開始時に、人間のプレーヤーとエージェントにはそれぞれ「プレーン」があります。自分のボードに配置します(下のアニメーションの人間のプレーヤーのボードのように「平面」を形成する 8 つの緑色のセル)。これらの「飛行機」ランダムに配置され、ボードの所有者にのみ表示され、対戦相手には非表示になります。
- 人間のプレーヤーとエージェントが、互いのボードの 1 つのセルを交互に攻撃します。人間のプレーヤーがエージェントのボードの任意のセルをタップすると、エージェントは ML モデルの予測に基づいて自動的に選択を行います。入力しようとしたセルが「plane」の場合、赤色になるセル (‘hit');正しくない場合は黄色(「ミス」)になります。
- 最初に 8 つの赤色のセルを達成した人がゲームの勝者です。その後 新しいボードでゲームを 再開します
ゲームのゲームプレイ例を次に示します。
3. Flutter の開発環境をセットアップする
Flutter 開発の場合、この Codelab を完了するには、Flutter SDK とエディタの 2 つのソフトウェアが必要です。
この Codelab は、次のいずれかのデバイスを使って実行できます。
- iOS シミュレータ(Xcode ツールのインストールが必要)
- Android Emulator(Android Studio でセットアップが必要)
- ブラウザ(デバッグには Chrome が必要)
- Windows、Linux、macOS のデスクトップ アプリケーション。開発はデプロイする予定のプラットフォームで行う必要があります。たとえば、Windows のデスクトップ アプリを開発する場合は、適切なビルドチェーンにアクセスできるように Windows で開発する必要があります。オペレーティング システム固有の要件については、docs.flutter.dev/desktop に詳しい説明があります。
4. セットアップする
この Codelab のコードをダウンロードするには:
- この Codelab の GitHub リポジトリに移動します。
- [Code] > [Download zip] をクリックして、この Codelab のすべてのコードをダウンロードします。
- ダウンロードした zip ファイルを解凍して、
codelabs-main
ルートフォルダを展開します。これには必要なリソースがすべて含まれています。
この Codelab に必要なのは、複数のフォルダを含むリポジトリの tfagents-flutter/
サブディレクトリにあるファイルのみです。
step0
フォルダからstep6
フォルダには、この Codelab の各ステップで構築するスターター コードが含まれています。finished
フォルダには、完成したサンプルアプリの完全なコードが含まれています。- 各フォルダには、バックエンド コードを含む
backbend
サブフォルダと、Flutter フロントエンド コードを含むfrontend
サブフォルダが含まれます。
5. プロジェクトの依存関係をダウンロードする
バックエンド
ターミナルを開いて、tfagents-flutter
サブフォルダに移動します。以下のコマンドを実行します。
pip install -r requirements.txt
フロントエンド
- VS Code で、[ファイル] >フォルダを開き、先ほどダウンロードしたソースコードの
step0
フォルダを選択します。 step0/frontend/lib/main.dart
ファイルを開きます。スターター アプリに必要なパッケージのダウンロードを求める VS Code ダイアログが表示されたら、[Get packages] をクリックします。- このダイアログが表示されない場合は、ターミナルを開いて
step0/frontend
フォルダでflutter pub get
コマンドを実行します。
6. ステップ 0: スターター アプリを実行する
- VS Code で
step0/frontend/lib/main.dart
ファイルを開き、Android Emulator または iOS Simulator が正しく設定され、ステータスバーに表示されていることを確認します。
たとえば、Android Emulator で Google Pixel 5 を使用する場合は次のようになります。
iOS シミュレータで iPhone 13 を使用する場合は次のようになります。
- [ Start debugging] をクリックします。
アプリを実行して操作する
Android Emulator または iOS シミュレータでアプリを起動します。UI は非常にシンプルです。ゲームボードは 2 つあります。ストライク ポジションとして、エージェントのボードの上のセルをタップできます。スマート エージェントをトレーニングして、人間のプレーヤーの板に基づいてストライクする場所を自動的に予測します。
内部で、Flutter アプリは人間のプレーヤーの現在のボードをバックエンドに送信します。バックエンドは強化学習モデルを実行し、次に打つべきセルの位置を予測して返します。フロントエンドは、レスポンスの受信後に UI に結果を表示します。
エージェントのボードのセルをクリックしても、アプリはまだバックエンドと通信できないため、何も起こりません。
7. ステップ 1: TensorFlow エージェントの Python 環境を作成する
この Codelab の主な目的は、環境とのやり取りによって学習するエージェントを設計することです。Plane Strike ゲームは比較的シンプルで、NPC エージェント向けのルールを手作りすることも可能です。ただし、強化学習を使用してエージェントをトレーニングすることで、スキルを習得し、将来的に他のゲームのエージェントを簡単に構築できるようになります。
標準の強化学習(RL)設定では、エージェントは時間ステップごとに観測結果を受け取り、アクションを選択します。アクションが環境に適用され、環境から報酬と新しい観察結果が返されます。エージェントは、報酬の合計(リターン)を最大化するアクションを選択するポリシーをトレーニングします。ゲームを何度もプレイすることで、エージェントはパターンを学習し、ゲームをマスターするためのスキルを磨くことができます。Plane Strike ゲームを RL 問題として定式化するには、盤面の状態を観察、ストライク ポジションを行動、ヒット/ミスのシグナルを報酬として考えます。
NPC エージェントをトレーニングするには、信頼性が高く、スケーラブルで使いやすい TensorFlow の強化学習ライブラリである TensorFlow エージェントを利用します。
TF エージェントは、すぐに利用できる Codelab、例、ドキュメントが豊富に用意されているため、強化学習に適しています。TF エージェントを使用すると、現実的で複雑な RL の問題をスケーラビリティとともに解決し、新しい RL アルゴリズムを迅速に開発できます。異なるエージェントやアルゴリズムを簡単に切り替えてテストできます。また、十分にテストされており、構成も簡単です。
OpenAI Gym には、事前構築済みのゲーム環境が数多く実装されています(Atari ゲームなど)、Mujuco などがあり、TF エージェントは簡単に利用できます。しかし、Plane Strike ゲームは完全なカスタムゲームであるため、まず新しい環境をゼロから実装する必要があります。
TF エージェントの Python 環境を実装するには、次のメソッドを実装する必要があります。
class YourGameEnv(py_environment.PyEnvironment): def __init__(self): """Initialize environment.""" def action_spec(self): """Return action_spec.""" def observation_spec(self): """Return observation_spec.""" def _reset(self): """Return initial_time_step.""" def _step(self, action): """Apply action and return new time_step."""
最も重要な関数は _step()
関数です。この関数は、アクションを受け取って新しい time_step
オブジェクトを返します。Plane Strike ゲームの場合は、ゲームボードが 1 つあります。新しいストライク ポジションに入ったとき、ゲーム盤の状態に基づいて、環境は次のように判断します。
- 次にゲームボードはどのように表示されますか(非表示の飛行機の位置に応じて、セルの色は赤と黄色のどちらに変わるでしょうか?)
- そのポジションで、プレーヤーはどのような報酬を受け取るべきですか(報酬にヒットするか、ペナルティを逃すか)。
- ゲームを終了すべきでしょうか(勝った人はいますか?)
_planestrike_py_environment.py
ファイルの_step()
関数に次のコードを追加します。
if self._hit_count == self._plane_size: self._episode_ended = True return self.reset() if self._strike_count + 1 == self._max_steps: self.reset() return ts.termination( np.array(self._visible_board, dtype=np.float32), UNFINISHED_GAME_REWARD ) self._strike_count += 1 action_x = action // self._board_size action_y = action % self._board_size # Hit if self._hidden_board[action_x][action_y] == HIDDEN_BOARD_CELL_OCCUPIED: # Non-repeat move if self._visible_board[action_x][action_y] == VISIBLE_BOARD_CELL_UNTRIED: self._hit_count += 1 self._visible_board[action_x][action_y] = VISIBLE_BOARD_CELL_HIT # Successful strike if self._hit_count == self._plane_size: # Game finished self._episode_ended = True return ts.termination( np.array(self._visible_board, dtype=np.float32), FINISHED_GAME_REWARD, ) else: self._episode_ended = False return ts.transition( np.array(self._visible_board, dtype=np.float32), HIT_REWARD, self._discount, ) # Repeat strike else: self._episode_ended = False return ts.transition( np.array(self._visible_board, dtype=np.float32), REPEAT_STRIKE_REWARD, self._discount, ) # Miss else: # Unsuccessful strike self._episode_ended = False self._visible_board[action_x][action_y] = VISIBLE_BOARD_CELL_MISS return ts.transition( np.array(self._visible_board, dtype=np.float32), MISS_REWARD, self._discount,
8. ステップ 2: TensorFlow エージェントを使用してゲーム エージェントをトレーニングする
TF エージェント環境が整ったら、ゲーム エージェントをトレーニングできます。この Codelab では、REINFORCE エージェントを使用します。REINFORCE は RL のポリシー勾配アルゴリズムです。基本的な考え方は、ゲームプレイ中に収集された報酬シグナルに基づいてポリシー ニューラル ネットワークのパラメータを調整することです。これにより、ポリシー ネットワークは将来のプレイで利益を最大化できます。
- まず、トレーニング環境と評価環境をインスタンス化する必要があります。
step2/backend/training.py
ファイルのtrain_agent()
関数に、次のコードを追加します。
train_py_env = planestrike_py_environment.PlaneStrikePyEnvironment( board_size=BOARD_SIZE, discount=DISCOUNT, max_steps=BOARD_SIZE**2 ) eval_py_env = planestrike_py_environment.PlaneStrikePyEnvironment( board_size=BOARD_SIZE, discount=DISCOUNT, max_steps=BOARD_SIZE**2 ) train_env = tf_py_environment.TFPyEnvironment(train_py_env) eval_env = tf_py_environment.TFPyEnvironment(eval_py_env)
- 次に、トレーニング対象の強化学習エージェントを作成する必要があります。この Codelab では、ポリシーベースのエージェントである REINFORCE エージェントを使用します。上記のコードのすぐ下に次のコードを追加します。
actor_net = tfa.networks.Sequential( [ tfa.keras_layers.InnerReshape([BOARD_SIZE, BOARD_SIZE], [BOARD_SIZE**2]), tf.keras.layers.Dense(FC_LAYER_PARAMS, activation="relu"), tf.keras.layers.Dense(BOARD_SIZE**2), tf.keras.layers.Lambda(lambda t: tfp.distributions.Categorical(logits=t)), ], input_spec=train_py_env.observation_spec(), ) optimizer = tf.keras.optimizers.Adam(learning_rate=LEARNING_RATE) train_step_counter = tf.Variable(0) tf_agent = reinforce_agent.ReinforceAgent( train_env.time_step_spec(), train_env.action_spec(), actor_network=actor_net, optimizer=optimizer, normalize_returns=True, train_step_counter=train_step_counter, )
- 最後に、トレーニング ループでエージェントをトレーニングします。ループ内では、まずゲームプレイのエピソードをバッファに収集し、バッファ内のデータでエージェントをトレーニングします。
step2/backend/training.py
ファイルのtrain_agent()
関数に、次のコードを追加します。
# Collect a few episodes using collect_policy and save to the replay buffer. collect_episode( train_py_env, collect_policy, COLLECT_EPISODES_PER_ITERATION, replay_buffer_observer, ) # Use data from the buffer and update the agent's network. iterator = iter(replay_buffer.as_dataset(sample_batch_size=1)) trajectories, _ = next(iterator) tf_agent.train(experience=trajectories) replay_buffer.clear()
- これで、トレーニングを開始できます。ターミナルでパソコンの
step2/backend
フォルダに移動し、次のコマンドを実行します。
python training.py
トレーニングが完了するまでには、ハードウェア構成に応じて 8 ~ 12 時間かかります(事前トレーニング済みモデルは step3
で提供されているため、自分でトレーニング全体を完了する必要はありません)。それまでの間、TensorBoard を使用して進行状況をモニタリングできます。新しいターミナルを開き、パソコンの step2/backend
フォルダに移動して次のコマンドを実行します。
tensorboard --logdir tf_agents_log/
tf_agents_log
は、トレーニング ログを含むフォルダです。トレーニング実行のサンプルは次のようになります。
トレーニングが進むにつれて、エピソードの平均長さは短くなり、平均収益は増加することがわかります。直感的には、エージェントがスマートで予測精度が高いほど、ゲームの長さが短くなり、エージェントがより多くの報酬を集められることが理解できます。エージェントは、後のステップで特典の大幅な割引を最小限に抑えるために、より少ないステップでゲームを終了させたいと考えているため、理にかなっています。
トレーニングが完了すると、トレーニング済みモデルが policy_model
フォルダにエクスポートされます。
9. ステップ 3: TensorFlow Serving を使用してトレーニング済みモデルをデプロイする
ゲーム エージェントをトレーニングしたので、TensorFlow Serving を使用してゲーム エージェントをデプロイできます。
- ターミナルで、パソコンの
step3/backend
フォルダに移動して、Docker による TensorFlow Serving を起動します。
docker run -t --rm -p 8501:8501 -p 8500:8500 -v "$(pwd)/backend/policy_model:/models/policy_model" -e MODEL_NAME=policy_model tensorflow/serving
Docker は、まず TensorFlow Serving のイメージを自動的にダウンロードします。これには 1 分ほどかかります。その後、TensorFlow Serving が起動します。次のようなログが出力されます。
2022-05-30 02:38:54.147771: I tensorflow_serving/model_servers/server.cc:89] Building single TensorFlow model file config: model_name: policy_model model_base_path: /models/policy_model 2022-05-30 02:38:54.148222: I tensorflow_serving/model_servers/server_core.cc:465] Adding/updating models. 2022-05-30 02:38:54.148273: I tensorflow_serving/model_servers/server_core.cc:591] (Re-)adding model: policy_model 2022-05-30 02:38:54.262684: I tensorflow_serving/core/basic_manager.cc:740] Successfully reserved resources to load servable {name: policy_model version: 123} 2022-05-30 02:38:54.262768: I tensorflow_serving/core/loader_harness.cc:66] Approving load for servable version {name: policy_model version: 123} 2022-05-30 02:38:54.262787: I tensorflow_serving/core/loader_harness.cc:74] Loading servable version {name: policy_model version: 123} 2022-05-30 02:38:54.265010: I external/org_tensorflow/tensorflow/cc/saved_model/reader.cc:38] Reading SavedModel from: /models/policy_model/123 2022-05-30 02:38:54.277811: I external/org_tensorflow/tensorflow/cc/saved_model/reader.cc:90] Reading meta graph with tags { serve } 2022-05-30 02:38:54.278116: I external/org_tensorflow/tensorflow/cc/saved_model/reader.cc:132] Reading SavedModel debug info (if present) from: /models/policy_model/123 2022-05-30 02:38:54.280229: I external/org_tensorflow/tensorflow/core/platform/cpu_feature_guard.cc:142] This TensorFlow binary is optimized with oneAPI Deep Neural Network Library (oneDNN) to use the following CPU instructions in performance-critical operations: AVX2 FMA To enable them in other operations, rebuild TensorFlow with the appropriate compiler flags. 2022-05-30 02:38:54.332352: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:206] Restoring SavedModel bundle. 2022-05-30 02:38:54.337000: I external/org_tensorflow/tensorflow/core/platform/profile_utils/cpu_utils.cc:114] CPU Frequency: 2193480000 Hz 2022-05-30 02:38:54.402803: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:190] Running initialization op on SavedModel bundle at path: /models/policy_model/123 2022-05-30 02:38:54.410707: I external/org_tensorflow/tensorflow/cc/saved_model/loader.cc:277] SavedModel load for tags { serve }; Status: success: OK. Took 145695 microseconds. 2022-05-30 02:38:54.412726: I tensorflow_serving/servables/tensorflow/saved_model_warmup_util.cc:59] No warmup data file found at /models/policy_model/123/assets.extra/tf_serving_warmup_requests 2022-05-30 02:38:54.417277: I tensorflow_serving/core/loader_harness.cc:87] Successfully loaded servable version {name: policy_model version: 123} 2022-05-30 02:38:54.419846: I tensorflow_serving/model_servers/server_core.cc:486] Finished adding/updating models 2022-05-30 02:38:54.420066: I tensorflow_serving/model_servers/server.cc:367] Profiler service is enabled 2022-05-30 02:38:54.428339: I tensorflow_serving/model_servers/server.cc:393] Running gRPC ModelServer at 0.0.0.0:8500 ... [warn] getaddrinfo: address family for nodename not supported 2022-05-30 02:38:54.431620: I tensorflow_serving/model_servers/server.cc:414] Exporting HTTP/REST API at:localhost:8501 ... [evhttp_server.cc : 245] NET_LOG: Entering the event loop ...
サンプル リクエストをエンドポイントに送信して、期待どおりに動作していることを確認できます。
curl -d '{"signature_name":"action","instances":[{"0/discount":0.0,"0/observation":[[0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]],"0/reward":0.0,"0/step_type":0}]}' -X POST http://localhost:8501/v1/models/policy_model:predict
エンドポイントは予測位置 45
を返します。これは盤面の中央に位置する (5, 5) です(興味がある方は、なぜ盤面の中央が最初のストライク位置として適切な位置なのかを考えてみてください)。
{ "predictions": [45] }
これで、NPC エージェントの次のストライク位置を予測するバックエンドが正常に構築されました。
10. ステップ 4: Android および iOS 用の Flutter アプリを作成する
バックエンドの準備が整いました。リクエストの送信を開始して、Flutter アプリからストライク位置の予測を取得できます。
- まず、送信する入力をラップするクラスを定義する必要があります。次のコードを
step4/frontend/lib/game_agent.dart
ファイルに追加します。
class Inputs { final List<double> _boardState; Inputs(this._boardState); Map<String, dynamic> toJson() { final Map<String, dynamic> data = <String, dynamic>{}; data['0/discount'] = [0.0]; data['0/observation'] = [_boardState]; data['0/reward'] = [0.0]; data['0/step_type'] = [0]; return data; } }
これで、TensorFlow Serving にリクエストを送信して予測を行えるようになりました。
step4/frontend/lib/game_agent.dart
ファイルのpredict()
関数に、次のコードを追加します。
var flattenedBoardState = boardState.expand((i) => i).toList(); final response = await http.post( Uri.parse('http://$server:8501/v1/models/policy_model:predict'), body: jsonEncode(<String, dynamic>{ 'signature_name': 'action', 'instances': [Inputs(flattenedBoardState)] }), ); if (response.statusCode == 200) { var output = List<int>.from( jsonDecode(response.body)['predictions'] as List<dynamic>); return output[0]; } else { throw Exception('Error response'); }
アプリがバックエンドからレスポンスを受信したら、ゲーム UI を更新してゲームの進行状況を反映します。
step4/frontend/lib/main.dart
ファイルの_gridItemTapped()
関数に、次のコードを追加します。
int agentAction = await _policyGradientAgent.predict(_playerVisibleBoardState); _agentActionX = agentAction ~/ _boardSize; _agentActionY = agentAction % _boardSize; if (_playerHiddenBoardState[_agentActionX][_agentActionY] == hiddenBoardCellOccupied) { // Non-repeat move if (_playerVisibleBoardState[_agentActionX][_agentActionY] == visibleBoardCellUntried) { _agentHitCount++; } _playerVisibleBoardState[_agentActionX][_agentActionY] = visibleBoardCellHit; } else { _playerVisibleBoardState[_agentActionX][_agentActionY] = visibleBoardCellMiss; } setState(() {});
実行
- [ Start debugging] をクリックして、アプリが読み込まれるまで待ちます。
- エージェントのボードのセルをタップすると、ゲームが始まります。
11. ステップ 5: デスクトップ プラットフォーム用に Flutter アプリを有効にする
Flutter は、Android と iOS だけでなく、Linux、Mac、Windows などのデスクトップ プラットフォームもサポートしています。
Linux
- VSCode のステータスバーで、対象デバイスが に設定されていることを確認します。
- [ Start debugging] をクリックして、アプリが読み込まれるまで待ちます。
- エージェントのボードのセルをクリックすると、ゲームが開始されます。
Mac
- Mac の場合、アプリがバックエンドに HTTP リクエストを送信するため、適切な利用資格を設定する必要があります。詳しくは、利用資格とアプリ サンドボックスをご覧ください。
このコードをそれぞれ step4/frontend/macOS/Runner/DebugProfile.entitlements
と step4/frontend/macOS/Runner/Release.entitlements
に追加します。
<key>com.apple.security.network.client</key>
<true/>
- VSCode のステータスバーで、対象デバイスが に設定されていることを確認します。
- [ Start debugging] をクリックして、アプリが読み込まれるまで待ちます。
- エージェントのボードのセルをクリックすると、ゲームが開始されます。
Windows
- VSCode のステータスバーで、対象デバイスが に設定されていることを確認します。
- [ Start debugging] をクリックして、アプリが読み込まれるまで待ちます。
- エージェントのボードのセルをクリックすると、ゲームが開始されます。
12. ステップ 6: ウェブ プラットフォーム用に Flutter アプリを有効にする
もう 1 つできることは、Flutter アプリにウェブサポートを追加することです。ウェブ プラットフォームは、デフォルトで Flutter アプリに対して自動的に有効になるため、ユーザーが行う作業は起動することだけです。
- VSCode のステータスバーで、対象デバイスが に設定されていることを確認します。
- [ Start debug] をクリックし、アプリが Chrome ブラウザに読み込まれるまで待ちます。
- エージェントのボードのセルをクリックすると、ゲームが開始されます。
13. 完了
人間のプレーヤーと対戦する ML 搭載エージェントで、ボードゲーム アプリを作成しました。