Copyright 2021 Os autores do TF-Agents.
 Ver no TensorFlow.org |  Executar no Google Colab |  Ver fonte no GitHub |  Baixar caderno |
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 intervalo 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.
Configurar
Se você ainda não instalou tf-agents ou gym, execute:
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
Ambientes Python
Ambientes Python tem um step(action) -> next_time_step método que se aplica uma ação para o meio ambiente, e retorna as seguintes informações sobre o próximo passo:
-
observation: Esta é a parte do estado ambiente que o agente pode observar a escolher suas ações na próxima etapa. -
reward: O agente é aprender a maximizar a soma dessas recompensas em várias etapas. -
step_type: Interacções com o ambiente são geralmente parte de uma sequência de / episódio. por exemplo, vários movimentos em um jogo de xadrez. step_type pode serFIRST,MIDouLASTpara indicar se este intervalo de tempo é a primeira, intermédia ou última etapa de uma sequência. -
discount: Este é um flutuador que representa o quanto de peso a recompensa na próxima vez que passo em relação à recompensa no passo de tempo atual.
Estes são agrupados em uma tupla chamado TimeStep(step_type, reward, discount, observation) .
A interface que todos os ambientes Python deve implementar é 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."""
Para além da step() método, ambientes também proporcionar uma reset() método que se inicia uma nova sequência e fornece uma inicial TimeStep . Não é necessário chamar a reset método 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.
Note-se que subclasses não implementar step() ou reset() diretamente. Eles vez substituir o _step() e _reset() métodos. Os passos de tempo retornados a partir destes métodos será armazenada e exposta através current_time_step() .
O observation_spec os e action_spec métodos retornar um ninho de (Bounded)ArraySpecs que descrevem o nome, a forma, o tipo de dados e intervalos das observações e acçõ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 agentes foi construído com invólucros para muitos ambientes padrão, como o Ginásio OpenAI, DeepMind-controle e Atari, para que sigam o nosso py_environment.PyEnvironment interface. 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 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')
Assim, vemos que o ambiente espera que as acções do tipo int64 em [0, 1] e retorna TimeSteps onde as observações são uma float32 vetor de comprimento 4 e fator de desconto é um float32 em [0.0, 1.0]. Agora, vamos tentar tirar uma ação fixa (1,) por 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(
{'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)})
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 das 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: obtenha 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_caras <= 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 a 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 como pretendido, 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(
{'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
Envoltórios de ambiente
Um wrapper de ambiente pega um ambiente Python e retorna uma versão modificada do ambiente. Tanto o ambiente original eo ambiente modificado são instâncias de py_environment.PyEnvironment , e vários invólucros podem ser encadeados.
Algumas embalagens comuns podem ser encontrados em environments/wrappers.py . Por exemplo:
-
ActionDiscretizeWrapper: converte um espaço de ação contínua para um espaço de acção discreta. -
RunStats: Captura executar estatísticas do ambiente, tais como o número de passos dados, número de episódios concluída etc. -
TimeLimit: Termina o episódio após um número fixo de passos.
Exemplo 1: Wrapper de Discretização de Ação
InvertedPendulum é um ambiente PyBullet que aceita acções contínuas no intervalo [-2, 2] . Se quisermos treinar um agente de ação discreto como o DQN neste ambiente, temos que discretizar (quantizar) o espaço de ação. Este é exatamente o que o ActionDiscretizeWrapper faz. Compare o action_spec antes e depois de acondicionamento:
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)
O enrolado discrete_action_env é um exemplo de py_environment.PyEnvironment e pode ser tratado como um ambiente normal Python.
Ambientes TensorFlow
A interface para ambientes TF é definida em environments/tf_environment.TFEnvironment e é muito parecido com os 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 contribui para uma replay_buffer , e um train_op que lê a partir do replay_buffer e treina o agente, e executá-los em paralelo naturalmente em 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 current_time_step() método retorna o time_step atual e inicializa o ambiente, se necessário.
Os reset() método força uma redefinição no ambiente e retorna o CURRENT_STEP.
Se a action não depende da anterior time_step um tf.control_dependency é necessária em Graph mode.
Por agora, vamos olhar como 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 é para implementar seu ambiente em Python e envolvê-la em TensorFlow usando o nosso TFPyEnvironment invólucro (veja abaixo).
Envolvendo um ambiente Python no TensorFlow
Podemos facilmente quebrar qualquer ambiente Python em um ambiente TensorFlow usando o TFPyEnvironment wrapper.
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))
Observe as especificações são agora 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(
{'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.]
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: 131 avg_length 26.2 avg_reward: 26.2