【強化学習】まとめてみた 第六回（１）（いよいよ登場！Q学習（基本））【TD学習】

はじめに

強化学習勉強会なるものがスタートしました！
なので，それらを勉強もかねてまとめていこうと思います．二番煎じ感がものすごいですが，自分の理解度向上のためにも！
予定ではQiitaで第7回分（Q学習ぐらいまで）ぐらいやろうかなと考えています．今回は第六回（１）です！

過去のもの

• 第一回：概要
• 第二回：n本腕バンディット
• 第三回：用語まとめ編
• 第四回：動的計画法
• 第五回（１）：モンテカルロ法（ES法）
• 第五回（２）：モンテカルロ法（方策ON型，方策OFF型）
• 第六回（１）：TD学習の基本（ランダムウォーク）
• 第六回（２）：TD学習（Q学習・Sarsa学習）

その他リンク

• github
  https://github.com/Shunichi09/
• twitter
  https://twitter.com/ShunichiSekigu1?lang=ja
• 参考書
  http://www.morikita.co.jp/books/book/1990
• 参考書（原本）
  http://incompleteideas.net/book/bookdraft2018jan1.pdf
• 原著プログラム
  https://github.com/ShangtongZhang

原著プログラムについてですが，基本的には見ずに書いてます．
原著のプログラムが分かりにくければ，僕のgithubで見ていただけると嬉しいです．

目的と結論

目的

• 強化学習勉強会の内容のまとめ
• 自分の理解度向上もかねて
• プログラムの書き方の練習もかねて

結論

• TD学習法を理解できます
• モンテカルロとの違いを理解できます！

おしながき

1. TD学習とは
2. ランダムウォークの問題

TD学習とは

前回、前々回と、動的計画法、モンテカルロ法を学んできましたがどちらにも問題点がありました

• 動的計画法

環境のモデルが必要

• モンテカルロ法

ランダムに，すべての状態をたくさん経験する必要がある

画像でみるとこんな感じですね

ちなみに動的計画法はすべての状態に対して予測して価値を計算していくので完全バックアップと言われます．
この話は本の最後あたりにでてきます
つまり，環境のモデルがあるのでそこからはじめようとなにしようと関係ないわけです

しかし，モデルがない場合モンテカルロ法のようにエピソード単位で学習を行う必要があります．

エピソード単位かどうか，実際のシナリオを考えるかどうかが大きな違いとも言えます

モチベーション

ここで，TD学習のモチベーションは，

• 環境のモデルを使いたくない
• でもエピソードがすべて終わるまで待ちたくない

という2つのモチベーションです．
なので式としてはこんな感じになります

V(s_t) ← V(s_t) + \alpha[r_{t+1} + \gamma V(s_{t+1}) - V(s_t)]

さて，このままではよくわからないので
モンテカルロ法を価値関数に置き換えた場合と比較してみます
モンテカルロ法は今まで行動価値関数Qでいろいろと考えていましたが，ここでは価値関数でいきしょう（本質的には変わりません）
（その状態の価値か，その状態で行動aをとった価値なのか，どちらを評価するのかの違いです）

上に示したようにモンテカルロ法はエピソード終了後の確定収益で更新します．
一方で，TD法は動的計画法を融合させ，その後の状態価値でも更新します．

この違いを図で確認します

この図をみると一目瞭然かと思います．
ちなみに動的計画法のようにすべての状態をバックアップするわけではないことに注意してください
つまり，モデルはないので，動的計画法（完全バックアップ）のようにすべての状態に対してこれはできません

基本的にはエピソードの中での話であるということを軸においておきます！
（サンプルバックアップです）

なのでどちらかというとモンテカルロ法的な発想です

まとめると

モンテカルロ法のように確定収益ではなく，報酬r+次状態の予測状態価値で更新をしていくということです．

では違いをどう考えればよいでしょうか

教科書の例がすごく分かりやすかったので紹介します
（この例では，バッチ更新と呼ばれる手法を使っています（更新分はすべての学習データの増加分をひとまとめで更新するという手法です））

このような条件（もちろんこのモデルはないです，分かりやすいように描画しているだけです！）

この時以下のような，状態が観測されたとします．

A:0⇒B:0　（状態AからスタートBにいって報酬は0）
B:1（状態Bから報酬は1）
B:1
B:1
B:1
B:1
B:1
B:0

まず，モンテカルロ的に考えてみましょう

状態Aにいて，Bにいって，報酬が0でした
そして，それ以外で状態Aにはいってないので，状態Aの価値は0になります

一方！

TDの場合
Bが状態価値3/4になりますね（8回中6回成功）
でしかも，A⇒Bの状態遷移が100％おきてます
よってAの価値も3/4になるというわけです

つまりTDはマルコフ過程でのもっともよい推定値（1ステップ前にのみ状態が依存する）を求めていて

モンテカルロ法は訓練データでのもっともよい推定値を求めていると言えます

どっちがいいのよという問題ですが
マルコフ過程をおくのであれば，TDのほうがよさそうです
過去の1ステップで未来を表現できそうなこの状態では．．
なぜなら，未来のデータに対して有効だからです

ただ，モンテカルロは現在のデータへの平均を取れます
なので今までの結果をより強く反映させる場合はモンテカルロの方がよいかと．．．

はっきりとはわかりませんが

では教科書の例にあげられているランダムウォークの問題をやってみましょう．

ランダムウォークの問題

これは以下のようなノードがあり，右端にいくと報酬が1
左端にいくと報酬が0です

このとき，ランダム方策での状態価値関数を求めてくださいという問題
ランダム方策とは，右に行くか左にいくかをランダムにとるということです

先ほどの更新式を使えば簡単に実装できます
ちなみに真値は確率になるので
左から

A:1/6
B:2/6
C:3/6
D:4/6
E:5/6

になります

まずTD法で学習させた後の結果をみてみましょう

になります
教科書通りですね

さて，各状態での予測誤差をとったものをモンテカルロ法と比較します
ちなみに，たくさん学習回すと，どちらも最後は同じ結果になります
モンテカルロ法の状態価値推定は上の式にのっとってます
（今回は行動価値を求める問題ではないのでそこはご注意ください）

どちらがいいとかそういう話ではないそうですが．．．
確率的なタスクについては

データの平均を取るモンテカルロ法よりも

次の状態遷移などを考えるTDの方が収束がはやいことが証明されています

githubにもあげてますが

一応のせときます

TDのみ

import numpy as np
import matplotlib.pyplot as plt
import random
from collections import OrderedDict

class RandomWalk():
    """
    environment model "Random model"
    Attributes
    ------------
    states : list of str
        state of environment
    """
    
    def __init__(self):
        self.states =  ["left_goal", "A", "B", "C", "D", "E", "right_goal"] 

    def state_update(self, state, action):
        """
        Parameters
        ------------
        state : char
            agent position in this game (in this case A-E)
        action : char
            agent action in this game (in this case "right" or "left")
        
        Returns
        ----------
        next_state : char
            agent next position
        temp_reward : int
            temporal (meaning not total) reward 
        """
        end_flg = False
        temp_reward = 0.         

        if action == "left":
            next_state_num = self.states.index(state) - 1
            if self.states[next_state_num] == "left_goal":
                end_flg = True
                temp_reward = 0. # end the game
            
            next_state = self.states[next_state_num]

        elif action == "right":
            next_state_num = self.states.index(state) + 1

            if self.states[next_state_num] == "right_goal":
                end_flg = True
                temp_reward = 1. # clear the game
        
            next_state = self.states[next_state_num]

        return end_flg, next_state, temp_reward

class Agent():
    """
    Agent model of reinforcement learning
    Attributes
    ------------
    state : str 
        now state of agent
    states : list of str
        state of environment
    action_list : list of str
        agent possibile action list 
    step rate : float
        learning rate of reinforcement learning
    discount rate : float
        discount rate of reward
    history_state : list 
        one episode state (this should not have the same state, first visit ES methods)
    values : OrderedDict
        value function
    """

    def __init__(self):
        # environment
        self.model = RandomWalk()
        
        self.state = None
        self.states = self.model.states # in this case we shoud know the all states
        self.action_list = ["left", "right"]

        self.step_rate = 0.1 # adjust subject
        self.discount_rate = 1.0
        
        # for bootstrap
        self.history_state = None
        self.values = OrderedDict()
        for state in self.states:
            self.values[state] = 0.5

            if state == "left_goal": # in this problem the final state does not have value
                self.values[state] = 0.
            if state == "right_goal":
                self.values[state] = 0.

    def train_TD(self, train_num):
        """training the agent TD methods
        Parameters
        -----------
        train_num : int
            training number of reinforcement learning
        """
        for _ in range(train_num):
            self.state = "C" # initial state has been decided

            while True: # Ending this while is worth of ending one episode

                end_flg, next_state, temp_reward = self._play(self.state) # play the game
                # for TD
                self._valuefunc_update(temp_reward, self.state, next_state)

                # update
                self.state = next_state

                if end_flg:
                    break

    def _play(self, state):
        """
        playing the game

        Parameters
        -----------
        state : str
            state of the agent
        """
        action = self._decide_action()
        end_flg, next_state, temp_reward = self.model.state_update(state, action)

        return end_flg, next_state, temp_reward

    def _decide_action(self):
        """
        Returns
        ---------
        action : str
            action of agent
        """

        action = random.choice(self.action_list)

        return action

    def _valuefunc_update(self, reward, state, next_state):
        """
        Parameters
        -----------
        reward : int
            reward the agent got
        state : str 
            now state of agent
        next_state : str 
            next state of agent
        """
        # TD
        self.values[state] = self.values[state] + self.step_rate * (reward + self.discount_rate * self.values[next_state] - self.values[state])

def main():
    # for fig 6.6 
    # 0
    agent = Agent()
    agent.train_TD(0)
    values = list(agent.values.values())
    plt.plot(range(len(agent.values.keys()) - 2), values[1:-1] , label="0", marker=".")
    # 1
    agent = Agent()
    agent.train_TD(1)
    values = list(agent.values.values())
    plt.plot(range(len(agent.values.keys()) - 2), values[1:-1] , label="1", marker=".")
    # 10
    agent = Agent()
    agent.train_TD(10)
    values = list(agent.values.values())
    plt.plot(range(len(agent.values.keys()) - 2), values[1:-1] , label="10", marker=".")
    # 100
    agent = Agent()
    agent.train_TD(100)
    values = list(agent.values.values())
    plt.plot(range(len(agent.values.keys()) - 2), values[1:-1] , label="100", marker=".")
    # true
    values = np.arange(1/6, 0.9, 1/6)
    print(values)
    plt.plot(range(len(agent.values.keys()) - 2), values, label="True", marker=".")

    plt.legend()
    plt.show()
    

if __name__ == '__main__':
    main()

比較

import numpy as np
import matplotlib.pyplot as plt
import math
import random
from collections import OrderedDict

class RandomWalk():
    """
    environment model "Random model"
    Attributes
    ------------
    states : list of str
        state of environment
    """
    
    def __init__(self):
        self.states =  ["left_goal", "A", "B", "C", "D", "E", "right_goal"] 

    def state_update(self, state, action):
        """
        Parameters
        ------------
        state : char
            agent position in this game (in this case A-E)
        action : char
            agent action in this game (in this case "right" or "left")
        
        Returns
        ----------
        next_state : char
            agent next position
        temp_reward : int
            temporal (meaning not total) reward 
        """
        end_flg = False
        temp_reward = 0.         

        if action == "left":
            next_state_num = self.states.index(state) - 1
            if self.states[next_state_num] == "left_goal":
                end_flg = True
                temp_reward = 0. # end the game
            
            next_state = self.states[next_state_num]

        elif action == "right":
            next_state_num = self.states.index(state) + 1

            if self.states[next_state_num] == "right_goal":
                end_flg = True
                temp_reward = 1. # clear the game
        
            next_state = self.states[next_state_num]

        return end_flg, next_state, temp_reward

class Agent():
    """
    Agent model of reinforcement learning
    Attributes
    ------------
    state : str 
        now state of agent
    states : list of str
        state of environment
    action_list : list of str
        agent possibile action list 
    step rate : float
        learning rate of reinforcement learning
    discount rate : float
        discount rate of reward
    history_state : list 
        one episode state (this should not have the same state, first visit ES methods)
    values : OrderedDict
        value function
    """

    def __init__(self, step_rate):
        # environment
        self.model = RandomWalk()
        
        self.state = None
        self.states = self.model.states # in this case we shoud know the all states
        self.action_list = ["left", "right"]

        self.step_rate = step_rate # adjust subject
        self.discount_rate = 1.0
        
        # for bootstrap
        self.history_state = None
        self.values = OrderedDict()
        for state in self.states:
            self.values[state] = 0.5

            if state == "left_goal": # in this problem the final state does not have value
                self.values[state] = 0.
            if state == "right_goal":
                self.values[state] = 0.
        
        self.total_loss = 0.0
        self.history_loss_state = []

    def train_TD(self, train_num):
        """training the agent TD methods
        Parameters
        -----------
        train_num : int
            training number of reinforcement learning
        """
        for count in range(train_num+1):
            self.state = "C" # initial state has been decided

            while True: # Ending this while is worth of ending one episode

                end_flg, next_state, temp_reward = self._play(self.state) # play the game
                # for TD
                self._valuefunc_update_TD(temp_reward, self.state, next_state)

                # save the state and update
                self.state = next_state

                if end_flg:
                    self._calc_RMS(count)
                    break

    def train_montecarlo(self, train_num):
        """training the agent montecarlo methods
        Parameters
        -----------
        train_num : int
            training number of reinforcement learning
        """
        for count in range(train_num+1):
            self.state = "C" # initial state has been decided
            self.history_state = []

            while True: # Ending this while is worth of ending one episode
                end_flg, next_state, reward = self._play(self.state) # play the game

                # save the state and update
                # first visit method
                """
                if not self.state in self.history_state:
                    self.history_state.append(self.state)
                """
                self.history_state.append(self.state)
                
                self.state = next_state

                if end_flg:
                    self._valuefunc_update_montecarlo(reward, self.history_state)
                    self._calc_RMS(count)
                    break

    def _play(self, state):
        """
        playing the game

        Parameters
        -----------
        state : str
            state of the agent
        """
        action = self._decide_action()
        end_flg, next_state, temp_reward = self.model.state_update(state, action)

        return end_flg, next_state, temp_reward

    def _decide_action(self):
        """
        Returns
        ---------
        action : str
            action of agent
        """

        action = random.choice(self.action_list)

        return action

    def _valuefunc_update_TD(self, reward, state, next_state):
        """
        Parameters
        -----------
        reward : int
            reward the agent got
        state : str 
            now state of agent
        next_state : str 
            next state of agent
        """
        # TD
        self.values[state] = self.values[state] + self.step_rate * (reward + self.discount_rate * self.values[next_state] - self.values[state])

    def _valuefunc_update_montecarlo(self, final_reward, history_state):
        """
        Parameters
        -----------
        reward : int
            reward the agent got
        state : str 
            now state of agent
        next_state : str 
            next state of agent
        """
        # montecarlo
        for state in history_state:
            self.values[state] = self.values[state] + self.step_rate * (final_reward - self.values[state])

        # print(self.values)

    def _calc_RMS(self, count):
        """calculating RMS
        Parameters
        -----------
        count : int
            training count
        """
        ans_rate = [1/6, 2/6, 3/6, 4/6, 5/6]
        for i, state in enumerate(self.states[1:-1]): # except for "left goal" and "right goal"
            loss = math.sqrt((self.values[state] - ans_rate[i])**2)
            self.total_loss += loss
        
        self.history_loss_state.append(self.total_loss/float(count + 1)/5.) # 5 means state num

def main():
    # for fig 6.6 
    monte_alphas = [0.01, 0.02, 0.03, 0.04]
    td_alphas = [0.15, 0.1, 0.05]

    for monte_alpha in monte_alphas:
        agent = Agent(monte_alpha)
        agent.train_montecarlo(100)
        
        plt.plot(range(len(agent.history_loss_state)), agent.history_loss_state, label="monte_{0}".format(monte_alpha), linestyle="dashed")

    for td_alpha in td_alphas:
        agent = Agent(td_alpha)
        agent.train_TD(100)
        plt.plot(range(len(agent.history_loss_state)), agent.history_loss_state, label="TD_{0}".format(td_alpha), linestyle="solid")

    plt.legend()
    plt.show()
    

if __name__ == '__main__':
    main()