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

June 11, 2021

前回、広げたマップに対してエージェントがスタート位置に移動したときのマイナス報酬を増やしてみましたが、ゴールにたどり着くことができませんでした。

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

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

最終報酬がプラスにならない問題の対策

これまでの結果を見るとプラス報酬になることがほとんどありませんでした。

そこで今回はゴール時の報酬を見直してみました。

具体的にはゴール時の報酬を3000から15000に変更してみました。

理由は、最終報酬が-20000となるケースが一番悪い結果のようなので、それ以前にゴールにたどり着けば最終報酬がプラスになるのではないかと考えたからです。

修正箇所は103～104行目となります。

[ソース]

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
                return 15000
            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)))))