Questa pagina è stata tradotta dall'API Cloud Translation.

Ambienti

Copyright 2021 Gli autori degli agenti TF.

Visualizza su TensorFlow.org

Esegui in Google Colab

Visualizza la fonte su GitHub

Scarica taccuino

introduzione

L'obiettivo del Reinforcement Learning (RL) è progettare agenti che apprendano interagendo con un ambiente. Nell'impostazione RL standard, l'agente riceve un'osservazione in 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 in genere 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 "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

Ambienti Python

Ambienti Python hanno una step(action) -> next_time_step metodo che si applica un'azione per l'ambiente, e restituisce le seguenti informazioni relative alla fase successiva:

observation : Questa è la parte dello stato dell'ambiente che l'agente può osservare a scegliere le sue azioni al passo successivo.
reward : L'agente è imparare a massimizzare la somma di questi premi in più fasi.
step_type : Interazioni con l'ambiente sono di solito parte di una sequenza / episodio. ad esempio più mosse in una partita a scacchi. step_type può essere FIRST , MID o LAST per indicare se questa fase temporale è il primo, intermedio o l'ultimo passo in una sequenza.
discount : Questo è un galleggiante che rappresenta quanto al peso ricompensa alla prossima volta che passo rispetto alla ricompensa alla fase temporale corrente.

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

L'interfaccia che tutti gli ambienti Python devono implementare sia in environments/py_environment.PyEnvironment . I principali metodi 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."""

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

Si noti che le sottoclassi non implementano step() o reset() direttamente. Hanno invece sostituiscono il _step() e _reset() metodi. Le fasi temporali tornati da questi metodi saranno memorizzati nella cache e esposti attraverso 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-Agents ci riferiamo ripetutamente a nidi che sono definiti come qualsiasi struttura ad albero composta da elenchi, tuple, tuple con nome o dizionari. Questi possono essere composti arbitrariamente per mantenere la struttura delle osservazioni e delle azioni. Abbiamo scoperto che questo è molto utile per ambienti più complessi in cui hai molte osservazioni e azioni.

Utilizzo di ambienti standard

Agenzia TF ha built-in wrapper per molti ambienti standard come l'OpenAI palestra, DeepMind-controllo e Atari, in modo che seguano il nostro py_environment.PyEnvironment interfaccia. Questi ambienti incapsulati possono essere facilmente caricati utilizzando le nostre suite di ambienti. 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')

Così vediamo che l'ambiente si aspetta che le azioni di tipo int64 in [0, 1] e restituisce TimeSteps in cui le osservazioni sono un float32 vettore di lunghezza 4 e fattore di sconto è un float32 in [0.0, 1.0]. Ora, proviamo a prendere un'azione fisso (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(
{'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)})

Creare il proprio ambiente Python

Per molti client, un caso d'uso comune consiste nell'applicare uno degli agent standard (vedi agent/) in TF-Agents al loro problema. Per fare questo, devono inquadrare il loro problema come un ambiente. Vediamo quindi come implementare un ambiente in Python.

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

Il gioco si gioca utilizzando un mazzo di carte infinito numerato da 1 a 10.
Ad ogni turno l'agente può fare 2 cose: ottenere una nuova carta a caso o interrompere il round in corso.
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:

Azioni: abbiamo 2 azioni. Azione 0: ottenere una nuova carta e Azione 1: terminare il round in corso.
Osservazioni: Somma delle carte nel turno in corso.
Ricompensa: L'obiettivo è avvicinarsi il più possibile a 21 senza superare, quindi possiamo raggiungere questo obiettivo utilizzando la seguente ricompensa alla fine del round: somma_di_carte - 21 se somma_di_carte <= 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 sopra. Quando crei il tuo ambiente, devi assicurarti che le osservazioni e i time_step generati seguano le forme e i tipi corretti come definito nelle tue specifiche. Questi sono usati per generare il grafico TensorFlow e come tali possono creare problemi di difficile debug se li sbagliamo.

Per convalidare il nostro ambiente, utilizzeremo una politica 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 chiudi 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(
{'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

Wrapper per l'ambiente

Un wrapper di ambiente prende un ambiente Python e restituisce una versione modificata dell'ambiente. Sia l'ambiente originale e l'ambiente modificato sono casi di py_environment.PyEnvironment , e molteplici involucri possono essere concatenati.

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

ActionDiscretizeWrapper : Converte uno spazio azione continua di uno spazio azione discontinua.
RunStats : Cattura eseguire statistiche dell'ambiente, come il numero di passi compiuti, numero di episodi completato etc.
TimeLimit : Termina l'episodio dopo un numero fisso di passaggi.

Esempio 1: Azione Discretize Wrapper

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

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)

Il avvolto 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 è definito in environments/tf_environment.TFEnvironment e sembra molto simile agli ambienti Python. Gli ambienti TF differiscono dagli ambienti Python in un paio di modi:

Generano oggetti tensori 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 parallelizzare le operazioni. Ad esempio, si potrebbe definire un collect_experience_op che raccoglie i dati dall'ambiente e aggiunge ad un replay_buffer , e un train_op che legge dal replay_buffer e allena l'agente, ed eseguirli 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 current_time_step() restituisce il time_step corrente e inizializza l'ambiente, se necessario.

I reset() metodo forze un RESET nell'ambiente e ritorna il CURRENT_STEP.

Se l' action non dipende precedente time_step un tf.control_dependency è necessaria Graph modalità.

Per ora, diamo un'occhiata a come TFEnvironments sono creati.

Creazione del proprio ambiente TensorFlow

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

Wrapping di un ambiente Python in TensorFlow

Possiamo facilmente avvolgere ogni ambiente Python in un ambiente tensorflow utilizzando il TFPyEnvironment involucro.

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))

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

Esempi di utilizzo

Esempio semplice

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.]

Interi episodi

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