PyTorchとGymnasiumを使用したProximal Policy Optimization

1. 環境の設定

PPOを実装する前に、必要なソフトウェアライブラリをインストールし、ポリシーを適用する適切な環境を選択する必要があります。

PyTorchと必要なライブラリのインストール

以下のソフトウェアをインストールする必要があります：

• PyTorchとその他のソフトウェアライブラリ、たとえば数学や統計関数向けのnumpyやグラフのプロット向けのmatplotlibなどです。
• OpenAIのオープンソースGymソフトウェアパッケージは、さまざまな環境やゲームをシミュレートするPythonライブラリで、強化学習を用いて解決することができます。Gym APIを使用してアルゴリズムを環境とやり取りさせることができます。機能がgymはアップグレードプロセスを通じて変更されることがあるため、この例ではバージョンを0.25.2に固定します。

サーバーまたはローカルマシンにインストールするには、次のコマンドを実行してください：

$ pip install torch numpy matplotlib gym==0.25.2

Google ColabやDataLabなどのノートブックを使用してインストールする場合は、次の手順を実行します：

!pip install torch numpy matplotlib gym==0.25.2

CartPole環境を作成します。

OpenAI Gymを使用してCartPole環境のトレーニング用とテスト用の2つのインスタンスを作成します：

env_train = gym.make('CartPole-v1')
env_test = gym.make('CartPole-v1')

2. PyTorchでPPOを実装する

さて、PyTorchを使用してPPOを実装しましょう。

ポリシーネットワークを定義する

先に説明したように、PPOはアクター・クリティックモデルとして実装されます。アクターはポリシーを実装し、クリティックはその推定値を予測します。アクターとクリティックのニューラルネットワークは、各タイムステップの状態を同じ入力として受け取ります。したがって、アクターとクリティックモデルは共通のニューラルネットワークを共有することができ、これはバックボーンアーキテクチャと呼ばれます。アクターとクリティックは、追加のレイヤーでバックボーンアーキテクチャを拡張できます。

バックボーンネットワークを定義する

次の手順でバックボーンネットワークを定義します：

• 入力、隠れ、出力の3つのレイヤーを持つネットワークを実装します。
• 入力レイヤーと隠れレイヤーの後には、活性化関数を使用します。このチュートリアルでは、計算効率が良いため、ReLUを選択します。
• 私たちは入力層と隠れ層の後にドロップアウト関数を導入して、堅牢なネットワークを得ています。 ドロップアウト関数はランダムに一部のニューロンをゼロにします。 これにより特定のニューロンへの依存が減少し、過学習を防ぐことができ、ネットワークをより堅牢にします。

以下のコードはバックボーンを実装します：

class BackboneNetwork(nn.Module):
    def __init__(self, in_features, hidden_dimensions, out_features, dropout):
        super().__init__()
        self.layer1 = nn.Linear(in_features, hidden_dimensions)
        self.layer2 = nn.Linear(hidden_dimensions, hidden_dimensions)
        self.layer3 = nn.Linear(hidden_dimensions, out_features)
        self.dropout = nn.Dropout(dropout)
    def forward(self, x):
        x = self.layer1(x)
        x = f.relu(x)
        x = self.dropout(x)
        x = self.layer2(x)
        x = f.relu(x)
        x = self.dropout(x)
        x = self.layer3(x)
        return x

アクター・クリティック・ネットワークを定義します

これで、このネットワークを使用してアクター・クリティック・クラスActorCriticを定義できます。 アクターはポリシーをモデル化し、アクションを予測します。 クリティックは価値関数をモデル化し、値を予測します。 両方とも状態を入力として受け取ります。

class ActorCritic(nn.Module):
    def __init__(self, actor, critic):
        super().__init__()
        self.actor = actor
        self.critic = critic
    def forward(self, state):
        action_pred = self.actor(state)
        value_pred = self.critic(state)
        return action_pred, value_pred

アクターとクリティックのネットワークをインスタンス化します

上記で定義したネットワークを使用してアクターとクリティックを作成します。 その後、アクターとクリティックを含むエージェントを作成します。

エージェントを作成する前に、ネットワークのパラメータを初期化します：

• 隠れ層の次元H、これは設定可能なパラメータです。 隠れ層のサイズと数は問題の複雑さに依存します。 64 x 64次元の隠れ層を使用します。
• 入力特徴量N、Nは状態配列のサイズです。 入力層はN x H次元です。 CartPole環境では、状態は4要素の配列です。 したがってNは4です。
• アクターネットワークの出力特徴量O、Oは環境内のアクション数です。 アクターの出力層はH x O次元です。 CartPole環境には2つのアクションがあります。
• クリティックネットワークの出力特徴量。 クリティックネットワークは単に期待値を予測するだけなので、出力特徴量の数は1です。
• ドロップアウトを分数で表す。

次のコードは、バックボーンネットワークに基づいてアクターと評価者のネットワークを宣言する方法を示しています。

def create_agent(hidden_dimensions, dropout):
    INPUT_FEATURES = env_train.observation_space.shape[0]
    HIDDEN_DIMENSIONS = hidden_dimensions
    ACTOR_OUTPUT_FEATURES = env_train.action_space.n
    CRITIC_OUTPUT_FEATURES = 1
    DROPOUT = dropout
    actor = BackboneNetwork(
            INPUT_FEATURES, HIDDEN_DIMENSIONS, ACTOR_OUTPUT_FEATURES, DROPOUT)
    critic = BackboneNetwork(
            INPUT_FEATURES, HIDDEN_DIMENSIONS, CRITIC_OUTPUT_FEATURES, DROPOUT)
    agent = ActorCritic(actor, critic)
    return agent

リターンの計算

環境は、エージェントの行動に応じて、各ステップから次のステップに報酬を与えます。報酬Rは以下のように表されます：

リターンは、期待される将来の報酬の累積価値と定義されます。将来のタイムステップからの報酬は、即座の報酬よりも価値が低いです。したがって、リターンは通常、割引リターンとして計算され、Gとして定義されます：

このチュートリアル（および他の多くの参照文献）では、リターンとは割引リターンを指します。

リターンを計算するには：

• すべての将来の状態からの期待報酬から始めます。
• 各将来の報酬に、割引率の指数を掛けます。例えば、現在から2つのタイムステップ後の期待報酬は2倍になります。
• すべての割引された将来の報酬を合計してリターンを計算します。
• リターンの値を正規化します。

関数calculate_returns()は、以下に示すようにこれらの計算を実行します：

def calculate_returns(rewards, discount_factor):
    returns = []
    cumulative_reward = 0
    for r in reversed(rewards):
        cumulative_reward = r + cumulative_reward * discount_factor
        returns.insert(0, cumulative_reward)
    returns = torch.tensor(returns)
    # リターンを正規化
    returns = (returns - returns.mean()) / returns.std()
    return returns

アドバンテージ関数の実装

アドバンテージは、評価者によって予測された値とポリシーに従ってアクターが選択したアクションからの期待リターンとの差として計算されます。特定のアクションに対して、アドバンテージはその特定のアクションを取る利点を平均的なアクションと比較したものを表します。

元のPPO論文（式10）では、時刻Tまで先を見る利点は次のように表されます：

アルゴリズムのコーディング中、一定数の時刻まで先を見る制約はバッチサイズを介して強制されます。したがって、上記の式は価値と期待されるリターンとの差として単純化できます。期待されるリターンは状態-行動価値関数Qで数量化されます。

したがって、以下の単純化された式は、次の行動を選択する利点を表します：

• 特定の行動
• を特定の状態で
• 特定の方策の下で
• 特定の時刻に

これは以下のように表されます：

OpenAIもこの式を使用して強化学習を実装します。 calculate_advantages() 以下に示す関数は利点を計算します：

def calculate_advantages(returns, values):
    advantages = returns - values
    # 利点を正規化
    advantages = (advantages - advantages.mean()) / advantages.std()
    return advantages

サロゲート損失とクリッピングメカニズム

ポリシーロスは、PPOのような特別なテクニックなしでの標準ポリシーグラディエントロスです。標準のポリシーグラディエントロスは以下の積として計算されます：

• ポリシーアクションの確率
• アドバンテージ関数、つまり次の差として計算されます：
• ポリシーリターン
• 期待値

標準ポリシーグラディエントロスは急激なポリシー変更に対して補正を行うことができません。サロゲートロスは、標準ロスを変更して各イテレーションでポリシーが変更できる量を制限します。次の2つの量の最小値です：

• 次の積:
• ポリシー比率。この比率は、古い行動確率と新しい行動確率の差を表します。
• アドバンテージ関数
• 次の積:
• ポリシー比率のクランプされた値。この比率は、更新されたポリシーが古いポリシーの一定の割合内に収まるようにクリップされます。
• アドバンテージ関数

最適化プロセスでは、サロゲート損失が実際の損失のプロキシとして使用されます。

クリッピングメカニズム

方策比、Rは、新旧の方策の違いであり、新旧のパラメーターのもとでの方策の対数確率の比率として与えられます：

クリップされた方策比、R’は、次のように制約されています：

与えられた利点、At、前節で示されているように、および上記の方針比率に基づいて、代替損失は次のように計算されます：

以下のコードは、クリッピングメカニズムと代替損失の実装方法を示しています。

def calculate_surrogate_loss(
        actions_log_probability_old,
        actions_log_probability_new,
        epsilon,
        advantages):
    advantages = advantages.detach()
    policy_ratio = (
            actions_log_probability_new - actions_log_probability_old
            ).exp()
    surrogate_loss_1 = policy_ratio * advantages
    surrogate_loss_2 = torch.clamp(
            policy_ratio, min=1.0-epsilon, max=1.0+epsilon
            ) * advantages
    surrogate_loss = torch.min(surrogate_loss_1, surrogate_loss_2)
    return surrogate_loss

3. エージェントのトレーニング

さて、エージェントのトレーニングを行いましょう。

方針と価値の損失の計算

今、方針と価値の損失を計算する準備が整いました：

• ポリシー損失は、代理損失とエントロピー報酬の合計です。
• 価値損失は、批評家によって予測された価値とポリシーによって生成されたリターン（累積報酬）の違いに基づいています。価値損失の計算には、スムーズL1損失関数が使用されます。これにより、損失関数が平滑化され、外れ値に対して敏感でなくなります。

上記のように計算された両方の損失はテンソルです。勾配降下法はスカラー値に基づいています。損失を表す単一のスカラー値を取得するには、.sum()関数を使用してテンソル要素を合計します。以下の関数は、これを行う方法を示しています：

def calculate_losses(
        surrogate_loss, entropy, entropy_coefficient, returns, value_pred):
    entropy_bonus = entropy_coefficient * entropy
    policy_loss = -(surrogate_loss + entropy_bonus).sum()
    value_loss = f.smooth_l1_loss(returns, value_pred).sum()
    return policy_loss, value_loss

トレーニングループの定義

トレーニングプロセスを開始する前に、空の配列としてバッファのセットを作成します。トレーニングアルゴリズムは、エージェントの行動、環境の状態、および各時間ステップでの報酬に関する情報をこれらのバッファに保存します。以下の関数はこれらのバッファを初期化します：

def init_training():
    states = []
    actions = []
    actions_log_probability = []
    values = []
    rewards = []
    done = False
    episode_reward = 0
    return states, actions, actions_log_probability, values, rewards, done, episode_reward

各トレーニングイテレーションでは、そのイテレーションのポリシーパラメータを使用してエージェントを実行します。エージェントは、タイムステップごとにループ内で環境とやり取りし、終了条件に達するまで続きます。

各タイムステップの後に、エージェントの行動、報酬、および値がそれぞれのバッファに追加されます。エピソードが終了すると、関数はエピソードの結果を要約した更新されたバッファのセットを返します。

トレーニングループを実行する前に：

• agent.train()を使用してモデルをトレーニングモードに設定します。
• env.reset()を使用して環境をランダムな状態にリセットします。これがこのトレーニングイテレーションの開始状態です。

トレーニングループ内の各タイムステップで何が起こるかを説明する以下の手順：

• エージェントに状態を渡します。
• エージェントが返す：
• ポリシー（アクター）に基づいて状態に基づいて予測されたアクションを返します。この予測されたアクションテンソルをソフトマックス関数を介してアクション確率のセットを取得します。
• 評論家に基づいて、状態の予測値。
• エージェントは取るべき行動を選択します:
• 行動確率を使用して確率分布を推定します。
• この分布からサンプルを選んでアクションをランダムに選択します。 dist.sample() 関数がこれを行います。
• このアクションを環境に渡して、この時間ステップの環境の応答をシミュレートするために env.step() 関数を使用します。エージェントの行動に基づいて、環境が生成するもの:
• 新しい状態
• 報酬
• ブール値done（これは環境が終端状態に達したかどうかを示します）
• エージェントの行動、報酬、予測値、および新しい状態の値をそれぞれのバッファに追加します。

トレーニングエピソードは、doneのブール値がenv.step()関数からtrueを返すときに終了します。

エピソードが終了した後、各タイムステップから蓄積された値を使用して、各タイムステップの報酬を加算してこのエピソードの累積リターンを計算します。これを行うには、以前に説明したcalculate_returns()関数を使用します。この関数の入力は割引率と各タイムステップからの報酬を含むバッファです。これらのリターンと各タイムステップからの蓄積された値を使用して、calculate_advantages()関数を使用してアドバンテージを計算します。

次のPython関数は、これらのステップを実装する方法を示しています:

def forward_pass(env, agent, optimizer, discount_factor):
    states, actions, actions_log_probability, values, rewards, done, episode_reward = init_training()
    state = env.reset()
    agent.train()
    while not done:
        state = torch.FloatTensor(state).unsqueeze(0)
        states.append(state)
        action_pred, value_pred = agent(state)
        action_prob = f.softmax(action_pred, dim=-1)
        dist = distributions.Categorical(action_prob)
        action = dist.sample()
        log_prob_action = dist.log_prob(action)
        state, reward, done, _ = env.step(action.item())
        actions.append(action)
        actions_log_probability.append(log_prob_action)
        values.append(value_pred)
        rewards.append(reward)
        episode_reward += reward
    states = torch.cat(states)
    actions = torch.cat(actions)
    actions_log_probability = torch.cat(actions_log_probability)
    values = torch.cat(values).squeeze(-1)
    returns = calculate_returns(rewards, discount_factor)
    advantages = calculate_advantages(returns, values)
    return episode_reward, states, actions, actions_log_probability, advantages, returns

モデルパラメータの更新

各トレーニングイテレーションでは、多くのタイムステップ（終了条件に達するまで）からなる完全なエピソードをモデルを実行します。各タイムステップで、ポリシーパラメータ、エージェントのアクション、リターン、およびアドバンテージを保存します。各イテレーションの後、そのイテレーション内のすべてのタイムステップを通じたポリシーのパフォーマンスに基づいてモデルを更新します。

CartPole環境における最大のタイムステップ数は500です。より複雑な環境では、数百万のタイムステップがあります。そのような場合、トレーニング結果のデータセットをバッチに分割する必要があります。各バッチのタイムステップ数は最適化バッチサイズと呼ばれます。

したがって、モデルパラメータを更新する手順は次のとおりです。

• トレーニング結果のデータセットをバッチに分割します。
• 各バッチについて：
• エージェントの行動と各状態の予測値を取得します。
• これらの予測された行動を使用して新しい行動確率分布を推定します。
• この分布を使用してエントロピーを計算します。
• この分布を使用して、トレーニング結果データセット内のアクションの対数確率を取得します。これは、トレーニング結果データセット内のアクションの新しい対数確率のセットです。同じアクションの古い対数確率のセットは、前のセクションで説明されたトレーニングループで計算されました。
• アクションの古い確率分布と新しい確率分布を使用してサロゲート損失を計算します。
• サロゲート損失、エントロピー、およびアドバンテージを使用してポリシー損失とバリュー損失を計算します。
• ポリシーおよびバリュー損失にそれぞれ.backward()を実行します。これにより、損失関数の勾配が更新されます。
• オプティマイザに.step()を実行してポリシーパラメータを更新します。この場合、速度と堅牢性のバランスを取るためにAdamオプティマイザを使用します。
• ポリシーとバリューロスを蓄積します。
• PPO_STEPSパラメータの値に応じて、各バッチでバックワードパス（上記の操作）を数回繰り返します。各バッチでバックワードパスを繰り返すことは、追加のフォワードパスを実行せずにトレーニングデータセットのサイズを効果的に増やすため、計算効率が良いです。サンプリングと最適化の間の交互に行われる環境ステップの数をイテレーションバッチサイズと呼びます。
• 平均ポリシーロスとバリューロスを返します。

以下のコードはこれらのステップを実装しています:

def update_policy(
        agent,
        states,
        actions,
        actions_log_probability_old,
        advantages,
        returns,
        optimizer,
        ppo_steps,
        epsilon,
        entropy_coefficient):
    BATCH_SIZE = 128
    total_policy_loss = 0
    total_value_loss = 0
    actions_log_probability_old = actions_log_probability_old.detach()
    actions = actions.detach()
    training_results_dataset = TensorDataset(
            states,
            actions,
            actions_log_probability_old,
            advantages,
            returns)
    batch_dataset = DataLoader(
            training_results_dataset,
            batch_size=BATCH_SIZE,
            shuffle=False)
    for _ in range(ppo_steps):
        for batch_idx, (states, actions, actions_log_probability_old, advantages, returns) in enumerate(batch_dataset):
            すべての入力状態に対してアクションの新しい対数確率を取得します
            action_pred, value_pred = agent(states)
            value_pred = value_pred.squeeze(-1)
            action_prob = f.softmax(action_pred, dim=-1)
            probability_distribution_new = distributions.Categorical(
                    action_prob)
            entropy = probability_distribution_new.entropy()
            古いアクションを使用して新しい対数確率を推定します
            actions_log_probability_new = probability_distribution_new.log_prob(actions)
            surrogate_loss = calculate_surrogate_loss(
                    actions_log_probability_old,
                    actions_log_probability_new,
                    epsilon,
                    advantages)
            policy_loss, value_loss = calculate_losses(
                    surrogate_loss,
                    entropy,
                    entropy_coefficient,
                    returns,
                    value_pred)
            optimizer.zero_grad()
            policy_loss.backward()
            value_loss.backward()
            optimizer.step()
            total_policy_loss += policy_loss.item()
            total_value_loss += value_loss.item()
    return total_policy_loss / ppo_steps, total_value_loss / ppo_steps

4. PPO エージェントの実行

PPO エージェントを最終的に実行しましょう。

パフォーマンスの評価

エージェントのパフォーマンスを評価するには、新しい環境を作成し、この新しい環境でエージェントを実行して得られる累積報酬を計算します。 エージェントを.eval()関数を使用して評価モードに設定する必要があります。手順はトレーニングループと同じです。 以下のコードスニペットは評価関数を実装しています。

def evaluate(env, agent):
    agent.eval()
    rewards = []
    done = False
    episode_reward = 0
    state = env.reset()
    while not done:
        state = torch.FloatTensor(state).unsqueeze(0)
        with torch.no_grad():
            action_pred, _ = agent(state)
            action_prob = f.softmax(action_pred, dim=-1)
        action = torch.argmax(action_prob, dim=-1)
        state, reward, done, _ = env.step(action.item())
        episode_reward += reward
    return episode_reward

トレーニング結果の視覚化

トレーニングプロセスの進捗状況を可視化するためにMatplotlibライブラリを使用します。 以下の関数は、トレーニングループとテストループからの報酬をプロットする方法を示しています：

def plot_train_rewards(train_rewards, reward_threshold):
    plt.figure(figsize=(12, 8))
    plt.plot(train_rewards, label='Training Reward')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Training Reward', fontsize=20)
    plt.hlines(reward_threshold, 0, len(train_rewards), color='y')
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

def plot_test_rewards(test_rewards, reward_threshold):
    plt.figure(figsize=(12, 8))
    plt.plot(test_rewards, label='Testing Reward')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Testing Reward', fontsize=20)
    plt.hlines(reward_threshold, 0, len(test_rewards), color='y')
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

以下の例のプロットでは、トレーニング環境とテスト環境でポリシーを適用して得られたトレーニングおよびテストの報酬を示しています。これらのプロットの形状は、コードを実行する度に異なるように見えることに注意してください。これは、トレーニングプロセス固有のランダム性によるものです。

トレーニング報酬（トレーニング環境でポリシーを適用して得られたもの）。著者による画像。

テスト報酬（テスト環境でポリシーを適用して得られたもの）。著者による画像。

上記の出力グラフで、トレーニングプロセスの進捗を観察してください：

• 報酬は低い値から始まります。トレーニングが進むにつれて、報酬が増加します。
• 報酬は増加しながらランダムに変動します。これはエージェントが方策空間を探索しているためです。
• トレーニングが終了し、テストの報酬はしばらくの間475のしきい値を安定させています。
• 報酬は500でキャップされています。これらは環境（Gym CartPole v1）によって課せられた制約です。

同様に、イテレーションを通じて値と方策の損失をプロットすることができます:

def plot_losses(policy_losses, value_losses):
    plt.figure(figsize=(12, 8))
    plt.plot(value_losses, label='Value Losses')
    plt.plot(policy_losses, label='Policy Losses')
    plt.xlabel('Episode', fontsize=20)
    plt.ylabel('Loss', fontsize=20)
    plt.legend(loc='lower right')
    plt.grid()
    plt.show()

以下は、トレーニングエピソードを通じて追跡された損失を示す例のプロットです:

トレーニングプロセスを通じた価値とポリシーの損失。著者による画像

プロットを観察し、次の点に注意してください：

• 損失はランダムに分布しており、特定のパターンに従っていません。
• これはRLトレーニングの典型的な特徴であり、損失を最小化するのではなく報酬を最大化することが目標です。

PPOアルゴリズムを実行

PPOを使用してエージェントをトレーニングするためのすべてのコンポーネントを持っています。すべてを組み合わせるためには、次の手順が必要です：

• 割引率、バッチサイズ、学習率などのハイパーパラメータを宣言する。
• 各イテレーションからの報酬と損失を格納するために、バッファーをヌル配列としてインスタンス化します。
• create_agent() 関数を使用してエージェントのインスタンスを作成します。
• forward_pass() および update_policy() 関数を使用して順伝播と逆伝播を反復的に実行します。
• evaluate() 関数を使用してポリシーのパフォーマンスをテストします。
• トレーニングおよび評価関数からのポリシー、価値損失、および報酬をそれぞれのバッファーに追加します。
• 過去数タイムステップの報酬と損失の平均を計算します。以下の例では、過去40タイムステップの報酬と損失の平均を取ります。
• 数ステップごとに評価結果を出力します。以下の例では、10ステップごとに出力します。
• 平均報酬が一定の閾値を超えたときにプロセスを終了します。

Pythonでこれを行う関数を宣言する方法は以下のコードに示されています:

def run_ppo():
    MAX_EPISODES = 500
    DISCOUNT_FACTOR = 0.99
    REWARD_THRESHOLD = 475
    PRINT_INTERVAL = 10
    PPO_STEPS = 8
    N_TRIALS = 100
    EPSILON = 0.2
    ENTROPY_COEFFICIENT = 0.01
    HIDDEN_DIMENSIONS = 64
    DROPOUT = 0.2
    LEARNING_RATE = 0.001
    train_rewards = []
    test_rewards = []
    policy_losses = []
    value_losses = []
    agent = create_agent(HIDDEN_DIMENSIONS, DROPOUT)
    optimizer = optim.Adam(agent.parameters(), lr=LEARNING_RATE)
    for episode in range(1, MAX_EPISODES+1):
        train_reward, states, actions, actions_log_probability, advantages, returns = forward_pass(
                env_train,
                agent,
                optimizer,
                DISCOUNT_FACTOR)
        policy_loss, value_loss = update_policy(
                agent,
                states,
                actions,
                actions_log_probability,
                advantages,
                returns,
                optimizer,
                PPO_STEPS,
                EPSILON,
                ENTROPY_COEFFICIENT)
        test_reward = evaluate(env_test, agent)
        policy_losses.append(policy_loss)
        value_losses.append(value_loss)
        train_rewards.append(train_reward)
        test_rewards.append(test_reward)
        mean_train_rewards = np.mean(train_rewards[-N_TRIALS:])
        mean_test_rewards = np.mean(test_rewards[-N_TRIALS:])
        mean_abs_policy_loss = np.mean(np.abs(policy_losses[-N_TRIALS:]))
        mean_abs_value_loss = np.mean(np.abs(value_losses[-N_TRIALS:]))
        if episode % PRINT_INTERVAL == 0:
            print(f'Episode: {episode:3} | \
                  Mean Train Rewards: {mean_train_rewards:3.1f} \
                  | Mean Test Rewards: {mean_test_rewards:3.1f} \
                  | Mean Abs Policy Loss: {mean_abs_policy_loss:2.2f} \
                  | Mean Abs Value Loss: {mean_abs_value_loss:2.2f}')
        if mean_test_rewards >= REWARD_THRESHOLD:
            print(f'Reached reward threshold in {episode} episodes')
            break
    plot_train_rewards(train_rewards, REWARD_THRESHOLD)
    plot_test_rewards(test_rewards, REWARD_THRESHOLD)
    plot_losses(policy_losses, value_losses)

プログラムを実行します:

run_ppo()

出力は以下のサンプルに似ているはずです:

Episode:  10 | Mean Train Rewards: 22.3 | Mean Test Rewards: 30.4 | Mean Abs Policy Loss: 0.37 | Mean Abs Value Loss: 0.39 
Episode:  20 | Mean Train Rewards: 38.6 | Mean Test Rewards: 69.8 | Mean Abs Policy Loss: 0.46 | Mean Abs Value Loss: 0.37 
.
.
.
Episode: 100 | Mean Train Rewards: 289.5 | Mean Test Rewards: 427.3 | Mean Abs Policy Loss: 1.73 | Mean Abs Value Loss: 0.21 
Episode: 110 | Mean Train Rewards: 357.7 | Mean Test Rewards: 461.4 | Mean Abs Policy Loss: 1.86 | Mean Abs Value Loss: 0.22 
Reached reward threshold in 116 episodes

このDataLabノートブックで作業プログラムを表示および実行できます！

5. ハイパーパラメータのチューニングと最適化

機械学習では、ハイパーパラメータがトレーニングプロセスを制御します。以下では、PPOで使用される重要なハイパーパラメータのいくつかを説明します：

• 学習率: 学習率は、各イテレーションでポリシーパラメータがどれだけ変化できるかを決定します。確率的勾配降下法では、ポリシーパラメータが各イテレーションで更新される量は、学習率と勾配の積で決まります。
• クリッピングパラメータ： これはεとしても言及されます。 ポリシー比率がクリップされる範囲を決定します。新しいポリシーと古いポリシーの比率は、 [1-ε、1+ε]の範囲で変動することが許可されています。この範囲を超えると、人工的に範囲内に収められます。
• バッチサイズ:これは、各勾配更新のために考慮するステップ数を指します。PPOでは、バッチサイズはポリシーを適用し、ポリシーパラメータを更新するためのサロゲート損失を計算するために必要なタイムステップの数です。この記事では、バッチサイズを64として使用しました。
• 反復ステップ:これは、各バッチをバックワードパスを実行するために何度再利用するかを示します。この記事のコードではこれをPPO_STEPSと呼んでいます。複雑な環境では、フォワードパスを何度も実行することは計算上コストがかかります。より効率的な代替手段は、各バッチを数回再実行することです。通常は、5から10の間の値を使用することが推奨されています。
• 割引率: これはγとも呼ばれます。 これは即時の報酬が将来の報酬よりも価値が高い程度を表します。これは、お金の時間価値を計算する際の利子率の概念に似ています。が0に近いほど、将来の報酬が少なく価値が低く、エージェントは即時の報酬を優先する必要があります。が1に近いほど、将来の報酬が重要であることを意味します。
• エントロピー係数:エントロピー係数は、分布のエントロピーとエントロピー係数の積で計算されるエントロピーボーナスを決定します。エントロピーボーナスの役割は、ポリシーにさらなるランダム性を導入することです。これにより、エージェントはポリシー空間を探索することが奨励されます。ただし、このランダム性が高すぎると、訓練が最適なポリシーに収束しなくなります。
• トレーニングの成功基準：トレーニングが成功したと見なすための基準を設定する必要があります。これを行う一般的な方法は、直近のN回（エピソード）の平均報酬が一定の閾値を超えている場合とすることです。上記のコード例では、これは変数N_TRIALSで表されています。これをより高い値に設定すると、トレーニングに時間がかかるようになります。なぜなら、ポリシーがより多くのエピソードで閾値報酬を達成しなければならなくなるからです。また、計算コストが高くなりますが、より堅牢なポリシーを生み出します。PPOは確率ポリシーであり、エージェントが閾値を超えないエピソードも発生します。したがって、N_TRIALSの値があまりにも高い場合、トレーニングが終了しない可能性があります。

PPOパフォーマンスを最適化する戦略

訓練PPOアルゴリズムのパフォーマンスを最適化するには、試行錯誤が必要であり、さまざまなハイパーパラメータ値を実験する必要があります。ただし、いくつかの一般的なガイドラインがあります：

• 割引率：長期的な報酬が重要な場合、たとえばポールが時間の経過とともに安定している必要があるCartPole環境などでは、0.99などの適度なガンマ値から始めます。
• エントロピー補正：複雑な環境では、エージェントは最適な方策を見つけるために行動空間を探索する必要があります。エントロピー補正は探索を促進します。エントロピー補正はサロゲート損失に追加されます。エントロピー係数を決定する前に、サロゲート損失と分布のエントロピーの大きさを確認してください。この記事では、エントロピー係数を0.01に設定しました。
• クリッピングパラメータ: クリッピングパラメータは、更新されたポリシーが現在のポリシーとどれだけ異なるかを決定します。 クリッピングパラメータの大きな値は環境の探索を促進しますが、トレーニングを不安定にするリスクがあります。 不安定な更新を防ぎながら、徐々に探索を許可するクリッピングパラメータを選択します。 この記事では、クリッピングパラメータを0.2に設定しました。
• 学習率: 学習率が高すぎると、ポリシーが大きなステップで更新され、トレーニングプロセスが不安定になる可能性があります。 低すぎると、トレーニングに時間がかかります。 このチュートリアルでは、環境に適した学習率0.001を使用しました。 多くの場合、学習率1e-5を使用することが推奨されています。

PPOにおける課題とベストプラクティス

PPOの概念と実装の詳細を説明した後、課題とベストプラクティスについて議論しましょう。

訓練中のPPOの一般的な課題

PPOは広く使用されているが、この技術を成功裏に使用して実世界の問題を解決するための潜在的な課題に注意する必要があります。そのような課題のいくつかは次のとおりです。

• 収束が遅い： 複雑な環境では、PPOはサンプル効率が悪く、最適ポリシーに収束するには環境との多くのやり取りが必要です。これにより、トレーニングが遅くて高価になります。
• ハイパーパラメータへの感度: PPOはポリシー空間を効率的に探索することに依存しています。トレーニングプロセスの安定性や収束速度はハイパーパラメータの値に敏感です。これらのハイパーパラメータの最適な値は、しばしば試行錯誤によってのみ決定されることがよくあります。
• 過学習: RL環境は通常、ランダムなパラメータで初期化されます。PPOのトレーニングは、エージェントの環境に基づいて最適なポリシーを見つけることに基づいています。トレーニングプロセスが特定の環境に最適なパラメータのセットに収束することがありますが、ランダム化された環境には収束しないことがあります。これは通常、異なるランダム化されたトレーニング環境を持つ多くの反復を行うことで対処されます。
• ダイナミックな環境: CartPole環境などのシンプルなRL環境は静的であり、ルールは時間と共に変わりません。一方、不安定な移動する地面で歩行を学習するロボットなど、他の多くの環境はダイナミックであり、環境のルールは時間と共に変化します。このような環境でうまく機能するためには、PPOは通常、追加の微調整が必要となります。
• 探索 vs 活用: PPOのクリッピングメカニズムにより、ポリシーの更新が信頼できる領域内で行われます。しかし、これはエージェントがアクション空間を探索することを妨げる一方、複雑な環境では特に局所最適解に収束する可能性があります。一方で、エージェントにあまりにも多く探索させると、最適ポリシーに収束させることができなくなる可能性があります。

PPOモデルのトレーニングにおけるベストプラクティス

PPOを使用して良い結果を得るためには、以下のベストプラクティスをお勧めします：

• 入力特徴の正規化：リターンやアドバンテージの値を正規化することで、データの変動を減らし、安定した勾配更新を実現します。データの正規化により、すべての値を一貫した数値範囲に揃えることができます。これにより、外れ値や極端な値の影響を軽減し、勾配更新の歪みや収束の遅れを防ぎます。
• 適切に大きなバッチサイズを使用する： 小さなバッチは更新とトレーニングを速く行うことができますが、局所最適解への収束やトレーニングプロセスの不安定化につながる可能性があります。大きなバッチサイズを使用すると、エージェントが頑健なポリシーを学習し、安定したトレーニングプロセスにつながります。ただし、あまりにも大きなバッチサイズも最適ではありません。計算コストが増加するだけでなく、大きなバッチで推定された平均値に基づいて勾配更新が行われるため、ポリシーアップデートが価値関数に対して反応しづらくなります。さらに、特定のバッチに過剰適合する可能性があります。
• イテレーションステップ： 通常、各バッチを5-10回のイテレーションで再利用することが望ましいです。これにより、トレーニングプロセスがより効率的になります。同じバッチをあまり多くの回数再利用すると、過剰適合につながります。コードではこのハイパーパラメータをPPO_STEPSとして参照しています。
• 定期的な評価を実施する:オーバーフィッティングを検出するために、方針の効果を定期的にモニターすることが重要です。特定のシナリオで方針が効果がないことが判明した場合、さらなるトレーニングや微調整が必要となるかもしれません。
• ハイパーパラメータを調整する: 先に説明した通り、PPOトレーニングはハイパーパラメータの値に敏感です。特定の問題に適した値のセットを決定するために、さまざまなハイパーパラメータの値を実験してみてください。
• 共有バックボーンネットワーク： この記事で説明されているように、共有バックボーンを使用すると、アクターネットワークとクリティックネットワークの間の不均衡を防ぐことができます。アクターとクリティックの間でバックボーンネットワークを共有することで、共有された特徴抽出と環境の共通理解が助けられます。これにより、学習プロセスがより効率的で安定します。また、アルゴリズムの計算空間と時間の複雑さを減らすのにも役立ちます。
• 隠れ層の数とサイズ： より複雑な環境には、隠れ層と次元数を増やしてください。CartPoleのようなシンプルな問題は単一の隠れ層で解決できます。この記事で使用されている隠れ層は64次元です。必要以上にネットワークを大きくすると、計算上の無駄が生じ、不安定になる可能性があります。
• 早期終了：評価メトリクスが満たされたときにトレーニングを停止することは、過学習を防ぎリソースの浪費を回避するのに役立ちます。エージェントが過去のN回のイベントで閾値リワードを超えたときが一般的な評価メトリクスです。

Conclusion

この記事では、RL問題を解決する方法としてPPOについて説明しました。その後、PyTorchを使用してPPOを実装する手順を詳細に説明しました。最後に、PPOのパフォーマンスのヒントとベストプラクティスを紹介しました。

学ぶ最良の方法は、コードを自分で実装することです。また、コードを他のクラシックな制御環境で動作するように変更することもできます。PythonとOpenAIのGymnasiumを使用してRLエージェントを実装する方法について学ぶには、「PythonでのGymnasiumを使った強化学習」のコースをフォローしてください！