Merken Sie den Termin vor! Google I / O kehrt vom 18. bis 20. Mai zurück Registrieren Sie sich jetzt
Diese Seite wurde von der Cloud Translation API übersetzt.
Switch to English

Tutorial zu mehrarmigen Banditen in TF-Agenten

Loslegen

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

Einrichten

Wenn Sie die folgenden Abhängigkeiten nicht installiert haben, führen Sie Folgendes aus:

pip install -q tf-agents

Importe

import abc
import numpy as np
import tensorflow as tf

from tf_agents.agents import tf_agent
from tf_agents.drivers import driver
from tf_agents.environments import py_environment
from tf_agents.environments import tf_environment
from tf_agents.environments import tf_py_environment
from tf_agents.policies import tf_policy
from tf_agents.specs import array_spec
from tf_agents.specs import tensor_spec
from tf_agents.trajectories import time_step as ts
from tf_agents.trajectories import trajectory
from tf_agents.trajectories import policy_step

# Clear any leftover state from previous colabs run.
# (This is not necessary for normal programs.)
tf.compat.v1.reset_default_graph()

tf.compat.v1.enable_resource_variables()
tf.compat.v1.enable_v2_behavior()
nest = tf.compat.v2.nest

Einführung

Das Multi-Armed Bandit-Problem (MAB) ist ein Sonderfall des Reinforcement Learning: Ein Agent sammelt Belohnungen in einer Umgebung, indem er einige Aktionen ausführt, nachdem er einen bestimmten Zustand der Umgebung beobachtet hat. Der Hauptunterschied zwischen allgemeinem RL und MAB besteht darin, dass wir in MAB davon ausgehen, dass die vom Agenten ergriffenen Maßnahmen keinen Einfluss auf den nächsten Zustand der Umgebung haben. Daher modellieren Agenten keine Zustandsübergänge, schreiben keine Belohnungen für vergangene Aktionen gut oder planen nicht voraus, um in belohnungsreiche Staaten zu gelangen.

Wie in anderen RL-Domänen besteht das Ziel eines MAB- Agenten darin, eine Richtlinie zu finden, die so viel Belohnung wie möglich sammelt. Es wäre jedoch ein Fehler, immer zu versuchen, die Aktion auszunutzen, die die höchste Belohnung verspricht, denn dann besteht die Möglichkeit, dass wir bessere Aktionen verpassen, wenn wir nicht genug erforschen. Dies ist das Hauptproblem, das in (MAB) gelöst werden muss und oft als Explorations-Exploitation-Dilemma bezeichnet wird .

Banditenumgebungen, Richtlinien und Agenten für MAB finden Sie in den Unterverzeichnissen von tf_agents / bandits .

Umgebungen

In TF-Agents dient die Umgebungsklasse dazu, Informationen über den aktuellen Status (dies wird als Beobachtung oder Kontext bezeichnet ) bereitzustellen, eine Aktion als Eingabe zu empfangen, einen Statusübergang durchzuführen und eine Belohnung auszugeben. Diese Klasse kümmert sich auch um das Zurücksetzen, wenn eine Episode endet, damit eine neue Episode beginnen kann. Dies wird durch Aufrufen einer reset Funktion realisiert, wenn ein Status als "letzter" der Episode gekennzeichnet ist.

Weitere Informationen finden Sie im Tutorial zu TF-Agents-Umgebungen .

Wie oben erwähnt, unterscheidet sich MAB von allgemeinem RL darin, dass Aktionen die nächste Beobachtung nicht beeinflussen. Ein weiterer Unterschied besteht darin, dass es bei Banditen keine "Episoden" gibt: Jeder Zeitschritt beginnt mit einer neuen Beobachtung, unabhängig von vorherigen Zeitschritten.

Um sicherzustellen, dass die Beobachtungen unabhängig sind und das Konzept der RL-Episoden abstrahieren, führen wir Unterklassen von PyEnvironment und TFEnvironment : BanditPyEnvironment und BanditTFEnvironment . Diese Klassen stellen zwei private Mitgliedsfunktionen bereit, die vom Benutzer noch implementiert werden müssen:

@abc.abstractmethod
def _observe(self):

und

@abc.abstractmethod
def _apply_action(self, action):

Die Funktion _observe gibt eine Beobachtung zurück. Anschließend wählt die Richtlinie eine Aktion aus, die auf dieser Beobachtung basiert. Die _apply_action empfängt diese Aktion als Eingabe und gibt die entsprechende Belohnung zurück. Diese privaten Mitgliedsfunktionen werden durch die Funktionen reset bzw. step aufgerufen.

class BanditPyEnvironment(py_environment.PyEnvironment):

  def __init__(self, observation_spec, action_spec):
    self._observation_spec = observation_spec
    self._action_spec = action_spec
    super(BanditPyEnvironment, self).__init__()

  # Helper functions.
  def action_spec(self):
    return self._action_spec

  def observation_spec(self):
    return self._observation_spec

  def _empty_observation(self):
    return tf.nest.map_structure(lambda x: np.zeros(x.shape, x.dtype),
                                 self.observation_spec())

  # These two functions below should not be overridden by subclasses.
  def _reset(self):
    """Returns a time step containing an observation."""
    return ts.restart(self._observe(), batch_size=self.batch_size)

  def _step(self, action):
    """Returns a time step containing the reward for the action taken."""
    reward = self._apply_action(action)
    return ts.termination(self._observe(), reward)

  # These two functions below are to be implemented in subclasses.
  @abc.abstractmethod
  def _observe(self):
    """Returns an observation."""

  @abc.abstractmethod
  def _apply_action(self, action):
    """Applies `action` to the Environment and returns the corresponding reward.
    """

Die obigen Zwischen abstrakter Klasse implementiert PyEnvironment ‚s _reset und _step Funktionen und macht die abstrakten Funktionen _observe und _apply_action durch Unterklassen implementiert werden.

Eine einfache Beispielumgebungsklasse

Die folgende Klasse gibt eine sehr einfache Umgebung an, für die die Beobachtung eine zufällige ganze Zahl zwischen -2 und 2 ist, es gibt 3 mögliche Aktionen (0, 1, 2) und die Belohnung ist das Produkt der Aktion und der Beobachtung.

class SimplePyEnvironment(BanditPyEnvironment):

  def __init__(self):
    action_spec = array_spec.BoundedArraySpec(
        shape=(), dtype=np.int32, minimum=0, maximum=2, name='action')
    observation_spec = array_spec.BoundedArraySpec(
        shape=(1,), dtype=np.int32, minimum=-2, maximum=2, name='observation')
    super(SimplePyEnvironment, self).__init__(observation_spec, action_spec)

  def _observe(self):
    self._observation = np.random.randint(-2, 3, (1,), dtype='int32')
    return self._observation

  def _apply_action(self, action):
    return action * self._observation

Jetzt können wir diese Umgebung nutzen, um Beobachtungen zu erhalten und Belohnungen für unsere Handlungen zu erhalten.

environment = SimplePyEnvironment()
observation = environment.reset().observation
print("observation: %d" % observation)

action = 2

print("action: %d" % action)
reward = environment.step(action).reward
print("reward: %f" % reward)
observation: 1
action: 2
reward: 2.000000

TF-Umgebungen

Man kann eine Banditenumgebung definieren, indem man BanditTFEnvironment , oder ähnlich wie in RL-Umgebungen kann man eine BanditPyEnvironment definieren und sie mit TFPyEnvironment . Der Einfachheit halber verwenden wir in diesem Tutorial die letztere Option.

tf_environment = tf_py_environment.TFPyEnvironment(environment)

Richtlinien

Eine Richtlinie in einem Banditenproblem funktioniert genauso wie in einem RL-Problem: Sie liefert eine Aktion (oder eine Verteilung von Aktionen), wenn eine Beobachtung als Eingabe gegeben ist.

Weitere Informationen finden Sie imLernprogramm zu TF-Agentenrichtlinien .

Wie in Umgebungen gibt es zwei Möglichkeiten, eine Richtlinie zu erstellen: Sie können eine PyPolicy erstellen und mit TFPyPolicy oder direkt eine TFPolicy erstellen. Hier entscheiden wir uns für die direkte Methode.

Da dieses Beispiel recht einfach ist, können wir die optimale Richtlinie manuell definieren. Die Aktion hängt nur vom Vorzeichen der Beobachtung ab, 0 wenn negativ und 2 wenn positiv.

class SignPolicy(tf_policy.TFPolicy):
  def __init__(self):
    observation_spec = tensor_spec.BoundedTensorSpec(
        shape=(1,), dtype=tf.int32, minimum=-2, maximum=2)
    time_step_spec = ts.time_step_spec(observation_spec)

    action_spec = tensor_spec.BoundedTensorSpec(
        shape=(), dtype=tf.int32, minimum=0, maximum=2)

    super(SignPolicy, self).__init__(time_step_spec=time_step_spec,
                                     action_spec=action_spec)
  def _distribution(self, time_step):
    pass

  def _variables(self):
    return ()

  def _action(self, time_step, policy_state, seed):
    observation_sign = tf.cast(tf.sign(time_step.observation[0]), dtype=tf.int32)
    action = observation_sign + 1
    return policy_step.PolicyStep(action, policy_state)

Jetzt können wir eine Beobachtung von der Umgebung anfordern, die Richtlinie aufrufen, um eine Aktion auszuwählen, und die Umgebung gibt die Belohnung aus:

sign_policy = SignPolicy()

current_time_step = tf_environment.reset()
print('Observation:')
print (current_time_step.observation)
action = sign_policy.action(current_time_step).action
print('Action:')
print (action)
reward = tf_environment.step(action).reward
print('Reward:')
print(reward)
Observation:
tf.Tensor([[2]], shape=(1, 1), dtype=int32)
Action:
tf.Tensor([2], shape=(1,), dtype=int32)
Reward:
tf.Tensor([[4.]], shape=(1, 1), dtype=float32)

Die Art und Weise, wie Banditenumgebungen implementiert werden, stellt sicher, dass wir bei jedem Schritt nicht nur die Belohnung für die von uns ergriffenen Maßnahmen erhalten, sondern auch die nächste Beobachtung.

step = tf_environment.reset()
action = 1
next_step = tf_environment.step(action)
reward = next_step.reward
next_observation = next_step.observation
print("Reward: ")
print(reward)
print("Next observation:")
print(next_observation)
Reward: 
tf.Tensor([[0.]], shape=(1, 1), dtype=float32)
Next observation:
tf.Tensor([[2]], shape=(1, 1), dtype=int32)

Agenten

Jetzt, da wir Banditenumgebungen und Banditenrichtlinien haben, ist es an der Zeit, auch Banditenagenten zu definieren, die sich darum kümmern, die Richtlinie basierend auf Trainingsbeispielen zu ändern.

Die API für Banditenagenten unterscheidet sich nicht von der von RL-Agenten: Der Agent muss _train Methoden _initialize und _train implementieren und eine policy und eine collect_policy .

Eine kompliziertere Umgebung

Bevor wir unseren Banditenagenten schreiben, müssen wir eine Umgebung haben, die etwas schwieriger herauszufinden ist. Um die Dinge ein wenig aufzupeppen, gibt die nächste Umgebung entweder immer reward = observation * action oder reward = -observation * action . Dies wird bei der Initialisierung der Umgebung entschieden.

class TwoWayPyEnvironment(BanditPyEnvironment):

  def __init__(self):
    action_spec = array_spec.BoundedArraySpec(
        shape=(), dtype=np.int32, minimum=0, maximum=2, name='action')
    observation_spec = array_spec.BoundedArraySpec(
        shape=(1,), dtype=np.int32, minimum=-2, maximum=2, name='observation')

    # Flipping the sign with probability 1/2.
    self._reward_sign = 2 * np.random.randint(2) - 1
    print("reward sign:")
    print(self._reward_sign)

    super(TwoWayPyEnvironment, self).__init__(observation_spec, action_spec)

  def _observe(self):
    self._observation = np.random.randint(-2, 3, (1,), dtype='int32')
    return self._observation

  def _apply_action(self, action):
    return self._reward_sign * action * self._observation[0]

two_way_tf_environment = tf_py_environment.TFPyEnvironment(TwoWayPyEnvironment())
reward sign:
1

Eine kompliziertere Politik

Eine kompliziertere Umgebung erfordert eine kompliziertere Politik. Wir brauchen eine Richtlinie, die das Verhalten der zugrunde liegenden Umgebung erkennt. Es gibt drei Situationen, mit denen die Richtlinie umgehen muss:

  1. Der Agent hat noch nicht erkannt, welche Version der Umgebung ausgeführt wird.
  2. Der Agent hat festgestellt, dass die ursprüngliche Version der Umgebung ausgeführt wird.
  3. Der Agent hat festgestellt, dass die gespiegelte Version der Umgebung ausgeführt wird.

Wir definieren eine tf_variable Namen _situation , um diese als Werte in [0, 2] codierten Informationen zu speichern, und sorgen dann dafür, dass sich die Richtlinie entsprechend verhält.

class TwoWaySignPolicy(tf_policy.TFPolicy):
  def __init__(self, situation):
    observation_spec = tensor_spec.BoundedTensorSpec(
        shape=(1,), dtype=tf.int32, minimum=-2, maximum=2)
    action_spec = tensor_spec.BoundedTensorSpec(
        shape=(), dtype=tf.int32, minimum=0, maximum=2)
    time_step_spec = ts.time_step_spec(observation_spec)
    self._situation = situation
    super(TwoWaySignPolicy, self).__init__(time_step_spec=time_step_spec,
                                           action_spec=action_spec)
  def _distribution(self, time_step):
    pass

  def _variables(self):
    return [self._situation]

  def _action(self, time_step, policy_state, seed):
    sign = tf.cast(tf.sign(time_step.observation[0, 0]), dtype=tf.int32)
    def case_unknown_fn():
      # Choose 1 so that we get information on the sign.
      return tf.constant(1, shape=(1,))

    # Choose 0 or 2, depending on the situation and the sign of the observation.
    def case_normal_fn():
      return tf.constant(sign + 1, shape=(1,))
    def case_flipped_fn():
      return tf.constant(1 - sign, shape=(1,))

    cases = [(tf.equal(self._situation, 0), case_unknown_fn),
             (tf.equal(self._situation, 1), case_normal_fn),
             (tf.equal(self._situation, 2), case_flipped_fn)]
    action = tf.case(cases, exclusive=True)
    return policy_step.PolicyStep(action, policy_state)

Der Agent

Jetzt ist es Zeit, den Agenten zu definieren, der das Vorzeichen der Umgebung erkennt und die Richtlinie entsprechend festlegt.

class SignAgent(tf_agent.TFAgent):
  def __init__(self):
    self._situation = tf.compat.v2.Variable(0, dtype=tf.int32)
    policy = TwoWaySignPolicy(self._situation)
    time_step_spec = policy.time_step_spec
    action_spec = policy.action_spec
    super(SignAgent, self).__init__(time_step_spec=time_step_spec,
                                    action_spec=action_spec,
                                    policy=policy,
                                    collect_policy=policy,
                                    train_sequence_length=None)

  def _initialize(self):
    return tf.compat.v1.variables_initializer(self.variables)

  def _train(self, experience, weights=None):
    observation = experience.observation
    action = experience.action
    reward = experience.reward

    # We only need to change the value of the situation variable if it is
    # unknown (0) right now, and we can infer the situation only if the
    # observation is not 0.
    needs_action = tf.logical_and(tf.equal(self._situation, 0),
                                  tf.not_equal(reward, 0))


    def new_situation_fn():
      """This returns either 1 or 2, depending on the signs."""
      return (3 - tf.sign(tf.cast(observation[0, 0, 0], dtype=tf.int32) *
                          tf.cast(action[0, 0], dtype=tf.int32) *
                          tf.cast(reward[0, 0], dtype=tf.int32))) / 2

    new_situation = tf.cond(needs_action,
                            new_situation_fn,
                            lambda: self._situation)
    new_situation = tf.cast(new_situation, tf.int32)
    tf.compat.v1.assign(self._situation, new_situation)
    return tf_agent.LossInfo((), ())

sign_agent = SignAgent()

Im obigen Code definiert der Agent die Richtlinie, und die variable situation wird vom Agenten und der Richtlinie gemeinsam genutzt.

Auch die Parameter experience der _train ist Funktion eine Trajektorie:

Flugbahnen

In TF-Agenten werden trajectories als Tupel bezeichnet, die Stichproben aus früheren Schritten enthalten. Diese Beispiele werden dann vom Agenten verwendet, um die Richtlinie zu trainieren und zu aktualisieren. In RL müssen Trajektorien Informationen über den aktuellen Status, den nächsten Status und darüber enthalten, ob die aktuelle Episode beendet wurde. Da wir in der Banditenwelt diese Dinge nicht brauchen, haben wir eine Hilfsfunktion eingerichtet, um eine Flugbahn zu erstellen:

# We need to add another dimension here because the agent expects the
# trajectory of shape [batch_size, time, ...], but in this tutorial we assume
# that both batch size and time are 1. Hence all the expand_dims.

def trajectory_for_bandit(initial_step, action_step, final_step):
  return trajectory.Trajectory(observation=tf.expand_dims(initial_step.observation, 0),
                               action=tf.expand_dims(action_step.action, 0),
                               policy_info=action_step.info,
                               reward=tf.expand_dims(final_step.reward, 0),
                               discount=tf.expand_dims(final_step.discount, 0),
                               step_type=tf.expand_dims(initial_step.step_type, 0),
                               next_step_type=tf.expand_dims(final_step.step_type, 0))

Schulung eines Agenten

Jetzt sind alle Teile bereit, um unseren Banditenagenten auszubilden.

step = two_way_tf_environment.reset()
for _ in range(10):
  action_step = sign_agent.collect_policy.action(step)
  next_step = two_way_tf_environment.step(action_step.action)
  experience = trajectory_for_bandit(step, action_step, next_step)
  print(experience)
  sign_agent.train(experience)
  step = next_step
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[0]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[-1]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[1]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[-1.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[0]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[1]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[0]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[1]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[-2]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[0]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[0]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[1]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[-2]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[0]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[-1]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[0]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[2]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[4.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[1]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[2.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)
Trajectory(step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, observation=<tf.Tensor: shape=(1, 1, 1), dtype=int32, numpy=array([[[2]]], dtype=int32)>, action=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, policy_info=(), next_step_type=<tf.Tensor: shape=(1, 1), dtype=int32, numpy=array([[2]], dtype=int32)>, reward=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[4.]], dtype=float32)>, discount=<tf.Tensor: shape=(1, 1), dtype=float32, numpy=array([[0.]], dtype=float32)>)

Aus der Ausgabe kann man ersehen, dass nach dem zweiten Schritt (es sei denn, die Beobachtung war im ersten Schritt 0) die Richtlinie die Aktion auf die richtige Weise auswählt und daher die gesammelte Belohnung immer nicht negativ ist.

Ein echtes kontextuelles Banditenbeispiel

Im Rest dieses Tutorials verwenden wir die vorimplementierten Umgebungen und Agenten der TF-Agents Bandits-Bibliothek.

# Imports for example.
from tf_agents.bandits.agents import lin_ucb_agent
from tf_agents.bandits.environments import stationary_stochastic_py_environment as sspe
from tf_agents.bandits.metrics import tf_metrics
from tf_agents.drivers import dynamic_step_driver
from tf_agents.replay_buffers import tf_uniform_replay_buffer

import matplotlib.pyplot as plt

Stationäre stochastische Umgebung mit linearen Auszahlungsfunktionen

Die in diesem Beispiel verwendete Umgebung ist die StationaryStochasticPyEnvironment . Diese Umgebung verwendet als Parameter eine (normalerweise verrauschte) Funktion zum Geben von Beobachtungen (Kontext) und für jeden Arm eine (ebenfalls verrauschte) Funktion, die die Belohnung basierend auf der gegebenen Beobachtung berechnet. In unserem Beispiel wird der Kontext gleichmäßig aus einem d-dimensionalen Würfel abgetastet, und die Belohnungsfunktionen sind lineare Funktionen des Kontexts sowie etwas Gaußsches Rauschen.

batch_size = 2 # @param
arm0_param = [-3, 0, 1, -2] # @param
arm1_param = [1, -2, 3, 0] # @param
arm2_param = [0, 0, 1, 1] # @param
def context_sampling_fn(batch_size):
  """Contexts from [-10, 10]^4."""
  def _context_sampling_fn():
    return np.random.randint(-10, 10, [batch_size, 4]).astype(np.float32)
  return _context_sampling_fn

class LinearNormalReward(object):
  """A class that acts as linear reward function when called."""
  def __init__(self, theta, sigma):
    self.theta = theta
    self.sigma = sigma
  def __call__(self, x):
    mu = np.dot(x, self.theta)
    return np.random.normal(mu, self.sigma)

arm0_reward_fn = LinearNormalReward(arm0_param, 1)
arm1_reward_fn = LinearNormalReward(arm1_param, 1)
arm2_reward_fn = LinearNormalReward(arm2_param, 1)

environment = tf_py_environment.TFPyEnvironment(
    sspe.StationaryStochasticPyEnvironment(
        context_sampling_fn(batch_size),
        [arm0_reward_fn, arm1_reward_fn, arm2_reward_fn],
        batch_size=batch_size))

Der LinUCB-Agent

Der folgende Agent implementiert den LinUCB- Algorithmus.

observation_spec = tensor_spec.TensorSpec([4], tf.float32)
time_step_spec = ts.time_step_spec(observation_spec)
action_spec = tensor_spec.BoundedTensorSpec(
    dtype=tf.int32, shape=(), minimum=0, maximum=2)

agent = lin_ucb_agent.LinearUCBAgent(time_step_spec=time_step_spec,
                                     action_spec=action_spec)

Bedauern Sie die Metrik

Die wichtigste Messgröße der Banditen ist das Bedauern , berechnet als Differenz zwischen der vom Agenten gesammelten Belohnung und der erwarteten Belohnung einer Orakelrichtlinie, die Zugriff auf die Belohnungsfunktionen der Umgebung hat. Das RegretMetric benötigt daher eine Baseline_Reward_Fn- Funktion, die die bestmögliche erwartete Belohnung bei einer Beobachtung berechnet. In unserem Beispiel müssen wir das Maximum der rauschfreien Äquivalente der Belohnungsfunktionen verwenden, die wir bereits für die Umgebung definiert haben.

def compute_optimal_reward(observation):
  expected_reward_for_arms = [
      tf.linalg.matvec(observation, tf.cast(arm0_param, dtype=tf.float32)),
      tf.linalg.matvec(observation, tf.cast(arm1_param, dtype=tf.float32)),
      tf.linalg.matvec(observation, tf.cast(arm2_param, dtype=tf.float32))]
  optimal_action_reward = tf.reduce_max(expected_reward_for_arms, axis=0)
  return optimal_action_reward

regret_metric = tf_metrics.RegretMetric(compute_optimal_reward)

Ausbildung

Jetzt stellen wir alle Komponenten zusammen, die wir oben eingeführt haben: die Umgebung, die Richtlinie und den Agenten. Wir führen die Richtlinien für die Umgebung aus und geben mithilfe eines Fahrers Trainingsdaten aus und schulen den Agenten anhand der Daten.

Beachten Sie, dass es zwei Parameter gibt, die zusammen die Anzahl der durchgeführten Schritte angeben. num_iterations gibt an, wie oft die Trainerschleife ausgeführt wird, während der Treiber pro Iteration Schritte_per_loop steps_per_loop . Der Hauptgrund für das Beibehalten dieser beiden Parameter besteht darin, dass einige Vorgänge pro Iteration ausgeführt werden, während andere vom Treiber in jedem Schritt ausgeführt werden. Zum Beispiel des Agenten train ist Funktion nur einmal pro Iteration genannt. Der Nachteil dabei ist, dass wenn wir öfter trainieren, unsere Politik "frischer" ist. Andererseits kann das Training in größeren Chargen zeiteffizienter sein.

num_iterations = 90 # @param
steps_per_loop = 1 # @param

replay_buffer = tf_uniform_replay_buffer.TFUniformReplayBuffer(
    data_spec=agent.policy.trajectory_spec,
    batch_size=batch_size,
    max_length=steps_per_loop)

observers = [replay_buffer.add_batch, regret_metric]

driver = dynamic_step_driver.DynamicStepDriver(
    env=environment,
    policy=agent.collect_policy,
    num_steps=steps_per_loop * batch_size,
    observers=observers)

regret_values = []

for _ in range(num_iterations):
  driver.run()
  loss_info = agent.train(replay_buffer.gather_all())
  replay_buffer.clear()
  regret_values.append(regret_metric.result())

plt.plot(regret_values)
plt.ylabel('Average Regret')
plt.xlabel('Number of Iterations')
WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.6/site-packages/tf_agents/drivers/dynamic_step_driver.py:203: calling while_loop_v2 (from tensorflow.python.ops.control_flow_ops) with back_prop=False is deprecated and will be removed in a future version.
Instructions for updating:
back_prop=False is deprecated. Consider using tf.stop_gradient instead.
Instead of:
results = tf.while_loop(c, b, vars, back_prop=False)
Use:
results = tf.nest.map_structure(tf.stop_gradient, tf.while_loop(c, b, vars))
WARNING:tensorflow:From <ipython-input-1-0cbf469acdfe>:21: ReplayBuffer.gather_all (from tf_agents.replay_buffers.replay_buffer) is deprecated and will be removed in a future version.
Instructions for updating:
Use `as_dataset(..., single_deterministic_pass=True)` instead.
Text(0.5, 0, 'Number of Iterations')

png

Nach dem Ausführen des letzten Code-Snippets zeigt das resultierende Diagramm (hoffentlich), dass das durchschnittliche Bedauern sinkt, wenn der Agent geschult wird, und die Richtlinie wird besser darin, herauszufinden, was die richtige Aktion ist, wenn man die Beobachtung berücksichtigt.

Was kommt als nächstes?

Weitere Arbeitsbeispiele finden Sie im Verzeichnis bandits / agentes / examples , das ausführbare Beispiele für verschiedene Agenten und Umgebungen enthält.

Die TF-Agents-Bibliothek kann auch mehrarmige Banditen mit Funktionen pro Arm verarbeiten. Zu diesem Zweck verweisen wir den Leser auf das Banditen- Tutorial pro Arm.