Questa pagina è stata tradotta dall'API Cloud Translation.
Switch to English

Ambienti

Visualizza su TensorFlow.org Esegui in Google Colab Visualizza sorgente su GitHub Scarica notebook

introduzione

L'obiettivo del Reinforcement Learning (RL) è progettare agenti che apprendono interagendo con un ambiente. Nell'impostazione RL standard, l'agente riceve un'osservazione a ogni fase temporale e sceglie un'azione. L'azione si applica all'ambiente e l'ambiente restituisce una ricompensa e una nuova osservazione. L'agente addestra una politica per scegliere le azioni per massimizzare la somma dei premi, nota anche come ritorno.

In TF-Agents, gli ambienti possono essere implementati in Python o TensorFlow. Gli ambienti Python sono generalmente più facili da implementare, comprendere ed eseguire il debug, ma gli ambienti TensorFlow sono più efficienti e consentono la parallelizzazione naturale. Il flusso di lavoro più comune consiste nell'implementare un ambiente in Python e utilizzare uno dei nostri wrapper per convertirlo automaticamente in TensorFlow.

Diamo prima un'occhiata agli ambienti Python. Gli ambienti TensorFlow seguono un'API molto simile.

Impostare

Se non hai ancora installato tf-agent o gym, esegui:

pip install -q tf-agents
pip install -q 'gym==0.10.11'
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

tf.compat.v1.enable_v2_behavior()

Ambienti Python

Gli ambienti Python hanno un metodo step(action) -> next_time_step che applica un'azione all'ambiente e restituisce le seguenti informazioni sul passaggio successivo:

  1. observation : questa è la parte dello stato ambientale che l'agente può osservare per scegliere le sue azioni nella fase successiva.
  2. reward : l'agente sta imparando a massimizzare la somma di queste ricompense attraverso più passaggi.
  3. step_type : le interazioni con l'ambiente fanno solitamente parte di una sequenza / episodio. es. più mosse in una partita a scacchi. step_type può essere FIRST , MID o LAST per indicare se questo passo temporale è il primo, intermedio o ultimo passo di una sequenza.
  4. discount : è un valore fluttuante che rappresenta quanto pesare la ricompensa nella fase temporale successiva rispetto alla ricompensa nella fase temporale corrente.

Questi sono raggruppati in una tupla TimeStep(step_type, reward, discount, observation) denominata TimeStep(step_type, reward, discount, observation) .

L'interfaccia che tutti gli ambienti python devono implementare è in environments/py_environment.PyEnvironment . I metodi principali sono:

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."""
    self._current_time_step = self._step(action)
    return self._current_time_step

Oltre al metodo step() , gli ambienti forniscono anche un metodo reset() che avvia una nuova sequenza e fornisce un TimeStep iniziale. Non è necessario chiamare esplicitamente il metodo di reset . Partiamo dal presupposto che gli ambienti si ripristinino automaticamente, o quando arrivano alla fine di un episodio o quando step () viene chiamato la prima volta.

Notare che le sottoclassi non implementano direttamente step() o reset() . _step() invece i _step() e _reset() . Le fasi temporali restituite da questi metodi verranno memorizzate nella cache ed esposte tramite current_time_step() .

observation_spec ei action_spec metodi restituiscono un nido di (Bounded)ArraySpecs che descrivono rispettivamente il nome, la forma, il tipo di dati e gli intervalli delle osservazioni e delle azioni.

In TF-Agent ci riferiamo ripetutamente a nidi che sono definiti come qualsiasi struttura ad albero composta da liste, tuple, tuple con nome o dizionari. Questi possono essere composti arbitrariamente per mantenere la struttura delle osservazioni e delle azioni. Lo abbiamo trovato molto utile per ambienti più complessi in cui sono presenti molte osservazioni e azioni.

Utilizzo di ambienti standard

TF Agents ha wrapper incorporati per molti ambienti standard come OpenAI Gym, DeepMind-control e Atari, in modo che seguano la nostra interfaccia py_environment.PyEnvironment . Questi ambienti avvolti possono essere facilmente caricati utilizzando le nostre suite ambientali. Carichiamo l'ambiente CartPole dalla palestra OpenAI e guardiamo l'azione e 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')

Quindi vediamo che l'ambiente si aspetta azioni di tipo int64 in [0, 1] e restituisce TimeSteps dove le osservazioni sono un vettore float32 di lunghezza 4 e il fattore di sconto è un float32 in [0.0, 1.0]. Ora proviamo a eseguire un'azione fissa (1,) per un intero episodio.

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(step_type=array(0, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([0.04087088, 0.03729887, 0.0153357 , 0.00713792], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.04161686,  0.23219757,  0.01547846, -0.28066722], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.04626081,  0.42709532,  0.00986511, -0.5684284 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.05480272,  0.6220775 , -0.00150346, -0.85798717], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.06724427,  0.8172199 , -0.0186632 , -1.1511425 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.08358867,  1.0125804 , -0.04168605, -1.4496187 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.10384028,  1.2081891 , -0.07067842, -1.7550292 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.12800406,  1.4040377 , -0.10577901, -2.0688307 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.1560848 ,  1.6000646 , -0.14715563, -2.3922687 ], dtype=float32))
TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.1880861 ,  1.7961396 , -0.19500099, -2.72631   ], dtype=float32))
TimeStep(step_type=array(2, dtype=int32), reward=array(1., dtype=float32), discount=array(0., dtype=float32), observation=array([ 0.2240089,  1.9920444, -0.2495272, -3.0715656], dtype=float32))

Creare il tuo ambiente Python

Per molti client, un caso d'uso comune è applicare uno degli agenti standard (vedere agenti /) in TF-Agents al loro problema. Per fare questo, devono inquadrare il loro problema come un ambiente. Quindi vediamo come implementare un ambiente in Python.

Supponiamo di voler addestrare un agente a giocare al seguente gioco di carte (ispirato a Black Jack):

  1. Il gioco si gioca utilizzando un mazzo infinito di carte numerate da 1 a 10.
  2. Ad ogni turno l'agente può fare 2 cose: ottenere una nuova carta casuale o interrompere il round in corso.
  3. L'obiettivo è ottenere la somma delle tue carte il più vicino possibile a 21 alla fine del round, senza andare oltre.

Un ambiente che rappresenta il gioco potrebbe assomigliare a questo:

  1. Azioni: abbiamo 2 azioni. Azione 0: prendi una nuova carta e Azione 1: termina il round corrente.
  2. Osservazioni: somma delle carte nel round corrente.
  3. Ricompensa: l'obiettivo è quello di avvicinarsi il più possibile a 21 senza andare oltre, quindi possiamo ottenere questo risultato utilizzando la seguente ricompensa alla fine del round: sum_of_cards - 21 if sum_of_cards <= 21, altrimenti -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)

Assicuriamoci di aver fatto tutto correttamente definendo l'ambiente di cui sopra. Quando crei il tuo ambiente, devi assicurarti che le osservazioni e i time_steps generati seguano le forme e i tipi corretti come definito nelle tue specifiche. Questi vengono utilizzati per generare il grafico TensorFlow e come tali possono creare problemi difficili da eseguire il debug se vengono sbagliati.

Per convalidare il nostro ambiente utilizzeremo una policy casuale per generare azioni e itereremo su 5 episodi per assicurarci che le cose funzionino come previsto. Viene generato un errore se riceviamo un time_step che non segue le specifiche dell'ambiente.

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

Ora che sappiamo che l'ambiente funziona come previsto, eseguiamo questo ambiente utilizzando una politica fissa: chiedi 3 carte e poi termina il round.

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(step_type=array(0, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([0], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([7], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([15], dtype=int32))
TimeStep(step_type=array(1, dtype=int32), reward=array(0., dtype=float32), discount=array(1., dtype=float32), observation=array([18], dtype=int32))
TimeStep(step_type=array(2, dtype=int32), reward=array(-3., dtype=float32), discount=array(0., dtype=float32), observation=array([18], dtype=int32))
Final Reward =  -3.0

Involucri ambientali

Un wrapper di ambiente accetta un ambiente Python e restituisce una versione modificata dell'ambiente. Sia l'ambiente originale che l'ambiente modificato sono istanze di py_environment.PyEnvironment e più wrapper possono essere concatenati insieme.

Alcuni wrapper comuni possono essere trovati in environments/wrappers.py . Per esempio:

  1. ActionDiscretizeWrapper : converte uno spazio di azioni continue in uno spazio di azioni discrete.
  2. RunStats : acquisisce le statistiche di esecuzione dell'ambiente come il numero di passaggi effettuati, il numero di episodi completati ecc.
  3. TimeLimit : termina l'episodio dopo un numero fisso di passaggi.

Esempio 1: azione Discretizza wrapper

InvertedPendulum è un ambiente PyBullet che accetta azioni continue nell'intervallo [-2, 2] . Se vogliamo addestrare un agente di azione discreto come DQN su questo ambiente, dobbiamo discretizzare (quantizzare) lo spazio di azione. Questo è esattamente ciò che fa ActionDiscretizeWrapper . Confronta action_spec prima e dopo il wrapping:

env = suite_gym.load('Pendulum-v0')
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)

Il wrapping discrete_action_env è un'istanza di py_environment.PyEnvironment e può essere trattato come un normale ambiente Python.

Ambienti TensorFlow

L'interfaccia per gli ambienti TF è definita negli environments/tf_environment.TFEnvironment e ha un aspetto molto simile agli ambienti Python. Gli ambienti TF differiscono dagli env python in un paio di modi:

  • Generano oggetti tensoriali invece di array
  • Gli ambienti TF aggiungono una dimensione batch ai tensori generati rispetto alle specifiche.

La conversione degli ambienti Python in TFEnvs consente a tensorflow di parellalizzare le operazioni. Ad esempio, si potrebbe definire un collect_experience_op che raccoglie i dati dall'ambiente e li aggiunge a un replay_buffer e un train_op che legge dal replay_buffer e addestra l'agente e li esegue in parallelo naturalmente in 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`."""

Il metodo current_time_step() restituisce il time_step corrente e, se necessario, inizializza l'ambiente.

Il metodo reset() forza un reset nell'ambiente e restituisce current_step.

Se l' action non dipende dal time_step precedente, è necessario un tf.control_dependency in modalità Graph .

Per ora, vediamo come vengono creati gli TFEnvironments .

Creazione del proprio ambiente TensorFlow

Questo è più complicato della creazione di ambienti in Python, quindi non lo tratteremo in questo colab. Un esempio è disponibile qui . Il caso d'uso più comune è implementare il proprio ambiente in Python e avvolgerlo in TensorFlow utilizzando il nostro wrapper TFPyEnvironment (vedi sotto).

Avvolgere un ambiente Python in TensorFlow

Possiamo facilmente avvolgere qualsiasi ambiente Python in un ambiente TensorFlow utilizzando il wrapper TFPyEnvironment .

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(step_type=TensorSpec(shape=(), dtype=tf.int32, name='step_type'), reward=TensorSpec(shape=(), dtype=tf.float32, name='reward'), 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)))
Action Specs: BoundedTensorSpec(shape=(), dtype=tf.int64, name='action', minimum=array(0), maximum=array(1))

Nota che le specifiche ora sono del tipo: (Bounded)TensorSpec .

Esempi di utilizzo

Semplice esempio

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(step_type=array([0], dtype=int32), reward=array([0.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.01064016, -0.0126703 , -0.02927821,  0.01341233]],
      dtype=float32)), array([0], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.01089357, -0.2073604 , -0.02900996,  0.29671577]],
      dtype=float32))]
[TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.01089357, -0.2073604 , -0.02900996,  0.29671577]],
      dtype=float32)), array([1], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.01504077, -0.01183716, -0.02307565, -0.00497343]],
      dtype=float32))]
[TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.01504077, -0.01183716, -0.02307565, -0.00497343]],
      dtype=float32)), array([0], dtype=int32), TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.01527752, -0.20662071, -0.02317511,  0.28034046]],
      dtype=float32))]
Total reward: [3.]

Episodi interi

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: 111
avg_length 22.2 avg_reward: 22.2