強化学習8 (Actor Critic法)

Actor Critic法による学習を試してみます。

Actor Critic法は、戦略担当(Actor)と価値評価担当(Critic)を相互に更新して学習する手法です。

まずはエージェントのベースになるクラスを実装します。(強化学習5・6 (モンテカルロ法・TD法)と同様です。)

el_agent.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
import numpy as np
import matplotlib.pyplot as plt

class ELAgent():

    def __init__(self, epsilon):
        self.Q = {}
        self.epsilon = epsilon
        self.reward_log = []

    def policy(self, s, actions):
        if np.random.random() < self.epsilon:
            # ランダムな行動（探索）
            return np.random.randint(len(actions))
        else:
            # self.Q => 状態における行動の価値
            # self.Q[s] => 状態sで行動aをとる場合の価値
            if s in self.Q and sum(self.Q[s]) != 0:
                # 価値評価に基づき行動（活用）
                return np.argmax(self.Q[s])
            else:
                # ランダムな行動（探索）
                return np.random.randint(len(actions))

    # 報酬の記録を初期化
    def init_log(self):
        self.reward_log = []

    # 報酬の記録
    def log(self, reward):
        self.reward_log.append(reward)

    def show_reward_log(self, interval=50, episode=-1):
        # そのepsilonの報酬をグラフ表示
        if episode > 0:
            rewards = self.reward_log[-interval:]
            mean = np.round(np.mean(rewards), 3)
            std = np.round(np.std(rewards), 3)
            print("At Episode {} average reward is {} (+/-{}).".format(
                   episode, mean, std))
        # 今までに獲得した報酬をグラフ表示
        else:
            indices = list(range(0, len(self.reward_log), interval))
            means = []
            stds = []
            for i in indices:
                rewards = self.reward_log[i:(i + interval)]
                means.append(np.mean(rewards))
                stds.append(np.std(rewards))
            means = np.array(means)
            stds = np.array(stds)
            plt.figure()
            plt.title("Reward History")
            plt.grid()
            plt.fill_between(indices, means - stds, means + stds,
                             alpha=0.1, color="g")
            plt.plot(indices, means, "o-", color="g",
                     label="Rewards for each {} episode".format(interval))
            plt.legend(loc="best")
            plt.show()

次に環境を扱うためのクラスを実装します。
(強化学習5・6 (モンテカルロ法・TD法)と同様です。)

frozen_lake_util.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
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.cm as cm
import gym
from gym.envs.registration import register
# スリップする設定(is_slippery)をオフ設定（学習に時間がかかるため）
register(id="FrozenLakeEasy-v0", entry_point="gym.envs.toy_text:FrozenLakeEnv", kwargs={"is_slippery": False})

def show_q_value(Q):
    """
    FrozenLake-v0環境での価値は下記の通り。
    各行動での評価を表しています。
    +----+------+----+
    |    |  上  |    |
    | 左 | 平均 | 右 |
    |    |  下  |    |
   +-----+------+----+
    """
    env = gym.make("FrozenLake-v0")
    nrow = env.unwrapped.nrow
    ncol = env.unwrapped.ncol
    state_size = 3
    q_nrow = nrow * state_size
    q_ncol = ncol * state_size
    reward_map = np.zeros((q_nrow, q_ncol))

    for r in range(nrow):
        for c in range(ncol):
            s = r * nrow + c
            state_exist = False
            if isinstance(Q, dict) and s in Q:
                state_exist = True
            elif isinstance(Q, (np.ndarray, np.generic)) and s < Q.shape[0]:
                state_exist = True

            if state_exist:
                # At the display map, the vertical index is reversed.
                _r = 1 + (nrow - 1 - r) * state_size
                _c = 1 + c * state_size
                reward_map[_r][_c - 1] = Q[s][0]  # 左 = 0
                reward_map[_r - 1][_c] = Q[s][1]  # 下 = 1
                reward_map[_r][_c + 1] = Q[s][2]  # 右 = 2
                reward_map[_r + 1][_c] = Q[s][3]  # 上 = 3
                reward_map[_r][_c] = np.mean(Q[s])  # 中央

    # 各状態・行動の評価を表示
    fig = plt.figure()
    ax = fig.add_subplot(1, 1, 1)
    plt.imshow(reward_map, cmap=cm.RdYlGn, interpolation="bilinear",
               vmax=abs(reward_map).max(), vmin=-abs(reward_map).max())
    # 報酬推移を表示
    ax.set_xlim(-0.5, q_ncol - 0.5)
    ax.set_ylim(-0.5, q_nrow - 0.5)
    ax.set_xticks(np.arange(-0.5, q_ncol, state_size))
    ax.set_yticks(np.arange(-0.5, q_nrow, state_size))
    ax.set_xticklabels(range(ncol + 1))
    ax.set_yticklabels(range(nrow + 1))
    ax.grid(which="both")
    plt.show()

Actor Critic法での学習を実行します。

53行目で行動評価(Q値)の更新を行い、54行目で状態価値の更新を行っています。
ValueベースとPolicyベース両方の特性を持っていることになります。

actor_critic.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
import numpy as np
import gym
from el_agent import ELAgent
from frozen_lake_util import show_q_value

class Actor(ELAgent):

    def __init__(self, env):
        super().__init__(epsilon=-1)
        nrow = env.observation_space.n
        ncol = env.action_space.n
        self.actions = list(range(env.action_space.n))
        self.Q = np.random.uniform(0, 1, nrow * ncol).reshape((nrow, ncol))

    def softmax(self, x):
        return np.exp(x) / np.sum(np.exp(x), axis=0)

    def policy(self, s):
        a = np.random.choice(self.actions, 1,
                             p=self.softmax(self.Q[s]))
        return a[0]

class Critic():

    def __init__(self, env):
        states = env.observation_space.n
        self.V = np.zeros(states)

class ActorCritic():

    def __init__(self, actor_class, critic_class):
        self.actor_class = actor_class
        self.critic_class = critic_class

    def train(self, env, episode_count=1000, gamma=0.9,
              learning_rate=0.1, render=False, report_interval=50):
        actor = self.actor_class(env)
        critic = self.critic_class(env)

        actor.init_log()
        for e in range(episode_count):
            s = env.reset()
            done = False
            while not done:
                if render:
                    env.render()
                a = actor.policy(s)
                n_state, reward, done, info = env.step(a)

                gain = reward + gamma * critic.V[n_state]
                estimated = critic.V[s]
                td = gain - estimated
                actor.Q[s][a] += learning_rate * td   # 行動評価(Q値)の更新
                critic.V[s] += learning_rate * td     # 状態価値の更新
                s = n_state

            else:
                actor.log(reward)

            if e != 0 and e % report_interval == 0:
                actor.show_reward_log(episode=e)

        return actor, critic

def train():
    trainer = ActorCritic(Actor, Critic)
    env = gym.make("FrozenLakeEasy-v0")
    actor, critic = trainer.train(env, episode_count=3000)
    show_q_value(actor.Q)
    actor.show_reward_log()

if __name__ == "__main__":
    train()
FrozenLake 各行動の評価

エピソード実行回数と獲得報酬平均の推移は次のようになります。
エピソード数と獲得報酬平均の推移

今まで試してきた手法より学習にかかるエピソード数は長くなっていますが、安定した報酬が得られるようになっています。

参考

Pythonで学ぶ強化学習 -入門から実践まで- サンプルコード