Copyright 2021 Os autores do TF-Agents.
 Ver no TensorFlow.org |  Executar no Google Colab |  Ver fonte no GitHub |  Baixar caderno |
Introdução
Na terminologia do Aprendizado por Reforço, as políticas mapeiam uma observação do ambiente para uma ação ou distribuição de ações. Em TF-agentes, observações do ambiente estão contidas em uma tupla chamado TimeStep('step_type', 'discount', 'reward', 'observation') , e as políticas de mapear Timesteps para ações ou distribuições sobre ações. A maioria das políticas usar timestep.observation , algumas políticas usar timestep.step_type (por exemplo, para redefinir o estado no início de um episódio em políticas stateful), mas timestep.discount e timestep.reward são geralmente ignoradas.
As políticas estão relacionadas a outros componentes em TF-Agents da seguinte maneira. A maioria das políticas tem uma rede neural para computar ações e / ou distribuições sobre ações de TimeSteps. Os agentes podem conter uma ou mais políticas para finalidades diferentes, por exemplo, uma política principal que está sendo treinada para implantação e uma política barulhenta para coleta de dados. As políticas podem ser salvas / restauradas e podem ser usadas independentemente do agente para coleta de dados, avaliação, etc.
Algumas políticas são mais fáceis de escrever no Tensorflow (por exemplo, aquelas com uma rede neural), enquanto outras são mais fáceis de escrever em Python (por exemplo, seguir um script de ações). Portanto, em agentes TF, permitimos políticas Python e Tensorflow. Além disso, as políticas escritas no TensorFlow podem ter que ser usadas em um ambiente Python ou vice-versa, por exemplo, uma política do TensorFlow é usada para treinamento, mas posteriormente implantada em um ambiente Python de produção. Para tornar isso mais fácil, fornecemos wrappers para conversão entre as políticas do Python e do TensorFlow.
Outra classe interessante de políticas são os wrappers de políticas, que modificam uma determinada política de uma determinada maneira, por exemplo, adicionam um tipo específico de ruído, fazem uma versão gananciosa ou gananciosa de uma política estocástica, misturam várias políticas aleatoriamente, etc.
Configurar
Se você ainda não instalou tf-agents, execute:
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 tensorflow_probability as tfp
import numpy as np
from tf_agents.specs import array_spec
from tf_agents.specs import tensor_spec
from tf_agents.networks import network
from tf_agents.policies import py_policy
from tf_agents.policies import random_py_policy
from tf_agents.policies import scripted_py_policy
from tf_agents.policies import tf_policy
from tf_agents.policies import random_tf_policy
from tf_agents.policies import actor_policy
from tf_agents.policies import q_policy
from tf_agents.policies import greedy_policy
from tf_agents.trajectories import time_step as ts
Políticas Python
A interface para políticas Python é definido em policies/py_policy.PyPolicy . Os principais métodos são:
class Base(object):
@abc.abstractmethod
def __init__(self, time_step_spec, action_spec, policy_state_spec=()):
self._time_step_spec = time_step_spec
self._action_spec = action_spec
self._policy_state_spec = policy_state_spec
@abc.abstractmethod
def reset(self, policy_state=()):
# return initial_policy_state.
pass
@abc.abstractmethod
def action(self, time_step, policy_state=()):
# return a PolicyStep(action, state, info) named tuple.
pass
@abc.abstractmethod
def distribution(self, time_step, policy_state=()):
# Not implemented in python, only for TF policies.
pass
@abc.abstractmethod
def update(self, policy):
# update self to be similar to the input `policy`.
pass
@property
def time_step_spec(self):
return self._time_step_spec
@property
def action_spec(self):
return self._action_spec
@property
def policy_state_spec(self):
return self._policy_state_spec
O método mais importante é action(time_step) que mapeia um time_step contendo uma observação a partir do ambiente para um tuplo PolicyStep chamado contendo os seguintes atributos:
-
action: A ação a ser aplicada ao ambiente. -
state: O estado da política (por exemplo, estado RNN) a ser alimentada para a próxima chamada a acção. -
info: informações laterais opcionais, tais como probabilidades de registro de ação.
O time_step_spec e action_spec estão as especificações para o passo de tempo de entrada e a ação de saída. Políticas também têm uma reset função que normalmente é usado para repor o estado em políticas stateful. A update(new_policy) função atualiza self direção new_policy .
Agora, vamos dar uma olhada em alguns exemplos de políticas Python.
Exemplo 1: política aleatória de Python
Um exemplo simples de um PyPolicy é o RandomPyPolicy que gera acções aleatórios para o discreta / dado action_spec contínua. A entrada time_step é ignorado.
action_spec = array_spec.BoundedArraySpec((2,), np.int32, -10, 10)
my_random_py_policy = random_py_policy.RandomPyPolicy(time_step_spec=None,
action_spec=action_spec)
time_step = None
action_step = my_random_py_policy.action(time_step)
print(action_step)
action_step = my_random_py_policy.action(time_step)
print(action_step)
PolicyStep(action=array([10, -4], dtype=int32), state=(), info=()) PolicyStep(action=array([7, 6], dtype=int32), state=(), info=())
Exemplo 2: política Python com script
Um script jogadas políticas trás um roteiro de acções representadas como uma lista de (num_repeats, action) tuplas. Toda vez que a action função é chamada, retorna a próxima ação na lista até que o número especificado de repetições é feito, e então se move para a próxima ação na lista. A reset método pode ser chamado para iniciar a execução a partir do início da lista.
action_spec = array_spec.BoundedArraySpec((2,), np.int32, -10, 10)
action_script = [(1, np.array([5, 2], dtype=np.int32)),
(0, np.array([0, 0], dtype=np.int32)), # Setting `num_repeats` to 0 will skip this action.
(2, np.array([1, 2], dtype=np.int32)),
(1, np.array([3, 4], dtype=np.int32))]
my_scripted_py_policy = scripted_py_policy.ScriptedPyPolicy(
time_step_spec=None, action_spec=action_spec, action_script=action_script)
policy_state = my_scripted_py_policy.get_initial_state()
time_step = None
print('Executing scripted policy...')
action_step = my_scripted_py_policy.action(time_step, policy_state)
print(action_step)
action_step= my_scripted_py_policy.action(time_step, action_step.state)
print(action_step)
action_step = my_scripted_py_policy.action(time_step, action_step.state)
print(action_step)
print('Resetting my_scripted_py_policy...')
policy_state = my_scripted_py_policy.get_initial_state()
action_step = my_scripted_py_policy.action(time_step, policy_state)
print(action_step)
Executing scripted policy... PolicyStep(action=array([5, 2], dtype=int32), state=[0, 1], info=()) PolicyStep(action=array([1, 2], dtype=int32), state=[2, 1], info=()) PolicyStep(action=array([1, 2], dtype=int32), state=[2, 2], info=()) Resetting my_scripted_py_policy... PolicyStep(action=array([5, 2], dtype=int32), state=[0, 1], info=())
Políticas do TensorFlow
As políticas do TensorFlow seguem a mesma interface que as políticas do Python. Vejamos alguns exemplos.
Exemplo 1: Política de TF aleatória
Um RandomTFPolicy pode ser usado para gerar acções aleatórios de acordo com uma dada discreta / contínua action_spec . A entrada time_step é ignorado.
action_spec = tensor_spec.BoundedTensorSpec(
(2,), tf.float32, minimum=-1, maximum=3)
input_tensor_spec = tensor_spec.TensorSpec((2,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
my_random_tf_policy = random_tf_policy.RandomTFPolicy(
action_spec=action_spec, time_step_spec=time_step_spec)
observation = tf.ones(time_step_spec.observation.shape)
time_step = ts.restart(observation)
action_step = my_random_tf_policy.action(time_step)
print('Action:')
print(action_step.action)
Action: tf.Tensor([-0.9448042 1.9039011], shape=(2,), dtype=float32)
Exemplo 2: Política de Ator
Uma política ator pode ser criado usando uma rede que mapeia time_steps a ações ou uma rede que mapeia time_steps a distribuições sobre ações.
Usando uma rede de ação
Vamos definir uma rede da seguinte maneira:
class ActionNet(network.Network):
def __init__(self, input_tensor_spec, output_tensor_spec):
super(ActionNet, self).__init__(
input_tensor_spec=input_tensor_spec,
state_spec=(),
name='ActionNet')
self._output_tensor_spec = output_tensor_spec
self._sub_layers = [
tf.keras.layers.Dense(
action_spec.shape.num_elements(), activation=tf.nn.tanh),
]
def call(self, observations, step_type, network_state):
del step_type
output = tf.cast(observations, dtype=tf.float32)
for layer in self._sub_layers:
output = layer(output)
actions = tf.reshape(output, [-1] + self._output_tensor_spec.shape.as_list())
# Scale and shift actions to the correct range if necessary.
return actions, network_state
No TensorFlow, a maioria das camadas de rede são projetadas para operações em lote, portanto, esperamos que os time_steps de entrada sejam agrupados e que a saída da rede também seja agrupada. Além disso, a rede é responsável por produzir ações no intervalo correto de determinado action_spec. Isto é convencionalmente feito utilizando, por exemplo uma activação tanh para a camada final de produto em acções [-1, 1] e, em seguida, escamação e deslocando este para o intervalo correcto como o action_spec de entrada (por exemplo, ver tf_agents/agents/ddpg/networks.actor_network() ).
Agora, podemos criar uma política de ator usando a rede acima.
input_tensor_spec = tensor_spec.TensorSpec((4,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
action_spec = tensor_spec.BoundedTensorSpec((3,),
tf.float32,
minimum=-1,
maximum=1)
action_net = ActionNet(input_tensor_spec, action_spec)
my_actor_policy = actor_policy.ActorPolicy(
time_step_spec=time_step_spec,
action_spec=action_spec,
actor_network=action_net)
Podemos aplicá-lo a qualquer lote de time_steps após time_step_spec:
batch_size = 2
observations = tf.ones([2] + time_step_spec.observation.shape.as_list())
time_step = ts.restart(observations, batch_size)
action_step = my_actor_policy.action(time_step)
print('Action:')
print(action_step.action)
distribution_step = my_actor_policy.distribution(time_step)
print('Action distribution:')
print(distribution_step.action)
Action:
tf.Tensor(
[[0.9318627 0.7770741 0.8645338]
[0.9318627 0.7770741 0.8645338]], shape=(2, 3), dtype=float32)
Action distribution:
tfp.distributions.Deterministic("Deterministic", batch_shape=[2, 3], event_shape=[], dtype=float32)
No exemplo acima, criamos a política usando uma rede de ação que produz um tensor de ação. Neste caso, policy.distribution(time_step) é um determinista (delta) de distribuição em torno da saída de policy.action(time_step) . Uma maneira de produzir uma política estocástica é envolver a política do ator em um wrapper de política que adiciona ruído às ações. Outra forma é criar a política de ator usando uma rede de distribuição de ação em vez de uma rede de ação, conforme mostrado abaixo.
Usando uma rede de distribuição de ação
class ActionDistributionNet(ActionNet):
def call(self, observations, step_type, network_state):
action_means, network_state = super(ActionDistributionNet, self).call(
observations, step_type, network_state)
action_std = tf.ones_like(action_means)
return tfp.distributions.MultivariateNormalDiag(action_means, action_std), network_state
action_distribution_net = ActionDistributionNet(input_tensor_spec, action_spec)
my_actor_policy = actor_policy.ActorPolicy(
time_step_spec=time_step_spec,
action_spec=action_spec,
actor_network=action_distribution_net)
action_step = my_actor_policy.action(time_step)
print('Action:')
print(action_step.action)
distribution_step = my_actor_policy.distribution(time_step)
print('Action distribution:')
print(distribution_step.action)
Action:
tf.Tensor(
[[ 0.96731853 1. 1. ]
[ 0.94488937 -0.29294527 1. ]], shape=(2, 3), dtype=float32)
Action distribution:
tfp.distributions.MultivariateNormalDiag("ActionNet_MultivariateNormalDiag", batch_shape=[2], event_shape=[3], dtype=float32)
Observe que no exemplo acima, as ações são cortadas no intervalo da especificação de ação dada [-1, 1]. Isso ocorre porque um argumento do construtor de ActorPolicy clip = True por padrão. Definir como falso retornará ações não cortadas produzidas pela rede.
Políticas estocásticos podem ser convertidos para as políticas determinísticos utilizando, por exemplo, um invólucro GreedyPolicy que escolhe stochastic_policy.distribution().mode() como a sua acção, e uma distribuição / delta determinista em torno desta acção ávido como a sua distribution() .
Exemplo 3: Política Q
A política AQ é usada em agentes como DQN e é baseada em uma rede Q que prevê um valor Q para cada ação discreta. Para um determinado intervalo de tempo, a distribuição de ação na Política Q é uma distribuição categórica criada usando os valores q como logits.
input_tensor_spec = tensor_spec.TensorSpec((4,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
action_spec = tensor_spec.BoundedTensorSpec((),
tf.int32,
minimum=0,
maximum=2)
num_actions = action_spec.maximum - action_spec.minimum + 1
class QNetwork(network.Network):
def __init__(self, input_tensor_spec, action_spec, num_actions=num_actions, name=None):
super(QNetwork, self).__init__(
input_tensor_spec=input_tensor_spec,
state_spec=(),
name=name)
self._sub_layers = [
tf.keras.layers.Dense(num_actions),
]
def call(self, inputs, step_type=None, network_state=()):
del step_type
inputs = tf.cast(inputs, tf.float32)
for layer in self._sub_layers:
inputs = layer(inputs)
return inputs, network_state
batch_size = 2
observation = tf.ones([batch_size] + time_step_spec.observation.shape.as_list())
time_steps = ts.restart(observation, batch_size=batch_size)
my_q_network = QNetwork(
input_tensor_spec=input_tensor_spec,
action_spec=action_spec)
my_q_policy = q_policy.QPolicy(
time_step_spec, action_spec, q_network=my_q_network)
action_step = my_q_policy.action(time_steps)
distribution_step = my_q_policy.distribution(time_steps)
print('Action:')
print(action_step.action)
print('Action distribution:')
print(distribution_step.action)
Action:
tf.Tensor([2 2], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.Categorical("Categorical", batch_shape=[2], event_shape=[], dtype=int32)
Wrappers de política
Um wrapper de política pode ser usado para embrulhar e modificar uma determinada política, por exemplo, adicionar ruído. Wrappers de política são uma subclasse de Política (Python / TensorFlow) e podem, portanto, ser usados como qualquer outra política.
Exemplo: Política gananciosa
Um wrapper ganancioso pode ser usado para embrulhar qualquer política TensorFlow que implementa distribution() . GreedyPolicy.action() retornará wrapped_policy.distribution().mode() E GreedyPolicy.distribution() é uma distribuição / delta determinista torno GreedyPolicy.action() :
my_greedy_policy = greedy_policy.GreedyPolicy(my_q_policy)
action_step = my_greedy_policy.action(time_steps)
print('Action:')
print(action_step.action)
distribution_step = my_greedy_policy.distribution(time_steps)
print('Action distribution:')
print(distribution_step.action)
Action:
tf.Tensor([1 1], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.DeterministicWithLogProb("Deterministic", batch_shape=[2], event_shape=[], dtype=int32)