Copyright 2021 Os autores do TF-Agents.
![]() | ![]() | ![]() | ![]() |
Introdução
O objetivo do Reinforcement Learning (RL) é projetar agentes que aprendem interagindo com um ambiente. Na configuração RL padrão, o agente recebe uma observação a cada passo de tempo e escolhe uma ação. A ação é aplicada ao meio ambiente e o meio ambiente retorna uma recompensa e uma nova observação. O agente treina uma política de escolha de ações para maximizar a soma das recompensas, também conhecido como retorno.
Em TF-Agents, os ambientes podem ser implementados em Python ou TensorFlow. Os ambientes Python geralmente são mais fáceis de implementar, entender e depurar, mas os ambientes TensorFlow são mais eficientes e permitem a paralelização natural. O fluxo de trabalho mais comum é implementar um ambiente em Python e usar um de nossos wrappers para convertê-lo automaticamente em TensorFlow.
Vejamos primeiro os ambientes Python. Os ambientes TensorFlow seguem uma API muito semelhante.
Estabelecer
Se você ainda não instalou tf-agents ou gym, execute:
pip install -q 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
tf.compat.v1.enable_v2_behavior()
Ambientes Python
Os ambientes Python têm um método step(action) -> next_time_step
que aplica uma ação ao ambiente e retorna as seguintes informações sobre a próxima etapa:
-
observation
: é a parte do estado do ambiente que o agente pode observar para escolher suas ações na próxima etapa. -
reward
: o agente está aprendendo a maximizar a soma dessas recompensas em várias etapas. -
step_type
: As interações com o ambiente geralmente fazem parte de uma sequência / episódio. por exemplo, vários movimentos em um jogo de xadrez. step_type pode serFIRST
,MID
ouLAST
para indicar se esta etapa de tempo é a primeira, intermediária ou última etapa em uma sequência. -
discount
: é um valor flutuante que representa quanto pesar a recompensa na próxima etapa de tempo em relação à recompensa na etapa de tempo atual.
Eles são agrupados em uma tupla nomeada TimeStep(step_type, reward, discount, observation)
.
A interface que todos os ambientes Python devem implementar está em environments/py_environment.PyEnvironment
. Os principais métodos são:
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."""
Além do método step()
, os ambientes também fornecem um método reset()
que inicia uma nova sequência e fornece um TimeStep
inicial. Não é necessário chamar o método reset
explicitamente. Assumimos que os ambientes são redefinidos automaticamente, seja quando chegam ao final de um episódio ou quando a etapa () é chamada pela primeira vez.
Observe que as subclasses não implementam step()
ou reset()
diretamente. Em vez disso, eles substituem os _step()
e _reset()
. Os intervalos de tempo retornados desses métodos serão armazenados em cache e expostos por meio de current_time_step()
.
Os métodos observation_spec
e action_spec
retornam um ninho de (Bounded)ArraySpecs
que descrevem o nome, a forma, o tipo de dados e os intervalos das observações e ações, respectivamente.
Em TF-Agents, repetidamente nos referimos a ninhos que são definidos como qualquer estrutura semelhante a uma árvore composta de listas, tuplas, tuplas nomeadas ou dicionários. Estes podem ser compostos arbitrariamente para manter a estrutura de observações e ações. Descobrimos que isso é muito útil para ambientes mais complexos, onde você tem muitas observações e ações.
Usando ambientes padrão
TF Agents tem wrappers integrados para muitos ambientes padrão, como OpenAI Gym, DeepMind-control e Atari, para que sigam nossa interface py_environment.PyEnvironment
. Esses ambientes agrupados podem ser carregados facilmente usando nossos pacotes de ambiente. Vamos carregar o ambiente CartPole do ginásio OpenAI e olhar para a ação e o 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')
Portanto, vemos que o ambiente espera ações do tipo int64
em [0, 1] e retorna TimeSteps
onde as observações são um vetor float32
de comprimento 4 e o fator de desconto é um float32
em [0,0, 1,0]. Agora, vamos tentar realizar uma ação fixa (1,)
para um episódio inteiro.
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.01285449, 0.04769544, 0.01983412, -0.00245379], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.0138084 , 0.24252741, 0.01978504, -0.2888134 ], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.01865895, 0.43736172, 0.01400878, -0.57519126], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.02740618, 0.6322845 , 0.00250495, -0.8634283 ], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.04005187, 0.82737225, -0.01476362, -1.1553226 ], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.05659932, 1.0226836 , -0.03787007, -1.452598 ], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.07705299, 1.2182497 , -0.06692202, -1.7568679 ], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.10141798, 1.4140631 , -0.10205939, -2.069591 ], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.12969925, 1.6100639 , -0.1434512 , -2.3920157 ], dtype=float32)) TimeStep(step_type=array(1, dtype=int32), reward=array(1., dtype=float32), discount=array(1., dtype=float32), observation=array([ 0.16190052, 1.8061239 , -0.19129153, -2.725115 ], dtype=float32)) TimeStep(step_type=array(2, dtype=int32), reward=array(1., dtype=float32), discount=array(0., dtype=float32), observation=array([ 0.198023 , 2.002027 , -0.24579382, -3.0695074 ], dtype=float32))
Criação de seu próprio ambiente Python
Para muitos clientes, um caso de uso comum é aplicar um dos agentes padrão (consulte agentes /) em Agentes TF ao problema. Para fazer isso, eles devem enquadrar seu problema como um ambiente. Então, vamos ver como implementar um ambiente em Python.
Digamos que queremos treinar um agente para jogar o seguinte jogo de cartas (inspirado no Black Jack):
- O jogo é jogado com um baralho infinito de cartas numeradas de 1 ... 10.
- A cada jogada, o agente pode fazer 2 coisas: obter uma nova carta aleatória ou interromper a rodada atual.
- O objetivo é fazer com que a soma de suas cartas seja o mais próximo possível de 21 no final da rodada, sem ultrapassar.
Um ambiente que representa o jogo pode ser assim:
- Ações: temos 2 ações. Ação 0: pegue uma nova carta e Ação 1: encerre a rodada atual.
- Observações: Soma das cartas na rodada atual.
- Recompensa: O objetivo é chegar o mais próximo possível de 21 sem ultrapassar, então podemos conseguir isso usando a seguinte recompensa no final da rodada: soma_de_cartas - 21 se soma_de_cartas <= 21, senão -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)
Vamos nos certificar de que fizemos tudo corretamente definindo o ambiente acima. Ao criar seu próprio ambiente, você deve se certificar de que as observações e time_steps gerados seguem as formas e tipos corretos, conforme definido em suas especificações. Eles são usados para gerar o gráfico do TensorFlow e, como tal, podem criar problemas difíceis de depurar se errarmos.
Para validar nosso ambiente, usaremos uma política aleatória para gerar ações e faremos uma iteração em mais de 5 episódios para garantir que as coisas estejam funcionando como pretendido. Um erro é gerado se recebermos um time_step que não segue as especificações do ambiente.
environment = CardGameEnv()
utils.validate_py_environment(environment, episodes=5)
Agora que sabemos que o ambiente está funcionando conforme o esperado, vamos executá-lo usando uma política fixa: peça 3 cartas e encerre a rodada.
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([2], 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([8], dtype=int32)) TimeStep(step_type=array(2, dtype=int32), reward=array(-13., dtype=float32), discount=array(0., dtype=float32), observation=array([8], dtype=int32)) Final Reward = -13.0
Envoltórios de ambiente
Um wrapper de ambiente pega um ambiente Python e retorna uma versão modificada do ambiente. Tanto o ambiente original quanto o ambiente modificado são instâncias de py_environment.PyEnvironment
e vários wrappers podem ser encadeados.
Alguns invólucros comuns podem ser encontrados em environments/wrappers.py
. Por exemplo:
-
ActionDiscretizeWrapper
: Converte um espaço de ação contínua em um espaço de ação discreto. -
RunStats
: as capturas executam estatísticas do ambiente, como número de etapas executadas, número de episódios concluídos, etc. -
TimeLimit
: termina o episódio após um número fixo de etapas.
Exemplo 1: Wrapper de Discretização de Ação
InvertedPendulum é um ambiente PyBullet que aceita ações contínuas no intervalo [-2, 2]
. Se quisermos treinar um agente de ação discreto como DQN neste ambiente, temos que discretizar (quantizar) o espaço de ação. Isso é exatamente o que ActionDiscretizeWrapper
faz. Compare o action_spec
antes e depois do empacotamento:
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)
O discrete_action_env
empacotado é uma instância de py_environment.PyEnvironment
e pode ser tratado como um ambiente Python normal.
Ambientes TensorFlow
A interface para ambientes TF é definida em environments/tf_environment.TFEnvironment
e é muito semelhante aos ambientes Python. Os ambientes TF diferem dos ambientes Python de duas maneiras:
- Eles geram objetos tensores em vez de matrizes
- Os ambientes TF adicionam uma dimensão de lote aos tensores gerados quando comparados às especificações.
Converter os ambientes Python em TFEnvs permite que o tensorflow paralelize as operações. Por exemplo, pode-se definir um collect_experience_op
que coleta dados do ambiente e adiciona a um replay_buffer
, e um train_op
que lê de replay_buffer
e treina o agente, e os executa em paralelo naturalmente no 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`."""
O método current_time_step()
retorna o time_step atual e inicializa o ambiente se necessário.
O método reset()
força um reset no ambiente e retorna current_step.
Se a action
não depender do time_step
anterior, um tf.control_dependency
será necessário no modo Graph
.
Por enquanto, vamos ver como os ambientes TFEnvironments
são criados.
Criar seu próprio ambiente TensorFlow
Isso é mais complicado do que criar ambientes em Python, então não vamos cobrir isso nesta colab. Um exemplo está disponível aqui . O caso de uso mais comum é implementar seu ambiente em Python e envolvê-lo no TensorFlow usando nosso wrapper TFPyEnvironment
(veja abaixo).
Envolvendo um ambiente Python no TensorFlow
Podemos facilmente envolver qualquer ambiente Python em um ambiente TensorFlow usando o 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))
Observe que as especificações agora são do tipo: (Bounded)TensorSpec
.
Exemplos de uso
Exemplo Simples
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.03501577, -0.04957427, 0.00623939, 0.03762257]], 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.03600726, -0.24478514, 0.00699184, 0.33226755]], dtype=float32))] [TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.03600726, -0.24478514, 0.00699184, 0.33226755]], 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.04090296, -0.0497634 , 0.01363719, 0.04179767]], dtype=float32))] [TimeStep(step_type=array([1], dtype=int32), reward=array([1.], dtype=float32), discount=array([1.], dtype=float32), observation=array([[-0.04090296, -0.0497634 , 0.01363719, 0.04179767]], 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.04189822, -0.24507822, 0.01447314, 0.33875188]], dtype=float32))] Total reward: [3.]
Episódios inteiros
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: 138 avg_length 27.6 avg_reward: 27.6