Kebijakan

Lihat di TensorFlow.org Jalankan di Google Colab Lihat sumber di GitHub Unduh buku catatan

pengantar

Dalam terminologi Reinforcement Learning, kebijakan memetakan pengamatan dari lingkungan ke suatu tindakan atau distribusi atas tindakan. Dalam TF-Agen, pengamatan dari lingkungan yang terkandung dalam tuple bernama TimeStep('step_type', 'discount', 'reward', 'observation') , dan kebijakan peta timesteps tindakan atau distribusi atas tindakan. Kebanyakan kebijakan menggunakan timestep.observation , beberapa kebijakan menggunakan timestep.step_type (misalnya untuk me-reset negara pada awal sebuah episode dalam kebijakan stateful), tapi timestep.discount dan timestep.reward biasanya diabaikan.

Kebijakan terkait dengan komponen lain di TF-Agents dengan cara berikut. Sebagian besar kebijakan memiliki jaringan saraf untuk menghitung tindakan dan/atau distribusi melalui tindakan dari TimeSteps. Agen dapat berisi satu atau lebih kebijakan untuk tujuan yang berbeda, misalnya kebijakan utama yang sedang dilatih untuk penerapan, dan kebijakan noise untuk pengumpulan data. Kebijakan dapat disimpan/dipulihkan, dan dapat digunakan secara mandiri oleh agen untuk pengumpulan data, evaluasi, dll.

Beberapa kebijakan lebih mudah ditulis dalam Tensorflow (misalnya, yang memiliki jaringan saraf), sedangkan yang lain lebih mudah ditulis dengan Python (misalnya, mengikuti skrip tindakan). Jadi di agen TF, kami mengizinkan kebijakan Python dan Tensorflow. Selain itu, kebijakan yang ditulis dalam TensorFlow mungkin harus digunakan di lingkungan Python, atau sebaliknya, misalnya kebijakan TensorFlow digunakan untuk pelatihan tetapi kemudian diterapkan di lingkungan produksi Python. Untuk mempermudah ini, kami menyediakan pembungkus untuk mengonversi antara kebijakan Python dan TensorFlow.

Kelas kebijakan lain yang menarik adalah pembungkus kebijakan, yang memodifikasi kebijakan tertentu dengan cara tertentu, misalnya menambahkan jenis kebisingan tertentu, membuat versi serakah atau epsilon-rakus dari kebijakan stokastik, mencampur beberapa kebijakan secara acak, dll.

Mempersiapkan

Jika Anda belum menginstal tf-agents, jalankan:

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

Kebijakan Python

Antarmuka untuk kebijakan Python didefinisikan dalam policies/py_policy.PyPolicy . Metode utamanya adalah:

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

Metode yang paling penting adalah action(time_step) yang memetakan sebuah time_step mengandung pengamatan dari lingkungan ke PolicyStep bernama tuple yang berisi atribut berikut:

  • action : Tindakan yang harus diterapkan untuk lingkungan.
  • state : Negara bagian kebijakan (misalnya negara RNN) untuk dimasukkan ke dalam panggilan berikutnya untuk bertindak.
  • info : Informasi sisi opsional seperti probabilitas tindakan log.

The time_step_spec dan action_spec adalah spesifikasi untuk input langkah waktu dan tindakan output. Kebijakan juga memiliki reset fungsi yang biasanya digunakan untuk mereset negara dalam kebijakan stateful. The update(new_policy) fungsi update self terhadap new_policy .

Sekarang, mari kita lihat beberapa contoh kebijakan Python.

Contoh 1: Kebijakan Python Acak

Contoh sederhana dari PyPolicy adalah RandomPyPolicy yang menghasilkan tindakan acak untuk diskrit / kontinu action_spec diberikan. Input time_step diabaikan.

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

Contoh 2: Kebijakan Python Berskrip

Sebuah drama kebijakan scripted kembali naskah tindakan direpresentasikan sebagai daftar (num_repeats, action) tupel. Setiap kali action fungsi disebut, ia mengembalikan tindakan selanjutnya dari daftar sampai jumlah tertentu mengulangi dilakukan, dan kemudian pindah ke tindakan berikutnya dalam daftar. The reset metode dapat dipanggil untuk mulai menjalankan dari awal daftar.

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

Kebijakan TensorFlow

Kebijakan TensorFlow mengikuti antarmuka yang sama dengan kebijakan Python. Mari kita lihat beberapa contoh.

Contoh 1: Kebijakan TF Acak

Sebuah RandomTFPolicy dapat digunakan untuk menghasilkan tindakan acak menurut diskrit / kontinu diberikan action_spec . Input time_step diabaikan.

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)

Contoh 2: Kebijakan Aktor

Kebijakan aktor dapat dibuat baik menggunakan jaringan yang memetakan time_steps tindakan atau jaringan yang memetakan time_steps untuk distribusi atas tindakan.

Menggunakan jaringan aksi

Mari kita definisikan jaringan sebagai berikut:

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

Di TensorFlow sebagian besar lapisan jaringan dirancang untuk operasi batch, jadi kami berharap input time_steps di-batch, dan output dari jaringan juga akan di-batch. Juga jaringan bertanggung jawab untuk menghasilkan tindakan dalam rentang yang benar dari action_spec yang diberikan. Ini secara konvensional dilakukan dengan menggunakan misalnya tanh aktivasi untuk lapisan akhir tindakan produksi di [-1, 1] dan kemudian scaling dan pergeseran ini untuk rentang yang benar sebagai action_spec input (misalnya lihat tf_agents/agents/ddpg/networks.actor_network() ).

Sekarang, kita dapat membuat kebijakan aktor menggunakan jaringan di atas.

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)

Kami dapat menerapkannya ke kumpulan time_steps apa pun yang mengikuti 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)

Pada contoh di atas, kami membuat kebijakan menggunakan jaringan tindakan yang menghasilkan tensor tindakan. Dalam hal ini, policy.distribution(time_step) adalah deterministik (delta) distribusi di seluruh output dari policy.action(time_step) . Salah satu cara untuk menghasilkan kebijakan stokastik adalah dengan membungkus kebijakan aktor dalam pembungkus kebijakan yang menambahkan noise pada tindakan. Cara lain adalah membuat kebijakan aktor menggunakan jaringan distribusi tindakan alih-alih jaringan tindakan seperti yang ditunjukkan di bawah ini.

Menggunakan jaringan distribusi aksi

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)

Perhatikan bahwa dalam contoh di atas, tindakan dijepit ke kisaran spesifikasi tindakan yang diberikan [-1, 1]. Ini karena argumen konstruktor dari ActorPolicy clip=True secara default. Menyetel ini ke false akan mengembalikan tindakan tidak terpotong yang dihasilkan oleh jaringan.

Kebijakan Stochastic dapat dikonversi ke kebijakan deterministik menggunakan, misalnya, pembungkus GreedyPolicy yang memilih stochastic_policy.distribution().mode() sebagai aksinya, dan distribusi / delta deterministik sekitar tindakan serakah ini sebagai yang distribution() .

Contoh 3: Kebijakan Q

Kebijakan AQ digunakan di agen seperti DQN dan didasarkan pada jaringan Q yang memprediksi nilai Q untuk setiap tindakan diskrit. Untuk langkah waktu tertentu, distribusi tindakan dalam Kebijakan Q adalah distribusi kategorikal yang dibuat menggunakan nilai q sebagai 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)

Pembungkus Kebijakan

Pembungkus kebijakan dapat digunakan untuk membungkus dan memodifikasi kebijakan yang diberikan, misalnya menambahkan noise. Pembungkus kebijakan adalah subkelas dari Kebijakan (Python/TensorFlow) dan oleh karena itu dapat digunakan sama seperti kebijakan lainnya.

Contoh: Kebijakan Greedy

Sebuah wrapper serakah dapat digunakan untuk membungkus setiap kebijakan TensorFlow yang mengimplementasikan distribution() . GreedyPolicy.action() akan kembali wrapped_policy.distribution().mode() Dan GreedyPolicy.distribution() adalah distribusi / delta deterministik sekitar 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)