bandits_tutorial.ipynb

Kernel: Python 3

Copyright 2023 The TF-Agents Authors.

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

TF-Agents における多腕バンディット問題のチュートリアル

はじめに

TensorFlow.org で表示

Google Colab で実行

GitHub でソースを表示

ノートブックをダウンロード

セットアップ

以下の依存関係をインストールしていない場合は、実行します。

!pip install tf-agents

インポート

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

nest = tf.nest

はじめに

タ腕バンディット問題（MAB）は、エージェントがある環境の状態を観察した後に何らかのアクションを取ることで、その環境における報酬を得るという強化学習の特殊なケースです。一般的な RL と MAB の主な違いは、MAB では、エージェントが環境の次の状態に影響を与えないことです。したがって、エージェントは状態遷移をモデル化したり、過去のアクションに対する報酬を与えたり、より多くの報酬を得るための「予測」を行いません。

ほかの RL 分野と同様に、MAB エージェントの目標は、できる限り多くの報酬を収集するポリシーを見つけ出すことです。ただし、十分に調べなかった場合により適したアクションを見逃す可能性があるため、最高の報酬を約束するアクションを常に使用しようとするのは間違いです。これが MAB で解決しなければならない主な問題であり、通常、探索と知識利用のジレンマと呼ばれています。

MAB のバンディット環境、ポリシー、およびエージェントは、tf_agents/bandits のサブディレクトリにあります。

環境

TF-Agents では、環境クラスは現在の状態（観測またはコンテキスト）に関する情報を提供し、アクションを入力として受け取って状態遷移を実行し、報酬を出力する役割があります。このクラスは、エピソードが終了したときに新しいエピソードが開始されるよう、リセットも行います。これは、状態のラベルがエピソードの「最後」となった時に reset 関数を呼び出して行われます。

詳細については、「TF-Agents 環境のチュートリアル」をご覧ください。

前述のとおり、MAB は、アクションが次の観測に影響を与えないという点で一般的な RL と異なりますが、もう一つの違いは、バンディットには、前の時間ステップから独立して新しい観測でステップが開始するたびに「エピソード」がないところにあります。

観測が確実に独立しており、RL エピソードの概念を中傷かするために、PyEnvironment と TFEnvironment のサブクラスである BanditPyEnvironment と BanditTFEnvironment を導入します。これらのクラスは、ユーザーが実装したままにする 2 つのプライベートメンバー関数を公開します。

@abc.abstractmethod
def _observe(self):

と

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

_observe 関数は観測を戻します。すると、ポリシーがこの観測に基づくアクションを選択します。_apply_action は、アクションを入力として受け取り、対応する報酬を戻します。これらのプライベートメンバー関数はそれぞれ、reset 関数と 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.
    """

上記の中間抽象クラスは PyEnvironment の _reset 関数と _step 関数を実装し、サブクラスが実装する抽象関数の _observe と _apply_action を公開します。

単純な環境クラスの例

以下のクラスは、観測が -2 から 2 のランダム整数で、3 つの可能なアクション (0, 1, 2) があり、報酬がこのアクションと観測の積である非常に単純な環境を提供します。

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

次に、この環境を使用して観測を取得し、アクションに対する報酬を受け取ります。

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

action = 2 #@param

print("action: %d" % action)
reward = environment.step(action).reward
print("reward: %f" % reward)

TF Environment

BanditTFEnvironment をサブクラス化するか、RL 環境と同様に BanditPyEnvironment を定義して TFPyEnvironment でラップすることでバンディット環境を定義できます。単純さを維持するために、このチュートリアルでは、後者を使用することにします。

tf_environment = tf_py_environment.TFPyEnvironment(environment)

ポリシー

バンディット問題におけるポリシーは、RL 問題と同様に機能し、観測を入力としてアクション（またはアクションの分布）を提供します。

詳細については、「TF-Agents ポリシーのチュートリアル」をご覧ください。

環境と同様に、ポリシーの構築には 2 つの方法があります。1 つは、PyPolicy を作成して TFPyPolicy でラップする方法で、もう 1 つは、TFPolicy を直接作成する方法です。ここでは、直接作成する方法を使用します。

この例は非常に単純であるため、最適なポリシーを手動で作成できます。アクションは観測の表示にのみ依存しており、負の場合は 0、正の場合は 2 となります。

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)

次に、環境に観測をリクエストし、ポリシーを呼び出してポリシーを選択すると、環境が報酬を出力します。

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)

バンディット環境の実装方法によって、ステップを取るたびに、選択したアクションに対する報酬が得られるだけでなく、次の観測も得られます。

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)

エージェント

バンディット環境とバンディットポリシーを準備したので、バンディットエージェントを定義することにしましょう。バンディットエージェントは、トレーニングサンプルに基づいてポリシーの変更を行います。

バンディットエージェントの API は RL のエージェントと同じですが、_initialize メソッドと _train メソッドを実装し、policy と collect_policy を定義する必要があります。

より複雑な環境

バンディットエージェントを記述する前に、少し理解しにくい環境を用意する必要があります。もう少し面白くするために、次の環境は、reward = observation * action または reward = -observation * action のいずれかを必ず与えるようにしましょう。どちらが与えられるかは、環境か初期化するときに決定されます。

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

より複雑なポリシー

より複雑なかんきょうには、より複雑なポリシーが伴います。基盤の環境の動作を検出するポリシーが必要です。ポリシーが処理する必要のある状況は 3 つあります。

1. エージェントが、実行している環境のバージョンを検出していない場合

2. エージェントが、実行している環境の元のバージョンを検出した場合

3. エージェントが、実行している環境の反転バージョンを検出した場合

_situation という tf_variable を、[0, 2] の値にエンコーディングされた情報を格納するように定義し、ポリシーが適宜に動作するようにします。

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)

エージェント

では、環境のサインを検出して、ポリシーを適切に設定するエージェントを定義することにしましょう。

class SignAgent(tf_agent.TFAgent):
  def __init__(self):
    self._situation = tf.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()

上記のコードでは、エージェントがポリシーを定義し、エージェントとポリシーが変数 situation を共有しています。

また、_train 関数のパラメータ experience はトラジェクトリです。

トラジェクトリ

TF-Agents では、trajectories は名前付きのタプルであり、前のステップで取得されたサンプルを含みます。これらのサンプルはエージェントによってポリシーのトレーニングと更新に使用されます。RL では、トラジェクトリには現在の状態、次の状態、そして現在のエピソードが終了したかどうかに関する情報が含まれている必要があります。バンディットの世界では、これらの情報は不要であるため、ヘルパー関数をセットアップしてトラジェクトリを作成します。

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

エージェントのトレーニング

これで、バンディットエージェントをトレーニングするためのピースがすべて用意できました。

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

出力から、2 つ目のステップ（最初のステップで観測が 0 でなければ）の後に、ポリシーが正しい方法でアクションを選択しており、したがって収集された報酬が常に非負であることがわかります。

実際の文脈付きバンディットの例

このチュートリアルの残りでは、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

線形ペイオフ関数を使った定常確率的環境

この例で使用する環境は、StationaryStochasticPyEnvironment です。この環境は、観測（コンテキスト）を提供する（非常にノイズの多い）関数をパラメータとして取り、アームごとに、与えられた観測に基づいて報酬を計算する（やはりノイズの多い）関数を取ります。このチュートリアルの例では、d 次元の立方体から均一にコンテキストをサンプリングすると、報酬関数はコンテキストの線形関数で、一部はガウスノイズです。

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

LinUCB エージェント

以下のエージェントは、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 基準

バンディットで最も重要な基準は regret です。エージェントが収集した報酬と、環境の報酬関数にアクセスできる予測ポリシーの期待報酬の差として計算されます。そのため、RegretMetric には、特定の観測があった場合に、達成可能な最大の期待報酬を計算する baseline_reward_fn 関数が必要です。このチュートリアルの例では、この環境に定義した報酬関数に相当するノイズのない関数の最大値を取る必要があります。

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)

トレーニング

上記で説明した環境、ポリシー、およびエージェントの要素をすべてを組み合わせましょう。ドライバーを使用して、環境でポリシーを実行してトレーニングデータを出力し、そのデータでエージェントをトレーニングします。

必要なステップ数を共に指定するパラメータが 2 つあることに注意してください。num_iterations はトレーナーループを実行する回数を指定し、ドライバーはイテレーションごとに steps_per_loop ステップを実行します。これらのパラメータを維持するのは、主に、イテレーションごとに実行される演算と、ステップごとにドライバーが行う演算があるためです。たとえば、エージェントの train 関数はイテレーションにつき一度しか呼び出されません。ここでは、トレーニングの頻度を高めると、ポリシーが「より新しく」なるのに対し、より大きなバッチでトレーニングすると時間の効率がよくなるというトレードオフがあります。

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

最後のコードスニペットを実行したら、生成されるプロットから、エージェントのトレーニングが増えて、特定の観測が与えられる場合にポリシーが適切なアクションを選択する確率が高まる過程で、平均 Regret が下降しているのが示されます（そうであることを願います）。

今後の内容

さらに機能例を確認する場合は、bandits/agents/examples をご覧ください。さまざまなエージェントと環境用にすぐに実行できる例が掲載されています。

TF-Agents ライブラリは、アームごとの特徴量でタ腕バンディットを処理することもできます。それについては、アームごとのバンディット問題のチュートリアルをご覧ください。