Copyright 2021 Los autores de TF-Agents.
 Ver en TensorFlow.org |  Ejecutar en Google Colab |  Ver fuente en GitHub |  Descargar cuaderno |
Introducción
En la terminología del aprendizaje por refuerzo, las políticas mapean una observación del entorno a una acción o una distribución sobre las acciones. En la carretera TF-agentes, observaciones desde el medio ambiente están contenidas en una tupla llamado TimeStep('step_type', 'discount', 'reward', 'observation') , y las políticas de mapa timesteps a acciones o distribuciones más acciones. La mayoría de las políticas de uso timestep.observation , algunas políticas utilizan timestep.step_type (por ejemplo, para restablecer el estado en el comienzo de un episodio en las políticas stateful), pero timestep.discount y timestep.reward son generalmente ignorados.
Las políticas están relacionadas con otros componentes en TF-Agents de la siguiente manera. La mayoría de las políticas tienen una red neuronal para calcular acciones y / o distribuciones sobre acciones de TimeSteps. Los agentes pueden contener una o más políticas para diferentes propósitos, por ejemplo, una política principal que se está capacitando para su implementación y una política ruidosa para la recopilación de datos. Las políticas se pueden guardar / restaurar y se pueden utilizar independientemente del agente para la recopilación de datos, evaluación, etc.
Algunas políticas son más fáciles de escribir en Tensorflow (por ejemplo, aquellas con una red neuronal), mientras que otras son más fáciles de escribir en Python (por ejemplo, siguiendo un script de acciones). Entonces, en los agentes TF, permitimos políticas de Python y Tensorflow. Además, es posible que las políticas escritas en TensorFlow deban usarse en un entorno de Python o viceversa, p. Ej., Se usa una política de TensorFlow para el entrenamiento, pero luego se implementa en un entorno de producción de Python. Para facilitar esto, proporcionamos contenedores para convertir entre políticas de Python y TensorFlow.
Otra clase interesante de políticas son los envoltorios de políticas, que modifican una política determinada de cierta manera, por ejemplo, agregan un tipo particular de ruido, hacen una versión codiciosa o épsilon-codiciosa de una política estocástica, mezclan al azar varias políticas, etc.
Configuración
Si aún no ha instalado tf-agents, ejecute:
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 de Python
La interfaz para las políticas de Python se define en policies/py_policy.PyPolicy . Los principales métodos son:
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
El método más importante es action(time_step) que mapea un time_step que contiene una observación del entorno para una tupla PolicyStep llamada que contiene los siguientes atributos:
-
action: La acción que ha de aplicarse al medio ambiente. -
state: El estado de la política (por ejemplo, estado RNN), que se alimenta en la siguiente llamada a la acción. -
info: la información lateral opcional tales como probabilidades de registro de la acción.
El time_step_spec y action_spec son especificaciones para el paso de tiempo de entrada y la acción de salida. Las políticas también tienen un reset función que normalmente se utiliza para restablecer el estado de las políticas con estado. La update(new_policy) función actualiza self hacia new_policy .
Ahora, veamos un par de ejemplos de políticas de Python.
Ejemplo 1: política de Python aleatoria
Un ejemplo sencillo de un PyPolicy es la RandomPyPolicy que genera acciones aleatorio para el discreta / continua action_spec dado. La entrada time_step se ignora.
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=())
Ejemplo 2: Política de Python con script
A desempeña la política guión copias de una secuencia de comandos de las acciones representadas como una lista de (num_repeats, action) tuplas. Cada vez que la action la función se llama, devuelve la siguiente acción de la lista hasta que se haga el número especificado de repeticiones, y luego pasa a la siguiente acción en la lista. El reset método puede ser llamado para comenzar a ejecutar desde el principio de la 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 de TensorFlow
Las políticas de TensorFlow siguen la misma interfaz que las políticas de Python. Veamos algunos ejemplos.
Ejemplo 1: Política de TF aleatoria
A RandomTFPolicy se puede utilizar para generar acciones al azar de acuerdo con un discreto / continuo dado action_spec . La entrada time_step se ignora.
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)
Ejemplo 2: Política de actor
Una política actor puede ser creado utilizando una red que asigna la time_steps a las acciones o una red que asigna la time_steps a las distribuciones más acciones.
Usando una red de acción
Definamos una red de la siguiente manera:
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
En TensorFlow, la mayoría de las capas de red están diseñadas para operaciones por lotes, por lo que esperamos que los time_steps de entrada se agrupen y la salida de la red también. Además, la red es responsable de producir acciones en el rango correcto de la action_spec dada. Esto se realiza convencionalmente utilizando por ejemplo una activación tanh para la capa final a acciones producen en [-1, 1] y luego escalamiento y el desplazamiento de este a la gama correcta como el action_spec de entrada (por ejemplo, véase tf_agents/agents/ddpg/networks.actor_network() ).
Ahora, podemos crear una política de actor utilizando la red anterior.
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 aplicarlo a cualquier lote de time_steps que siga a 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)
En el ejemplo anterior, creamos la política utilizando una red de acción que produce un tensor de acción. En este caso, policy.distribution(time_step) es un determinista (delta) de distribución alrededor de la salida de policy.action(time_step) . Una forma de producir una política estocástica es envolver la política del actor en un envoltorio de política que agregue ruido a las acciones. Otra forma es crear la política de actores utilizando una red de distribución de acciones en lugar de una red de acciones como se muestra a continuación.
Usando una red de distribución de acciones
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)
Tenga en cuenta que en lo anterior, las acciones se recortan al rango de la especificación de acción dada [-1, 1]. Esto se debe a que un argumento de constructor de ActorPolicy clip = True de forma predeterminada. Establecer esto en falso devolverá las acciones no recortadas producidas por la red.
Políticas estocásticos se pueden convertir a las políticas deterministas utilizando, por ejemplo, una envoltura GreedyPolicy que elige stochastic_policy.distribution().mode() como su acción, y una distribución / delta determinista alrededor de esta acción codicioso como su distribution() .
Ejemplo 3: Política Q
La política AQ se utiliza en agentes como DQN y se basa en una red Q que predice un valor Q para cada acción discreta. Para un paso de tiempo dado, la distribución de acciones en la Política Q es una distribución categórica creada usando los 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)
Envoltorios de políticas
Se puede utilizar un contenedor de políticas para ajustar y modificar una política determinada, por ejemplo, agregar ruido. Los contenedores de políticas son una subclase de políticas (Python / TensorFlow) y, por lo tanto, se pueden usar como cualquier otra política.
Ejemplo: política codiciosa
Una envoltura codicioso puede ser utilizado para envolver cualquier política TensorFlow que implementa distribution() . GreedyPolicy.action() volverá wrapped_policy.distribution().mode() Y GreedyPolicy.distribution() es una distribución / delta determinista alrededor 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)