Richtlinien

Ansicht auf TensorFlow.org In Google Colab ausführen Quelle auf GitHub anzeigen Notizbuch herunterladen

Einführung

In der Terminologie des Reinforcement Learning ordnen Richtlinien eine Beobachtung aus der Umgebung einer Aktion oder einer Verteilung über Aktionen zu. In TF-Agenten sind Beobachtungen aus der Umgebung in einem benannten Tupel TimeStep('step_type', 'discount', 'reward', 'observation') , und Richtlinien ordnen Zeitschritte Aktionen oder Verteilungen über Aktionen zu. Die meisten Richtlinien verwenden timestep.observation , einige Richtlinien verwenden timestep.step_type (z. B. um den Status zu Beginn einer Episode in statusbehafteten Richtlinien zurückzusetzen), aber timestep.discount und timestep.reward werden normalerweise ignoriert.

Richtlinien beziehen sich auf folgende Weise auf andere Komponenten in TF-Agents. Die meisten Richtlinien verfügen über ein neuronales Netzwerk zum Berechnen von Aktionen und / oder Verteilungen über Aktionen von TimeSteps. Agenten können eine oder mehrere Richtlinien für verschiedene Zwecke enthalten, z. B. eine Hauptrichtlinie, die für die Bereitstellung geschult wird, und eine verrauschte Richtlinie für die Datenerfassung. Richtlinien können gespeichert / wiederhergestellt und unabhängig vom Agenten für die Datenerfassung, Auswertung usw. verwendet werden.

Einige Richtlinien sind in Tensorflow einfacher zu schreiben (z. B. solche mit einem neuronalen Netzwerk), während andere in Python einfacher zu schreiben sind (z. B. nach einem Aktionsskript). Daher erlauben wir in TF-Agenten sowohl Python- als auch Tensorflow-Richtlinien. Darüber hinaus müssen in TensorFlow geschriebene Richtlinien möglicherweise in einer Python-Umgebung verwendet werden oder umgekehrt, z. B. wird eine TensorFlow-Richtlinie für Schulungen verwendet, später jedoch in einer Python-Produktionsumgebung bereitgestellt. Um dies zu vereinfachen, stellen wir Wrapper für die Konvertierung zwischen Python- und TensorFlow-Richtlinien bereit.

Eine weitere interessante Klasse von Richtlinien sind Richtlinien-Wrapper, die eine bestimmte Richtlinie auf eine bestimmte Weise ändern, z. B. eine bestimmte Art von Rauschen hinzufügen, eine gierige oder epsilon-gierige Version einer stochastischen Richtlinie erstellen, mehrere Richtlinien zufällig mischen usw.

Einrichten

Wenn Sie tf-agent noch nicht installiert haben, führen Sie Folgendes aus:

pip install -q 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

tf.compat.v1.enable_v2_behavior()

Python-Richtlinien

Die Schnittstelle für Python-Richtlinien ist in policies/py_policy.PyPolicy . Die Hauptmethoden sind:

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

Die wichtigste Methode ist action(time_step) die einen time_step mit einer Beobachtung aus der Umgebung einem time_step mit dem Namen tuple mit den folgenden Attributen time_step :

  • action : Die Aktion, die auf die Umgebung angewendet werden soll.
  • state : Der Status der Richtlinie (z. B. RNN-Status), die in den nächsten Aufruf zum Handeln eingespeist werden soll.
  • info : Optionale Nebeninformationen wie Aktionsprotokollwahrscheinlichkeiten.

time_step_spec und action_spec sind Spezifikationen für den Eingabezeitschritt und die Ausgabeaktion. Richtlinien verfügen auch über eine reset , die normalerweise zum Zurücksetzen des Status in statusbehafteten Richtlinien verwendet wird. Die Funktion update(new_policy) aktualisiert sich self Richtung new_policy .

Schauen wir uns nun einige Beispiele für Python-Richtlinien an.

Beispiel 1: Zufällige Python-Richtlinie

Ein einfaches Beispiel für eine PyPolicy ist die RandomPyPolicy , die zufällige Aktionen für die diskrete / kontinuierliche gegebene action_spec generiert. Die Eingabe time_step wird ignoriert.

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([-6,  2], dtype=int32), state=(), info=())
PolicyStep(action=array([-7,  7], dtype=int32), state=(), info=())

Beispiel 2: Skriptgesteuerte Python-Richtlinie

Eine (num_repeats, action) ein Skript mit Aktionen wieder, das als Liste von (num_repeats, action) Tupeln dargestellt wird. Jedes Mal, wenn die action aufgerufen wird, gibt sie die nächste Aktion aus der Liste zurück, bis die angegebene Anzahl von Wiederholungen ausgeführt ist, und fährt dann mit der nächsten Aktion in der Liste fort. Die reset Methode kann aufgerufen werden, um die Ausführung am Anfang der Liste zu starten.

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

TensorFlow-Richtlinien

TensorFlow-Richtlinien folgen derselben Schnittstelle wie Python-Richtlinien. Schauen wir uns einige Beispiele an.

Beispiel 1: Zufällige TF-Richtlinie

Eine RandomTFPolicy kann verwendet werden, um zufällige Aktionen gemäß einer bestimmten diskreten / kontinuierlichen action_spec . Die Eingabe time_step wird ignoriert.

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([ 1.8538141  -0.95033884], shape=(2,), dtype=float32)

Beispiel 2: Akteursrichtlinie

Eine time_steps kann entweder mithilfe eines Netzwerks erstellt werden, das time_steps Aktionen time_steps , oder mithilfe eines Netzwerks, das time_steps Verteilungen über Aktionen time_steps .

Verwenden eines Aktionsnetzwerks

Definieren wir ein Netzwerk wie folgt:

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 sind die meisten Netzwerkschichten für Stapeloperationen ausgelegt. Daher erwarten wir, dass die Eingabezeitschritte gestapelt werden und die Ausgabe des Netzwerks ebenfalls gestapelt wird. Das Netzwerk ist auch dafür verantwortlich, Aktionen im richtigen Bereich der angegebenen action_spec zu erstellen. Dies erfolgt üblicherweise unter Verwendung einer z. B. einer tanh-Aktivierung für die letzte Ebene, um Aktionen in [-1, 1] zu erzeugen, und anschließendem Skalieren und Verschieben dieser in den richtigen Bereich als Eingabe action_spec (z. B. siehe tf_agents/agents/ddpg/networks.actor_network() ).

Jetzt können wir mithilfe des oben genannten Netzwerks eine Akteursrichtlinie erstellen.

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)

Wir können es auf jeden Stapel von Zeitschritten anwenden, die auf time_step_spec folgen:

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.06422257  0.6671298   0.734995  ]
 [-0.06422257  0.6671298   0.734995  ]], shape=(2, 3), dtype=float32)
Action distribution:
tfp.distributions.Deterministic("Deterministic", batch_shape=[2, 3], event_shape=[], dtype=float32)

Im obigen Beispiel haben wir die Richtlinie mithilfe eines Aktionsnetzwerks erstellt, das einen Aktionstensor erzeugt. In diesem Fall ist policy.distribution(time_step) eine deterministische (Delta) Verteilung um die Ausgabe von policy.action(time_step) . Eine Möglichkeit, eine stochastische Richtlinie zu erstellen, besteht darin, die Akteurrichtlinie in einen Richtlinienumbruch zu verpacken, der den Aktionen Rauschen hinzufügt. Eine andere Möglichkeit besteht darin, die Akteursrichtlinie mithilfe eines Aktionsverteilungsnetzwerks anstelle eines Aktionsnetzwerks zu erstellen, wie unten gezeigt.

Verwenden eines Aktionsverteilungsnetzwerks

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(
[[ 1.         1.        -0.810334 ]
 [-0.5075199  1.         0.1654225]], shape=(2, 3), dtype=float32)
Action distribution:
tfp.distributions.MultivariateNormalDiag("ActionNet_MultivariateNormalDiag", batch_shape=[2], event_shape=[3], dtype=float32)

Beachten Sie, dass oben Aktionen auf den Bereich der angegebenen Aktionsspezifikation [-1, 1] beschränkt sind. Dies liegt daran, dass ein Konstruktorargument von ActorPolicy clip = True standardmäßig ist. Wenn Sie dies auf false setzen, werden nicht abgeschnittene Aktionen zurückgegeben, die vom Netzwerk erstellt wurden.

Stochastische Richtlinien können beispielsweise mithilfe eines GreedyPolicy-Wrappers, der stochastic_policy.distribution().mode() als Aktion auswählt, und einer deterministischen / Delta-Verteilung um diese gierige Aktion als distribution() in deterministische Richtlinien konvertiert werden.

Beispiel 3: Q-Richtlinie

Die AQ-Richtlinie wird in Agenten wie DQN verwendet und basiert auf einem Q-Netzwerk, das für jede einzelne Aktion einen Q-Wert vorhersagt. Für einen bestimmten Zeitschritt ist die Aktionsverteilung in der Q-Richtlinie eine kategoriale Verteilung, die unter Verwendung der q-Werte als Protokolle erstellt wird.

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([0 2], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.Categorical("Categorical", batch_shape=[2], event_shape=[], dtype=int32)

Policy Wrapper

Ein Richtlinien-Wrapper kann verwendet werden, um eine bestimmte Richtlinie zu verpacken und zu ändern, z. B. Rauschen hinzuzufügen. Richtlinien-Wrapper sind eine Unterklasse von Richtlinien (Python / TensorFlow) und können daher wie jede andere Richtlinie verwendet werden.

Beispiel: Gierige Politik

Ein gieriger Wrapper kann verwendet werden, um jede TensorFlow-Richtlinie zu verpacken, die distribution() implementiert. GreedyPolicy.action() gibt wrapped_policy.distribution().mode() und GreedyPolicy.distribution() ist eine deterministische / Delta-Verteilung um 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([0 0], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.DeterministicWithLogProb("Deterministic", batch_shape=[2], event_shape=[], dtype=int32)