מדיניות

הצג באתר TensorFlow.org הפעל בגוגל קולאב צפה במקור ב-GitHub הורד מחברת

מבוא

במינוח של למידת חיזוק, מדיניות ממפה תצפית מהסביבה לפעולה או התפלגות על פעולות. In-סוכנים TF, תצפיות מהסביבה מוכלים בתוך tuple בשם TimeStep('step_type', 'discount', 'reward', 'observation') , ומדיניות למפות timesteps לפעולות או הפצות על מעשיו. רוב פוליסות להשתמש timestep.observation , מדיניות כלשהי להשתמש timestep.step_type (למשל כדי לאפס את המדינה בתחילת פרק במדיניות מצבי), אבל timestep.discount ו timestep.reward בדרך כלל מתעלמים.

מדיניות קשורה לרכיבים אחרים ב-TF-Agents באופן הבא. לרוב המדיניות יש רשת עצבית לחישוב פעולות ו/או התפלגות על פני פעולות מ-TimeSteps. סוכנים יכולים להכיל מדיניות אחת או יותר למטרות שונות, למשל מדיניות עיקרית המתאמנת לפריסה ומדיניות רועשת לאיסוף נתונים. ניתן לשמור/לשחזר מדיניות וניתן להשתמש בהם ללא תלות בסוכן לצורך איסוף נתונים, הערכה וכו'.

חלק מהמדיניות קלה יותר לכתוב ב-Tensorflow (למשל אלה עם רשת עצבית), בעוד שאחרים קל יותר לכתוב ב-Python (למשל בעקבות סקריפט של פעולות). אז בסוכני TF, אנו מאפשרים גם מדיניות Python וגם Tensorflow. יתרה מכך, ייתכן שיהיה צורך להשתמש במדיניות שנכתבת ב-TensorFlow בסביבת Python, או להיפך, למשל, מדיניות TensorFlow משמשת לאימון אך נפרסת מאוחר יותר בסביבת Python של ייצור. כדי להקל על כך, אנו מספקים עטיפות להמרה בין מדיניות Python ו- TensorFlow.

סוג נוסף של פוליסות מעניין הוא עוטפי מדיניות, אשר משנים מדיניות נתונה בצורה מסוימת, למשל מוסיפים סוג מסוים של רעש, יוצרים גרסה חמדנית או אפסילון חמדנית של מדיניות סטוכסטית, מערבבים באופן אקראי מספר פוליסות וכו'.

להכין

אם עדיין לא התקנת tf-agents, הרץ:

pip install tf-agents

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import abc
import tensorflow as tf
import tensorflow_probability as tfp
import numpy as np

from tf_agents.specs import array_spec
from tf_agents.specs import tensor_spec
from tf_agents.networks import network

from tf_agents.policies import py_policy
from tf_agents.policies import random_py_policy
from tf_agents.policies import scripted_py_policy

from tf_agents.policies import tf_policy
from tf_agents.policies import random_tf_policy
from tf_agents.policies import actor_policy
from tf_agents.policies import q_policy
from tf_agents.policies import greedy_policy

from tf_agents.trajectories import time_step as ts

מדיניות פייתון

הממשק עבור פוליסות Python מוגדר policies/py_policy.PyPolicy . השיטות העיקריות הן:

class Base(object):

  @abc.abstractmethod
  def __init__(self, time_step_spec, action_spec, policy_state_spec=()):
    self._time_step_spec = time_step_spec
    self._action_spec = action_spec
    self._policy_state_spec = policy_state_spec

  @abc.abstractmethod
  def reset(self, policy_state=()):
    # return initial_policy_state.
    pass

  @abc.abstractmethod
  def action(self, time_step, policy_state=()):
    # return a PolicyStep(action, state, info) named tuple.
    pass

  @abc.abstractmethod
  def distribution(self, time_step, policy_state=()):
    # Not implemented in python, only for TF policies.
    pass

  @abc.abstractmethod
  def update(self, policy):
    # update self to be similar to the input `policy`.
    pass

  @property
  def time_step_spec(self):
    return self._time_step_spec

  @property
  def action_spec(self):
    return self._action_spec

  @property
  def policy_state_spec(self):
    return self._policy_state_spec

השיטה החשובה ביותר היא action(time_step) אשר ממפה time_step המכיל תצפית מהסביבה כדי tuple בשם PolicyStep המכיל את התכונות הבאות:

  • action : הפעולה שיש להחיל על הסביבה.
  • state : המדינה של המדיניות (המדינה למשל RNN) כדי להיות מוזנת לתוך השיחה הבאה לפעולה.
  • info : מידע צד אופציונלי כגון הסתברויות יומן פעולה.

time_step_spec ו action_spec הם מפרטים את צעד זמן הקלט ופעולת הפלט. מדיניות יש גם reset פונקציה אשר בדרך כלל משמש לאיפוס המדינה במדיניות מצבי. update(new_policy) הפונקציה מעדכנת self כלפי new_policy .

כעת, הבה נסתכל על כמה דוגמאות למדיניות של Python.

דוגמה 1: מדיניות Python אקראית

דוגמא פשוט של PyPolicy היא RandomPyPolicy אשר יוצר פעולות אקראיות עבור action_spec הנתון הדיסקרטי / הרציף. הקלט time_step מתעלם.

action_spec = array_spec.BoundedArraySpec((2,), np.int32, -10, 10)
my_random_py_policy = random_py_policy.RandomPyPolicy(time_step_spec=None,
    action_spec=action_spec)
time_step = None
action_step = my_random_py_policy.action(time_step)
print(action_step)
action_step = my_random_py_policy.action(time_step)
print(action_step)

PolicyStep(action=array([10, -4], dtype=int32), state=(), info=())
PolicyStep(action=array([7, 6], dtype=int32), state=(), info=())

דוגמה 2: מדיניות Python עם סקריפט

מחזות מדיניות תסריט לגבות תסריט של פעולות המיוצגות כרשימה (num_repeats, action) tuples. בכל פעם action פונקציה נקראת, היא מחזירה את הפעולה הבאה מהרשימה עד מספר מסוים של חזרות נעשה, ועובר הלאה לפעולה הבאה ברשימה. reset השיטה יכולה להיקרא להתחיל בביצוע מתחילת הרשימה.

action_spec = array_spec.BoundedArraySpec((2,), np.int32, -10, 10)
action_script = [(1, np.array([5, 2], dtype=np.int32)), 
                 (0, np.array([0, 0], dtype=np.int32)), # Setting `num_repeats` to 0 will skip this action.
                 (2, np.array([1, 2], dtype=np.int32)), 
                 (1, np.array([3, 4], dtype=np.int32))]

my_scripted_py_policy = scripted_py_policy.ScriptedPyPolicy(
    time_step_spec=None, action_spec=action_spec, action_script=action_script)

policy_state = my_scripted_py_policy.get_initial_state()
time_step = None
print('Executing scripted policy...')
action_step = my_scripted_py_policy.action(time_step, policy_state)
print(action_step)
action_step= my_scripted_py_policy.action(time_step, action_step.state)
print(action_step)
action_step = my_scripted_py_policy.action(time_step, action_step.state)
print(action_step)

print('Resetting my_scripted_py_policy...')
policy_state = my_scripted_py_policy.get_initial_state()
action_step = my_scripted_py_policy.action(time_step, policy_state)
print(action_step)

Executing scripted policy...
PolicyStep(action=array([5, 2], dtype=int32), state=[0, 1], info=())
PolicyStep(action=array([1, 2], dtype=int32), state=[2, 1], info=())
PolicyStep(action=array([1, 2], dtype=int32), state=[2, 2], info=())
Resetting my_scripted_py_policy...
PolicyStep(action=array([5, 2], dtype=int32), state=[0, 1], info=())

מדיניות TensorFlow

מדיניות TensorFlow פועלת באותו ממשק כמו מדיניות Python. הבה נסתכל על כמה דוגמאות.

דוגמה 1: מדיניות TF אקראית

RandomTFPolicy יכול לשמש כדי ליצור פעולות אקראיות פי דיסקרטי / רציף נתון action_spec . הקלט time_step מתעלם.

action_spec = tensor_spec.BoundedTensorSpec(
    (2,), tf.float32, minimum=-1, maximum=3)
input_tensor_spec = tensor_spec.TensorSpec((2,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)

my_random_tf_policy = random_tf_policy.RandomTFPolicy(
    action_spec=action_spec, time_step_spec=time_step_spec)
observation = tf.ones(time_step_spec.observation.shape)
time_step = ts.restart(observation)
action_step = my_random_tf_policy.action(time_step)

print('Action:')
print(action_step.action)

Action:
tf.Tensor([-0.9448042  1.9039011], shape=(2,), dtype=float32)

דוגמה 2: מדיניות שחקנים

פוליסת שחקן יכולה להיווצר גם באמצעות רשת ממפה time_steps לפעולות או רשת ממפת time_steps חלוק על מעשיו.

שימוש ברשת פעולה

הבה נגדיר רשת באופן הבא:

class ActionNet(network.Network):

  def __init__(self, input_tensor_spec, output_tensor_spec):
    super(ActionNet, self).__init__(
        input_tensor_spec=input_tensor_spec,
        state_spec=(),
        name='ActionNet')
    self._output_tensor_spec = output_tensor_spec
    self._sub_layers = [
        tf.keras.layers.Dense(
            action_spec.shape.num_elements(), activation=tf.nn.tanh),
    ]

  def call(self, observations, step_type, network_state):
    del step_type

    output = tf.cast(observations, dtype=tf.float32)
    for layer in self._sub_layers:
      output = layer(output)
    actions = tf.reshape(output, [-1] + self._output_tensor_spec.shape.as_list())

    # Scale and shift actions to the correct range if necessary.
    return actions, network_state

ב-TensorFlow רוב שכבות הרשת מיועדות לפעולות אצווה, ולכן אנו מצפים שהקלט time_steps יהיה אצווה, והפלט של הרשת יהיה אצווה גם כן. כמו כן, הרשת אחראית לייצר פעולות בטווח הנכון של מפרט ה-action_נתון. הדבר נעשה כמקובל באמצעות למשל הפעלת TANH עבור השכבה הסופית לפעולות התוצרת ב [-1, 1] ולאחר מכן קנה המידה ועל הסטה זו בטווח הנכון כמו action_spec קלט (למשל לראות tf_agents/agents/ddpg/networks.actor_network() ).

כעת, אנו יכולים ליצור מדיניות שחקנים באמצעות הרשת שלמעלה.

input_tensor_spec = tensor_spec.TensorSpec((4,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
action_spec = tensor_spec.BoundedTensorSpec((3,),
                                            tf.float32,
                                            minimum=-1,
                                            maximum=1)

action_net = ActionNet(input_tensor_spec, action_spec)

my_actor_policy = actor_policy.ActorPolicy(
    time_step_spec=time_step_spec,
    action_spec=action_spec,
    actor_network=action_net)

אנו יכולים להחיל אותו על כל אצווה של time_steps העוקבת אחר time_step_spec:

batch_size = 2
observations = tf.ones([2] + time_step_spec.observation.shape.as_list())

time_step = ts.restart(observations, batch_size)

action_step = my_actor_policy.action(time_step)
print('Action:')
print(action_step.action)

distribution_step = my_actor_policy.distribution(time_step)
print('Action distribution:')
print(distribution_step.action)

Action:
tf.Tensor(
[[0.9318627 0.7770741 0.8645338]
 [0.9318627 0.7770741 0.8645338]], shape=(2, 3), dtype=float32)
Action distribution:
tfp.distributions.Deterministic("Deterministic", batch_shape=[2, 3], event_shape=[], dtype=float32)

בדוגמה לעיל, יצרנו את המדיניות באמצעות רשת פעולה המייצרת טנזור פעולה. במקרה זה, policy.distribution(time_step) היא הפצה דטרמיניסטית (דלתא) ברחבי הפלט של policy.action(time_step) . אחת הדרכים לייצר מדיניות סטוכסטית היא לעטוף את מדיניות השחקן בעטיפת מדיניות שמוסיפה רעש לפעולות. דרך נוספת היא ליצור את מדיניות השחקנים באמצעות רשת הפצת פעולה במקום רשת פעולה כפי שמוצג להלן.

שימוש ברשת הפצת פעולה

class ActionDistributionNet(ActionNet):

  def call(self, observations, step_type, network_state):
    action_means, network_state = super(ActionDistributionNet, self).call(
        observations, step_type, network_state)

    action_std = tf.ones_like(action_means)
    return tfp.distributions.MultivariateNormalDiag(action_means, action_std), network_state


action_distribution_net = ActionDistributionNet(input_tensor_spec, action_spec)

my_actor_policy = actor_policy.ActorPolicy(
    time_step_spec=time_step_spec,
    action_spec=action_spec,
    actor_network=action_distribution_net)

action_step = my_actor_policy.action(time_step)
print('Action:')
print(action_step.action)
distribution_step = my_actor_policy.distribution(time_step)
print('Action distribution:')
print(distribution_step.action)

Action:
tf.Tensor(
[[ 0.96731853  1.          1.        ]
 [ 0.94488937 -0.29294527  1.        ]], shape=(2, 3), dtype=float32)
Action distribution:
tfp.distributions.MultivariateNormalDiag("ActionNet_MultivariateNormalDiag", batch_shape=[2], event_shape=[3], dtype=float32)

שים לב שברשימה לעיל, הפעולות נגזרות לטווח של מפרט הפעולה הנתון [-1, 1]. הסיבה לכך היא ארגומנט בנאי של ActorPolicy clip=True כברירת מחדל. הגדרה של זה ל-false תחזיר פעולות שלא נקטעו שנוצרו על ידי הרשת.

ניתן להמיר מדיניות סטוכסטיים למדיניות דטרמיניסטית באמצעות, למשל, עטיפת GreedyPolicy אשר בוחרת stochastic_policy.distribution().mode() כמו הפעולה שלה, וכן הפצת דלתא / דטרמיניסטית סביב פעולת חמדן זה כמו שלה distribution() .

דוגמה 3: מדיניות Q

מדיניות AQ משמשת בסוכנים כמו DQN ומבוססת על רשת Q המנבאת ערך Q עבור כל פעולה בדידה. עבור שלב זמן נתון, התפלגות הפעולה ב-Q Policy היא התפלגות קטגורית שנוצרה באמצעות ערכי q כלוגיטים.

input_tensor_spec = tensor_spec.TensorSpec((4,), tf.float32)
time_step_spec = ts.time_step_spec(input_tensor_spec)
action_spec = tensor_spec.BoundedTensorSpec((),
                                            tf.int32,
                                            minimum=0,
                                            maximum=2)
num_actions = action_spec.maximum - action_spec.minimum + 1


class QNetwork(network.Network):

  def __init__(self, input_tensor_spec, action_spec, num_actions=num_actions, name=None):
    super(QNetwork, self).__init__(
        input_tensor_spec=input_tensor_spec,
        state_spec=(),
        name=name)
    self._sub_layers = [
        tf.keras.layers.Dense(num_actions),
    ]

  def call(self, inputs, step_type=None, network_state=()):
    del step_type
    inputs = tf.cast(inputs, tf.float32)
    for layer in self._sub_layers:
      inputs = layer(inputs)
    return inputs, network_state


batch_size = 2
observation = tf.ones([batch_size] + time_step_spec.observation.shape.as_list())
time_steps = ts.restart(observation, batch_size=batch_size)

my_q_network = QNetwork(
    input_tensor_spec=input_tensor_spec,
    action_spec=action_spec)
my_q_policy = q_policy.QPolicy(
    time_step_spec, action_spec, q_network=my_q_network)
action_step = my_q_policy.action(time_steps)
distribution_step = my_q_policy.distribution(time_steps)

print('Action:')
print(action_step.action)

print('Action distribution:')
print(distribution_step.action)

Action:
tf.Tensor([2 2], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.Categorical("Categorical", batch_shape=[2], event_shape=[], dtype=int32)

עטיפות מדיניות

ניתן להשתמש במעטפת מדיניות כדי לעטוף ולשנות מדיניות נתונה, למשל להוסיף רעש. עטיפות מדיניות הן תת-סיווג של מדיניות (Python/TensorFlow) ולכן ניתן להשתמש בהן בדיוק כמו כל מדיניות אחרת.

דוגמה: מדיניות חמדנית

מעטפת חמדן יכול לשמש כדי לעטוף כל מדיניות TensorFlow המיישמת distribution() . GreedyPolicy.action() יחזיר wrapped_policy.distribution().mode() ו GreedyPolicy.distribution() היא הפצת דלתא / דטרמיניסטית סביב GreedyPolicy.action() :

my_greedy_policy = greedy_policy.GreedyPolicy(my_q_policy)

action_step = my_greedy_policy.action(time_steps)
print('Action:')
print(action_step.action)

distribution_step = my_greedy_policy.distribution(time_steps)
print('Action distribution:')
print(distribution_step.action)

Action:
tf.Tensor([1 1], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.DeterministicWithLogProb("Deterministic", batch_shape=[2], event_shape=[], dtype=int32)