Google I / O คืนวันที่ 18-20 พ.ค. จองพื้นที่และสร้างตารางเวลาของคุณ ลงทะเบียนตอนนี้

บทช่วยสอนเรื่อง Multi Armed Bandits ใน TF-Agents

เริ่ม

ดูใน TensorFlow.org เรียกใช้ใน Google Colab ดูแหล่งที่มาบน GitHub ดาวน์โหลดสมุดบันทึก

ติดตั้ง

หากคุณยังไม่ได้ติดตั้งการอ้างอิงต่อไปนี้ให้รัน:

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

# 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

บทนำ

ปัญหา Multi-Armed Bandit (MAB) เป็นกรณีพิเศษของ Reinforcement Learning: ตัวแทนรวบรวมรางวัลในสภาพแวดล้อมโดยดำเนินการบางอย่างหลังจากสังเกตสภาพแวดล้อมบางอย่าง ความแตกต่างที่สำคัญระหว่าง RL ทั่วไปและ MAB คือใน MAB เราถือว่าการกระทำของตัวแทนไม่มีผลต่อสถานะถัดไปของสภาพแวดล้อม ดังนั้นตัวแทนจะไม่จำลองการเปลี่ยนสถานะการให้รางวัลเครดิตกับการกระทำในอดีตหรือ "วางแผนล่วงหน้า" เพื่อไปสู่สถานะที่มีรางวัลมากมาย

เช่นเดียวกับในโดเมน RL อื่น ๆ เป้าหมายของ ตัวแทน MAB คือการค้นหา นโยบาย ที่รวบรวมรางวัลให้ได้มากที่สุด อย่างไรก็ตามจะเป็นความผิดพลาดที่จะพยายามใช้ประโยชน์จากการกระทำที่สัญญาว่าจะได้รับรางวัลสูงสุดอยู่เสมอเพราะจะมีโอกาสที่เราจะพลาดการกระทำที่ดีกว่าหากเราไม่สำรวจให้เพียงพอ นี่เป็นปัญหาหลักที่ต้องแก้ไขใน (MAB) ซึ่งมักเรียกว่า ภาวะที่กลืนไม่เข้าคายไม่ออกของการแสวงหาประโยชน์จากการสำรวจ

สภาพแวดล้อม Bandit นโยบายและตัวแทนสำหรับ 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

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

สภาพแวดล้อม TF

เราสามารถกำหนดสภาพแวดล้อมแบบ bandit โดยการ subclassing BanditTFEnvironment หรือในทำนองเดียวกันกับสภาพแวดล้อม RL เราสามารถกำหนด BanditPyEnvironment และรวมเข้ากับ TFPyEnvironment เพื่อความเรียบง่ายเราเลือกใช้ตัวเลือกหลังในบทช่วยสอนนี้

tf_environment = tf_py_environment.TFPyEnvironment(environment)

นโยบาย

นโยบาย ในปัญหาโจรทำงานในลักษณะเดียวกับปัญหา 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)
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)

วิธีการใช้สภาพแวดล้อมของโจรทำให้มั่นใจได้ว่าทุกครั้งที่เราก้าวไปข้างหน้าเราไม่เพียง แต่จะได้รับรางวัลสำหรับการกระทำที่เราทำเท่านั้น แต่ยังรวมถึงการสังเกตการณ์ครั้งต่อไปด้วย

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)

ตัวแทน

ตอนนี้เรามีสภาพแวดล้อมแบบโจรและนโยบายโจรแล้วก็ถึงเวลากำหนดตัวแทนโจรที่ดูแลการเปลี่ยนแปลงนโยบายตามตัวอย่างการฝึกอบรม

API สำหรับ bandit agent ไม่แตกต่างจาก RL agent: เอเจนต์จำเป็นต้องใช้ _initialize และ _train และกำหนด policy และ collect_policy

สภาพแวดล้อมที่ซับซ้อนมากขึ้น

ก่อนที่เราจะเขียนตัวแทนโจรเราจำเป็นต้องมีสภาพแวดล้อมที่ยากต่อการเข้าใจ เพื่อเพิ่มชีวิตชีวาให้กับสิ่งต่างๆเพียงเล็กน้อยสภาพแวดล้อมถัดไปจะให้ reward = observation * action หรือ reward = observation * action ให้ 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())
reward sign:
1

นโยบายที่ซับซ้อนมากขึ้น

สภาพแวดล้อมที่ซับซ้อนมากขึ้นเรียกร้องให้มีนโยบายที่ซับซ้อนมากขึ้น เราต้องการนโยบายที่ตรวจจับพฤติกรรมของสภาพแวดล้อมพื้นฐาน มีสามสถานการณ์ที่นโยบายต้องจัดการ:

  1. เอเจนต์ตรวจไม่พบทราบว่าสภาพแวดล้อมกำลังรันเวอร์ชันใด
  2. เอเจนต์ตรวจพบว่าเวอร์ชันดั้งเดิมของสภาพแวดล้อมกำลังทำงานอยู่
  3. เอเจนต์ตรวจพบว่าเวอร์ชันที่พลิกกลับของสภาพแวดล้อมกำลังทำงานอยู่

เรากำหนด tf_variable ชื่อ _situation เพื่อจัดเก็บข้อมูลนี้ที่เข้ารหัสเป็นค่าใน [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.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()

ในโค้ดด้านบนตัวแทนจะกำหนดนโยบายและ situation ตัวแปรจะแบ่งใช้โดยเอเจนต์และนโยบาย

นอกจากนี้ experience พารามิเตอร์ของฟังก์ชัน _train ยังเป็นวิถี:

วิถี

ใน TF-Agents trajectories จะถูกตั้งชื่อสิ่งที่มีตัวอย่างจากขั้นตอนก่อนหน้านี้ จากนั้นตัวแทนจะใช้ตัวอย่างเหล่านี้เพื่อฝึกอบรมและปรับปรุงนโยบาย ใน 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))

การฝึกอบรมตัวแทน

ตอนนี้ทุกชิ้นพร้อมสำหรับการฝึกอบรมตัวแทนโจรของเรา

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

จากผลลัพธ์จะเห็นได้ว่าหลังจากขั้นตอนที่สอง (เว้นแต่การสังเกตจะเป็น 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

สภาพแวดล้อม Stochastic ที่อยู่กับที่พร้อมฟังก์ชั่น Linear Payoff

สภาพแวดล้อมที่ใช้ในตัวอย่างนี้คือ 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)

เสียดายเมตริก

เมตริกที่สำคัญที่สุดของ Bandits คือ ความเสียใจ ซึ่งคำนวณจากความแตกต่างระหว่างรางวัลที่ตัวแทนรวบรวมและรางวัลที่คาดหวังของนโยบาย Oracle ที่สามารถเข้าถึงฟังก์ชันการให้รางวัลของสภาพแวดล้อมได้ ดังนั้น 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')
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

หลังจากเรียกใช้ข้อมูลโค้ดล่าสุดพล็อตผลลัพธ์ (หวังว่า) แสดงให้เห็นว่าความเสียใจโดยเฉลี่ยจะลดลงเมื่อตัวแทนได้รับการฝึกฝนและนโยบายจะดีขึ้นในการหาว่าการกระทำที่ถูกต้องคืออะไรจากการสังเกต

อะไรต่อไป?

หากต้องการดูตัวอย่างการทำงานเพิ่มเติมโปรดดูไดเร็กทอรี bandits / agent / samples ที่มี ตัวอย่าง พร้อมรันสำหรับเอเจนต์และสภาพแวดล้อมที่แตกต่างกัน

ไลบรารี TF-Agents ยังสามารถจัดการ Multi-Armed Bandits ด้วยคุณสมบัติต่อแขน ด้วยเหตุนี้เราจึงแนะนำผู้อ่านให้ไปที่ บทแนะนำเกี่ยวกับ โจรต่อแขน