LoslegenKostenlos loslegen

Priorisierter Erfahrungswiedergabepuffer

Du stellst die Klasse PrioritizedExperienceReplay vor, eine Datenstruktur, die du später verwenden wirst, um DQN mit Prioritized Experience Replay zu implementieren.

PrioritizedExperienceReplay ist eine Verfeinerung der Klasse ExperienceReplay, die du bisher zum Trainieren deiner DQN Agenten verwendet hast. Ein priorisierter Erfahrungswiedergabepuffer stellt sicher, dass die Übergänge, die aus ihm entnommen werden, für den Agenten wertvoller sind als bei einer gleichmäßigen Stichprobe.

Implementiere erst einmal die Methoden .__init__(), .push(), .update_priorities(), .increase_beta() und .__len__(). Die letzte Methode, .sample(), wird der Schwerpunkt der nächsten Übung sein.

Diese Übung ist Teil des Kurses

Deep Reinforcement Learning in Python

Kurs anzeigen

Anleitung zur Übung

  • In .push() initialisierst du die Priorität des Übergangs mit der maximalen Priorität im Puffer (oder 1, wenn der Puffer leer ist).
  • Setze in .update_priorities() die Priorität auf den absoluten Wert des entsprechenden TD Fehlers; füge self.epsilon hinzu, um Randfälle abzudecken.
  • In .increase_beta() erhöhe beta um self.beta_increment; stelle sicher, dass beta nie größer als 1 ist.

Interaktive Übung

Vervollständige den Beispielcode, um diese Übung erfolgreich abzuschließen.

class PrioritizedReplayBuffer:
    def __init__(
        self, capacity, alpha=0.6, beta=0.4, beta_increment=0.001, epsilon=0.01
    ):
        self.memory = deque(maxlen=capacity)
        self.alpha, self.beta, self.beta_increment, self.epsilon = (alpha, beta, beta_increment, epsilon)
        self.priorities = deque(maxlen=capacity)

    def push(self, state, action, reward, next_state, done):
        experience_tuple = (state, action, reward, next_state, done)
        # Initialize the transition's priority
        max_priority = ____
        self.memory.append(experience_tuple)
        self.priorities.append(max_priority)
    
    def update_priorities(self, indices, td_errors):
        for idx, td_error in zip(indices, td_errors.tolist()):
            # Update the transition's priority
            self.priorities[idx] = ____

    def increase_beta(self):
        # Increase beta if less than 1
        self.beta = ____

    def __len__(self):
        return len(self.memory)
      
buffer = PrioritizedReplayBuffer(capacity=3)
buffer.push(state=[1,3], action=2, reward=1, next_state=[2,4], done=False)
print("Transition in memory buffer:", buffer.memory)
print("Priority buffer:", buffer.priorities)
Code bearbeiten und ausführen