[PYTHON] [Lernen stärken] Ich habe R2D3 (Keras-RL) implementiert / erklärt.

Es scheint, dass die nächste Methode von R2D2 namens R2D3 angekündigt wurde. Ich war neugierig und habe es umgesetzt.

Ganzer Code

Der in diesem Artikel erstellte Code ist unten aufgeführt.

Inhaltsverzeichnis

Einführung

R2D3 ist eine sogenannte Verstärkungslernmethode der DQN-Serie. Die technischen Erklärungen bis zu diesem Punkt werden in der folgenden Reihe erläutert. Nehmen Sie Kontakt mit uns auf.

Über R2D3

Mit der von Google DeepMind im September 2019 angekündigten erweiterten Lernmethode Grob gesagt ist es eine Methode, die R2D2 und DQfD kombiniert.

Grob gesagt lernt DQfD besser (DQN-Basis), indem es sich auf das Spiel (Demonstration) eines guten Menschen bezieht, und Letztendlich ist es eine Möglichkeit, eine bessere Leistung als eine Demonstration zu erlernen.

Übrigens scheint der nicht abgekürzte Name Recurrent Replay Distributed DQN from Demonstrations (R2D3) zu sein.

·Referenz

Gesamtfluss des Lernens

20191012205515.png

Das Obige ist die Gesamtansicht von R2D3. (* Zitiert aus dem Papier) Die rechte Seite der Figur (lila und blauer Teil) entspricht R2D2, der Unterschied ist der rote Teil auf der linken Seite.

Als Voraussetzung wird die Demo-Wiedergabe zunächst mit Referenzspieldaten im Voraus geliefert.

Bis R2D2 wurden die für das Training verwendeten Stapeldaten für die Stapelgröße basierend auf den Agentenwiedergabedaten erstellt. R2D3 erstellt diese Batchdaten aus der Demo-Wiedergabe und der Agentenwiedergabe gemäß dem Demo-Verhältnis.

Über das Demo-Verhältnis

In der Arbeit haben wir 1/16, 1/32, 1/64 1/128, 1/256 mit festen Werten verglichen. Es wurde festgestellt, dass 1/256 der genaueste Teil der Aufgabe ist.

Von hier an ist es meine Meinung, aber im eigentlichen menschlichen Sinne ist das Demo-Spiel zunächst hilfreich, aber wenn man sich daran gewöhnt, sieht man es nicht. In meiner Implementierung habe ich hier das Demo-Verhältnis implementiert, damit es geglüht werden kann. (Das Tempern entspricht dem festen Fall, wenn Sie die Einstellung ändern.)

Implementierung einer Demonstrationsumgebung

Zunächst müssen wir die Daten für die Demonstration vorbereiten. Ich habe es unter Bezugnahme auf den von OpenAI unten bereitgestellten Code für das manuelle Spielen erstellt.

Demo-Play-Datenstruktur

Die zu speichernde Datenstruktur ist zweigeteilt, eine zum Lernen und die andere zur Wiedergabe. Sie hat die folgende Form.

・ Zum Lernen (für jeden Frame speichern)

Name Inhalt
action Aktion
observation Status
reward Belohnung
done Ob es fertig ist

・ Zur Wiedergabe (allgemeine Informationen)

Name Inhalt
episode Episodennummer
rgb_size Bildgröße
states Array jeder Frame-Information(Enthält die folgenden Informationen)

・ Für die Wiedergabe (für jedes Bild speichern)

Name Inhalt
step Rahmennummer
reward_total Aktuelle Gesamtbelohnung
info Frame-Info-Informationen(gym)
rgb Bild

Fügen Sie dem Speicher Demo-Wiedergabedaten hinzu

(Der Code ist die Funktion add_memory in env_play.py.)

Wenn Sie Demo-Wiedergabedaten zum Speicher hinzufügen, müssen Sie die gleichen Schritte wie der eigentliche Agent ausführen, um sie zu speichern. (Referenz: [Erklärung zur Implementierung von [DQN (Rainbow)]](https://qiita.com/pocokhc/items/408f0f818140924ad4c4#dqnrainbow%E3%81%AE%E5%AE%9F%E8%A3%85%E8%A7 % A3% E8% AA% AC))

Es ist ein bisschen redundant, aber wir werden den gleichen Mechanismus separat erstellen und ihn dem Speicher hinzufügen. Unten ist der Ablauf mit Pseudocode. (Da es kompliziert ist, wird es im Fall von Stateful LSTM nicht beschrieben.)

add_memory


def add_memory(episode_file, memory, agent):
・ Folge_Holen Sie sich Demo-Play-Informationen aus der Datei

  #Erstellen Sie Variablen für die empirische Datenerstellung
  recent_actions =Array mit der Anzahl der zu speichernden Aktionen
  recent_rewards =Eine Reihe von Belohnungen zum Speichern
  recent_rewards_multistep =Zur mehrstufigen Berechnung
  recent_observations =Eine Reihe von Situationen, die gespeichert werden sollen

für Schritt in Folge:
    observation =Rahmeninformationen[step]["observation"]
    action      =Rahmeninformationen[step]["action"]
    reward      =Rahmeninformationen[step]["reward"]

    #Status hinzufügen
    recent_observations.pop(0)
    recent_observations.append(observation)

    #Erstellen Sie eine Erfahrung
    exp = (
      recent_observations[:agent.input_sequence],  #Vorheriger Status
      recent_actions[0],                           #Aktion im vorherigen Zustand
      recent_rewards_multistep,                    #Belohnung
      recent_observations[-agent.input_sequence:]) #Nächster Zustand
    )
    
    #Erfahrung in Erinnerung behalten
    memory.add(exp)

    #Fügen Sie Aktion und Belohnung hinzu
    recent_actions.pop(0)
    recent_actions.append(action)
    recent_rewards.pop(0)
    recent_rewards.append(reward)

    recent_rewards_multistep =Mehrstufige Lernberechnung

Implementierung der Spielumgebung

(Der Code ist die EpisodeSave-Klasse in env_play.py)

Dies ist die Klasse, die Sie tatsächlich spielen. Es hat die folgenden Funktionen.

Der Bildschirm ist wie folgt.

play1.PNG

Beispiel für einen Ausführungscode

Der auszuführende Code sieht wie folgt aus.

import gym
from src.env_play import EpisodeSave

def run_play():
  env = gym.make("MountainCar-v0")
  processor = None  #Wenn ja, geben Sie es beliebig an

  es = EpisodeSave(
    env,
    episode_save_dir="tmp",
    processor=processor
  )
  es.play()
  env.close()

run_play()

Spielschlüsselbindungen

Die Schlüsselbindung des Spiels kann vom Prozessor festgelegt werden. Wenn der Prozessor über eine Methode get_keys_to_action verfügt, wird diese geladen.

get_keys_to_action


import rl
class MyProcessor(rl.core.Processor):
  def get_keys_to_action(self):
    return {
      ():0,           #0, wenn nicht gedrückt
      (ord('d'),):1,  #d Taste ist 1
      (ord('a'),):2,  #Ein Schlüssel ist 2
    }

Wiedergabe gespeicherter Wiedergabedaten

(Der Code ist die EpisodeReplay-Klasse in env_play.py)

Ich habe auch einen Mechanismus zum Abspielen der von EpisodeSave gespeicherten Episode erstellt. Hauptsächlich zur Bestätigung.

Beispiel für die Codeausführung

from src.env_play import EpisodeReplay

def replay():
    r = EpisodeReplay(episode_save_dir="tmp")
    r.play()

replay()

Implementierung des DemoReplay-Speichers

Implementierung auf Rainbow

Wie üblich werden wir es aus der Rainbow-Version implementieren, die ohne parallele Verarbeitung leicht zu verstehen ist.

  1. Fügen Sie die folgenden neuen Parameter hinzu.
Name Inhalt
demo_memory Speichertyp(Ähnlich wie beim Wiedergabespeicher)
demo_episode_dir Verzeichnispfad gespeichert durch Episode Speichern oben
demo_ratio_initial Anfangsrate der Demo
demo_ratio_final Demo-Endstatusrate
demo_ratio_steps Anzahl der Schritte, um die endgültige Rate zu erreichen

demo_memory kann aus ReplayMemory, PERGreedyMemory, PERProportionalMemory, PERRankBaseMemory sowie [ReplayMemory] ausgewählt werden Ich kann es schaffen

  1. Fügen Sie dem DemoReplay-Speicher während der Initialisierungsphase Demoplay-Erfahrung hinzu.

rainbow


def __init__(self):
(Kürzung)

  # add_Demo mit Speicherfunktion_Demo-Wiedergabe zum Speicher hinzugefügt
  add_memory(demo_episode_dir, self.demo_memory, self)

  # demo_Stellen Sie Variablen für das Verhältnisglühen ein
  self.demo_ratio_initial = demo_ratio_initial
  if demo_ratio_final is None:
    self.demo_ratio_final = self.demo_ratio_initial
  else:
    self.demo_ratio_final = demo_ratio_final
  self.demo_ratio_step = (self.demo_ratio_initial - self.demo_ratio_final) / demo_ratio_steps

(Kürzung)
  1. Holen Sie sich Batch-Daten von replay_memory und demo_memory. Aktualisieren Sie dann die Priorität.

rainbow


import random

def forward(self, observation):
  #Es ist der Zeitpunkt zum Zeitpunkt des Lernens
(Kürzung)

  #Berechnen Sie das Verhältnis von Demo-Verhältnis
  ratio_demo = self.demo_ratio_initial - self.local_step * self.demo_ratio_step
  if ratio_demo < self.demo_ratio_final:
    ratio_demo = self.demo_ratio_final

  #Berechnen Sie die Anzahl der Chargen anhand des Verhältnisses
  batch_replay = 0
  batch_demo = 0
  for _ in range(self.batch_size):
    r = random.random()
    if r < ratio_demo:
      batch_demo += 1
      continue
    batch_replay += 1

  #Erstellen Sie einen Stapel basierend auf dem Verhältnis
  indexes = []
  batchs = []
  weights = []
  memory_types = []  #Speichern Sie den erfassten Speichertyp
  if batch_replay > 0:
    (i, b, w) = self.memory.sample(batch_replay, self.local_step)
    indexes.extend(i)
    batchs.extend(b)
    weights.extend(w)
    #0 ist Wiedergabe_memory
    memory_types.extend([0 for _ in range(batch_replay)])
  if batch_demo > 0:
    (i, b, w) = self.demo_memory.sample(batch_demo, self.local_step)
    indexes.extend(i)
    batchs.extend(b)
    weights.extend(w)
    #1 ist Demo_memory
    memory_types.extend([1 for _ in range(batch_demo)])

(Kürzung)

  for i in range(self.batch_size):
(Lernen)

    #Priorität aktualisieren
    if memory_types[i] == 0:
      # replay_Speicher aktualisieren
      self.memory.update(indexes[i], batchs[i], priority)
    elif memory_types[i] == 1:
      # demo_Speicher aktualisieren
      self.demo_memory.update(indexes[i], batchs[i], priority)
    else:
      assert False
  
(Kürzung)

Implementierung auf R2D3

Es kann genauso implementiert werden wie Rainbow. Implementieren Sie es einfach auf der Lernerseite.

EpisodeMemory

Dies ist meine eigene Implementierung. Ich kam auf die Idee, nachdem ich das Demo-Spiel gesehen hatte, aber wenn das Demo-Spiel sinnvoll ist, warum nicht beim Lernen? Ich dachte.

Insbesondere wird das Spiel mit der höchsten Gesamtbelohnung der Episode separat im Speicher gespeichert. Die Idee ist, dies auf die gleiche Weise wie diesmal zu einer Charge zu mischen.

Als Bild fühlt es sich an, als würde ich mich an das Stück erinnern, das zufällig erfolgreich war, und es viele Male wiederholen.

Unten ist die Implementierung.

Definition von EpisodeMemory

Erstellen Sie ein EpisodeMemory, das ReplayMemory enthält. Dies ist eine Wrapper-Klasse, die ReplayMemory pro Episode mehr Erfahrung bietet.

EpisodeMemory


class EpisodeMemory():
  def __init__(self, memory):
    self.max_reward = None
    self.memory = memory
  
  def add_episode(self, episode, total_reward):
    # max_Füge eine Episode zum Speicher hinzu, wenn die Belohnung aktualisiert wird
    if self.max_reward is None:
      self.max_reward = total_reward
    elif self.max_reward <= total_reward:  #Zum Speicher hinzufügen, auch wenn es sich in derselben Zeile befindet
      self.max_reward = total_reward
    else:
      return
    
    #Tatsächliche Speicher zusätzliche Verarbeitung
    for e in episode_recent:
      if len(e) == 5:  #Verarbeitung bei Priorität
        self.memory.add(e, e[4])
      else:
        self.memory.add(e)

Implementierung von EpisodeMemory

Implementierung auf Rainbow

  1. Zusätzliche Parameter sind wie folgt. Im Gegensatz zu Demori Play ist es kein Tempern.
Name Inhalt
episode_memory Speichertyp(Ähnlich wie beim Wiedergabespeicher)
episode_ratio Episodenerinnerungsrate
  1. Initialisierung des Episodenspeichers

rainbow


from src.memory.EpisodeMemory import EpisodeMemory

def __init__(self):
(Kürzung)

  #Wrap in EpisodeMemory-Klasse
  self.episode_memory = EpisodeMemory(episode_memory)
  self.episode_ratio = episode_ratio

(Kürzung)
  1. Speichern Sie die Erfahrung für die Episode

reset_states


#Wird zu Beginn der Episode aufgerufen
def reset_states(self):
(Kürzung)

  #Zum Speichern von Episodenerfahrung
  self.episode_exp = []
  self.total_reward = 0

  #Zur Überprüfung des Endstatus
  self.recent_terminal = False

forward


#Wird vor der Ausführung der Aktion in jedem Schritt aufgerufen
def forward(self, observation):
(Kürzung)

  #Wenn es fertig ist, Folge_Zum Speicher hinzufügen
  if self.recent_terminal:
    self.episode_memory.add_episode(self.episode_exp, self.total_reward)

(Kürzung)

  exp =(Erfahrungsdaten erstellen)
  self.memory.add(exp)   # replay_Speicher hinzufügen
  self.episode_exp.append(exp)  # episode_Erfahrung für das Gedächtnis hinzufügen

(Kürzung)

backward


#Wird aufgerufen, nachdem in jedem Schritt eine Aktion ausgeführt wurde
def backward(self, reward, terminal):
(Kürzung)

  #Berechnen Sie die Gesamterfahrung
  self.total_reward += reward

  #Exit-Status speichern
  self.recent_terminal = terminal

(Kürzung)
  1. Mischen Sie episodenspeicher mit der Erstellung von Stapeldaten

rainbow


import random

def forward(self, observation):
  #Es ist der Zeitpunkt zum Zeitpunkt des Lernens
(Kürzung)

  ratio_demo =(Berechnung des Demoverhältnisses)
  
  # episode_Wenn Speicher Speicher hat, mischen Sie ihn in einem Stapel
  if len(self.episode_memory) < self.batch_size:
    ratio_epi = 0
  else:
    ratio_epi = self.episode_ratio

  #Berechnen Sie die Anzahl der Chargen anhand des Verhältnisses
  batch_replay = 0
  batch_demo = 0
  batch_episode = 0
  for _ in range(self.batch_size):
    r = random.random()
    if r < ratio_demo:
      batch_demo += 1
      continue
    r -= ratio_demo
    if r < ratio_epi:
      batch_episode += 1
      continue
    batch_replay += 1

  #Erstellen Sie einen Stapel basierend auf dem Verhältnis
  indexes = []
  batchs = []
  weights = []
  memory_types = []  #Speichern Sie den erfassten Speichertyp
  if batch_replay > 0:
    （replay_Stapelerstellung des Speichers)
  if batch_demo > 0:
    （demo_Stapelerstellung des Speichers)
  if batch_episode > 0:
    (i, b, w) = self.episode_memory.sample(batch_episode, self.local_step)
    indexes.extend(i)
    batchs.extend(b)
    weights.extend(w)
    # episode_Speicher ist 2
    memory_types.extend([2 for _ in range(batch_episode)])

(Kürzung)

  for i in range(self.batch_size):
(Lernen)

    #Priorität aktualisieren
    if memory_types[i] == 0:
      （replay_Speicher aktualisieren)
    elif memory_types[i] == 1:
      （demo_Speicher aktualisieren)
    elif memory_types[i] == 2:
      # episode_Speicher aktualisieren
      self.episode_memory.update(indexes[i], batchs[i], priority)
    else:
      assert False
  
(Kürzung)

Implementierung auf R2D3

Es ist fast das gleiche wie die Implementierung auf Rainbow. Es werden jedoch Episodendaten für jeden Schauspieler erstellt. Es wird auf der Lernerseite verwaltet, um den Kommunikationsaufwand zwischen Prozessen zu verringern.

Learner


class Learner():
  def __init__():
(Kürzung)

    #Erstellen Sie Variablen für das Episodenmanagement für jeden Schauspieler, indem Sie Learner initialisieren
    self.episode_exp = [ [] for _ in range(self.actors_num)]
    self.total_reward = [ 0 for _ in range(self.actors_num)]

  def train(self):
(Kürzung)

    #Schauspieler → Erfahrung zum Lernenden hinzufügen
    for _ in range(self.exp_q.qsize()):
      exp = self.exp_q.get(timeout=1)

      # add memory
      self.memory.add(exp[0], exp[0][4])

      # add episode_exp
      self.total_reward[exp[1]] += exp[0][2]
      self.episode_exp[exp[1]].append(exp[0])
      
      if exp[2]:  # terminal
          self.episode_memory.add_episode(
              self.episode_exp[exp[1]],
              self.total_reward[exp[1]]
          )
          self.episode_exp[exp[1]] = []
          self.total_reward[exp[1]] = 0

(Kürzung)

Actor


class Actor():
  def forward(self, observation):
(Kürzung)

    #An den Lernenden senden
    # actor_Übergeben Sie auch Index- und Terminalinformationen
    self.exp_q.put((exp, self.actor_index, self.recent_terminal))

(Kürzung)

Vergleichsergebnis mit der herkömmlichen Methode

MountainCar

Dieses Mal werde ich es mit MountainCar versuchen.

mountaincar.gif

Mountain Car ist ein Spiel, bei dem Sie das Auto nach links und rechts bewegen, um auf die Flagge oben rechts zu zielen.

Die Belohnung ist immer -1. Kurz gesagt, je früher Sie die Flagge erreichen, desto höher ist Ihre Punktzahl.

Wenn Sie es als Q-Lernen betrachten, ist es eine Aufgabe, die Sie erst belohnen können, wenn Sie das Ziel erreicht haben (es ist nicht bekannt, ob es gut oder schlecht ist), und es ist eine ziemlich schwierige Aufgabe.

Traditionelle Ergebnisse

Processor ist ein reines MountainCar-Training, das vom Fitnessstudio ohne Definition angeboten wird. Das Protokoll wird alle 2000 Schritte erfasst. (Dies ist die Operation in Rainbow)

Die ersten 50.000 Schritte werden aufgewärmt und anschließend 100.000 Mal trainiert. ·Ergebnis

Figure_1_replay2.png

Es ist schwer zu lernen, bis das Spiel, das das Ziel erreicht hat, zu einem gewissen Grad im Speicher gespeichert ist. Die Ergebnisse beginnen mit rund 130.000 Schritten.

Ergebnisse, wenn der DemoReplay-Speicher aktiviert ist

Die anderen Parameter als der DemoReplay-Speicher sind dieselben wie die vorherigen Ergebnisse. Die Demo hat nur eine Episode unten vorbereitet.

MountainCar-v0.gif

Die Parameter des DemoReplay-Speichers sind wie folgt.

demo_memory = PERProportionalMemory(100_000, alpha=0.8)
demo_episode_dir = episode_save_dir
demo_ratio_initial = 1.0
demo_ratio_final = 1.0/512.0
demo_ratio_steps = warmup + 50_000

·Ergebnis

Figure_2_demo2.png

Die Ergebnisse liegen bereits bei rund 70.000.

Ergebnisse, wenn der Episodenspeicher aktiviert ist

Andere Parameter als EpisodeMemory sind dieselben wie die vorherigen Ergebnisse. Die Parameter von EpisodeMemory sind wie folgt.

episode_memory = PERProportionalMemory(2_000, alpha=0.8),
episode_ratio = 1.0/8.0,

·Ergebnis

Figure_3_episode3.png

Die Ergebnisse beginnen bei rund 80.000.

Bonus (sowohl der Demo-Wiederholungsspeicher als auch der Episodenspeicher sind gültig)

Figure_4_mix2.png

Bonus 2 (R2D3)

Sowohl DemoReplay-Speicher als auch EpisodeMemory sind gültig. (Die x-Achse enthält nicht die Anzahl der Aufwärmübungen.)

Figure_5_2.png

Nachwort

Es war einfacher zu implementieren als ich erwartet hatte. Ich denke, dass es eine sehr effektive Methode ist, da es beim Lernen nicht viele Aufgaben ohne Demo-Spiel gibt. Die Entwicklung des verstärkenden Lernens hat noch nicht aufgehört. Ich freue mich darauf zu sehen, was als nächstes kommt.

Recommended Posts

[Lernen stärken] Ich habe R2D3 (Keras-RL) implementiert / erklärt.
[Lernen stärken] Rache-Kommentar (Keras-RL), der versucht hat, R2D2 zu implementieren / zu erklären
[Lernen stärken] R2D2 implementiert / erklärt Revenge Hyper Parameter Explanation (Keras-RL)
Ich habe versucht, die Extreme-Lernmaschine zu implementieren
Ich habe versucht, mit PyBrain verstärkt zu lernen
[Python] Probieren Sie mit Keras-RL ganz einfach erweitertes Lernen (DQN) aus
[Stärkung des Lernens] Endlich die Menschen übertroffen! ?? Ich habe versucht, Agent57 (Keras-RL) zu erklären / zu implementieren.
[Einführung] Stärkung des Lernens
Ich habe CycleGAN (1) implementiert.
Zukünftiges Verstärkungslernen_2
Zukünftiges Verstärkungslernen_1
Ich habe ResNet implementiert!
[Mac] Ich habe versucht, das Lernen mit Open AI Baselines zu stärken
Ich möchte mit verstärkendem Lernen einen Berg besteigen
Ich untersuchte den stärkenden Lernalgorithmus des Algorithmushandels
Erweitertes Lernen 1 Python-Installation
Stärkung des Lernens 3 OpenAI-Installation
Stärkung des Lernens der dritten Zeile
Qiskit: Ich habe VQE implementiert
Ich habe versucht, tief zu lernen
[Lernen stärken] Banditenaufgabe
Python + Unity Enhanced Learning (Lernen)
Stärkung des Lernens 1 Einführungsausgabe