前編のあらすじ

この記事は「Pytorchで自作ゲームにDQNのAIを組み込もう！(前編)」の続きとなります。

前編ではpygameでゲームを作成し、DQNのAIを組み込むためにゲームの環境をAIに入力する仕組みを用意しました。

この記事ではいよいよ、使用するモデルや学習する方法について解説していきます。

AIのモデル作成

AIに入力するデータの準備ができたので、ここからAIの説明に入ります。

モデルの定義

今回のゲームで使用するAIのモデルの定義部のソースコードは以下になります。

1import torch.nn as nn
2Outputs = 24
3
4class Agent(nn.Module):
5    def __init__(self):
6        super().__init__()
7        self.relu = nn.ReLU()
8        self.conv1 = nn.Conv2d(12, 8, 3)
9        self.conv2 = nn.Conv2d(8, 4, 3)
10        self.pool = nn.MaxPool2d(2, stride=2)
11        self.fc1 = nn.Linear(1564,32)
12        self.fc2 = nn.Linear(32, Outputs)
13    
14    def forward(self, x):
15        x = self.conv1(x)
16        x = self.relu(x)
17        x = self.pool(x)
18        x = self.conv2(x)
19        x = self.relu(x)
20        x = self.pool(x)
21        x = x.view(x.size()[0], -1)
22        x = self.fc1(x)
23        x = self.relu(x)
24        x = self.fc2(x)
25
26        return x

Pytorchでは、torch.nn.Moduleを継承したクラスを定義することでディープラーニングのモデルを作成します。Pythonではクラスの定義時にclass クラス名(継承したいクラスのクラス名): とすることでクラスを継承できます。モデルの構造や処理はこのクラス内で実装します。続いて、クラス内の関数について説明します。

まずは__init__関数です。__init__関数は他の言語では「コンストラクタ」などと呼ばれるもので、このクラスのインスタンス生成と同時に実行されます。Pytorchでは訓練や推論時に使用する線形変換の層や活性化関数の定義に使用します。それでは、__init__関数の中身について説明します。

• super().__init__()：super()は継承したクラス(親クラス)を返す関数で、super().__init__()とすることで継承したクラスのコンストラクタを呼ぶことができます。これを書き忘れると継承したクラスで本来必要だった初期化が行われずに処理が進んでしまい、思わぬエラーや不具合の原因になるので忘れずに書きましょう。
• self.relu = nn.ReLU()：Agentクラスのメンバ変数に「活性化関数」を追加しています。活性化関数はニューラルネットワークの各層の間に入り、次の層へ渡す値を調整する役割を持っています。この活性化関数はどんなものでも良いというわけではなく、「非線形(グラフが直線ではない)関数であること」「微分可能な(関数が途切れていたり、尖っている点(尖点)や接線が垂直になるような点を含まない)関数であること」の２つの条件を満たす必要があります。今回使用しているのは入力値が0以下の時は0、それ以上の時は入力値をそのまま返すReLU関数です。ReLU関数はニューラルネットワークの活性化関数としてよく使用される関数で、活性化関数に求められる性質を全て満たしています。
• self.conv1 = nn.Conv2d(12, 8, 3)：入力層(AIへの入力を受け取る層)を定義しています。Conv2dはPytorchが提供する２次元の畳み込みニューラルネットワーク(CNN)です。今回の入力であるゲーム画面のような画像データを扱う時はCNN、と覚えておけば良いかと思います。Conv2dでは、主に３つ引数を指定します。１個目は入力のチャンネル数、２個目はこのCNNを通過後に出力するデータのチャンネル数、３個目はカーネルと呼ばれる畳み込み演算を行う時に使用するフィルタのサイズです。今回の場合、入力される画像データのチャンネル数は１フレームにつき赤・緑・青の各成分が1次元ずつの合計3次元、それが連続4フレーム入力されるので3×4 = 12 チャンネルとなるため、１個目の引数は12となります。一方、２個目の引数である出力チャンネル数に特に縛りはないようです。好きな値を指定しましょう。３個目のカーネルのサイズについても自由に値を決めて大丈夫なようです(カーネルのサイズに１を指定すると処理として意味合いが変わってくるようなので、気になる方は１以外を指定しましょう)。今回は3を指定し、3×3のカーネルで畳み込みを行います。
• self.conv2 = nn.Conv2d(8, 4, 3)：最初の中間層(入力層と出力層の間に入る層)を定義しています。今回は入力層のCNNの出力チャンネル数に8を指定しているので、１個目の引数には8を指定する必要があります(合っていないとエラーになります)。基本的にどんな種類の層を使用する場合でも、「前の層の出力の次元数と次の層の入力の次元数を同じにする」ということさえ分かっていればディープラーニングのモデルを組む時に困ることはないと思います。２個目の引数やカーネルのサイズについては自由な値を入れてOKです。
• self.pool = nn.MaxPool2d(2, stride=2)：「プーリング」と呼ばれる、画像を縮小する処理を行う層です。位置のズレによってAIが同じ物体を「同じ」とみなせなくなるといった事態を防いだり、画像の縮小によって処理を軽量化する役割があります。今回定義しているのはMaxプーリングと呼ばれ、１個目の引数で指定されたサイズの領域内の最大値をその領域の代表値として取り出すことで縮小を行います。今回は2を指定しているので、2×2の領域でMaxプーリングを行います。stride=2はこの領域を何マス飛ばしで行うか(ストライド)の指定で、stride=2は２マスずつ飛ばして領域を作ります。
• self.fc1 = nn.Linear(1564,32)：2番目の中間層であり、全結合層を定義します。前のCNNの層で学習した結果を受け取り、最後の出力層に渡します。１個目の引数はこの層に入力されるデータの次元数、２個目の引数は出力するデータの次元数です。先ほどと同様、１個目の引数の次元数を前の層の出力の次元数と揃える必要があるのですが、「CNN層と全結合層で種類が違うけど、どうやって揃えればいいの？」と思った方もいらっしゃるかもしれません。これについては公式が存在しており計算でも求められますが、公式を使わなくても入力次元数の指定が誤っていた場合には「mat1 and mat2 shapes cannot be multiplied (A × B and C × D)(A、B、C、Dの位置に入る実際の数値は条件によって異なります)」というエラーが出るため、このエラーメッセージ内のBの数値を１個目の引数に指定すれば大丈夫です(モデルによっては内部処理の微妙な違いなどで入ってくる次元数が変わってくることもあるため、実際にエラーメッセージから出力された値を使ったほうが確実な印象です)。
• self.fc2 = nn.Linear(32, Outputs)：出力層(AIの実行結果を出力する層)の全結合層です。今回のAIの入力はゲーム画面の画像データですが、必要な出力はAIの行動であるため全結合層を出力とします。今回のゲームで行える行動は全部で24通りあるため、出力も24とします。

次はforward関数です。forward関数では、__init__関数で定義した各層と活性化関数、その他に必要な処理を順に実行します。基本的には入力層→活性化関数→その他の処理(必要な場合)→中間層→...→出力層 という順に処理を進めていきます。そして、x = x.view(x.size()[0], -1)の処理を２つ目のプーリングと全結合層の間に入れることでプーリングした結果の行列をベクトルに引き延ばしています。これを忘れると後でモデルの出力を元にAIを行動させる時やモデルの学習時にデータの形状がおかしくなります。CNN層から全結合層にデータを渡す時は、その直前に引き伸ばしの処理を入れておきましょう(次元数が合っていない場合などとは異なり、書き忘れてもそのまま処理が進んでしまうので特に注意しましょう！)。

学習用環境の用意

入力するデータや使用するモデルの説明が終わったので、次はモデルを学習させる方法について説明していきます。

1. 報酬の設定

DQNでは、行動した結果の得点(報酬)を設定することで行動を学習させていきます。報酬はAIにどのように行動すべきかを教える指針になります。

「ところで、報酬はどのように設定したらいいの？」と思った方もいらっしゃるかもしれません。ゲームタイトルやジャンルによって得点条件や得られる得点の数値は異なっていますが、強化学習に関して言えば得点の付け方には以下のようなセオリーが存在しているので、それらに従えばOKです。

• ゲーム1回ごとの報酬を-1から1の範囲に収まるように調整(報酬クリッピング)する(学習の円滑化及び他のゲームでもハイパーパラメーターを変えずにそのまま学習可能にするため)
• 目標を達成したら1、失敗したら-1にする(強化学習は報酬を最大化するように学習するため)

上記を鑑みて、今回のゲームでは以下のように報酬を設定しました。

• ゲームに勝ったら報酬1、負けたら報酬-1
• １分経っても決着が付かない時は報酬-1とする(弾を撃たずに相手の自滅を待つような消極的なプレイを防止するため)

2. 訓練の方法

報酬の設定ができたので、実際に訓練を行う方法について実際のソースコードを交えながら、順に説明します。

(0) 訓練用環境の概要

今回AIを作成するゲームは対戦型のゲームです。ゲームをプレイするには対戦相手が必要になります。今回強化学習でAIを作ったのはAIの行動パターン作成を自動化するためなので、わざわざ人間のプレイヤーに対戦してもらったのでは本末転倒です。また、強化学習でゲームをプレイできるようにするには何万回もプレイしないといけないので、何万回もずっとプレイし続けるというのも無理があります。

強化学習に頼らない「⚪︎⚪︎なら××する」の条件分岐によるAIと対戦させるという方法もありますが、条件分岐のAIとの対戦ばかり続けていると強さがどこかで頭打ちになったり、作成したAIの行動に最適化されすぎてしまうなどの問題が発生します。

そこで今回は全く同じ構造の強化学習モデルをもう1個用意し、２つの強化学習モデル同士で対戦してもらう形にしました。実際に強化学習AIをゲームに組み込む時は片方のAIのみを使用します(もう一方は人間のプレイヤーが操作します)。

(1) モデルの準備

まずは学習するためのモデルを準備します。

最初に、以下の6行で必要なライブラリをインポートします。AgentモジュールはAIのモデル作成にて解説したAIのモデルを定義したモジュールで、Memoryモジュールは経験再生用に用意したメモリを定義するモジュールです。

1import Agent
2import torch
3import torch.optim as optim
4import torch.nn as nn
5import Memory
6from collections import deque

続いて、以下の記述でGPUが使用可能かをチェックします。

1DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")

torch.cuda.is_available()関数はこのプログラムを実行しているPCでPytorchがGPUを利用できるならTrue、そうでないならFalseを返します。この記述により、torch.cuda.is_available()がTrue(GPUを利用できる)ならGPUを計算処理用デバイスとして設定し、そうでない時はCPUを設定します。

次に、以下の記述で経験記憶用のメモリと機械学習のモデルのインスタンスを生成します。

1memsize = 1000
2Model1P = Agent.Agent().to(DEVICE)
3Model2P = Agent.Agent().to(DEVICE)
4Target_Model1P = Agent.Agent().to(DEVICE)
5Target_Model2P = Agent.Agent().to(DEVICE)
6Memory1P = Memory.ReplayMemory(memsize)
7Memory2P = Memory.ReplayMemory(memsize)

まずはmemsizeによって経験記憶で記憶する件数の上限を指定し、経験記憶用のクラスであるMemory.ReplayMemoryクラスのインスタンス生成時にmemsizeを引数に渡すことで指定した上限を設定します。そして次に、AIのモデルのインスタンスを生成します。今回はDQNを使用するため、実際に学習するネットワークとターゲットネットワークの２つを用意します。さらに(0) 訓練用環境の概要でも説明したように、２つの強化学習モデル同士で対戦するため、実際に学習するネットワークとターゲットネットワークをもう1組用意するため、合計4つのモデルを用意します。モデルのインスタンスを生成した後に.to(DEVICE)を呼び出すことで作成したモデルのインスタンスをGPUに転送します。GPUが利用可能かチェックした後に明示的に転送処理を書く必要がある点に注意しましょう。

そして最後に、損失関数と最適化関数を用意します。ソースコードは以下です。

1criterion1P = nn.HuberLoss()
2criterion2P = nn.HuberLoss()
3optimizer1P = optim.Adam(Model1P.parameters(),lr=0.001,weight_decay=0.005)
4optimizer2P = optim.Adam(Model2P.parameters(),lr=0.001,weight_decay=0.005)

まずは損失関数についてです。損失関数は目標とする数値とモデルが出力した予測値のズレを求めるための関数です。よく使用される損失関数には平均二乗誤差(MSE)がありますが、平均二乗誤差には誤差が大きすぎると学習が安定しなくなるという欠点があります。そのため、今回は誤差が大きくても安定して学習できるHuber関数(nn.HuberLoss()クラス)を使用しています。

次は最適化関数についてです。最適化関数とは、損失関数で求めた予測値のズレを元に、ズレが小さくなるようにモデル内のパラメーターをどのように調整すべきかを求める関数です。今回は広く使用されているAdamを使用します。

torch.optim.Adamクラスのコンストラクタに設定する引数は以下のようになります。

• 1個目の引数：最適化対象のモデルのパラメーター(モデルのインスタンス.parameters()で取得可能)
• 2個目の引数lr：学習率と呼ばれるパラメーターです。0.01や0.001などを設定することが多いようです。今回は0.001を設定しています。
• 3個目の引数weight_decay：重み減衰のパラメーターです。これはモデルが訓練用のデータに過剰に適合してしまって、実際のデータに対する予測の精度が悪化することを防ぐための設定です。今回は0.005を指定しています。

損失関数と最適化関数はそれぞれ強化学習モデル1つにつき1個、計2個ずつ用意します。

(2) ハイパーパラメーター・入力の準備

次はモデルを訓練する上で、ハイパーパラメーター(プログラマー側で決定しておく必要がある定数類)やモデルに入力するデータの準備について説明します。

1batch_size = 32
2gamma = 0.99
3max_episode = 10000
4eps_start = 1.0
5eps_end = 0.01
6eps_reduce_rate = 0.001
7current_episode = 0
8step = 0
9total_step = 0
10input_frames = 4
11window_dim = 3

ソースコードの内容について、順に説明していきます。

• batch_size：バッチサイズ(訓練を一回行う時に一度に投入するデータの件数)を指定しています。バッチサイズの指定を大きくするほど学習時間の短縮や外れ値の影響を受けにくくなるなどのメリットが得られますが、大きくし過ぎるとその分メモリの消費量も多くなります。
• gamma：Q学習における割引率の設定です。基本的には0より大きく１未満の値を設定します。割引率は遠い将来の報酬よりも近い将来の報酬を重視して行動させるためのもので、この値を小さくするほどその傾向が強くなります。
• max_episode：訓練のために最大で何エピソード訓練させるか(何回対戦するか)の設定です。
• eps_start・eps_end・eps_reduce_rate：強化学習で重要になる活用と探索のバランスを取るための手法である「ε-greedy法」に関連するパラメーターです。ε-greedy法では、確率εで探索(ランダムな行動を取る)を行い、それ以外では活用(強化学習のモデルが予測する最適な行動を取る)を行なって行動を決定します。今回のプログラムでは訓練が進むほど確率εの値は減っていく(活用を増やす)ようにしています。最初はゲームについて何も知らないのでランダムな行動を多くして経験を積み、十分な経験を積んだところで強化学習に任せるようなイメージです。eps_startは確率εの初期値、eps_endは確率εの最終的な値、eps_reduce_rateは確率εの減少スピードの設定です。
• current_episode・step・total_step：何エピソード・何ステップ(何フレーム)訓練しているかを記録するための変数です。
• input_frames・window_dim：input_framesは連続何フレームを入力にするかの設定で、window_dimは1フレームあたりの入力画像のチャンネル数です。

続いては入力データの準備です。以下のソースコードで準備を行なっています。

1InputDeque = deque(maxlen=input_frames)
2InputDeque.extend(np.zeros((input_frames, window_dim, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale))))

まずはPython標準ライブラリのcollections.dequeを利用して入力用データを格納するための領域を作成します。ここに連続するゲーム画面のフレームを格納していきますが、この段階ではゲームが始まっていないのでゲーム画面のフレームのデータもありません。そこで、代わりにダミーのフレームデータをinput_framesの数だけ格納します。さらにcollections.dequeを用意する時の引数にinput_framesを指定することで、ゲーム画面のフレームのデータを格納するだけで自動的に一番古いフレームのデータが削除されるようになります。

(3) 訓練ループ開始

いよいよ、学習のメイン部分である訓練ループについての解説です。引き続きソースコードを交えながら説明します。まずはpygame関連の処理についてです。

1while current_episode < max_episode:
2        #最大フレームレートを30fpsで固定
3        clock = pygame.time.Clock()
4        clock.tick(30)
5        for event in pygame.event.get():
6            if event.type == pygame.QUIT:
7                SaveModel()
8                pygame.quit()
9                sys.exit()
10        pygame.display.update()

訓練ループではゲームの対戦をmax_episode回実行します。ソースコードではwhileでループを回すことで実現しています。以降の処理の説明は以下の通りです。

• clock = pygame.time.Clock()・clock.tick(30)：フレームレートの設定です。詳しくは前編の②フレームレートへの対応にて解説しています。
• for event in pygame.event.get()〜sys.exit()：プログラム終了についての処理です。詳しくは前編の① ゲームループについてにて解説しておりますが、訓練可能かどうか動作確認だけしたい、あるいは途中で学習を止めた場合でも訓練済みモデルが欲しいというケースにも対応するため、訓練済みモデルを保存するSaveModel関数の呼び出しも追加しております。SaveModel関数については後ほど詳しく解説します。
• pygame.display.update()：pygameで作成したゲーム画面を更新します。

(4) AIの行動を反映

次はAIの行動をゲームに反映させるための処理です。ソースコードは以下です。なお、以降の処理については基本的にwhileループの中で行なっているものとして話を進めます。

1epsilon = eps_end + (eps_start - eps_end) * np.exp(-eps_reduce_rate * total_step)        
2step += 1
3total_step += 1
4State = np.array(InputDeque)
5if epsilon > np.random.rand():
6    action1P = np.random.randint(0, Agent.Outputs)
7    action2P = np.random.randint(0, Agent.Outputs)
8else:
9    Model1P.eval()
10    Model2P.eval()
11    with torch.no_grad():
12        Input = State.reshape((1, -1, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))
13        Input = torch.from_numpy(Input).float().to(DEVICE)
14        action1P = torch.argmax(Model1P(Input)).cpu().detach().numpy()
15        action2P = torch.argmax(Model2P(Input)).cpu().detach().numpy()
16        
17    NextObservation, finishedFlag, p1reward, p2reward = Game.update(Player1, Player2, action1P, action2P)
18    # 1分(30FPS × 60秒 = 1800フレーム)経っても決着が付かない時は両者負けとみなして次のエピソードへ
19    if finishedFlag == False and step >= 1800:
20        finishedFlag = True
21        p1reward = -1
22        p2reward = -1

• epsilon = eps_end + (eps_start - eps_end) * np.exp(-eps_reduce_rate * total_step)：探索の行動をとる確率εを更新する処理です。更新のたびに確率εは減少し、かわりに活用の行動を取る可能性が上がります。
• step += 1・total_step += 1：新しいフレームに入るので、まずはstepとtotal_stepを1加算します。
• State = np.array(InputDeque)：InputDequeをnumpy配列に変換しています。
• if epsilon > np.random.rand():〜action2P = np.random.randint(0, Agent.Outputs)：探索の行動を取る時の行動を決定するための処理です。ランダムな整数を生成するnp.random.randint関数によりランダムな行動を1つ選んで行動します。
• else:〜Model2P.eval()：活用の行動を取る場合の処理です。このタイミングではパラメータの更新などを行う必要はないため、(強化学習モデル).eval()でモデルを推論モードに切り替えます。
• with torch.no_grad():：モデルからの出力を得るための準備です。このwith句の中でモデルからの予測値を受け取ることで不要な計算を省きます。
• Input = State.reshape((1, -1, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))：入力の変換処理を行います。Stateの形状を(バッチサイズ, チャンネル数, 画像の幅, 画像の高さ)に変換します。基本的にPytorchでCNNを使用する時は、(バッチサイズ, チャンネル数, 画像の幅, 画像の高さ)(幅と高さは逆になっていてもよい)の形状に変形する必要があります。Stateの中身をそのままモデルに渡しているため、バッチサイズは1になります。形状の変形はreshape関数で行なっていますが、reshape関数では変形先の形状の指定において1個だけ-1を指定することができます。-1を指定した部分については他の形状指定から自動で設定されます。便利な機能なので覚えておきましょう。
• Input = torch.from_numpy(State).float().to(DEVICE)：入力の変換処理の続きです。Pytorchのモデルに入力を渡すには、入力したいデータをtorch.Tensorという特別なデータ型に変換する必要があります。numpy配列をtorch.Tensorに変換したい場合はtorch.from_numpy関数の引数にnumpy配列を渡し、その返り値を利用すればOKです。また、PytorchのCNNは基本的にtorch.float32のtorch.Tensorのみに対応しているため、float関数でtorch.float32のtorch.Tensorに変換します。さらに、to関数によって入力用のデータをモデルの計算処理用デバイスに転送します。
• action1P = torch.argmax(Model1P(Input)).cpu().detach().numpy()・action2P = torch.argmax(Model2P(Input)).cpu().detach().numpy()：モデルの出力を受け取るための処理です。Pytorchでは、モデルのインスタンス(入力値)とすることでモデルの予測値を取得できます。モデルの予測値は実行可能な行動それぞれについて対応する行動を取った時に得られる報酬の数値となっているため、引数のtorch.Tensor内の最大値を持つ添え字を求めるtorch.argmax関数に予測値を渡すことでモデルが最適と判断した行動がわかります。GPUで訓練している場合、そのままではPytorch以外のライブラリに予測値を渡せないため、cpu関数でデータをCPU側のメモリに転送します(CPUで訓練している場合は何もしません)。さらにtorch.Tensorは「計算グラフ」というニューラルネットワークを訓練する時に使用する特別なデータを保持していますが、numpyなど他のライブラリは計算グラフのデータを扱えないためdetach関数で計算グラフを破棄した上でnumpy関数を呼び出してnumpy配列に変換します。
• NextObservation, finishedFlag, p1reward, p2reward = Game.update(Player1, Player2, action1P, action2P)：AIの行動をゲームに反映させる部分です。Game.update関数に先ほどのaction1P・action2Pを渡すことで、AIが指示した行動の通りにゲームを動かします。ゲームを動かした後の結果はGame.update関数の返り値から確認することができます。この時の結果とモデルへの入力を利用して学習を行います。
• if finishedFlag == False and step >= 1800: 〜 p2reward = -1：1. 報酬の設定で説明したように、「１分経っても決着が付かない時は報酬-1とする」を実現するための記述です。stepは1フレームにつき1加算されていくため、これにより何フレーム経過したかを確認できます。さらに、finishedFlagはゲームが終了した時にTrue、していない時にFalseになるので、決着がついているかの確認に使えます。

(5) 経験記憶・「次の状態」関連の処理

続いて、経験記憶や「次の状態」に関連した処理です。このフレームでゲームが終了した時とそうでない時で処理が少し異なります。

まずはゲームが終了した時の処理です。「次の状態」をゲーム終了に対応したデータにします。

1if finishedFlag == True:
2    current_episode += 1
3    # NextObservationを「状態なし」に
4    NextObservation = np.full((window_dim, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)), -1)
5    InputDeque.append(NextObservation)
6    NextState = np.array(InputDeque)
7    Target_Model1P.load_state_dict(Model1P.state_dict())
8    Target_Model2P.load_state_dict(Model2P.state_dict())
9
10    Memory1P.append((State, action1P, p1reward, NextState))
11    Memory2P.append((State, action2P, p2reward, NextState))
12    step = 0
13    InputDeque.extend(np.zeros((input_frames, window_dim, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale))))
14    Game.start(Player1, Player2)

• current_episode += 1：ゲーム終了に伴い1エピソードが終了するので、エピソード数に1加算します。
• NextObservation = np.full((window_dim, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)), -1) ：ゲームが終了すると「次のフレーム」もなくなるため、NextObservationにダミーの「次のフレーム」をセットします。
• InputDeque.append(NextObservation)・NextState = np.array(InputDeque)：ダミーの「次のフレーム」を含む形で「次の状態」を作成します。
• Target_Model1P.load_state_dict(Model1P.state_dict())・Target_Model2P.load_state_dict(Model2P.state_dict())：ターゲットネットワークの更新処理です。今回はエピソード終了時に更新するため、ゲームが終了した時に処理しています。モデルのパラメータはモデルのインスタンス.state_dict()で取得できます。
• Memory1P.append((State, action1P, p1reward, NextState))・Memory2P.append((State, action2P, p2reward, NextState))：経験記憶用のメモリにデータを追加します。
• step = 0〜Game.start(Player1, Player2)：ゲームをリセットし、新しいゲームを開始させる処理です。ステップ数を0にして、入力用データとゲームを初期状態にリセットしています。

続いて、ゲームが終了しない時の処理です。次のフレームに向けての準備をしています。

1else:
2    InputDeque.append(Agent.convertStateToAgent(NextObservation, Agent.scale))
3    NextState = np.array(InputDeque)
4    Memory1P.append((State, action1P, p1reward, NextState))
5    Memory2P.append((State, action2P, p2reward, NextState))
6    State = NextState

• InputDeque.append(Agent.convertStateToAgent(NextObservation, Agent.scale))・NextState = np.array(InputDeque)：InputDequeにAIが行動した後のゲーム画面の1フレームを追加して、それを「次の状態」にします。Agent.convertStateToAgent関数を通すことで画像のリサイズ・正規化を行なった形で格納します。また、InputDequeはcollections.dequeなので格納を行った時点で古いデータは自動的に削除されます。
• Memory1P.append((State, action1P, p1reward, NextState))・Memory2P.append((State, action2P, p2reward, NextState))：経験記憶用のメモリにデータを追加します。
• State = NextState：次のフレームでの処理に備え、NextStateをStateにコピーします。

(6) モデルの訓練

モデルを訓練させる処理です。ここでは例として1P用のモデルについてのみ掲載しておりますが、実際は2P用のモデルでも同様の処理を行っております。また、訓練は経験記憶用のメモリ内のデータ件数がバッチサイズを超えている時のみ行います。

1if Memory1P.length() > batch_size:
2    Model1P.train()
3    miniBatch = Memory1P.sample(batch_size)
4    targets = np.empty((batch_size, Agent.Outputs))
5    inputs = np.empty((batch_size, (input_frames * window_dim), int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))
6    for i, (state, action, reward, nextState) in enumerate(miniBatch):
7        if reward == 0:
8            nextState = nextState.reshape((1, -1, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))
9            with torch.no_grad():
10                maxQ = Target_Model1P(torch.from_numpy(nextState).float().to(DEVICE)).flatten()
11            target = reward + gamma * torch.max(maxQ).detach().cpu().numpy()
12        else:
13            target = reward
14        inputs[i] = state.reshape((-1, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))
15        targets[i] = Model1P(torch.from_numpy(inputs[i].reshape((1, -1, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))).float().to(DEVICE)).flatten().detach().cpu().numpy()
16        targets[i][action] = target
17    optimizer1P.zero_grad()
18    outputs = Model1P(torch.from_numpy(inputs).float().to(DEVICE))
19    loss = criterion1P(outputs, torch.from_numpy(targets).float().to(DEVICE))
20    loss.backward()
21    optimizer1P.step()

• Model1P.train()：訓練に合わせてパラメータの更新を行う必要があるため、この1文でモデルを訓練モードに切り替えます。
• miniBatch = Memory1P.sample(batch_size)：経験記憶用のメモリから訓練に使用するデータをバッチサイズの個数分ランダムに取り出します。
• targets = np.empty((batch_size, Agent.Outputs))・inputs = np.empty((batch_size, (input_frames * window_dim), int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))：訓練のために入力値と目標とする出力を保持するnumpy配列を用意します。基本的に、numpy配列を作成する時に必要なサイズがわかっている時はnp.empty関数などで必要なサイズをあらかじめ一括で用意した方が高速に処理できます。
• for i, (state, action, reward, nextState) in enumerate(miniBatch):〜target = reward：経験記憶用のメモリからデータを取り出して1件ずつ処理するループです。enumerate()関数により、インデックス番号とminiBatchの中身を同時に取得することができます。このfor文では取り出したデータがゲーム終了時のデータかどうか(次の行動が存在するか)に応じて処理を行い、その結果を正解データにします。reward == 0の時(ゲーム終了時のデータでない場合)、取り出したnextStateについてターゲットネットワークで予測を行い、それを元に以下の式で取った行動の価値を求めます。

取った行動の価値 = 次のステップでの報酬 + 割引率 × 次のステップの行動価値関数の最大値

「次のステップでの報酬」は取り出したデータのreward、「次のステップの行動価値関数の最大値」はターゲットネットワークの予測値の最大値になります。一方、reward != 0の時(ゲーム終了時のデータの場合)は取り出したデータのrewardをそのまま使います。

• inputs[i] = state.reshape((-1, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))〜targets[i][action] = target：訓練用の入力値と正解データの準備です。入力値は取り出したデータのstateを形状変化した上でinputsに格納します。正解データは実際に学習するネットワークの訓練前の予測値をtargetsに格納した上で、targetsの内、取り出したデータのactionに該当する部分(実際に取った行動の結果)についてはtargetに上書きします。
• optimizer1P.zero_grad()：モデルの訓練を行うにあたり勾配のデータを初期化します。これを忘れると訓練がうまくいかなくなるので、忘れずに書いておきましょう。
• outputs = Model1P(torch.from_numpy(inputs).float().to(DEVICE))：訓練用の入力値から実際に学習するネットワークの予測値を取得します。
• loss = criterion1P(outputs, torch.from_numpy(targets).float().to(DEVICE))：「(1) モデルの準備」で用意した損失関数で実際に学習するネットワークの予測値と正解データの間の誤差を計算します。
• loss.backward()：誤差逆伝播という手法を使って、モデル内部のパラメータをどのように調整すれば誤差が小さくなるかを求めます。この段階ではモデル内部のパラメータを実際に調整していないことに注意してください。
• optimizer1P.step()：loss.backward()で求めた情報を元に、実際にモデル内部のパラメータを調整します。

ここまでの処理によってAIがゲームを繰り返しプレイしながら学習し、パラメータを修正して上達していきます。

(7) モデルの保存・読み込み

どんなに訓練したモデルであっても、そのままではプログラム終了とともに消えてしまいます。そのため、訓練したモデルを他で利用したい場合は訓練したモデルを保存することが必要になります。今回のプログラムではSaveModel関数で訓練したモデルを保存しており、訓練が終了した時、あるいはユーザーが「×」ボタンをクリックするなどによりプログラムが終了した時に自動で保存処理が行われるようになっています。SaveModel関数の定義は以下となっております。

1def SaveModel():
2    #モデル保存処理
3    torch.save(Model1P.state_dict(),"Model1P.pth")
4    torch.save(Model2P.state_dict(),"Model2P.pth")

Pytorchではtorch.save関数に保存したいモデルとその保存先のパスを引数に指定することで訓練済みのモデルを保存することができます。ただしここで一点注意が必要なのは、モデルをそのまま保存するのではなく、モデルのインスタンス.state_dict()を保存対象にして、モデルのパラメータを保存する必要があるということです。Pytorchでモデルをそのまま保存して読み込んだ場合、訓練に使用した計算処理用デバイスに自動で転送されるため、訓練に使用していた計算処理用デバイスが何かの事情で使用できない場合や別のPCで訓練したモデルを使おうとした時にロードできなくなる恐れがあるからです。必ずモデルのパラメータだけを保存するようにしましょう。

訓練したモデルのファイル名は好きな名前を指定して良いのですが、拡張子は「.pt」や「.pth」 とするのが慣例となっているようです。特に理由がなければファイルの拡張子は「.pt」か「.pth」 にしましょう。

なお、保存したモデルは以下のようにすれば読み込むことができます。

1. 保存したモデルと同じクラスのインスタンスを用意する。
2. (1.のインスタンス).load_state_dict(torch.load("読み込みたい訓練済みモデルのファイル名"))として、訓練済みモデルのパラメータを1.のインスタンスにコピーする。

訓練させたAIとの対戦

本記事のテーマは対戦型ゲームの相手AIの行動パターン作成を強化学習で自動化することですから、実際に訓練させたAIと対戦できるようにしましょう。

ここでは主に、学習用環境の用意のソースコードと違う部分を中心に説明します。対戦用環境とモデル学習用の環境のソースコードで大きく異なるのはメインループ内部の以下の部分です。

1with torch.no_grad():
2    Input = np.array(InputDeque).reshape((1, -1, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale)))
3    Input = torch.from_numpy(Input).float().to(DEVICE)
4    action2P = torch.argmax(Model2P(Input)).cpu().detach().numpy()
5NextObservation, finishedFlag, _, _ = Game.update(Player1, Player2, P2Input=action2P)
6InputDeque.append(NextObservation)
7if finishedFlag == True:
8    InputDeque.extend(np.zeros((input_frames, window_dim, int(Game.Width * Agent.scale), int(Game.Height * Agent.scale))))

まず、対戦用環境では訓練を行う必要がなく、訓練済みのモデルでは活用と探索のバランスを取る必要がないためランダムな行動は一切取らず全てAIの判断で動くようにします。また、今回は1Pを人間操作、2PをAI操作とするため、Game.update関数には2Pに対してのみAIの行動を引数として渡します(Game.update関数に行動を引数として渡さない場合、そのプレイヤーは人間操作となります)。経験記憶用のメモリも用意していないので、ゲームが終了した場合もInputDequeを初期化するのみです。

これでソースコードの解説は終わりです。ついに強化学習AIの組み込みに成功しました！

少し工夫は必要になりますが、強化学習のAIは行動パターン作成の自動化だけでなく最適なプレイ方法の分析やゲームバランスの調整などにも応用できます。強化学習を使いこなしてゲーム開発を効率化させましょう！

今後やりたいこと

ここまでの内容を踏まえて、今後さらに試していきたいことを2点ほど挙げたいと思います。

① モデルの改善

今回はDQNを利用してAIを実装しました。DQNは強化学習の中でも基本的な手法であり、様々な発展形がすでに考案されています。それらを実装することにより、学習の安定化や訓練の高速化など、様々なメリットを得ることができます。このゲームにおいては、特に以下の手法を実装することで更なる性能向上が見込めそうです。

1. RNNなどを利用して、ゲーム画面の連続数フレームをまとめて入力にしなくてもAIが物体の移動方向を把握できるようにする
2. モデル訓練時の正解データの作り方を工夫することで、取った行動の価値が実際より大きく見積もられてしまう現象を回避する(Double DQN)
3. 経験再生用のメモリからのデータの取り出し方を工夫して、より効率よく学習させる(優先度つき経験再生)

もし時間があれば、これらの改良も加えてみたいです。

② ゲームルールを変えた時の振る舞いの比較

こちらはAIを変えるのではなく、環境であるゲームを変えます。具体的には、以下のような変更が考えられます。

1. 弾の威力やスピードを変更する
2. 弾を撃った時に消費するエネルギー量を多くする/少なくする
3. エネルギーの回復速度や上限の変更
4. 自分の弾をUFOに当てた時の無敵状態になる時間やエイリアンに当てた時の威力の上昇幅の調整

これらを変えるということはゲームのルールが変わるということであり、最適な行動もそれに合わせて変わります。これらの振る舞いを比較することで、ゲームのルールやパラメーターがプレイにどのような影響を与えるかについても調べてみたいです。

最後に

ここまでいかがだったでしょうか？

「pygameによるゲーム作成」・「PytorchによるDQNの強化学習モデルの自作」・「ゲーム画面から環境を観測する」・「連続数フレームを入力に使用する」の4つを同時に満たす記事は非常に少なく、その点において役立つ記事になったと考えております。ぜひ皆さんも自作した対戦型ゲームのAIに強化学習を組み込んでみてください。

この記事が何かの参考になりましたら幸いです。

それではまたどこかでお会いしましょう。