優先度付き経験再生 (PER)

Pythonで学ぶDeep Reinforcement Learning

Timothée Carayol

Principal Machine Learning Engineer, Komment

すべての経験が同等ではない

 

  • 経験再生:
    • 一様サンプリングでは重要な記憶を見落とすことがある
  • 優先度付き経験再生:
    • TD 誤差に基づき各経験に優先度を付与
    • 学習効果の高い経験に集中

 

学生が図書館で勉強している

Pythonで学ぶDeep Reinforcement Learning

優先度付き経験再生 (PER)

 

for step = 1 to T do:
    # 価値関数に基づく最適な行動を取る
    # 次状態と報酬を観測
    # 遷移をリプレイバッファに追加

      # 最高優先度 (1) を付与

    # 過去の遷移をバッチでサンプリング

      # 優先度に基づく (2)

    # バッチの TD 誤差を計算

    # 損失を計算し Q ネットワークを更新

      # 重要度サンプリング重みを使用 (4)

    # サンプルした遷移の優先度を更新 (3)

# 時間とともに重要度サンプリングを増加 (5)

(1) 新しい遷移は最高優先度で追加: $p_i = \max_k(p_k)$

(2) 遷移 $i$ のサンプル確率 $$P(i) = p_i^{\alpha} / \sum_k p_k^{\alpha}\ \ \ \ \ \ \ \ (0<\alpha<1)$$

(3) サンプル済み遷移の優先度を TD 誤差に更新: $p_i = |\delta_i| + \varepsilon$

(4) 重要度サンプリング重み $$w_i = \left( \frac{1}{N} \cdot \frac{1}{P(i)} \right)^\beta\ \ \ \ \ \ \ \ (0<\beta<1)$$

(5) $\beta$ を徐々に 1 に近づける

Pythonで学ぶDeep Reinforcement Learning

PER の実装

def __init__(self, capacity, alpha=0.6, beta=0.4, beta_increment=0.001, epsilon=0.001):
    # メモリバッファを初期化
    self.memory = deque(maxlen=capacity)

    # パラメータを保持し優先度を初期化
    self.alpha, self.beta, self.beta_increment, self.epsilon = (alpha, beta, beta_increment, epsilon)
    self.priorities = deque(maxlen=capacity)


...
Pythonで学ぶDeep Reinforcement Learning

PER の実装

...

def push(self, state, action, reward, next_state, done):
    # 経験をメモリに追加
    experience_tuple = (state, action, reward, next_state, done)
    self.memory.append(experience_tuple)

    # 新規遷移の優先度を最大値に設定
    max_priority = max(self.priorities) if self.memory else 1.0
    self.priorities.append(max_priority)


...
Pythonで学ぶDeep Reinforcement Learning

PER の実装

def sample(self, batch_size):
    priorities = np.array(self.priorities)
    # サンプリング確率を計算
    probabilities = priorities**self.alpha / np.sum(priorities**self.alpha)

    # インデックスを確率的にサンプリング
    indices = np.random.choice(len(self.memory), batch_size, p=probabilities)

    # 重みを計算
    weights = (1 / (len(self.memory) * probabilities)) ** self.beta
    weights /= np.max(weights)
    states, actions, rewards, next_states, dones = zip(*[self.memory[idx] for idx in indices])
    weights = [weights[idx] for idx in indices]
    states, actions, rewards, next_states, dones = (zip(*[self.memory[idx] for idx in indices]))

    # テンソルを返す
    states = torch.tensor(states, dtype=torch.float32)
    ... # rewards, next_states, dones, weights も同様に
    actions = torch.tensor(actions, dtype=torch.long).unsqueeze(1)    
    return (states, actions, rewards, next_states, dones, indices, weights)
Pythonで学ぶDeep Reinforcement Learning

PER の実装

...

def update_priorities(self, indices, td_errors: torch.Tensor):
    # サンプルした遷移の優先度を更新
    for idx, td_error in zip(indices, td_errors.abs()):
        self.priorities[idx] = abs(td_error.item()) + self.epsilon


def increase_beta(self):
    # beta を 1 に近づけて増加
    self.beta = min(1.0, self.beta + self.beta_increment)
Pythonで学ぶDeep Reinforcement Learning

DQN 学習ループでの PER

 

  1. ループ前のコード:

    buffer = PrioritizedReplayBuffer(capacity)

  2. 各エピソードの開始時:

    buffer.increase_beta()

3. 各ステップで:

# 行動選択後
buffer.push(state, action, reward, 
            next_state, done)
...

# TD 誤差を計算する前:
replay_buffer.sample(batch_size)
...

# TD 誤差を計算した後
buffer.update_priorities(indices, td_errors)

loss = torch.sum(weights * (td_errors ** 2))
Pythonで学ぶDeep Reinforcement Learning

PER の実例: Cartpole

Cartpole 環境での 100 回の学習:

  1. 優先度付き経験再生あり
  2. 一様サンプリングの経験再生
  • 一様再生より、PER の方が学習が速く、性能も高い

学習曲線: PER はより速く学習

 

100 エポック後: Cartpole、100 エポック後は不安定

 

400 エポック後: Cartpole、400 エポック後は安定

Pythonで学ぶDeep Reinforcement Learning

PER の実例: Atari 環境

 

  • Atari 環境で PER により大幅な性能向上

棒グラフ。人間、DQN、DDQN、Dueling DDQN、Prioritized DDQN、Prioritized Dueling DQN を比較。最後の手法で Prioritized Experience Replay により DDQN の性能が向上。

1 https://arxiv.org/abs/2303.11634
Pythonで学ぶDeep Reinforcement Learning

Laten we oefenen!

Pythonで学ぶDeep Reinforcement Learning

Preparing Video For Download...