Stratégies

Voir sur TensorFlow.org Exécuter dans Google Colab Voir la source sur GitHub Télécharger le cahier

introduction

Dans la terminologie de l'apprentissage par renforcement, les stratégies mappent une observation de l'environnement à une action ou à une distribution sur les actions. Dans TF-agents, les observations de l'environnement sont contenues dans un tuple nommé TimeStep('step_type', 'discount', 'reward', 'observation') , et les politiques carte à des actions ou pas de temps distributions sur les actions. La plupart des politiques utilisent timestep.observation , certaines politiques utilisent timestep.step_type (par exemple pour réinitialiser l'état au début d'un épisode dans les politiques stateful), mais timestep.discount et timestep.reward sont généralement ignorés.

Les politiques sont liées à d'autres composants dans TF-Agents de la manière suivante. La plupart des politiques ont un réseau de neurones pour calculer les actions et/ou les distributions sur les actions de TimeSteps. Les agents peuvent contenir une ou plusieurs stratégies à des fins différentes, par exemple une stratégie principale en cours de formation pour le déploiement et une stratégie bruyante pour la collecte de données. Les politiques peuvent être enregistrées/restaurées et peuvent être utilisées indépendamment de l'agent pour la collecte de données, l'évaluation, etc.

Certaines politiques sont plus faciles à écrire dans Tensorflow (par exemple celles avec un réseau de neurones), tandis que d'autres sont plus faciles à écrire en Python (par exemple en suivant un script d'actions). Ainsi, dans les agents TF, nous autorisons à la fois les politiques Python et Tensorflow. De plus, les stratégies écrites dans TensorFlow peuvent devoir être utilisées dans un environnement Python, ou vice versa, par exemple une stratégie TensorFlow est utilisée pour la formation mais déployée plus tard dans un environnement Python de production. Pour faciliter cela, nous fournissons des wrappers pour la conversion entre les stratégies Python et TensorFlow.

Une autre classe intéressante de politiques sont les wrappers de politiques, qui modifient une politique donnée d'une certaine manière, par exemple en ajoutant un type particulier de bruit, en créant une version gourmande ou epsilon-greedy d'une politique stochastique, en mélangeant au hasard plusieurs politiques, etc.

Installer

Si vous n'avez pas encore installé tf-agents, exécutez :

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

Politiques Python

L'interface pour les politiques Python est défini dans les policies/py_policy.PyPolicy . Les principales méthodes sont :

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

La méthode la plus importante est l' action(time_step) qui associe une time_step contenant une observation de l'environnement à un tuple nommé PolicyStep contenant les attributs suivants:

  • l' action : L'action à appliquer à l'environnement.
  • state : L'état de la politique (par exemple l'état RNN) à introduire dans le prochain appel à l' action.
  • info : informations latérales optionnelles telles que les probabilités de journaux d'action.

Le time_step_spec et action_spec sont des spécifications pour le pas de temps d'entrée et l'action de sortie. Les politiques ont également une reset à reset fonction qui est généralement utilisé pour réinitialiser l'état dans les politiques stateful. La update(new_policy) à self new_policy update(new_policy) fonction met à jour l' self vers new_policy .

Voyons maintenant quelques exemples de politiques Python.

Exemple 1 : politique Python aléatoire

Un exemple simple d'un PyPolicy est le RandomPyPolicy qui génère des actions aléatoires pour la action_spec donnée discret / continu. L'entrée time_step est ignorée.

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=())

Exemple 2 : politique Python scriptée

A joue la politique scriptées un script dos d'actions représentées comme une liste de (num_repeats, action) tuples. Chaque fois que l' action fonction est appelée, elle renvoie l'action suivante de la liste jusqu'à ce que le nombre spécifié de répétitions est fait, puis passe à l'action suivante dans la liste. La reset à reset méthode peut être appelée à commencer à exécuter à partir du début de la liste.

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=())

Règles TensorFlow

Les règles TensorFlow suivent la même interface que les règles Python. Voyons quelques exemples.

Exemple 1 : politique de TF aléatoire

Un RandomTFPolicy peut être utilisé pour générer des actions aléatoires selon un discret / continu donné action_spec . L'entrée time_step est ignorée.

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)

Exemple 2 : Politique d'acteur

Une politique d'acteur peut être créé à l' aide d'un réseau qui associe time_steps à des actions ou à un réseau qui associe time_steps aux distributions sur les actions.

Utiliser un réseau d'action

Définissons un réseau comme suit :

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

Dans TensorFlow, la plupart des couches réseau sont conçues pour les opérations par lots. Nous nous attendons donc à ce que les time_steps d'entrée soient groupés et que la sortie du réseau le soit également. Le réseau est également responsable de la production d'actions dans la plage correcte de l'action_spec donnée. Ceci est classiquement réalisé en utilisant par exemple un tanh activation de la couche finale pour produire des actions de [-1, 1], puis mise à l' échelle et le décalage de cette la plage correcte comme action_spec d'entrée (par exemple , voir tf_agents/agents/ddpg/networks.actor_network() ).

Maintenant, nous pouvons créer une politique d'acteur en utilisant le réseau ci-dessus.

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)

Nous pouvons l'appliquer à n'importe quel lot de time_steps qui suivent 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)

Dans l'exemple ci-dessus, nous avons créé la politique à l'aide d'un réseau d'action qui produit un tenseur d'action. Dans ce cas, policy.distribution(time_step) est une distribution déterministe (delta) autour de la sortie de policy.action(time_step) . Une façon de produire une politique stochastique consiste à envelopper la politique d'acteur dans un wrapper de politique qui ajoute du bruit aux actions. Une autre façon consiste à créer la politique d'acteur en utilisant un réseau de distribution d'action au lieu d'un réseau d'action comme indiqué ci-dessous.

Utiliser un réseau de distribution d'actions

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)

Notez que dans ce qui précède, les actions sont découpées dans la plage de la spécification d'action donnée [-1, 1]. C'est parce qu'un argument de constructeur d'ActorPolicy clip=True par défaut. Définir ce paramètre sur false renverra les actions non écrêtées produites par le réseau.

Les politiques Stochastic peuvent être converties en politiques déterministes en utilisant, par exemple, une enveloppe GreedyPolicy qui choisit stochastic_policy.distribution().mode() le distribution() stochastic_policy.distribution().mode() Comme son action, et une distribution déterministe / delta autour de cette action gourmande comme la distribution() .

Exemple 3 : politique Q

La politique AQ est utilisée dans des agents tels que DQN et est basée sur un réseau Q qui prédit une valeur Q pour chaque action discrète. Pour un pas de temps donné, la distribution d'action dans la politique Q est une distribution catégorielle créée en utilisant les valeurs q comme 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 politique

Un wrapper de politique peut être utilisé pour encapsuler et modifier une politique donnée, par exemple ajouter du bruit. Les wrappers de stratégie sont une sous-classe de Policy (Python/TensorFlow) et peuvent donc être utilisés comme n'importe quelle autre stratégie.

Exemple : politique gourmande

Une enveloppe gourmande peut être utilisée pour envelopper toute politique de tensorflow qui met en œuvre la distribution() . GreedyPolicy.action() retournera en wrapped_policy.distribution().mode() Et GreedyPolicy.distribution() est une distribution déterministe / delta autour 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)