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의 다중 슬롯머신(Multi Armed Bandits)에 대한 튜토리얼

시작하기

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과 다릅니다. 또 다른 차이점은 슬롯머신(Bandits)에는 "에피소드"가 없다는 것입니다. 매 시간의 단계는 이전 시간의 단계와 관련 없는 독립적인 새로운 관찰로 시작됩니다.

관찰이 독립적인지 확인하고 RL 에피소드의 개념을 추상화하기 위해 PyEnvironment 및 TFEnvironment의 하위 클래스의 하위 클래스인 BanditPyEnvironment와 BanditTFEnvironment를 도입합니다. 이러한 클래스들은 사용자가 구현해야 하는 두 개의 비공개 멤버 함수를 노출합니다.

@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 환경

BanditTFEnvironment를 하위 클래스화하여 슬롯머신(Bandit) 환경을 정의하거나 RL 환경과 유사하게 정의하거나, BanditPyEnvironment를 정의하고 TFPyEnvironment로 래핑할 수 있습니다. 간단하게 보여주기 위해 이 튜토리얼에서는 후자 옵션을 사용합니다.

tf_environment = tf_py_environment.TFPyEnvironment(environment)

정책

슬롯머신(bandit) 문제의 정책은 RL 문제와 동일한 방식으로 작동합니다. 즉, 입력으로 관찰이 주어지면 행동(또는 행동 분포)을 제공합니다.

자세한 내용은 TF-Agents 정책 튜토리얼을 참조합니다.

환경과 마찬가지로 정책을 구성하는 두 가지 방법이 있습니다. 하나는 PyPolicy를 생성하고 이를 TFPyPolicy로 래핑하거나 직접 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)

슬롯머신(bandit) 환경이 구현되는 방식은 우리가 단계를 밟아갈 때마다 수행한 행동에 대한 보상뿐만 아니라 다음 관찰에 대한 보상도 받을 수 있도록 합니다.

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 에이전트의 API와 다르지 않습니다. 에이전트에 대해 _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())

더 복잡한 정책의 경우

더 복잡한 환경은 더 복잡한 정책을 요구합니다. 우리는 기본 환경의 동작을 감지하는 정책이 필요합니다. 이 정책에서 처리해야 하는 세 가지 상황은 다음과 같습니다.

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는 이전 단계에서 수행한 샘플을 포함하는 명명된 튜플(named tuples)입니다. 에이전트는 이러한 샘플을 사용하여 정책을 훈련하고 업데이트합니다. RL의 궤적은 현재 상태, 다음 상태, 현재 에피소드의 종료 여부에 대한 정보를 포함해야 합니다. 슬롯머신(bandit) 세계에서는 이런 정보가 필요하지 않기 때문에 우리는 궤적을 생성하는 도우미 함수를 설정합니다.

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

에이전트 훈련하기

이제 우리의 슬롯머신(bandit) 에이전트를 훈련할 모든 준비가 완료되었습니다.

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

출력에서, 두 번째 단계 이후(첫 번째 단계에서 관찰 결과가 0이 아닌 한) 정책이 올바른 방식으로 행동을 선택하므로 수집된 보상이 항상 음수가 아닌 것을 확인할 수 있습니다.

실제 상황별 슬롯머신 예시

이 튜토리얼의 나머지 부분에서는 사전에 구현된 TF-Agents 슬롯머신 라이브러리의 환경과 에이전트를 사용합니다.

# 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차원 큐브의 맥락(context)를 균일하게 샘플링하며, 보상 함수는 맥락의 선형 함수에 가우시안 노이즈를 더한 것입니다.

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

슬롯머신에서 가장 중요한 메트릭은 에이전트가 수집한 보상과 환경의 보상 기능에 액세스하는 예측 시스템 정책의 예상 보상 간의 차이로 계산되는 후회(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)

훈련하기

이제 위에서 소개한 모든 구성 요소인 환경, 정책, 에이전트를 한곳에 모았습니다. 우리는 드라이버를 사용하여 환경 정책을 실행하고 훈련 데이터를 출력하고, 에이전트로 하여금 데이터를 학습시킵니다.

수행한 단계 수를 함께 지정하는 두 개의 매개변수가 있습니다. 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')

마지막 코드 스니펫을 실행한 후 결과 플롯(희망적으로)은 에이전트 훈련이 진행됨에 따라 평균 후회 횟수가 줄어들고, 관찰이 제공되는 경우 올바른 행동이 무엇인지 파악하는 정책이 향상되는 모습을 보여줍니다.

다음은?

더 많은 작업 예시를 보려면 여러 에이전트와 환경을 바로 실행할 수 있는 예제가 있는 bandits/agents/examples를 참조해 주시길 바랍니다.

TF-Agents 라이브러리는 독립적인 슬롯머신 손잡이(per-arm) 특성을 지닌 다중 슬롯머신(Multi-Armed Bandits)도 처리할 수 있습니다. 해당 내용은 독립적인 슬롯머신 손잡이 튜토리얼을 참조합니다.