強化学習2 (マルコフ決定過程)

前回作成した環境(環境構築に関する記事)を使ってマルコフ決定過程に従う環境を作成します。

  • 遷移先の状態は直前の状態とそこでの行動だけに依存する。
  • 報酬は直前の状態と遷移先に依存する。

構成要素は次の４つとなります。

  1. 状態（State）
  2. 行動（Action）
  3. 状態遷移の確率（遷移関数／Transition function）
  4. 即時報酬（報酬関数／Reward function）

コードにコメントをいっぱい書いてみましたので参考にしてください

environment.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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
from enum import Enum
import numpy as np

# エージェントのいる位置を表すクラス
class State():

    def __init__(self, row=-1, column=-1):      # コンストラクタ
        self.row = row
        self.column = column

    def __repr__(self):                    # printした時に表示される
        return "<State: [{}, {}]>".format(self.row, self.column)

    def clone(self):
        return State(self.row, self.column)

    def __hash__(self):
        return hash((self.row, self.column))

    def __eq__(self, other):
        return self.row == other.row and self.column == other.column

# 上下左右の行動を表すクラス（対になる行動は -1を掛けた値となる）
class Action(Enum):
    UP = 1
    DOWN = -1
    LEFT = 2
    RIGHT = -2

# 環境クラス（遷移関数と報酬関数あり）・・・一番長くそして核となるクラス
class Environment():

    def __init__(self, grid, move_prob=0.8):
        # ２次元のグリッド。
        # [種類]
        #   0: 通常（エージェント移動可能）
        #  -1: ダメージ (ゲーム終了)
        #   1: 報酬 (ゲーム終了)
        #   9: 壁 (エージェント移動不可)
        self.grid = grid
        self.agent_state = State()  # エージェントの位置

        # デフォルトの報酬はマイナスとする。 （毒沼のような感じ）
        # エージェントが早くゴール向かうようにするため。
        self.default_reward = -0.04

        # エージェントは遷移確率(%)で指定された方向に進む。
        # 遷移確率から1を引いた確率(%)で違う方向に進む。
        self.move_prob = move_prob
        self.reset()

    @property
    def row_length(self):
        return len(self.grid)

    @property
    def column_length(self):
        return len(self.grid[0])

    @property
    def actions(self):
        return [Action.UP, Action.DOWN,
                Action.LEFT, Action.RIGHT]

    @property
    def states(self):
        states = []
        for row in range(self.row_length):
            for column in range(self.column_length):
                # 壁(9)はステータスに含まない。
                if self.grid[row][column] != 9:
                    states.append(State(row, column))
        return states

    # 遷移関数
    def transit_func(self, state, action):
        transition_probs = {} # 遷移確率を初期化
        if not self.can_action_at(state):
            # すでに終端にいる
            return transition_probs
        # 反対方向
        opposite_direction = Action(action.value * -1)

        for a in self.actions:
            prob = 0
            if a == action:
                prob = self.move_prob             # 遷移確率(0.8=80%)
            elif a != opposite_direction:
                prob = (1 - self.move_prob) / 2   # 遷移確率(0.1=10%)

            # 移動位置ごとに遷移確率を設定する
            next_state = self._move(state, a)
            if next_state not in transition_probs:
                transition_probs[next_state] = prob
            else:
                transition_probs[next_state] += prob

        return transition_probs

    # 移動可能な場所かどうか
    def can_action_at(self, state):
        if self.grid[state.row][state.column] == 0:
            return True
        else:
            return False

    # 移動する
    def _move(self, state, action):
        if not self.can_action_at(state):
            raise Exception("Can't move from here!")
        # 移動先用にグリッドをもう１つ用意
        next_state = state.clone()
        # アクションに応じて移動する
        if action == Action.UP:
            next_state.row -= 1
        elif action == Action.DOWN:
            next_state.row += 1
        elif action == Action.LEFT:
            next_state.column -= 1
        elif action == Action.RIGHT:
            next_state.column += 1

        # グリッドの外かどうかをチェックする。（外だったらもとの位置に戻す）
        if not (0 <= next_state.row < self.row_length):
            next_state = state
        # グリッドの外かどうかをチェックする。（外だったらもとの位置に戻す）
        if not (0 <= next_state.column < self.column_length):
            next_state = state
        # 壁にぶつかったかどうかをチェックする。（ぶつかったらもとの位置に戻す）
        if self.grid[next_state.row][next_state.column] == 9:
            next_state = state
        # 移動先のグリッド位置を返す
        return next_state

    # 報酬関数（ゲーム終了判定と報酬を返す）
    def reward_func(self, state):
        reward = self.default_reward
        done = False

        # 次の状態の属性をチェックする
        attribute = self.grid[state.row][state.column]
        if attribute == 1:
            # 報酬を得てゲーム終了
            reward = 1
            done = True
        elif attribute == -1:
            # ダメージを受けてゲーム終了.
            reward = -1
            done = True

        return reward, done

    def reset(self):        # エージェントを左上に戻す
        self.agent_state = State(self.row_length - 1, 0)
        return self.agent_state

    # エージェントから行動を受け取り、遷移関数／報酬関数を用いて、
    # 次の遷移先と即時報酬を計算する
    def step(self, action):
        next_state, reward, done = self.transit(self.agent_state, action)
        if next_state is not None:
            self.agent_state = next_state

        return next_state, reward, done

    def transit(self, state, action):
        # 遷移関数で遷移確率を取得する。
        transition_probs = self.transit_func(state, action)
        if len(transition_probs) == 0:
            return None, None, True

        next_states = []
        probs = []
        for s in transition_probs:
            next_states.append(s)
            probs.append(transition_probs[s])

        # 遷移関数の出力した確率に沿って遷移先を得る（ここで実際の行動が決まる！）
        next_state = np.random.choice(next_states, p=probs)
        # 報酬関数で報酬とゲーム終了判定を取得する
        reward, done = self.reward_func(next_state)
        return next_state, reward, done

上記で実装した環境を実行するためのコードは下記の通りです。

environment_demo.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
import random
from environment import Environment

# エージェントを表すクラス（戦略を定義する）
class Agent():

    def __init__(self, env):
        self.actions = env.actions

    # 戦略（最初なのでただのランダム選択）
    def policy(self, state):
        return random.choice(self.actions)

# 実行関数
def main():
    # グリッド環境を作成
    grid = [
        [0, 0, 0, 1],
        [0, 9, 0, -1],
        [0, 0, 0, 0]
    ]
    env = Environment(grid)   # グリッドから環境を作成
    agent = Agent(env)        # 環境をエージェントに設定

    # ゲームを１０回実行する
    for i in range(10):
        state = env.reset()   # エージェントの位置リセット
        total_reward = 0      # トータル報酬初期化
        done = False          # ゲーム終了状態初期化

        while not done:       # ゲーム終了まで続ける
            # 戦略に沿ったアクション（移動方向）を取得。（現状はランダム選択）
            action = agent.policy(state)
            # アクションによる移動位置、報酬、ゲーム終了状態を取得。
            next_state, reward, done = env.step(action)
            # 今回移動分の報酬をトータル報酬に加算・減算する
            total_reward += reward
            # エージェントの位置を更新する
            state = next_state

        print("エピソード {}: エージェント取得報酬 {}.".format(i, total_reward))

if __name__ == "__main__":
    main()

environment_demo.pyを実行すると下記のような結果となりました。
(乱数を使ってるので実行するたびに結果は異なります。)
結果

最初よくわからなったのですがエージェントの戦略で決めた行動が必ず実行されるわけではなく、その行動をもとに遷移確率を算出し、その確率に応じて実際の動作(Action)が決まるということです。
（Policyで決めた行動と違う方向に進んでいて「なぜなんだ？！」と悩んでました。。。）

今回の例では、まず戦略で決めた行動がUPだとしたらそれが実行される確率が80%で、残り20%の半分10%がRIGHTかLEFTが実行される確率に割り当てられるということになります。（反対行動のDOWNは0%）