Chainerを使ったDeep Q Networkの実装でのType Checkエラー

Question

強化学習について成果物を作る必要があり、三目並べを参考にプログラムを実装しました。ゲームの仕様はdungeon.pyに実装しており、定義したマス数＋周囲分の１マスを対象のボードとします。(N=3なら　5×5のボードで内周1マスが落とし穴となる)この質問ではN=3つまり5×5を想定させてください。

ユーザははじめ(1, 1)にいて、そのままランダムアクションと学習ずみのアクションを用いてゴール(3×3)を目指します。動き方は(右左下上)の4パターンで1マスずつ移動します。何度か思考を繰り返したのち、(Goal, Miss, Starvation)のいずれかの報酬をえます。Goalはゴールマスに到着した場合、Missは内周1マスの落とし穴にハマった場合、Starvationは10アクション以内にゴールできない場合です。Goal時のみreward=50を受け取ります。

以下にコードを掲載いたします。なお前述の三目並べを参考にしております。探索に必要なパラメータは次元数を(5*5の25)とし、アクションの次元数を4(右左下上)としました。dqn.pyにつきましては参照元と一緒のため割愛させていただきたいです。

#dungeon.py
import numpy as np


#ゲームボード
class Dungeon():
    def __init__(self, n):
        self.n = n
        self.obs_size = (n + 2) * (n + 2)
        self.action_size = 4
        # board の初期化
        board = []
        for i in range(n + 2):
            line = np.array([0] * (n + 2), dtype=np.float32)
            board.append(line)
        self.board = np.array(board, dtype=np.float32)
        ## board の外周に落とし穴(-1)を設置
        self.board[0] = np.array([-1] * (n + 2), dtype=np.float32)
        self.board[n + 1] = np.array([-1] * (n + 2), dtype=np.float32)
        for pos in self.board:
            pos[0] = -1
            pos[n + 1] = -1
        self.board[n][n] = 10

        self.pos = [1, 1]
        self.random_count = 0
        self.goal = False
        self.missed = False


    def reset(self):
        self.pos = [1, 1]
        self.goal = False
        self.missed = False


    def move(self, direction):
        if direction == "right":
            # right
            x_pos = self.pos[0] + 1
            y_pos = self.pos[1]
            self.pos = [x_pos, y_pos]
            if self.board[x_pos][y_pos] < 0:
                self.missed = True

            elif self.board[x_pos][y_pos] == 10:
                self.goal = True

        elif direction == "left":
            # left
            x_pos = self.pos[0] - 1
            y_pos = self.pos[1]
            self.pos = [x_pos, y_pos]
            if self.board[x_pos][y_pos] < 0:
                self.missed = True

            elif self.board[x_pos][y_pos] == 10:
                self.goal = True

        elif direction == "upper":
            # upper
            x_pos = self.pos[0]
            y_pos = self.pos[1] - 1
            self.pos = [x_pos, y_pos]
            if self.board[x_pos][y_pos] < 0:
                self.missed = True

            elif self.board[x_pos][y_pos] == 10:
                self.goal = True

        elif direction == "lower":
            # lower
            x_pos = self.pos[0]
            y_pos = self.pos[1] + 1
            self.pos = [x_pos, y_pos]

            if self.board[x_pos][y_pos] < 0:
                self.missed = True

            elif self.board[x_pos][y_pos] == 10:
                self.goal = True

    def get_random_action(self):
        self.random_count += 1
        directions = ["right", "left", "upper", "lower"]
        return np.random.choice(directions)

    def copy(self):
        line_board = self.board.copy()
        x_pos = self.pos[0]
        y_pos = self.pos[1]
        line_board[x_pos][y_pos] = 1
        return line_board.flatten()

    def show(self):
        line_board = self.board.copy()
        x_pos = self.pos[0]
        y_pos = self.pos[1]
        line_board[x_pos][y_pos] = 1
        print(line_board)

#train_dungeon.py
from dqn import QFunctions
from dungeon import Dungeon

import chainer
import chainerrl
import numpy as np
from tqdm import tqdm


if __name__ == "__main__":
   N = 3 # 3 * 3のマスであることを宣言
   d = Dungeon(N)
   # 環境の次元数
   obs_size = d.obs_size
   # 行動の次元数
   n_actions = d.action_size

   q_func = QFunctions(obs_size, n_actions)
   optimizer = chainer.optimizers.Adam(eps=1e-2)
   optimizer.setup(q_func)
   # 報酬の割引率
   gamma = 0.95
   # epsilon-greedyを使ってたまに冒険
   explorer = chainerrl.explorers.LinearDecayEpsilonGreedy(
       start_epsilon=1.0, end_epsilon=0.3, decay_steps=50000, random_action_func=d.get_random_action
   )
   replay_buffer = chainerrl.replay_buffer.ReplayBuffer(capacity=10**6)
   # agentの生成
   phi = lambda x:x.astype(np.float32, copy=False)
   agent_p1 = chainerrl.agents.DoubleDQN(q_func, optimizer, replay_buffer, gamma, explorer, replay_start_size=500, update_interval=1, target_update_interval=100, phi=phi)


   print("Start train...")
  # 学習ゲーム回数
   n_episodes = 20000
   miss = 0
   goal = 0
   starvation = 0
   # 繰り返し実行
   for i in range(1, n_episodes + 1):
       d.reset()
       reward = 0
       count = 0
       while not d.goal and not d.missed:
           print("before: ", d.pos)
           print(d.show(), reward)
           action = agent_p1.act_and_train(d.copy(), reward)
           print("action choice")
           d.move(action)
           print("after: ", d.pos)
           print(d.show(), reward)

           #ゲームが終わった場合
           if d.goal == True:
                print("-----Goal-----")
                goal += 1
                reward = 50
                if agent_p1.last_state is not None:
                    continue

                else:
                    agent_p1.stop_episode_and_train(d.copy(), reward, True)

           elif d.missed == True:
                print("-----Miss-----")
                miss += 1
                reward = 0
                if agent_p1.last_state is not None:
                    continue

                else:
                    agent_p1.stop_episode_and_train(d.copy(), reward, True)

           else:
                print("-----Stay-----")
                count += 1

           if count > 10:
                print("-----Starvation-----")
                starvation += 1
                if agent_p1.last_state is not None:
                    continue

                else:
                    agent_p1.stop_episode_and_train(d.copy(), 0, True)
                    continue


       if i % 100 == 0:
           print("episode:", i, " / rnd:", d.random_count, " / miss:", miss, " / goal:", goal, " / starvation:", starvation, " / statistics:", agent_p1.get_statistics(), " / epsilon:", agent_p1.explorer.epsilon)
           #カウンタの初期化
           miss = 0
           goal = 0
           starvation = 0
           d.random_count = 0


agent_p1.save("result_" + str(i))
print("Training finished.")

このコードを実行したところ、100回程度ループが回り学習が進むのですが、その後、以下のエラーが発生してしまいます。

Traceback (most recent call last):
  File "train_dungeon.py", line 51, in <module>
    action = agent_p1.act_and_train(d.copy(), reward)
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainerrl/agents/dqn.py", line 424, in act_and_train
    self.replay_updater.update_if_necessary(self.t)
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainerrl/replay_buffer.py", line 545, in update_if_necessary
    self.update_func(transitions)
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainerrl/agents/dqn.py", line 237, in update
    loss = self._compute_loss(exp_batch, errors_out=errors_out)
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainerrl/agents/dqn.py", line 353, in _compute_loss
    y, t = self._compute_y_and_t(exp_batch)
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainerrl/agents/dqn.py", line 334, in _compute_y_and_t
    batch_actions), (batch_size, 1))
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainerrl/action_value.py", line 79, in evaluate_actions
    return F.select_item(self.q_values, actions)
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainer/functions/array/select_item.py", line 120, in select_item
    return SelectItem().apply((x, t))[0]
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainer/function_node.py", line 297, in apply
    self._check_data_type_forward(in_data)
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainer/function_node.py", line 400, in _check_data_type_forward
    self.check_type_forward(in_type)
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainer/functions/array/select_item.py", line 23, in check_type_forward
    x_type.shape[0] == t_type.shape[0],
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainer/utils/type_check.py", line 550, in expect
    expr.expect()
  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainer/utils/type_check.py", line 483, in expect
    '{0} {1} {2}'.format(left, self.inv, right))
chainer.utils.type_check.InvalidType:
Invalid operation is performed in: SelectItem (Forward)

Expect: t.dtype.kind == i
Actual: U != i
コード

型系のエラーと判断しデバッグを進めたところ、

  File "/home/vagrant/source_code/ML/dqn_test/lib/python3.7/site-packages/chainer/functions/array/select_item.py", line 120, in select_item
    return SelectItem().apply((x, t))[0]

に注目して、原因はSelectItem()をコールする際の、tとxの型の不一致が原因とわかりました。またagent_p1.act_and_train(d.copy(), reward)を実行する場合には本来はx,tは
(1, 4) (1,)
となるはずなのに、
(32, 4) (32,)となっていることもわかりました。

さらにデバッグを進めたところ、本来while文内で1ループにつきact_and_trainは一度しか呼ばれないはずなのに、2度呼ばれていることがわかりました。また呼ばれているのはprint("action choice")の前の段階であることもわかりました。つまり二度続けて関数が呼ばれてしまっていることがわかりました。

ここまでデバッグして先に進めなくなってしまいました。非常に困っています....原因の御心あたりのあるかたや強化学習の実装経験があるかたがございましたらぜひアドバイスをいただきたく思います。なお、Python 3.7.0を使用しています。

参考

cached-property==1.5.1
chainer==6.2.0
chainerrl==0.7.0
cloudpickle==1.2.1
filelock==3.0.12
future==0.17.1
gym==0.14.0
numpy==1.17.0
Pillow==6.1.0
protobuf==3.7.1
pyglet==1.3.2
scipy==1.3.0
six==1.12.0
tqdm==4.32.2
typing==3.6.6
typing-extensions==3.6.6