Cette page a été traduite par l'API Cloud Translation.

Environnements

Copyright 2021 Les auteurs TF-Agents.

Voir sur TensorFlow.org

Exécuter dans Google Colab

Voir la source sur GitHub

Télécharger le cahier

introduction

Le but de l'apprentissage par renforcement (RL) est de concevoir des agents qui apprennent en interagissant avec un environnement. Dans le paramètre RL standard, l'agent reçoit une observation à chaque pas de temps et choisit une action. L'action est appliquée à l'environnement et l'environnement renvoie une récompense et une nouvelle observation. L'agent forme une politique pour choisir des actions afin de maximiser la somme des récompenses, également appelée retour.

Dans TF-Agents, les environnements peuvent être implémentés en Python ou TensorFlow. Les environnements Python sont généralement plus faciles à implémenter, à comprendre et à déboguer, mais les environnements TensorFlow sont plus efficaces et permettent une parallélisation naturelle. Le workflow le plus courant consiste à implémenter un environnement en Python et à utiliser l'un de nos wrappers pour le convertir automatiquement en TensorFlow.

Voyons d'abord les environnements Python. Les environnements TensorFlow suivent une API très similaire.

Installer

Si vous n'avez pas encore installé tf-agents ou gym, exécutez :

pip install "gym>=0.21.0"
pip install tf-agents

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import abc
import tensorflow as tf
import numpy as np

from tf_agents.environments import py_environment
from tf_agents.environments import tf_environment
from tf_agents.environments import tf_py_environment
from tf_agents.environments import utils
from tf_agents.specs import array_spec
from tf_agents.environments import wrappers
from tf_agents.environments import suite_gym
from tf_agents.trajectories import time_step as ts

Environnements Python

Environnements ont un python step(action) -> next_time_step procédé qui applique une action de l'environnement, et renvoie les informations suivantes concernant l'étape suivante:

observation : Ceci est la partie de l'état de l' environnement que l'agent peut observer de choisir ses actions à l'étape suivante.
reward : L'agent apprend à maximiser la somme de ces récompenses à travers plusieurs étapes.
step_type : Les interactions avec l'environnement font généralement partie d'une séquence / épisode. par exemple plusieurs coups dans une partie d'échecs. step_type peut être soit FIRST , MID ou LAST pour indiquer si cette étape de temps est la première, intermédiaire ou dernière étape dans une séquence.
discount : Ceci est un flottant représentant la quantité de pondérer la récompense à la prochaine étape par rapport à la récompense à l'étape de l' heure actuelle.

Ceux - ci sont regroupés en un tuple nommé TimeStep(step_type, reward, discount, observation) .

L'interface que tous les environnements Python doivent mettre en œuvre est dans des environments/py_environment.PyEnvironment . Les principales méthodes sont :

class PyEnvironment(object):

  def reset(self):
    """Return initial_time_step."""
    self._current_time_step = self._reset()
    return self._current_time_step

  def step(self, action):
    """Apply action and return new time_step."""
    if self._current_time_step is None:
        return self.reset()
    self._current_time_step = self._step(action)
    return self._current_time_step

  def current_time_step(self):
    return self._current_time_step

  def time_step_spec(self):
    """Return time_step_spec."""

  @abc.abstractmethod
  def observation_spec(self):
    """Return observation_spec."""

  @abc.abstractmethod
  def action_spec(self):
    """Return action_spec."""

  @abc.abstractmethod
  def _reset(self):
    """Return initial_time_step."""

  @abc.abstractmethod
  def _step(self, action):
    """Apply action and return new time_step."""

En plus de l' step() méthode, les environnements fournissent également une reset() à TimeStep reset() méthode qui commence une nouvelle séquence et fournit un premier TimeStep . Il ne faut pas appeler la reset méthode explicitement. Nous supposons que les environnements se réinitialisent automatiquement, soit lorsqu'ils arrivent à la fin d'un épisode, soit lorsque step() est appelé la première fois.

Notez que les sous - classes ne mettent pas en oeuvre l' step() ou reset() directement. Ils remplacent plutôt le _step() et _reset() méthodes. Les pas de temps de retour ces méthodes seront mises en cache et exposées par current_time_step() .

Le observation_spec et les action_spec méthodes retournent un nid de (Bounded)ArraySpecs qui décrivent le nom, la forme, le type de données et les plages des observations et des mesures respectivement.

Dans TF-Agents, nous nous référons à plusieurs reprises aux nids qui sont définis comme n'importe quelle structure arborescente composée de listes, de tuples, de tuples nommés ou de dictionnaires. Ceux-ci peuvent être arbitrairement composés pour maintenir la structure des observations et des actions. Nous avons trouvé cela très utile pour les environnements plus complexes où vous avez de nombreuses observations et actions.

Utilisation d'environnements standard

TF Agents a intégré dans des emballages pour de nombreux environnements standards comme le OpenAI Gym, DeepMind contrôle et Atari, afin qu'ils suivent notre py_environment.PyEnvironment interface. Ces environnements enveloppés peuvent être facilement chargés à l'aide de nos suites d'environnement. Chargeons l'environnement CartPole depuis le gymnase OpenAI et regardons l'action et time_step_spec.

environment = suite_gym.load('CartPole-v0')
print('action_spec:', environment.action_spec())
print('time_step_spec.observation:', environment.time_step_spec().observation)
print('time_step_spec.step_type:', environment.time_step_spec().step_type)
print('time_step_spec.discount:', environment.time_step_spec().discount)
print('time_step_spec.reward:', environment.time_step_spec().reward)

action_spec: BoundedArraySpec(shape=(), dtype=dtype('int64'), name='action', minimum=0, maximum=1)
time_step_spec.observation: BoundedArraySpec(shape=(4,), dtype=dtype('float32'), name='observation', minimum=[-4.8000002e+00 -3.4028235e+38 -4.1887903e-01 -3.4028235e+38], maximum=[4.8000002e+00 3.4028235e+38 4.1887903e-01 3.4028235e+38])
time_step_spec.step_type: ArraySpec(shape=(), dtype=dtype('int32'), name='step_type')
time_step_spec.discount: BoundedArraySpec(shape=(), dtype=dtype('float32'), name='discount', minimum=0.0, maximum=1.0)
time_step_spec.reward: ArraySpec(shape=(), dtype=dtype('float32'), name='reward')

Nous voyons donc que l'environnement prévoit des actions de type int64 dans [0, 1] et renvoie TimeSteps où les observations sont un float32 vecteur de longueur 4 et le facteur de réduction est un float32 dans [0.0, 1.0]. Maintenant, nous allons essayer de prendre une action fixe (1,) , pour un épisode.

action = np.array(1, dtype=np.int32)
time_step = environment.reset()
print(time_step)
while not time_step.is_last():
  time_step = environment.step(action)
  print(time_step)

TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.0138565 , -0.03582913,  0.04861612, -0.03755046], dtype=float32),
 'reward': array(0., dtype=float32),
 'step_type': array(0, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.01313992,  0.15856317,  0.0478651 , -0.3145069 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.01631118,  0.35297176,  0.04157497, -0.5917188 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.02337062,  0.54748774,  0.02974059, -0.87102115], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.03432037,  0.74219286,  0.01232017, -1.1542072 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.04916423,  0.93715197, -0.01076398, -1.4430016 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.06790727,  1.1324048 , -0.03962401, -1.7390285 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.09055536,  1.327955  , -0.07440457, -2.04377   ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.11711447,  1.523758  , -0.11527998, -2.3585167 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([ 0.14758962,  1.7197047 , -0.16245031, -2.6843033 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([ 0.18198372,  1.9156038 , -0.21613638, -3.0218334 ], dtype=float32),
 'reward': array(1., dtype=float32),
 'step_type': array(2, dtype=int32)})

Créer votre propre environnement Python

Pour de nombreux clients, un cas d'utilisation courant consiste à appliquer l'un des agents standard (voir agents/) dans TF-Agents à leur problème. Pour ce faire, ils doivent formuler leur problème comme un environnement. Voyons donc comment implémenter un environnement en Python.

Disons que nous voulons former un agent à jouer au jeu de cartes suivant (inspiré du Black Jack) :

Le jeu se joue à l'aide d'un jeu de cartes infini numéroté de 1 à 10.
À chaque tour, l'agent peut faire 2 choses : obtenir une nouvelle carte aléatoire ou arrêter le tour en cours.
Le but est d'obtenir la somme de vos cartes aussi proche que possible de 21 à la fin du tour, sans dépasser.

Un environnement qui représente le jeu pourrait ressembler à ceci :

Actions : Nous avons 2 actions. Action 0 : obtenez une nouvelle carte, et Action 1 : terminez le tour en cours.
Observations : Somme des cartes du tour en cours.
Récompense : L'objectif est de s'approcher le plus possible de 21 sans dépasser, nous pouvons donc y parvenir en utilisant la récompense suivante à la fin du tour : sum_of_cards - 21 if sum_of_cards <= 21, else -21

class CardGameEnv(py_environment.PyEnvironment):

  def __init__(self):
    self._action_spec = array_spec.BoundedArraySpec(
        shape=(), dtype=np.int32, minimum=0, maximum=1, name='action')
    self._observation_spec = array_spec.BoundedArraySpec(
        shape=(1,), dtype=np.int32, minimum=0, name='observation')
    self._state = 0
    self._episode_ended = False

  def action_spec(self):
    return self._action_spec

  def observation_spec(self):
    return self._observation_spec

  def _reset(self):
    self._state = 0
    self._episode_ended = False
    return ts.restart(np.array([self._state], dtype=np.int32))

  def _step(self, action):

    if self._episode_ended:
      # The last action ended the episode. Ignore the current action and start
      # a new episode.
      return self.reset()

    # Make sure episodes don't go on forever.
    if action == 1:
      self._episode_ended = True
    elif action == 0:
      new_card = np.random.randint(1, 11)
      self._state += new_card
    else:
      raise ValueError('`action` should be 0 or 1.')

    if self._episode_ended or self._state >= 21:
      reward = self._state - 21 if self._state <= 21 else -21
      return ts.termination(np.array([self._state], dtype=np.int32), reward)
    else:
      return ts.transition(
          np.array([self._state], dtype=np.int32), reward=0.0, discount=1.0)

Assurons-nous que nous avons tout fait correctement en définissant l'environnement ci-dessus. Lors de la création de votre propre environnement, vous devez vous assurer que les observations et les time_steps générés suivent les formes et les types corrects tels que définis dans vos spécifications. Ceux-ci sont utilisés pour générer le graphique TensorFlow et, en tant que tels, peuvent créer des problèmes difficiles à déboguer si nous nous trompons.

Pour valider notre environnement, nous utiliserons une politique aléatoire pour générer des actions et nous itérerons sur 5 épisodes pour nous assurer que les choses fonctionnent comme prévu. Une erreur est générée si nous recevons un time_step qui ne respecte pas les spécifications de l'environnement.

environment = CardGameEnv()
utils.validate_py_environment(environment, episodes=5)

Maintenant que nous savons que l'environnement fonctionne comme prévu, exécutons cet environnement en utilisant une politique fixe : demandez 3 cartes, puis terminez le tour.

get_new_card_action = np.array(0, dtype=np.int32)
end_round_action = np.array(1, dtype=np.int32)

environment = CardGameEnv()
time_step = environment.reset()
print(time_step)
cumulative_reward = time_step.reward

for _ in range(3):
  time_step = environment.step(get_new_card_action)
  print(time_step)
  cumulative_reward += time_step.reward

time_step = environment.step(end_round_action)
print(time_step)
cumulative_reward += time_step.reward
print('Final Reward = ', cumulative_reward)

TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([0], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(0, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([9], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(1., dtype=float32),
 'observation': array([12], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(1, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([21], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(2, dtype=int32)})
TimeStep(
{'discount': array(0., dtype=float32),
 'observation': array([21], dtype=int32),
 'reward': array(0., dtype=float32),
 'step_type': array(2, dtype=int32)})
Final Reward =  0.0

Emballages d'environnement

Un wrapper d'environnement prend un environnement Python et renvoie une version modifiée de l'environnement. L'environnement d' origine et de l'environnement modifié sont des instances de py_environment.PyEnvironment et de multiples emballages peuvent être enchaînées.

Certains emballages de communes se trouvent dans des environments/wrappers.py . Par exemple:

ActionDiscretizeWrapper : convertit un professionnel Espace d'action continue sur un espace d'action discrète.
RunStats : Capture exécuter les statistiques de l'environnement telles que le nombre de mesures prises, le nombre d'épisodes etc. terminé
TimeLimit : après l'épisode Met fin à un nombre fixe d'étapes.

Exemple 1 : Action Discrétiser Wrapper

InvertedPendulum est un environnement PyBullet qui accepte des actions continues dans la plage [-2, 2] . Si nous voulons former un agent d'action discret tel que DQN sur cet environnement, nous devons discrétiser (quantifier) l'espace d'action. C'est exactement ce que le ActionDiscretizeWrapper fait. Comparez le action_spec avant et après l' emballage:

env = suite_gym.load('Pendulum-v1')
print('Action Spec:', env.action_spec())

discrete_action_env = wrappers.ActionDiscretizeWrapper(env, num_actions=5)
print('Discretized Action Spec:', discrete_action_env.action_spec())

Action Spec: BoundedArraySpec(shape=(1,), dtype=dtype('float32'), name='action', minimum=-2.0, maximum=2.0)
Discretized Action Spec: BoundedArraySpec(shape=(), dtype=dtype('int32'), name='action', minimum=0, maximum=4)

Le enveloppé discrete_action_env est une instance de py_environment.PyEnvironment et peut être traité comme un environnement Python régulier.

Environnements TensorFlow

L'interface pour les environnements de TF est défini dans des environments/tf_environment.TFEnvironment et ressemble beaucoup aux environnements Python. Les environnements TF diffèrent des environnements Python de plusieurs manières :

Ils génèrent des objets tenseurs au lieu de tableaux
Les environnements TF ajoutent une dimension de lot aux tenseurs générés par rapport aux spécifications.

La conversion des environnements Python en TFEnvs permet à tensorflow de paralléliser les opérations. Par exemple, on pourrait définir une collect_experience_op qui collecte des données de l'environnement et ajoute à un replay_buffer et un train_op qui lit la replay_buffer et forme l'agent, et de les exécuter en parallèle naturellement dans tensorflow.

class TFEnvironment(object):

  def time_step_spec(self):
    """Describes the `TimeStep` tensors returned by `step()`."""

  def observation_spec(self):
    """Defines the `TensorSpec` of observations provided by the environment."""

  def action_spec(self):
    """Describes the TensorSpecs of the action expected by `step(action)`."""

  def reset(self):
    """Returns the current `TimeStep` after resetting the Environment."""
    return self._reset()

  def current_time_step(self):
    """Returns the current `TimeStep`."""
    return self._current_time_step()

  def step(self, action):
    """Applies the action and returns the new `TimeStep`."""
    return self._step(action)

  @abc.abstractmethod
  def _reset(self):
    """Returns the current `TimeStep` after resetting the Environment."""

  @abc.abstractmethod
  def _current_time_step(self):
    """Returns the current `TimeStep`."""

  @abc.abstractmethod
  def _step(self, action):
    """Applies the action and returns the new `TimeStep`."""

Le current_time_step() méthode renvoie le time_step courant et initialise l'environnement en cas de besoin.

Les reset() forces de procédé un reset de l'environnement et retourne le CURRENT_STEP.

Si l' action ne dépend pas de la précédente time_step un tf.control_dependency est nécessaire dans le Graph en mode.

Pour l' instant, nous regardons comment TFEnvironments sont créés.

Créer votre propre environnement TensorFlow

C'est plus compliqué que de créer des environnements en Python, nous ne le couvrirons donc pas dans cette collaboration. Un exemple est disponible ici . Le cas d'utilisation plus courante consiste à mettre en œuvre votre environnement en Python et l' envelopper dans tensorflow en utilisant notre TFPyEnvironment emballage (voir ci - dessous).

Envelopper un environnement Python dans TensorFlow

Nous pouvons facilement envelopper tout environnement Python dans un environnement tensorflow utilisant le TFPyEnvironment emballage.

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

print(isinstance(tf_env, tf_environment.TFEnvironment))
print("TimeStep Specs:", tf_env.time_step_spec())
print("Action Specs:", tf_env.action_spec())

True
TimeStep Specs: TimeStep(
{'discount': BoundedTensorSpec(shape=(), dtype=tf.float32, name='discount', minimum=array(0., dtype=float32), maximum=array(1., dtype=float32)),
 'observation': BoundedTensorSpec(shape=(4,), dtype=tf.float32, name='observation', minimum=array([-4.8000002e+00, -3.4028235e+38, -4.1887903e-01, -3.4028235e+38],
      dtype=float32), maximum=array([4.8000002e+00, 3.4028235e+38, 4.1887903e-01, 3.4028235e+38],
      dtype=float32)),
 'reward': TensorSpec(shape=(), dtype=tf.float32, name='reward'),
 'step_type': TensorSpec(shape=(), dtype=tf.int32, name='step_type')})
Action Specs: BoundedTensorSpec(shape=(), dtype=tf.int64, name='action', minimum=array(0), maximum=array(1))

Notez que les spécifications sont désormais de type: (Bounded)TensorSpec .

Exemples d'utilisation

Exemple simple

env = suite_gym.load('CartPole-v0')

tf_env = tf_py_environment.TFPyEnvironment(env)
# reset() creates the initial time_step after resetting the environment.
time_step = tf_env.reset()
num_steps = 3
transitions = []
reward = 0
for i in range(num_steps):
  action = tf.constant([i % 2])
  # applies the action and returns the new TimeStep.
  next_time_step = tf_env.step(action)
  transitions.append([time_step, action, next_time_step])
  reward += next_time_step.reward
  time_step = next_time_step

np_transitions = tf.nest.map_structure(lambda x: x.numpy(), transitions)
print('\n'.join(map(str, np_transitions)))
print('Total reward:', reward.numpy())

[TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.0078796 , -0.04736348, -0.04966116,  0.04563603]],
      dtype=float32),
 'reward': array([0.], dtype=float32),
 'step_type': array([0], dtype=int32)}), array([0], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.00882687, -0.24173944, -0.04874843,  0.32224613]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
[TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.00882687, -0.24173944, -0.04874843,  0.32224613]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)}), array([1], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01366166, -0.04595843, -0.04230351,  0.01459712]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
[TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01366166, -0.04595843, -0.04230351,  0.01459712]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)}), array([0], dtype=int32), TimeStep(
{'discount': array([1.], dtype=float32),
 'observation': array([[-0.01458083, -0.24044897, -0.04201157,  0.2936384 ]],
      dtype=float32),
 'reward': array([1.], dtype=float32),
 'step_type': array([1], dtype=int32)})]
Total reward: [3.]

Épisodes entiers

env = suite_gym.load('CartPole-v0')
tf_env = tf_py_environment.TFPyEnvironment(env)

time_step = tf_env.reset()
rewards = []
steps = []
num_episodes = 5

for _ in range(num_episodes):
  episode_reward = 0
  episode_steps = 0
  while not time_step.is_last():
    action = tf.random.uniform([1], 0, 2, dtype=tf.int32)
    time_step = tf_env.step(action)
    episode_steps += 1
    episode_reward += time_step.reward.numpy()
  rewards.append(episode_reward)
  steps.append(episode_steps)
  time_step = tf_env.reset()

num_steps = np.sum(steps)
avg_length = np.mean(steps)
avg_reward = np.mean(rewards)

print('num_episodes:', num_episodes, 'num_steps:', num_steps)
print('avg_length', avg_length, 'avg_reward:', avg_reward)

num_episodes: 5 num_steps: 131
avg_length 26.2 avg_reward: 26.2