Relecture d'expériences priorisée

Deep Reinforcement Learning en Python

Timothée Carayol

Principal Machine Learning Engineer, Komment

Toutes les expériences ne se valent pas

Experience Replay :
- L'échantillonnage uniforme peut ignorer des souvenirs importants
Prioritized Experience Replay :
- Attribuer une priorité à chaque expérience, selon les erreurs TD
- Mettre l'accent sur les expériences au fort potentiel d'apprentissage

Des étudiant·e·s étudient à la bibliothèque

Prioritized Experience Replay (PER)

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

      # Lui donner la priorité la plus élevée (1)

    # Échantillonner un lot d'anciennes transitions

      # Selon la priorité (2)

    # Calculer les erreurs TD pour le lot

    # Calculer la perte et mettre à jour le réseau Q

      # Utiliser des poids d'échantillonnage par importance (4)

    # Mettre à jour la priorité des transitions échantillonnées (3)

# Augmenter l'échantillonnage par importance au fil du temps. (5)

(1) Les nouvelles transitions sont ajoutées avec la priorité maximale $p_i = \max_k(p_k)$

(2) Échantillonner la transition $i$ avec la probabilité $$P(i) = p_i^{\alpha} / \sum_k p_k^{\alpha}\ \ \ \ \ \ \ \ (0<\alpha<1)$$

(3) Les transitions échantillonnées voient leur priorité mise à jour avec leur erreur TD : $p_i = |\delta_i| + \varepsilon$

(4) Utiliser des poids d'échantillonnage par importance $$w_i = \left( \frac{1}{N} \cdot \frac{1}{P(i)} \right)^\beta\ \ \ \ \ \ \ \ (0<\beta<1)$$

(5) Augmenter progressivement $\beta$ vers 1

Implémenter PER

def __init__(self, capacity, alpha=0.6, beta=0.4, beta_increment=0.001, epsilon=0.001):
    # Initialiser le tampon de mémoire
    self.memory = deque(maxlen=capacity)

    # Stocker les paramètres et initialiser les priorités
    self.alpha, self.beta, self.beta_increment, self.epsilon = (alpha, beta, beta_increment, epsilon)
    self.priorities = deque(maxlen=capacity)


...

Implémenter PER

...

def push(self, state, action, reward, next_state, done):
    # Ajouter l'expérience au tampon de mémoire
    experience_tuple = (state, action, reward, next_state, done)
    self.memory.append(experience_tuple)

    # Définir la priorité de la nouvelle transition à la priorité maximale
    max_priority = max(self.priorities) if self.memory else 1.0
    self.priorities.append(max_priority)


...

Implémenter PER

def sample(self, batch_size):
    priorities = np.array(self.priorities)
    # Calculer les probabilités d'échantillonnage
    probabilities = priorities**self.alpha / np.sum(priorities**self.alpha)

    # Sélectionner aléatoirement les indices échantillonnés
    indices = np.random.choice(len(self.memory), batch_size, p=probabilities)

    # Calculer les poids
    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]))

    # Retourner des tenseurs
    states = torch.tensor(states, dtype=torch.float32)
    ... # Répéter pour rewards, next_states, dones, weights
    actions = torch.tensor(actions, dtype=torch.long).unsqueeze(1)    
    return (states, actions, rewards, next_states, dones, indices, weights)

Implémenter PER

...

def update_priorities(self, indices, td_errors: torch.Tensor):
    # Mettre à jour les priorités des transitions échantillonnées
    for idx, td_error in zip(indices, td_errors.abs()):
        self.priorities[idx] = abs(td_error.item()) + self.epsilon


def increase_beta(self):
    # Incrémenter beta vers 1
    self.beta = min(1.0, self.beta + self.beta_increment)

PER dans la boucle d'entraînement DQN

Dans le code avant la boucle :

buffer = PrioritizedReplayBuffer(capacity)

Au début de chaque épisode :
```
buffer.increase_beta()
```

À chaque étape :

# Après avoir choisi une action
buffer.push(state, action, reward, 
            next_state, done)
...

# Avant de calculer les erreurs TD :
replay_buffer.sample(batch_size)
...

# Après avoir calculé les erreurs TD
buffer.update_priorities(indices, td_errors)

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

PER en action : Cartpole

100 entraînements dans l'environnement Cartpole :

avec Prioritized Experience Replay
avec Experience Replay uniforme

Apprentissage plus rapide et meilleure performance avec PER qu'avec l'uniforme

Courbes d'apprentissage : PER apprend plus vite

Après 100 époques : Cartpole, instable après 100 époques

Après 400 époques : Cartpole, stable après 400 époques

PER en action : environnements Atari

Gain de performance notable avec PER dans les environnements Atari

Diagramme à barres comparant la performance des humains, DQN, DDQN, Dueling DDQN, Prioritized DDQN et Prioritized Dueling DQN. Les quatre premiers sont identiques au diagramme à barres de la leçon précédente sur Dueling DQN. Le dernier montre que l'ajout de Prioritized Experience Replay améliore la performance de DDQN.

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

Passons à la pratique !

Deep Reinforcement Learning en Python