নীতিমালা

TensorFlow.org এ দেখুন Google Colab-এ চালান GitHub-এ উৎস দেখুন নোটবুক ডাউনলোড করুন

ভূমিকা

রিইনফোর্সমেন্ট লার্নিং পরিভাষায়, নীতিগুলি পরিবেশ থেকে একটি ক্রিয়া বা কর্মের উপর বন্টন পর্যন্ত একটি পর্যবেক্ষণকে ম্যাপ করে। মেমরি-এজেন্ট, পরিবেশ থেকে পর্যবেক্ষণ একটি নামাঙ্কিত tuple অন্তর্ভুক্ত করা হয় TimeStep('step_type', 'discount', 'reward', 'observation') , এবং নীতি ক্রিয়া বা কর্ম উপর ডিস্ট্রিবিউশন থেকে timesteps মানচিত্র। সর্বাধিক নীতি ব্যবহার timestep.observation , কিছু নীতি ব্যবহার timestep.step_type (যেমন stateful নীতির একটি পর্বে শুরুতে রাষ্ট্র পুনরায় সেট করতে), কিন্তু timestep.discount এবং timestep.reward সাধারণত উপেক্ষা করা হয়।

নীতিগুলি নিম্নলিখিত উপায়ে TF-এজেন্টের অন্যান্য উপাদানগুলির সাথে সম্পর্কিত৷ টাইমস্টেপস থেকে অ্যাকশন এবং/অথবা ডিস্ট্রিবিউশন গণনা করার জন্য বেশিরভাগ নীতিতে একটি নিউরাল নেটওয়ার্ক থাকে। এজেন্টদের বিভিন্ন উদ্দেশ্যে এক বা একাধিক নীতি থাকতে পারে, যেমন একটি প্রধান নীতি যা স্থাপনার জন্য প্রশিক্ষণ দেওয়া হচ্ছে, এবং ডেটা সংগ্রহের জন্য একটি গোলমাল নীতি। নীতিগুলি সংরক্ষণ/পুনরুদ্ধার করা যেতে পারে এবং ডেটা সংগ্রহ, মূল্যায়ন ইত্যাদির জন্য এজেন্টের স্বাধীনভাবে ব্যবহার করা যেতে পারে।

কিছু নীতি টেনসরফ্লোতে লেখা সহজ (যেমন একটি নিউরাল নেটওয়ার্কের সাথে), যেখানে অন্যগুলি পাইথনে লেখা সহজ (যেমন অ্যাকশনের স্ক্রিপ্ট অনুসরণ করে)। তাই TF এজেন্টে, আমরা Python এবং Tensorflow উভয় নীতিই অনুমোদন করি। অধিকন্তু, TensorFlow-এ লিখিত নীতিগুলি একটি 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) যা একটি Maps time_step একটি PolicyStep নামে নিম্নলিখিত বৈশিষ্ট্য ধারণকারী tuple করার পরিবেশ থেকে একটি পর্যবেক্ষণ ধারণকারী:

  • 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) 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=())

টেনসরফ্লো নীতি

টেনসরফ্লো নীতিগুলি পাইথন নীতিগুলির মতো একই ইন্টারফেস অনুসরণ করে৷ আসুন কয়েকটি উদাহরণ দেখি।

উদাহরণ 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 এর সঠিক পরিসরে ক্রিয়া তৈরি করার জন্য দায়ী। এই সাধারনত এ [-1, 1] এবং তারপর স্কেলিং এবং (ইনপুট action_spec যেমন সঠিক পরিসরের এই নাড়াচাড়া উত্পাদন কর্মের চূড়ান্ত স্তরের জন্য যেমন একটি TANH অ্যাক্টিভেশন ব্যবহার করা যাবে যেমন দেখতে 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)

নীতির মোড়ক

একটি নীতির মোড়ক একটি প্রদত্ত নীতি মোড়ানো এবং সংশোধন করতে ব্যবহার করা যেতে পারে, যেমন শব্দ যোগ করুন। পলিসি র‍্যাপার হল পলিসির একটি সাবক্লাস (পাইথন/টেনসরফ্লো) এবং তাই অন্য যেকোনো পলিসির মতোই ব্যবহার করা যেতে পারে।

উদাহরণ: লোভী নীতি

একজন লোভী মোড়কের যে কার্যকরী কোন 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)