モンテカルロ法による倒立振子 (Ｑ値を見える化)

gymの倒立振子を使って強化学習 モンテカルロ法

モンテカルロ法とQ-learning、SARSA法の違い

モンテカルロ法は、アクションごとにQ値を更新するQ-learningやSARSA法と違い、CartPoleが倒れるまで行動しその行動履歴から一気にＱ値を更新します。

モンテカルロ法は一度未来を見てから一気にQ値の更新をかけるため、遠い未来の報酬も比較的早く学習できるというメリットがあります。

f:id:Start_python:20191129163705g:plain
20試行回数くらいですでに安定しています。1試行が終わったときに一気にQ値が更新されます。

プログラムのコード

# coding:utf-8
# [0]ライブラリのインポート
import gym  # 倒立振子(cartpole)の実行環境
from gym import wrappers  #gymの画像保存
import numpy as np
import time
from collections import deque
import os


# [1]Q関数を離散化して定義する関数　------------
# 観測した状態を離散値にデジタル変換する
def bins(clip_min, clip_max, num):
    return np.linspace(clip_min, clip_max, num + 1)[1:-1]


# 各値を離散値に変換
def digitize_state(observation):
    cart_pos, cart_v, pole_angle, pole_v = observation
    digitized = [
        np.digitize(pole_angle, bins=bins(-0.5, 0.5, num_split1)),
        np.digitize(pole_v, bins=bins(-2.0, 2.0, num_split2))
    ]
    return sum([x * (num_split1**i) for i, x in enumerate(digitized)])


# [2]行動a(t)を求める関数 -------------------------------------
def get_action(next_state, episode):    # 徐々に最適行動のみをとる、ε-greedy法
    epsilon = 0.5 * (1 / (episode + 1))
    if epsilon <= np.random.uniform(0, 1):
        next_action = np.argmax(q_table[next_state])
    else:
        next_action = np.random.choice([0, 1])
    return next_action


# [3]1試行の各ステップの行動を保存しておくメモリクラス
class Memory:
    def __init__(self, max_size=200):
        self.buffer = deque(maxlen=max_size)

    def add(self, experience):
        self.buffer.append(experience)

    def sample(self):
        return self.buffer.pop()  # 最後尾のメモリを取り出す

    def len(self):
        return len(self.buffer)


# [4]Qテーブルを更新する(モンテカルロ法) ＊Qlearningと異なる＊ -------------------------------------
def update_Qtable_montecarlo(q_table, memory):
    gamma = 0.99
    alpha = 0.5
    total_reward_t = 0

    while (memory.len() > 0):
        (state, action, reward) = memory.sample()
        total_reward_t = gamma * total_reward_t       # 時間割引率をかける
        # Q関数を更新
        q_table[state, action] = q_table[state, action] + alpha*(reward+total_reward_t-q_table[state, action])
        total_reward_t = total_reward_t + reward    # ステップtより先でもらえた報酬の合計を更新

    return q_table


# [5]. メイン関数開始 パラメータ設定--------------------------------------------------------
env = gym.make('CartPole-v0')
max_number_of_steps = 200  #1試行のstep数
num_consecutive_iterations = 100  #学習完了評価に使用する平均試行回数
num_episodes = 100  #総試行回数
goal_average_reward = 195  #この報酬を超えると学習終了（中心への制御なし）
num_render = 80  #表示開始の試行回数
# 状態を分割してQ関数（表）を作成
num_split1 = 8  #分割数1
num_split2 = 6  #分割数2
memory_size = max_number_of_steps            # バッファーメモリの大きさ
memory = Memory(max_size=memory_size)
q_table = np.random.uniform(low=-1, high=1, size=(num_split1*num_split2, env.action_space.n))
total_reward_vec = np.zeros(num_consecutive_iterations)  #各試行の報酬を格納
np.set_printoptions(precision=1, suppress=True)  #print用フォーマット


# [5] メインルーチン--------------------------------------------------
for episode in range(num_episodes):  #試行数分繰り返す
    # 環境の初期化
    observation = env.reset()
    state = digitize_state(observation)
    action = np.argmax(q_table[state])
    episode_reward = 0

    for t in range(max_number_of_steps):  #1試行のループ
        if episode > num_render:
            #cartPoleを描画する
            env.render()
            time.sleep(0.001)

        # 行動a_tの実行により、s_{t+1}, r_{t}などを計算する
        observation, reward, done, info = env.step(action)

        # 報酬を設定し与える
        if done:
            if t < 195:
                reward = -200  #こけたら罰則
            else:
                reward = 1  #立ったまま終了時は罰則はなし

        else:
            reward = 1  #各ステップで立ってたら報酬追加


        # メモリに、現在の状態と行った行動、得た報酬を記録する
        memory.add((state, action, reward))

        # 次ステップへ行動と状態を更新
        next_state = digitize_state(observation)  # t+1での観測状態を、離散値に変換
        next_action = get_action(next_state, episode)  # 次の行動a_{t+1}を求める
        action = next_action  # a_{t+1}
        state = next_state  # s_{t+1}

        episode_reward += reward  #報酬を追加

        if episode > num_render:
            # ここにq_tableを表示
            q_table_print = np.reshape(q_table, (1,num_split1*num_split2*2),'F')
            os.system('cls')
            print(np.reshape(q_table_print, (2,num_split2,num_split1)))

        # 終了時の処理
        if done:
            # これまでの行動の記憶と、最終的な結果からQテーブルを更新していく
            q_table = update_Qtable_montecarlo(q_table, memory)

            print('%d Episode finished after %f time steps / mean %f' %
                  (episode, t + 1, total_reward_vec.mean()))
            total_reward_vec = np.hstack((total_reward_vec[1:],
                                          episode_reward))  #報酬を記録
            if episode > num_render:
                time.sleep(1)
            break