سیاست های

مشاهده در TensorFlow.org در Google Colab اجرا شود مشاهده منبع در GitHub دانلود دفترچه یادداشت

معرفی

در اصطلاحات یادگیری تقویتی، خط مشی ها یک مشاهده از محیط را به یک عمل یا توزیع بر روی کنش ها ترسیم می کنند. در TF-نمایندگی ها، مشاهدات از محیط در یک تاپل به نام موجود TimeStep('step_type', 'discount', 'reward', 'observation') ، و سیاست های نقشه timesteps به اقدامات و یا توزیع بیش از اقدامات. اکثر سیاست های استفاده از timestep.observation ، برخی از سیاست های استفاده از timestep.step_type (به عنوان مثال برای تنظیم مجدد دولت در آغاز یک قسمت که در سیاست های stateful به)، اما timestep.discount و timestep.reward معمولا نادیده گرفته میشود.

خط مشی ها به روش زیر با سایر اجزای موجود در TF-Agents مرتبط می شوند. بیشتر خط‌مشی‌ها دارای یک شبکه عصبی برای محاسبه اقدامات و/یا توزیع‌ها بر روی اقدامات از TimeSteps هستند. عامل ها می توانند شامل یک یا چند خط مشی برای اهداف مختلف باشند، به عنوان مثال یک خط مشی اصلی که برای استقرار آموزش داده می شود و یک خط مشی پر سر و صدا برای جمع آوری داده ها. خط‌مشی‌ها را می‌توان ذخیره/بازیابی کرد و به‌طور مستقل از عامل برای جمع‌آوری داده‌ها، ارزیابی و غیره استفاده کرد.

نوشتن برخی از خط مشی ها در تنسورفلو آسان تر است (مثلاً آنهایی که دارای شبکه عصبی هستند)، در حالی که نوشتن برخی دیگر در پایتون آسان تر است (مثلاً دنبال کردن یک اسکریپت از اقدامات). بنابراین در عامل های TF، ما هر دو خط مشی Python و Tensorflow را مجاز می کنیم. علاوه بر این، سیاست‌های نوشته شده در TensorFlow ممکن است در یک محیط پایتون استفاده شوند، یا برعکس، به عنوان مثال، یک خط‌مشی 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

سیاست های پایتون

رابط کاربری را برای سیاست های پایتون در تعریف 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 حاوی یک مشاهده از محیط زیست برای یک تاپل PolicyStep نام حاوی ویژگی های زیر:

  • action : عملی به به محیط زیست استفاده شود.
  • state : دولت از سیاست (به عنوان مثال RNN دولت) به تماس بعدی به عمل تغذیه می شود.
  • info : اطلاعات جانبی اختیاری مانند احتمال اقدام ورود به سیستم.

time_step_spec و action_spec مشخصات ورودی گام زمان و عمل خروجی می باشد. سیاست همچنین یک reset تابع است که به طور معمول برای بازنشانی دولت در سیاست های stateful به استفاده می شود. 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: خط مشی پایتون اسکریپت شده

نمایشنامه سیاست اسکریپت یک اسکریپت از اقدامات به عنوان نماینده یک لیست از (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 از رابطی مشابه با سیاست های پایتون پیروی می کنند. اجازه دهید به چند نمونه نگاه کنیم.

مثال 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_spec است. طبق قرارداد انجام با استفاده از به عنوان مثال یک 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 یک توزیع طبقه بندی شده است که با استفاده از مقادیر q به عنوان logit ایجاد می شود.

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)

لفاف های سیاست

برای بسته بندی و اصلاح یک خط مشی معین، به عنوان مثال اضافه کردن نویز، می توان از یک پوشش سیاست استفاده کرد. پوشش‌های سیاست یک زیر کلاس از Policy (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)