Redarea prioritizată a experienței

Deep Reinforcement Learning în Python

Timothée Carayol

Principal Machine Learning Engineer, Komment

Nu toate experiențele sunt egale

Redarea experienței:
- Eșantionarea uniformă poate neglija experiențele importante
Redarea prioritizată a experienței:
- Atribuie prioritate fiecărei experiențe, pe baza erorilor TD
- Se concentrează pe experiențele cu potențial ridicat de învățare

Studenți care studiază într-o bibliotecă

Redarea Prioritizată a Experienței (PER)

for step = 1 to T do:
    # Take optimal action according to value function
    # Observe next state and reward
    # Append transition to replay buffer

      # Give it highest priority (1)

    # Sample a batch of past transitions

      # Based on priority (2)

    # Calculate TD errors for the batch

    # Calculate the loss and update the Q Network

      # Use importance sampling weights (4)

    # Update priority of sampled transitions (3)

# Increase importance sampling over time. (5)

(1) Tranzițiile noi sunt adăugate cu prioritatea maximă $p_i = \max_k(p_k)$

(2) Eșantionarea tranziției $i$ cu probabilitatea $$P(i) = p_i^{\alpha} / \sum_k p_k^{\alpha}\ \ \ \ \ \ \ \ (0<\alpha<1)$$

(3) Prioritatea tranzițiilor eșantionate este actualizată la eroarea TD: $p_i = |\delta_i| + \varepsilon$

(4) Utilizarea ponderilor de eșantionare prin importanță $$w_i = \left( \frac{1}{N} \cdot \frac{1}{P(i)} \right)^\beta\ \ \ \ \ \ \ \ (0<\beta<1)$$

(5) Creșterea progresivă a $\beta$ către 1

Implementarea PER

def __init__(self, capacity, alpha=0.6, beta=0.4, beta_increment=0.001, epsilon=0.001):
    # Initialize memory buffer
    self.memory = deque(maxlen=capacity)

    # Store parameters and initialize priorities
    self.alpha, self.beta, self.beta_increment, self.epsilon = (alpha, beta, beta_increment, epsilon)
    self.priorities = deque(maxlen=capacity)


...

Implementarea PER

...

def push(self, state, action, reward, next_state, done):
    # Append experience to memory buffer
    experience_tuple = (state, action, reward, next_state, done)
    self.memory.append(experience_tuple)

    # Set priority of new transition to maximum priority
    max_priority = max(self.priorities) if self.memory else 1.0
    self.priorities.append(max_priority)


...

Implementarea PER

def sample(self, batch_size):
    priorities = np.array(self.priorities)
    # Calculate sampling probabilities
    probabilities = priorities**self.alpha / np.sum(priorities**self.alpha)

    # Randomly select sampled indices
    indices = np.random.choice(len(self.memory), batch_size, p=probabilities)

    # Calculate weights
    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]))

    # Return tensors
    states = torch.tensor(states, dtype=torch.float32)
    ... # Repeat for rewards, next_states, dones, weights
    actions = torch.tensor(actions, dtype=torch.long).unsqueeze(1)    
    return (states, actions, rewards, next_states, dones, indices, weights)

Implementarea PER

...

def update_priorities(self, indices, td_errors: torch.Tensor):
    # Update priorities for sampled transitions
    for idx, td_error in zip(indices, td_errors.abs()):
        self.priorities[idx] = abs(td_error.item()) + self.epsilon


def increase_beta(self):
    # Increment beta towards 1
    self.beta = min(1.0, self.beta + self.beta_increment)

PER în bucla de antrenament DQN

În codul anterior buclei:

buffer = PrioritizedReplayBuffer(capacity)

La începutul fiecărui episod:
```
buffer.increase_beta()
```

3. La fiecare pas:

# After selecting an action
buffer.push(state, action, reward, 
            next_state, done)
...

# Before calculating the TD errors:
replay_buffer.sample(batch_size)
...

# After calculating the TD errors
buffer.update_priorities(indices, td_errors)

loss = torch.sum(weights * (td_errors ** 2))

PER în acțiune: Cartpole

100 de rulări de antrenament în mediul Cartpole:

cu Redare Prioritizată a Experienței
cu Redare Uniformă a Experienței

Învățare mai rapidă și performanță mai bună cu PER față de redarea uniformă

Curbele de învățare arată că PER învață mai rapid

După 100 de epoci: Cartpole, instabil după 100 de epoci

După 400 de epoci: Cartpole, stabil după 400 de epoci

PER în acțiune: medii Atari

Îmbunătățire semnificativă a performanței cu PER în mediile Atari

Diagramă cu bare comparând performanța oamenilor, DQN, DDQN, Dueling DDQN, Prioritized DDQN și Prioritized Dueling DQN. Primele patru sunt identice cu diagrama din lecția anterioară despre Dueling DQN. Ultima arată că introducerea Redării Prioritizate a Experienței îmbunătățește performanța DDQN.

¹ https://arxiv.org/abs/2303.11634

Să exersăm!

Deep Reinforcement Learning în Python