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

6月 10, 2021

前回、広げたマップに対してエージェントが移動しない場合のマイナス報酬を修正して攻略を試みましたが、ゴールにたどり着くことができませんでした。

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

今回は再度報酬を見直して、攻略を目指します。

ゴール近くから動かないことの対策

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

そのため前回、移動しない場合のマイナス報酬をよりマイナスにしたのですが、それでもスタート位置付近にいるケースが多々ありました。

という訳で今回はスタート地点に移動したときの報酬を-10としてみました。

修正箇所は52行目、99～100行目となります。

[ソース]

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.start = self._find_pos('Ｓ')[0]
        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 (self.start == pos).all():   # スタート位置に戻ってきたときのマイナス報酬
            return -10
        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)))))