カスタムGym環境作成(18) - 広げたマップを強化学習で攻略(報酬を改善1)

6月 9, 2021

前回、広げたマップに対していろいろな学習率で攻略を試みましたが、ゴールにたどり着くことができませんでした。

[広くしたマップイメージ]

今回はカスタム環境の報酬を改善して、攻略を目指します。

報酬を改善

前回の結果を見ていて一番問題だと思ったのが、スタート地点付近からエージェントが全く動かないことでした。

これを改善するために、エージェントが動かない場合のマイナス報酬を増やして(-1 ⇒ -10)みたいと思います。

修正箇所は108～109行目となります。

[ソース]

env7.py

# activate openai_gym
import sys

import gym
import numpy as np
import gym.spaces

class MyEnv(gym.Env):
    FIELD_TYPES = [
        'Ｓ',  # 0: スタート
        'Ｇ',  # 1: ゴール
        '　',  # 2: 平地
        '山',  # 3: 山(歩けない)
        '☆',  # 4: プレイヤー
        '三',  # 5: 橋
        '川',  # 6: 川
        '林',  # 7: 林
    ]
    MAP = np.array([
        [0, 3, 3, 3, 2, 3, 7, 7, 3, 2, 2, 3],
        [2, 2, 2, 2, 2, 2, 7, 3, 2, 2, 2, 2],
        [3, 2, 3, 2, 3, 7, 7, 2, 2, 2, 2, 3],
        [3, 2, 2, 2, 7, 7, 6, 6, 3, 3, 2, 2],
        [7, 7, 3, 2, 6, 6, 6, 3, 7, 7, 3, 3],
        [3, 2, 3, 2, 3, 7, 3, 1, 2, 7, 2, 2],
        [3, 7, 2, 2, 3, 2, 3, 3, 3, 3, 2, 7],
        [2, 2, 3, 2, 2, 3, 2, 3, 2, 3, 3, 2],
        [7, 3, 2, 7, 7, 2, 2, 2, 7, 6, 6, 7],
        [2, 3, 2, 6, 6, 6, 2, 3, 2, 7, 2, 2],
        [3, 3, 2, 3, 2, 3, 6, 2, 3, 2, 2, 3],
        [7, 2, 2, 2, 3, 3, 6, 2, 7, 3, 2, 2],
    ])
    MAX_STEPS = 2000

    def __init__(self):
        super().__init__()
        # action_space, observation_space, reward_range を設定する
        self.action_space = gym.spaces.Discrete(4)  # 上下左右
        self.observation_space = gym.spaces.Box(
            low=0,
            high=len(self.FIELD_TYPES),
            shape=self.MAP.shape
        )
        self.reset()

    def reset(self):
        # 諸々の変数を初期化する
        self.pos = self._find_pos('Ｓ')[0]
        self.goal = self._find_pos('Ｇ')[0]
        self.river = self._find_pos('川')
        self.trees = self._find_pos('林')
        self.done = False
        self.steps = 0
        return self._observe()

    def step(self, action):
        # 1ステップ進める処理を記述。戻り値は observation, reward, done(ゲーム終了したか), info(追加の情報の辞書)
        # 左上の座標を(0, 0)とする
        if action == 0:     # 右移動
            next_pos = self.pos + [0, 1]
        elif action == 1:   # 左移動
            next_pos = self.pos + [0, -1]
        elif action == 2:   # 下移動
            next_pos = self.pos + [1, 0]
        elif action == 3:   # 上移動
            next_pos = self.pos + [-1, 0]

        if self._is_movable(next_pos):
            self.pos = next_pos
            moved = True
        else:
            moved = False

        self.steps += 1
        observation = self._observe()
        reward = self._get_reward(self.pos, moved)
        self.done = self._is_done()
        return observation, reward, self.done, {}

    def render(self, mode='console', close=False):
        for row in self._observe():
            for elem in row:
                print(self.FIELD_TYPES[elem], end='')
            print()

    def _close(self):
        pass

    def _seed(self, seed=None):
        pass

    def _get_reward(self, pos, moved):
        # 報酬を返す。
        # - ゴールにたどり着くと 3000 ポイント
        # - 川に入ったら -10 ポイント
        # - 林に入ったら -3 ポイント
        # - 1ステップごとに-1ポイント(できるだけ短いステップでゴールにたどり着きたい)
        if moved:
            if (self.goal == pos).all():
                return 3000
            for x in self.river:
                if (x == pos).all():
                    return -10
            for x in self.trees:
                if (x == pos).all():
                    return -3
            return -1
        else:   # エージェントが動かなかった場合
            return -10

    def _is_movable(self, pos):
        # マップの中にいるか、歩けない場所にいないか
        return (
            0 <= pos[0] < self.MAP.shape[0]
            and 0 <= pos[1] < self.MAP.shape[1]
            and self.FIELD_TYPES[self.MAP[tuple(pos)]] != '山'
        )

    def _observe(self):
        # マップにプレイヤーの位置を重ねて返す
        observation = self.MAP.copy()
        observation[tuple(self.pos)] = self.FIELD_TYPES.index('☆')
        return observation

    def _is_done(self):
        # 最大で self.MAX_STEPS まで
        if (self.pos == self.goal).all():
            return True
        elif self.steps > self.MAX_STEPS:
            return True
        else:
            return False

    def _find_pos(self, field_type):
        return np.array(list(zip(*np.where(self.MAP == self.FIELD_TYPES.index(field_type)))))