नीतियों

TensorFlow.org पर देखें Google Colab में चलाएं GitHub पर स्रोत देखें नोटबुक डाउनलोड करें

परिचय

सुदृढीकरण सीखने की शब्दावली में, नीतियां पर्यावरण से किसी क्रिया या कार्यों पर वितरण के लिए एक अवलोकन को मैप करती हैं। TF-एजेंटों में, पर्यावरण से टिप्पणियों नामांकित टपल में निहित हैं TimeStep('step_type', 'discount', 'reward', 'observation') , और नीतियों क्रिया या कार्यों पर वितरण के लिए timesteps मैप करें। अधिकांश नीतियों का उपयोग timestep.observation , कुछ नीतियों का उपयोग timestep.step_type (जैसे स्टेटफुल नीतियों में एक प्रकरण की शुरुआत में राज्य पुनर्स्थापित करने के लिए), लेकिन timestep.discount और timestep.reward आमतौर पर ध्यान नहीं दिया जाता।

नीतियाँ TF-Agents के अन्य घटकों से निम्न प्रकार से संबंधित हैं। अधिकांश नीतियों में टाइमस्टेप्स से क्रियाओं और/या वितरण की गणना करने के लिए एक तंत्रिका नेटवर्क होता है। एजेंटों में विभिन्न उद्देश्यों के लिए एक या अधिक नीतियां हो सकती हैं, उदाहरण के लिए एक मुख्य नीति जिसे परिनियोजन के लिए प्रशिक्षित किया जा रहा है, और डेटा संग्रह के लिए एक शोर नीति। नीतियों को सहेजा/बहाल किया जा सकता है, और डेटा संग्रह, मूल्यांकन आदि के लिए एजेंट के स्वतंत्र रूप से उपयोग किया जा सकता है।

कुछ नीतियों को 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 एक 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: स्क्रिप्टेड पायथन पॉलिसी

एक पटकथा नीति नाटकों की एक सूची के रूप में प्रतिनिधित्व कार्यों की एक स्क्रिप्ट वापस (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_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)

हम इसे टाइम_स्टेप्स के किसी भी बैच पर लागू कर सकते हैं जो टाइम_स्टेप_स्पेक का पालन करता है:

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] की सीमा से जुड़ी हुई हैं। ऐसा इसलिए है क्योंकि अभिनेता नीति क्लिप का एक निर्माता तर्क = डिफ़ॉल्ट रूप से सही है। इसे गलत पर सेट करने से नेटवर्क द्वारा निर्मित अनक्लिप्ड कार्रवाइयां वापस आ जाएंगी।

स्टोकेस्टिक नीतियों नियतात्मक नीतियों का उपयोग कर, उदाहरण के लिए, एक GreedyPolicy आवरण जो चुनता में बदला जा सकता stochastic_policy.distribution().mode() अपनी कार्रवाई के रूप में है, और उसके रूप में इस लालची कार्रवाई के चारों ओर एक नियतात्मक / डेल्टा वितरण distribution() ।

उदाहरण 3: क्यू नीति

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)