カスタムGym環境作成(1) - ランダム実行

強化学習とは

強化学習とは、エージェントが環境の状態に応じてどのように行動すれば報酬を多くもらえるかを求める手法です。

学習データなしに自身の試行錯誤のみで学習するのが特徴になります。

効率のいい学習アルゴリズムは日々進化・公開されており、それを理解するよりもうまく使いこなす方が重要ではないかと感じております。
（アルゴリズムを理解するために必要な呪文のような数式に挫折したという理由もあります😥）

学習アルゴリズムを使いこなすために、独自の学習環境・・・つまりカスタム環境をいろいろと作成できるようなったほうがよいのではないかと思い今回の記事を書くことにしました。

（環境構築等は省略します）

カスタムGym環境作成

今回は２次元マップを独自で作成し、スタート地点からゴール地点までたどり着く環境を作成してみます。

できるだけ単純なマップとするために、移動可能な平地と移動できない山だけを並べてみました。

[ソース]

env3.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
# 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: プレイヤー
    ]
    MAP = np.array([
        [0, 2, 3, 2],  # ■　山　
        [2, 3, 1, 2],  # 　山Ｇ　
        [2, 2, 3, 2],  # 　　山　
        [3, 2, 2, 2],  # 山　　　
    ])
    MAX_STEPS = 50

    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.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

        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):
        # 報酬を返す。
        # - ゴールにたどり着くと 100 ポイント
        # - 1ステップごとに-1ポイント(できるだけ短いステップでゴールにたどり着かせるため必要)
        if moved and (self.goal == pos).all():
            return 100
        else:
            return -1

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

ランダム実行

上記で作成したカスタム環境を読み込み、ランダムで実行してみます。

[ソース]

test3.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import gym
from env3 import MyEnv

# カスタム環境の生成
env = MyEnv()

# 環境リセット
state = env.reset()

total_reward = 0
while True:
    # ランダム行動の取得
    action = env.action_space.sample()
    # １ステップの実行
    state, reward, done, info = env.step(action)
    total_reward += reward
    # 環境の描画
    env.render()
    print('reward:', reward, 'total_reward', total_reward)
    print('-----------')
    # エピソード完了
    if done:
        print('done')
        break

結果は次のように表示されます。

星マーク(☆)が移動するプレイヤーを現しています。

[結果（１部略）]

☆　山　
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -1
-----------
☆　山　
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -2
-----------
Ｓ☆山　
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -3
-----------
Ｓ☆山　
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -4
-----------
Ｓ☆山　
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -5
-----------
☆　山　
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -6
-----------
Ｓ　山　
☆山Ｇ　
　　山　
山　　　
reward: -1 total_reward -7
-----------

（略）

-----------
Ｓ　山☆
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -146
-----------
Ｓ　山　
　山Ｇ☆
　　山　
山　　　
reward: -1 total_reward -147
-----------
Ｓ　山☆
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -148
-----------
Ｓ　山☆
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -149
-----------
Ｓ　山☆
　山Ｇ　
　　山　
山　　　
reward: -1 total_reward -150
-----------
Ｓ　山　
　山Ｇ☆
　　山　
山　　　
reward: -1 total_reward -151
-----------
Ｓ　山　
　山☆　
　　山　
山　　　
reward: 100 total_reward -51
-----------
done

ランダムでプレイヤーを移動させるので、結果は毎回変わりますがトータル報酬(total_reward)は-450～70の範囲でした。

次回はStable BaselinesのPPO2アルゴリズムを使って、強化学習を行ってみます。