Gymで強化学習④手動エージェント

今回は手動でコントロールするエージェントを使って考察します。そうすることで環境がより理解でき、またその難しさが見えてくるからです。

そこで準備編で登場した月面着陸を取り上げます。実際に宇宙船を動かしてみながら報酬や観測値の変化を調べましょう。

今後はさまざまなエージェントが登場するので、エージェントをクラスとして定義します。最初はランダムに行動するランダム・エージェントをPythonのクラスとして定義し直し環境とエージェントが正常に動作するかを確認します。

次に、手動で動く手動エージェントを作成して報酬や観測値を見ながら考察します。月面着陸では報酬が与えられる条件が詳しく決められており、それに従って手動で操作すれば容易に目標達成できそうですが、実際はかなり難しいのがわかるはずです。

このようにして強化学習が直面する問題に対する理解を深めましょう。また、前回の記事で紹介したマルコフ性について具体例を通して再び解説します。

では、さっそく始めましょう。

Pythonの環境設定

準備編と同様ですが、ここに再掲します。以下のように、Pythonの環境を設定します。

# テスト用のフォルダを作る
mkdir rl_test
cd rl_test

＃Pythonの環境を作る
python3 -m venv venv
source venv/bin/activate

# Pipを最新のものにしておく
pip install --upgrade pip

# ジムと2Dゲーム環境の依存関係をインストール
pip install gymnasium[box2d]

ランダム・エージェント

準備編で登場したランダム・エージェントは、ランダムに選ばれた行動を返すもので次の一行で実装されていました。

action = env.action_space.sample()

しかし、より複雑なエージェントはPythonのクラスとして定義した方が理解しやすくなります。また、これからさまざまなエージェントを作成していくので、Pythonクラスにしておいた方がエージェント間での比較をしやすくなるという利便性もあります。

ランダム・エージェントは何も学ばないエージェントであり、強化学習エージェントとは言えませんが、強化学習の環境で実行することはできます。つまり、エージェントとしては最も単純なものと言えます。

ではランダム・エージェントをPythonクラスに書き直します。エージェントのクラスの条件としては、以下に示すように4つのメソッドを設定する必要があるとします。

class RandomAgent:
    def __init__(self, env):
        # エージェントの初期化
        self.env = env

    def reset(self, observation, info):
        # 環境からの初期観測値を受け取る
        pass

    def act(self):
        # 環境からランダムな行動を取得
        action = self.env.action_space.sample()
        return action

    def learn(self, observation, reward, terminated, truncated, info):
        pass

一つ一つ見ていきます。

    def __init__(self, env):
        # エージェントの初期化
        self.env = env

__init__メソッドはエージェントのオブジェクトを生成し初期化する時に呼ばれます。ランダム・エージェントはエージェントの参照を保持する以外には特に初期化することはありません。

    def reset(self, observation, info):
        # 環境からの初期観測値を受け取る
        pass

resetメソッドは環境がリセットされた時に呼ばれ初期の観測値（observation）を受け取ります。infoは月面着陸では空のディクショナリです。ここでもランダム・エージェントは特に何もしません。

    def act(self):
        # 環境からランダムな行動を取得
        action = self.env.action_space.sample()
        return action

actメソッドは次の行動（アクション）を返します。ランダム・エージェントは環境のアクション・スペースからランダムに選んだ行動を返します。強化学習エージェントならば、ここで状態（State）をもとに行動を返すところです。

    def learn(self, observation, reward, terminated, truncated, info):
        pass

learnメソッドは行動の結果として環境から返された観測値や報酬などの情報を受け取ります。ランダム・エージェントは何も学習を行いません。強化学習エージェントならば、ここで状態を更新したり、学習して内部パラメータの調整を行なったりします。

エピソード実行環境

ランダム・エージェントがPythonクラスとして定義されましたので、次にエージェントとGymの環境を使ってエピソードを繰り返し実行するファンクションを定義します。

準備編で紹介したコードに改良を加えたものをファンクションにしました。環境オブジェクト（env）とエージェントのオブジェクト（agent）を渡すとmax_episodesで与えられた回数までエピソードを繰り返します。

def run_episodes(env, agent, max_episodes):

    # エピソードのループをmax_episode回繰り返す
    for episode in range(max_episodes):

        # ゲーム環境を初期化してエージェントに初期観測値を渡す
        observation, info = env.reset()
        agent.reset(observation, info)

        # 報酬の総和をリセット
        total_reward = 0

        # 1つのエピソードのループ
        while True:
            # エージェントの行動を決定
            action = agent.act()

            # 行動を実行すると、環境の状態が更新される
            observation, reward, terminated, truncated, info = env.step(action)
            total_reward += reward

            # エージェントの学習
            agent.learn(observation, reward, terminated, truncated, info)

            # ゲームが終了したら、次のエピソードへ
            if terminated or truncated:
                break

        # エピソード終了時の処理
        print('-' * 20)
        print(f'エピソード: {episode:4d}')
        print(f'報酬総和: {total_reward:6.3f}')

    # 環境を閉じる
    env.close()

順を追って解説します。

エピソードはmax_episodesで指定さえた回数まで実行されます。

def run_episodes(env, agent, max_episodes):

    # エピソードのループをmax_episode回繰り返す
    for episode in range(max_episodes):

       ...

エピソードが始まる前に環境とエージェントのリセットを行います。ここでエージェントのresetメソッドが呼ばれています。また、報酬の総和も0にリセットします。

    # エピソードのループをmax_episodes回繰り返す
    for episode in range(max_episodes):

        # ゲーム環境を初期化してエージェントに初期観測値を渡す
        observation, info = env.reset()
        agent.reset(observation, info)

        # 報酬の総和をリセット
        total_reward = 0

一つのエピソードは宇宙船が着陸や墜落したり、画面からはみ出たりするまで続くので以下のように回数を決めずにステップを繰り返します。

        # 1つのエピソードのループ
        while True:
            # エージェントの行動を決定
            action = agent.act()

            ...

エージェントの行動はactメソッドから返されます。これを環境へと渡すと観測値や報酬などが返されます。ここで報酬の総和も更新します。

            # エージェントの行動を決定
            action = agent.act()

            # 行動を実行すると、環境の状態が更新される
            observation, reward, terminated, truncated, info = env.step(action)
            total_reward += reward

新しい観測値や報酬などがlearnメソッドへ渡され、エージェントは学習を行う機会を与えられます。もちろんランダム・エージェントは何もしません。

            # エージェントの学習
            agent.learn(observation, reward, terminated, truncated, info)

そしてゲームが終了したのであればエピソードのループから抜け出します。

        # 1つのエピソードのループ
        while True:
            # エージェントの行動を決定
            action = agent.act()

            ...

            # ゲームが終了したら、次のエピソードへ
            if terminated or truncated:
                break

エピソードが終了したら報酬総和をプリントし、次のエピソードへと続きます。

        # エピソード終了時の処理
        print('-' * 20)
        print(f'エピソード: {episode:4d}')
        print(f'報酬総和: {total_reward:6.3f}')

以上でエピソードの実行を行うファンクションができました。あとは、環境とエージェントのオブジェクトを作成して呼び出すだけです。

def main():
    # 月着陸(Lunar Lander)ゲームの環境を作成
    env = gym.make("LunarLander-v2", render_mode="human")

    # ランダムエージェントを作成
    agent = RandomAgent(env)

    # エピソードを実行
    run_episodes(env, agent, max_episodes=10)


if __name__ == "__main__":
   main()

すべてのコードをagent.pyに保存してください（ファイル名はなんでも構いません）。記事の最後にすべてのコードをまとめて掲載します。

Pythonでファイルを実行できます。

python agent.py

下のような画面が確認に出来れば実行はOKです。

また、コンソールには下のようにエピソードごとの報酬総和をみることができます。

--------------------
エピソード:    0
報酬総和: -88.967
--------------------
エピソード:    2
報酬総和: -275.430
--------------------
エピソード:    3
報酬総和: -244.139
--------------------
エピソード:    4
報酬総和: -257.675
--------------------
エピソード:    5
報酬総和: -158.187
--------------------
エピソード:    6
報酬総和: -110.795
--------------------
エピソード:    7
報酬総和: -97.614
--------------------

...

手動エージェントの定義

本題の手動エージェントを定義します。手動エージェントもランダム・エージェントと同様に4つのメソッドを定義します。

class ManualAgent:
    def __init__(self, observation):
        pass

    def reset(self, observation, info):
        ... 省略 ...

    def act(self):
        ... 省略 ...

    def learn(self, observation, reward, terminated, truncated, info):
        ... 省略 ...

__init__メソッドは何もしません。

resetメソッドはself.actionを0に設定します。これについては後述します。また、初期の観測値を表示します。print_observationメソッドについても後述します。

    def reset(self, observation, info):
        # 環境からの初期観測値を受け取る
        self.action = 0

        # 初期観測値を表示
        self.print_observation(observation)

actionメソッドはユーザからの入力を処理して次のアクションを決めます。

    def act(self):
        # 手動でアクションを入力。何も入力しなければ前回のアクションを継続
        # 0: 何もしない
        # 1: 左へ噴射
        # 2: 下へ噴射
        # 3: 右へ噴射
        while True:
            action = input(f"アクションを入力(現在のアクション{self.action}): ")

            if action == "":
                action = self.action
                break

            if action in ["0", "1", "2", "3"]:
                action = int(action)
                break

            print(f"Invalid action: {action}")

        # アクションを保存
        print(f"アクション: {action}")
        self.action = action

        return action

順番に解説します。

while True:となっているのは正当なアクションの値が取得できるまでループを繰り返すためです。ループの中では、まずユーザからの入力をinput(…)で取得します。

    def act(self):
        # 手動でアクションを入力。何も入力しなければ前回のアクションを継続
        # 0: 何もしない
        # 1: 右へ噴射
        # 2: 下へ噴射
        # 3: 左へ噴射
        while True:
            action = input(f"アクションを入力(現在のアクション{self.action}): ")

            ...

これが空の値（Enterキーを押すだけ）ならば前回のアクションを続けるのでループから抜けます。

            action = input(f"アクションを入力(現在のアクション{self.action}): ")

            if action == "":
                action = self.action
                break

0から3が入力されたならならばそれをアクションとして更新してループから抜けます。ここでself.actionが登場します。それがアクションの値を保持しているわけです。

            if action in ["0", "1", "2", "3"]:
                self.action = int(action)
                break

            print(f"Invalid action: {action}")

また、0から3以外のものが入力されたらエラーを表示してループの頭に戻ります。こうして正当な入力が取得できるまで入力のループが繰り返されます。

アクションが決まったらそれをプリントして返します。

        print(f"アクション: {self.action}")
        return self.action

以上から、エージェントのアクションが手動で決められることになります。

最後に、learnメソッドを見ましょう。ここでは環境から返された観測値や報酬などを表示するだけです。

    def learn(self, observation, reward, terminated, truncated, info):
        # 観測値を表示
        self.print_observation(observation)

        # 報酬などを表示
        print(f'報酬　: {reward:6.3f}')
        print(f'終了　: {terminated}')
        print(f'切断　: {truncated}')
        print('-' * 20)

最後にprint_observationメソッドを見ましょう。

    def print_observation(self, observation):
        # 宇宙船のx軸の位置(-1.5 ~ 1.5)
        # 宇宙船のy軸の位置(-1.5 ~ 1.5)
        # 宇宙船のx軸の速度(-5 ~ 5)
        # 宇宙船のy軸の速度(-5 ~ 5)
        # 宇宙船の角度(-3.14 ~ 3.14)
        # 宇宙船の角速度(-5 ~ 5)
        # 左の足が地面についているか(0か1)
        # 右の足が地面についているか(0か1)
        x, y, v_x, v_y, angle, angular_velocity, left_leg, right_leg = observation

        #　ラジアンから度に変換
        angle = angle / math.pi * 180
        angular_velocity = angular_velocity / math.pi * 180

        print(f'横軸　: {x:6.3f}')
        print(f'縦軸　: {y:6.3f}')
        print(f'横速度: {v_x:6.3f}')
        print(f'縦速度: {v_y:6.3f}')
        print(f'角度　: {angle:6.3f}')
        print(f'角速度: {angular_velocity:6.3f}')
        print(f'左足　: {left_leg:4.1f}')
        print(f'右足　: {right_leg:4.1f}')
        print('-' * 20)

長く見えますが、observationから値を取り出してバラバラにしたものをプリントしているだけです。角度と角速度はラジアンから度に変換しています。

この出力を見ることで手動でアクションを決めるのに役立てます。

手動エージェントを実行

手動エージェントを動かすにはmain()ファンクションを以下のように変更します。ランダム・エージェントはコメントして手動エージェントを使います。

def main():
    # 月着陸(Lunar Lander)ゲームの環境を作成
    env = gym.make("LunarLander-v2", render_mode="human")

    # ランダムエージェントを作成
    # agent = RandomAgent(env)

    # 手動エージェントを作成
    agent = ManualAgent(env)

    # エピソードを実行
    run_episodes(env, agent, max_episodes=10)

ではPythonでファイルを実行して実際にゆっくり目的地に着陸できるかやってみでください。全部のコードは最後にあります。

python agent.py

スタートしたらコンソールで観測値を見てアクションを入力します。

横軸　: -0.005
縦軸　:  1.418
横速度: -0.545
縦速度:  0.331
角度　:  0.358
角速度:  7.076
左足　:  0.0
右足　:  0.0
--------------------
アクションを入力(現在のアクション0): 

例えば2（下に噴射）を入れます。

アクションを入力(現在のアクション0): 2
アクション: 2
横軸　: -0.011
縦軸　:  1.426
横速度: -0.540
縦速度:  0.349
角度　:  0.722
角速度:  7.289
左足　:  0.0
右足　:  0.0
--------------------
報酬　: -2.216
終了　: False
切断　: False
--------------------
アクションを入力(現在のアクション2): 

物体の移動には重力による加速やその他の慣性が働くので下へ噴射してすぐに速度が遅くなるわけでもありません。また、角度によって下への噴射が垂直でなければ横速度にも影響します。

とにかく、やってみるとかなり難しいのがわかります。ぜひ挑戦してください。強化学習が解く問題の難しさを実感できるはずです。

マルコフ性について

もう気づいた方もいるとは思いますが、手動エージェントは人間そのものです。そして人間は観測値の値を見て行動を決めることになります。よってここでは観測値そのものが状態（State）になります。

もし、過去三個の観測値を記憶してそれをもとに行動を決めている人は過去三個の観測値が状態になります。そしてそれ以前の観測値は必要なくなります。

もちろん現在の状態は過去からの影響で決まるわけですが、次の状態を決める要因は現在の状態と行動のみになります。

この依存関係を数式で表現すると以下のようになります。

$$
s_{t+1} \leftarrow f(s_t, a_t)
$$

実際の$${s_{t+1}}$$の状態値は次の観測値などにより決まります。その際、さまざまなランダム性が影響してきます。

例えば、エンジンの噴射が一定でない場合、アクションがどのように環境に影響するかも一定ではありません。

つまり、$${s_{t+1}}$$は$${s_t}$$と$${a_t}$$に依存するのですが、実際の値は一意に決まるとは限りません。よって通常は$${s_t}$$と$${a_t}$$を条件とした$${s_{t+1}}$$の確率分布があると考えます。

$$
P(s_{t+1} | s_t, a_t )
$$

すると、現在の状態と各アクションによって次の状態の分布が決まることになります。次の状態とその報酬には相関があるので、現在のアクションのどれが最も高い報酬を得る可能性が高いのかを判断することができます。

次回はこの辺りの理論をもっと掘り下げていきます。

（続く）

すべてのコード

import gymnasium as gym
import math


class RandomAgent:
    def __init__(self, env):
        # エージェントの初期化
        self.env = env

    def reset(self, observation, info):
        # 環境からの初期観測値を受け取る
        pass

    def act(self):
        # 環境からランダムな行動を取得
        action = self.env.action_space.sample()
        return action

    def learn(self, observation, reward, terminated, truncated, info):
        pass


class ManualAgent:
    def __init__(self, observation):
        pass

    def reset(self, observation, info):
        # 環境からの初期観測値を受け取る
        self.action = 0

        # 初期観測値を表示
        self.print_observation(observation)

    def act(self):
        # 手動でアクションを入力。何も入力しなければ前回のアクションを継続
        # 0: 何もしない
        # 1: 右へ噴射
        # 2: 下へ噴射
        # 3: 左へ噴射
        while True:
            action = input(f"アクションを入力(現在のアクション{self.action}): ")

            if action == "":
                break

            if action in ["0", "1", "2", "3"]:
                self.action = int(action)
                break

            print(f"Invalid action: {action}")

        # アクションを保存
        print(f"アクション: {self.action}")
        return self.action

    def learn(self, observation, reward, terminated, truncated, info):
        # 観測値を表示
        self.print_observation(observation)

        # 報酬などを表示
        print(f'報酬　: {reward:6.3f}')
        print(f'終了　: {terminated}')
        print(f'切断　: {truncated}')
        print('-' * 20)

    def print_observation(self, observation):
        # 宇宙船のx軸の位置(-1.5 ~ 1.5)
        # 宇宙船のy軸の位置(-1.5 ~ 1.5)
        # 宇宙船のx軸の速度(-5 ~ 5)
        # 宇宙船のy軸の速度(-5 ~ 5)
        # 宇宙船の角度(-3.14 ~ 3.14)
        # 宇宙船の角速度(-5 ~ 5)
        # 左の足が地面についているか(0か1)
        # 右の足が地面についているか(0か1)
        x, y, v_x, v_y, angle, angular_velocity, left_leg, right_leg = observation

        #　ラジアンから度に変換
        angle = angle / math.pi * 180
        angular_velocity = angular_velocity / math.pi * 180

        print(f'横軸　: {x:6.3f}')
        print(f'縦軸　: {y:6.3f}')
        print(f'横速度: {v_x:6.3f}')
        print(f'縦速度: {v_y:6.3f}')
        print(f'角度　: {angle:6.3f}')
        print(f'角速度: {angular_velocity:6.3f}')
        print(f'左足　: {left_leg:4.1f}')
        print(f'右足　: {right_leg:4.1f}')
        print('-' * 20)


def run_episodes(env, agent, max_episodes):

    # エピソードのループをmax_episodes回繰り返す
    for episode in range(max_episodes):

        # ゲーム環境を初期化してエージェントに初期観測値を渡す
        observation, info = env.reset()
        agent.reset(observation, info)

        # 報酬の総和をリセット
        total_reward = 0

        # 1つのエピソードのループ
        while True:
            # エージェントの行動を決定
            action = agent.act()

            # 行動を実行すると、環境の状態が更新される
            observation, reward, terminated, truncated, info = env.step(action)
            total_reward += reward

            # エージェントの学習
            agent.learn(observation, reward, terminated, truncated, info)

            # ゲームが終了したら、次のエピソードへ
            if terminated or truncated:
                break

        # エピソード終了時の処理
        print('-' * 20)
        print(f'エピソード: {episode:4d}')
        print(f'報酬総和: {total_reward:6.3f}')

    # 環境を閉じる
    env.close()


def main():
    # 月着陸(Lunar Lander)ゲームの環境を作成
    env = gym.make("LunarLander-v2", render_mode="human")

    # ランダムエージェントを作成
    #agent = RandomAgent(env)

    # 手動エージェントを作成
    agent = ManualAgent(env)

    # エピソードを実行
    run_episodes(env, agent, max_episodes=10)


if __name__ == "__main__":
    main()