Copyright 2021 Gli autori degli agenti TF.
 Visualizza su TensorFlow.org |  Esegui in Google Colab |  Visualizza la fonte su GitHub |  Scarica taccuino |
introduzione
Nella terminologia di Reinforcement Learning, le politiche mappano un'osservazione dall'ambiente a un'azione oa una distribuzione sulle azioni. Nel TF-agenti, osservazioni dell'ambiente sono contenute in una tupla di nome TimeStep('step_type', 'discount', 'reward', 'observation') , e le politiche mappa Timesteps alle azioni o distribuzioni sulle azioni. La maggior parte delle politiche di utilizzo timestep.observation , alcune politiche usano timestep.step_type (ad esempio, per ripristinare lo stato all'inizio di un episodio in politiche stateful), ma timestep.discount e timestep.reward sono di solito ignorati.
Le policy sono correlate ad altri componenti in TF-Agents nel modo seguente. La maggior parte dei criteri dispone di una rete neurale per calcolare azioni e/o distribuzioni su azioni da TimeSteps. Gli agenti possono contenere una o più policy per scopi diversi, ad esempio una policy principale che viene addestrata per l'implementazione e una policy rumorosa per la raccolta dei dati. I criteri possono essere salvati/ripristinati e possono essere utilizzati indipendentemente dall'agente per la raccolta dei dati, la valutazione, ecc.
Alcune policy sono più facili da scrivere in Tensorflow (es. quelle con una rete neurale), mentre altre sono più facili da scrivere in Python (es. seguendo uno script di azioni). Quindi negli agenti TF, consentiamo sia le politiche Python che Tensorflow. Inoltre, le policy scritte in TensorFlow potrebbero dover essere utilizzate in un ambiente Python, o viceversa, ad esempio una policy TensorFlow viene utilizzata per la formazione ma successivamente distribuita in un ambiente Python di produzione. Per semplificare questa operazione, forniamo wrapper per la conversione tra le policy Python e TensorFlow.
Un'altra classe interessante di policy sono i policy wrapper, che modificano una data policy in un certo modo, ad esempio aggiungono un particolare tipo di rumore, creano una versione greedy o epsilon-greedy di una policy stocastica, mescolano casualmente più policy ecc.
Impostare
Se non hai ancora installato tf-agent, esegui:
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
Politiche di Python
L'interfaccia per le politiche di Python è definito in policies/py_policy.PyPolicy . I principali metodi sono:
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
Il metodo più importante è action(time_step) che associa un time_step contenente un'osservazione dall'ambiente a una tupla PolicyStep di nome che contiene i seguenti attributi:
-
action: L'azione da applicata all'ambiente. -
state: Lo stato della politica (es RNN state) da alimentare nel prossimo invito all'azione. -
info: Informazioni laterali opzionali come ad esempio le probabilità di log azione.
Il time_step_spec e action_spec sono specifiche per la fase di tempo di ingresso e l'azione di uscita. Politiche hanno anche un reset funzione che viene in genere utilizzato per il ripristino dello stato nelle politiche stateful. L' update(new_policy) funzione aggiorna self verso new_policy .
Ora, diamo un'occhiata a un paio di esempi di policy Python.
Esempio 1: criterio Python casuale
Un semplice esempio di un PyPolicy rappresenta l' RandomPyPolicy che genera azioni casuali per discreto / continuo proposta action_spec. L'ingresso time_step viene ignorato.
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=())
Esempio 2: policy Python con script
Suona un policy script indietro uno script di azioni rappresentate come una lista di (num_repeats, action) tuple. Ogni volta che action funzione viene chiamata, restituisce l'azione successiva nella lista fino a quando viene fatto il numero specificato di ripetizioni, e quindi passa alla prossima azione nella lista. Il reset metodo può essere chiamato per iniziare l'esecuzione dall'inizio della 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=())
Politiche TensorFlow
Le policy TensorFlow seguono la stessa interfaccia delle policy Python. Vediamo alcuni esempi.
Esempio 1: politica TF casuale
Un RandomTFPolicy può essere utilizzato per generare le azioni casuali secondo una data discreto / continuo action_spec . L'ingresso time_step viene ignorato.
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)
Esempio 2: politica dell'attore
Una politica attore può essere creato utilizzando una rete che mappa time_steps ad azioni o di una rete che associa time_steps alle distribuzioni più azioni.
Utilizzo di una rete d'azione
Definiamo una rete come segue:
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
In TensorFlow la maggior parte dei livelli di rete è progettata per operazioni batch, quindi prevediamo che i time_steps di input vengano raggruppati in batch e anche l'output della rete. Anche la rete è responsabile della produzione di azioni nell'intervallo corretto di action_spec. Questo è convenzionalmente fatto utilizzando ad esempio un tanh attivazione per lo strato finale di azioni producono in [-1, 1] e quindi ridimensionamento e spostando questo per l'intervallo corretto come action_spec ingresso (vedi ad esempio tf_agents/agents/ddpg/networks.actor_network() ).
Ora, possiamo creare una policy dell'attore usando la rete sopra.
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)
Possiamo applicarlo a qualsiasi batch di time_steps che segue 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)
Nell'esempio sopra, abbiamo creato la policy utilizzando una rete di azioni che produce un tensore di azione. In questo caso, policy.distribution(time_step) è deterministico (delta) distribuzione attorno all'uscita del policy.action(time_step) . Un modo per produrre una policy stocastica è avvolgere la policy dell'attore in un wrapper di policy che aggiunge rumore alle azioni. Un altro modo consiste nel creare la policy dell'attore utilizzando una rete di distribuzione delle azioni invece di una rete di azioni come mostrato di seguito.
Utilizzo di una rete di distribuzione di azioni
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)
Si noti che in quanto sopra, le azioni sono ritagliate nell'intervallo della specifica azione data [-1, 1]. Questo perché un argomento del costruttore di ActorPolicy clip=True per impostazione predefinita. L'impostazione di questo su false restituirà le azioni non ritagliate prodotte dalla rete.
Politiche stocastici possono essere convertiti in politiche deterministici utilizzando, per esempio, un involucro GreedyPolicy che sceglie stochastic_policy.distribution().mode() come la sua azione, e una distribuzione / delta deterministico intorno questa azione avido come distribution() .
Esempio 3: criterio Q
Il criterio AQ viene utilizzato in agenti come DQN e si basa su una rete Q che prevede un valore Q per ogni azione discreta. Per una determinata fase temporale, la distribuzione dell'azione nella politica Q è una distribuzione categoriale creata utilizzando i valori q come logit.
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)
Wrapper delle politiche
Un wrapper di policy può essere utilizzato per avvolgere e modificare una data policy, ad esempio aggiungere rumore. I wrapper di policy sono una sottoclasse di Policy (Python/TensorFlow) e possono quindi essere utilizzati come qualsiasi altra policy.
Esempio: politica avida
Un involucro avido può essere utilizzato per avvolgere qualsiasi politica tensorflow che implementa distribution() . GreedyPolicy.action() restituirà wrapped_policy.distribution().mode() E GreedyPolicy.distribution() è una distribuzione / triangolo deterministico intorno 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)