Odpowiedz już dziś na lokalne wydarzenie TensorFlow Everywhere!
Ta strona została przetłumaczona przez Cloud Translation API.
Switch to English

Samouczek o Multi Armed Bandits w TF-Agents

Zaczynaj

Zobacz na TensorFlow.org Uruchom w Google Colab Wyświetl źródło na GitHub Pobierz notatnik

Ustawiać

Jeśli nie zainstalowałeś następujących zależności, uruchom:

pip install -q tf-agents

Import

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

Wprowadzenie

Problem Multi-Armed Bandit (MAB) to szczególny przypadek uczenia się ze wzmocnieniem: agent zbiera nagrody w środowisku, podejmując pewne działania po zaobserwowaniu pewnego stanu otoczenia. Główną różnicą między ogólnym RL i MAB jest to, że w MAB zakładamy, że działanie podjęte przez agenta nie wpływa na kolejny stan środowiska. Dlatego agenci nie modelują zmian stanu, nagród kredytowych dla przeszłych działań ani nie „planują z wyprzedzeniem”, aby dostać się do stanów bogatych w nagrody.

Podobnie jak w innych domenach RL, celem agenta MAB jest znalezienie zasady, która zapewnia jak największe wynagrodzenie. Błędem byłoby jednak zawsze próbować wykorzystywać akcję, która obiecuje najwyższą nagrodę, ponieważ wtedy jest szansa, że ​​przegapimy lepsze działania, jeśli nie będziemy wystarczająco eksplorować. To jest główny problem do rozwiązania w (MAB), często nazywany dylematem eksploracyjno-eksploatacyjnym .

Środowiska, zasady i agenty bandytów dla MAB można znaleźć w podkatalogach tf_agents / bandits .

Środowiska

W TF-Agents, klasa środowiska pełni rolę dostarczania informacji o aktualnym stanie (nazywa się to obserwacją lub kontekstem ), odbierania działania jako danych wejściowych, przeprowadzania przejścia między stanami i wydawania nagrody. Ta klasa dba również o resetowanie po zakończeniu odcinka, dzięki czemu można rozpocząć nowy odcinek. Jest to realizowane przez wywołanie funkcji reset gdy stan jest oznaczony jako „ostatni” odcinka.

Aby uzyskać więcej informacji, zobacz samouczek dotyczący środowisk TF-Agents .

Jak wspomniano powyżej, MAB różni się od ogólnego RL tym, że działania nie wpływają na następną obserwację. Inną różnicą jest to, że w Bandits nie ma „epizodów”: każdy krok czasowy rozpoczyna się od nowej obserwacji, niezależnie od poprzednich kroków czasowych.

Aby upewnić się, że obserwacje są niezależne i odciąć koncepcję epizodów RL, wprowadzamy podklasy PyEnvironment i TFEnvironment : BanditPyEnvironment i BanditTFEnvironment . Te klasy udostępniają dwie prywatne funkcje składowe, które pozostają do zaimplementowania przez użytkownika:

@abc.abstractmethod
def _observe(self):

i

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

Funkcja _observe zwraca obserwację. Następnie polityka wybiera działanie na podstawie tej obserwacji. _apply_action odbiera tę akcję jako dane wejściowe i zwraca odpowiednią nagrodę. Te prywatne funkcje członkowskie są wywoływane odpowiednio przez funkcje reset i step .

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.
    """

Powyższa tymczasowa klasa abstrakcyjna implementuje PyEnvironment _reset i _step i udostępnia funkcje abstrakcyjne _observe i _apply_action do zaimplementowania przez podklasy.

Prosta przykładowa klasa środowiska

Poniższa klasa daje bardzo proste środowisko, dla którego obserwacja jest losową liczbą całkowitą z przedziału od -2 do 2, istnieją 3 możliwe akcje (0, 1, 2), a nagroda jest iloczynem działania i obserwacji.

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

Teraz możemy używać tego środowiska do zbierania obserwacji i otrzymywania nagród za nasze działania.

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

Środowiska TF

Środowisko bandytów można zdefiniować przez podklasę BanditTFEnvironment lub podobnie do środowisk RL można zdefiniować środowisko BanditPyEnvironment i opakować je w TFPyEnvironment . Dla uproszczenia korzystamy z tej drugiej opcji w tym samouczku.

tf_environment = tf_py_environment.TFPyEnvironment(environment)

Zasady

Polityka w problemie bandyty działa tak samo, jak w przypadku problemu RL: zapewnia akcję (lub rozkład działań), biorąc pod uwagę obserwację jako dane wejściowe.

Aby uzyskać więcej informacji, zobaczsamouczek dotyczący zasad TF-Agents .

Podobnie jak w przypadku środowisk, istnieją dwa sposoby konstruowania zasad: jeden można utworzyć PyPolicy i opakować go w TFPyPolicy lub bezpośrednio utworzyć TFPolicy . Tutaj wybieramy metodę bezpośrednią.

Ponieważ ten przykład jest dość prosty, możemy ręcznie zdefiniować optymalną politykę. Działanie zależy tylko od znaku obserwacji, 0, gdy jest ujemne, a 2, gdy jest dodatnie.

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)

Teraz możemy poprosić środowisko o obserwację, zadzwonić do polityki, aby wybrać działanie, a następnie środowisko wypisze nagrodę:

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([[-1]], shape=(1, 1), dtype=int32)
Action:
tf.Tensor([0], shape=(1,), dtype=int32)
Reward:
tf.Tensor([[0.]], shape=(1, 1), dtype=float32)

Sposób zaimplementowania środowisk bandyckich sprawia, że ​​za każdym razem, gdy robimy krok, otrzymujemy nie tylko nagrodę za wykonaną akcję, ale także kolejną obserwację.

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([[-1.]], shape=(1, 1), dtype=float32)
Next observation:
tf.Tensor([[1]], shape=(1, 1), dtype=int32)

Agenci

Teraz, gdy mamy środowiska bandytów i politykę bandytów, czas również zdefiniować agentów bandytów, którzy zajmą się zmianą polityki w oparciu o próbki szkoleniowe.

API dla agentów bandytów nie różni się od interfejsu agentów RL: agent musi po prostu zaimplementować metody _initialize i _train oraz zdefiniować policy i collect_policy .

Bardziej skomplikowane środowisko

Zanim napiszemy do naszego agenta bandytów, musimy stworzyć nieco trudniejsze do rozgryzienia środowisko. Aby trochę urozmaicić, następne środowisko zawsze będzie dawało reward = observation * action lub reward = -observation * action . Decyzja zostanie podjęta po zainicjowaniu środowiska.

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

Bardziej skomplikowana polityka

Bardziej skomplikowane środowisko wymaga bardziej skomplikowanej polityki. Potrzebujemy polityki, która wykrywa zachowanie środowiska bazowego. Istnieją trzy sytuacje, w których polityka musi uwzględniać:

  1. Agent nie wykrył jeszcze, która wersja środowiska jest uruchomiona.
  2. Agent wykrył, że działa oryginalna wersja środowiska.
  3. Agent wykrył, że odwrócona wersja środowiska jest uruchomiona.

Definiujemy tf_variable nazwie _situation aby przechowywać te informacje zakodowane jako wartości w [0, 2] , a następnie sprawiamy, że polityka zachowuje się odpowiednio.

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)

Agent

Teraz nadszedł czas, aby zdefiniować agenta, który wykrywa znak środowiska i odpowiednio ustawia zasady.

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

W powyższym kodzie agent definiuje politykę, a zmienna situation jest wspólna dla agenta i polityki.

Również experience parametrów funkcji _train jest trajektorią:

Trajektorie

W TF-Agentach trajectories nazywane są krotkami, które zawierają próbki z poprzednich kroków. Te przykłady są następnie używane przez agenta do szkolenia i aktualizowania zasad. W RL trajektorie muszą zawierać informacje o obecnym stanie, następnym stanie i tym, czy bieżący odcinek się zakończył. Ponieważ w świecie Bandytów nie potrzebujemy tych rzeczy, ustawiliśmy funkcję pomocniczą do tworzenia trajektorii:

# 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))

Szkolenie agenta

Teraz wszystkie elementy są gotowe do szkolenia naszego agenta bandytów.

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

Z danych wyjściowych widać, że po drugim kroku (chyba, że ​​w pierwszym kroku obserwacja wynosiła 0), polityka wybiera działanie we właściwy sposób, a zatem zebrana nagroda jest zawsze nieujemna.

Prawdziwy kontekstowy przykład bandyty

W pozostałej części tego samouczka używamy wstępnie zaimplementowanych środowisk i agentów biblioteki TF-Agents Bandits.

# 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

Stacjonarne środowisko stochastyczne z liniowymi funkcjami wypłaty

Środowisko użyte w tym przykładzie to StationaryStochasticPyEnvironment . Środowisko to przyjmuje jako parametr (zwykle hałaśliwą) funkcję dającą obserwacje (kontekst), a dla każdego ramienia przyjmuje (również hałaśliwą) funkcję, która oblicza nagrodę na podstawie danej obserwacji. W naszym przykładzie pobieramy kontekst jednolicie z sześcianu o wymiarach d, a funkcje nagrody są funkcjami liniowymi kontekstu oraz pewnym szumem Gaussa.

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

Agent LinUCB

Poniższy agent implementuje algorytm LinUCB .

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)

Regret Metric

Najważniejszą miarą bandytów jest żal , obliczany jako różnica między nagrodą zebraną przez agenta a oczekiwaną nagrodą polityki wyroczni, która ma dostęp do funkcji nagrody w środowisku. RegretMetric potrzebuje zatem funkcji baseline_reward_fn, która oblicza najlepszą możliwą do osiągnięcia oczekiwaną nagrodę na podstawie obserwacji. W naszym przykładzie musimy przyjąć maksimum bezszumowych odpowiedników funkcji nagrody, które już zdefiniowaliśmy dla środowiska.

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)

Trening

Teraz zebraliśmy wszystkie komponenty, które wprowadziliśmy powyżej: środowisko, politykę i agenta. Prowadzimy politykę dotyczącą środowiska i wyprowadzamy dane szkoleniowe z pomocą kierowcy oraz szkolimy agenta w zakresie tych danych.

Zauważ, że istnieją dwa parametry, które razem określają liczbę wykonanych kroków. num_iterations określa, ile razy uruchamiamy pętlę trenera, podczas gdy sterownik będzie wykonywał kroki steps_per_loop na iterację. Głównym powodem zachowania obu tych parametrów jest to, że niektóre operacje są wykonywane na iterację, a niektóre są wykonywane przez sterownik na każdym kroku. Na przykład funkcja train agenta jest wywoływana tylko raz na iterację. Kompromis polega na tym, że jeśli trenujemy częściej, nasza polityka jest „świeższa”, z drugiej strony szkolenie w większych seriach może być bardziej efektywne czasowo.

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

Po uruchomieniu ostatniego fragmentu kodu wynikowy wykres (miejmy nadzieję) pokazuje, że średni żal spada, gdy agent jest szkolony, a polityka staje się lepsza w określaniu właściwego działania, biorąc pod uwagę obserwację.

Co dalej?

Aby zobaczyć więcej działających przykładów, zapoznaj się z katalogiem bandits / agent / examples, który zawiera gotowe do uruchomienia przykłady dla różnych agentów i środowisk.

Biblioteka TF-Agents jest również w stanie obsługiwać wielorękich bandytów z funkcjami na ramię. W tym celu odsyłamy czytelnika do samouczka dotyczącego bandytów na ramię.