سياسات

عرض على TensorFlow.org تشغيل في Google Colab عرض المصدر على جيثب تحميل دفتر

مقدمة

في مصطلحات التعلم التعزيزي ، تحدد السياسات ملاحظة من البيئة إلى إجراء أو توزيع على الإجراءات. في وكلاء 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 حتى الآن ، فقم بتشغيل:

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

سياسات بايثون

يتم تعريف واجهة لسياسات بيثون في 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 .

الآن ، دعونا نلقي نظرة على بعض الأمثلة على سياسات بايثون.

مثال 1: سياسة بايثون العشوائية

وهناك مثال بسيط من 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) الصفوف. في كل مرة 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: سياسة تمويل الإرهاب العشوائية

A 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 ، تم تصميم معظم طبقات الشبكة للعمليات المجمعة ، لذلك نتوقع أن يتم تجميع خطوات الإدخال ، كما سيتم تجميع إخراج الشبكة أيضًا. كما أن الشبكة مسئولة عن إنتاج الإجراءات في النطاق الصحيح من نوع action_spec المحدد. ويتم ذلك باستخدام تقليديا على سبيل المثال تان تنشيط طبقة نهائية لإجراءات المنتجات في [-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_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 افتراضيًا. سيؤدي تعيين هذا إلى "خطأ" إلى إرجاع الإجراءات غير المقصودة التي تنتجها الشبكة.

ويمكن تحويل السياسات العشوائية لسياسات حتمية استخدام، على سبيل المثال، مجمع GreedyPolicy الذي يختار stochastic_policy.distribution().mode() كما عملها، وتوزيع / دلتا القطعية حول هذا العمل الجشع كما في distribution() .

مثال 3: سياسة Q

تُستخدم سياسة AQ في وكلاء مثل DQN وتستند إلى شبكة Q التي تتوقع قيمة Q لكل إجراء منفصل. بالنسبة لخطوة زمنية معينة ، فإن توزيع الإجراء في سياسة Q هو توزيع قاطع تم إنشاؤه باستخدام قيم 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) ويمكن بالتالي استخدامها تمامًا مثل أي سياسة أخرى.

مثال: سياسة الجشع

A المجمع الجشع يمكن أن تستخدم لالتفاف أي سياسة 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)