Авторские права 2021 Авторы TF-Agents.
 Посмотреть на TensorFlow.org |  Запускаем в Google Colab |  Посмотреть исходный код на GitHub |  Скачать блокнот |
Вступление
В терминологии обучения с подкреплением политики сопоставляют наблюдение из среды с действием или распределение по действиям. В TF-агентов, наблюдения из окружающей среды, содержатся в именованный кортеж TimeStep('step_type', 'discount', 'reward', 'observation') , и политика карта временных шагов к действиям или распределения по действиям. Большинство политика использует timestep.observation , некоторые политики используют timestep.step_type (например , для сброса состояния в начале эпизода в сохраняющей состоянии политики), но timestep.discount и timestep.reward , как правило , игнорируются.
Политики связаны с другими компонентами в TF-Agents следующим образом. Большинство политик имеют нейронную сеть для вычисления действий и / или распределений по действиям из TimeSteps. Агенты могут содержать одну или несколько политик для разных целей, например, основную политику, которая обучается для развертывания, и шумную политику для сбора данных. Политики могут быть сохранены / восстановлены и могут использоваться независимо от агента для сбора данных, оценки и т. Д.
Некоторые политики легче написать в Tensorflow (например, с нейронной сетью), тогда как другие проще написать на Python (например, следуя сценарию действий). Таким образом, в агентах TF мы разрешаем политики как Python, так и Tensorflow. Более того, политики, написанные на TensorFlow, возможно, придется использовать в среде Python или наоборот, например, политика TensorFlow используется для обучения, но позже развертывается в производственной среде Python. Чтобы упростить эту задачу, мы предоставляем оболочки для преобразования между политиками Python и TensorFlow.
Другой интересный класс политик - это оболочки политик, которые изменяют данную политику определенным образом, например, добавляют определенный тип шума, создают жадную или эпсилон-жадную версию стохастической политики, случайным образом смешивают несколько политик и т. Д.
Настраивать
Если вы еще не установили tf-agent, запустите:
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
Политики Python
Интерфейс для политики Python определяется в 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: состояние политики (например , РНН состояния) , которые будут подаваться в следующий призыв к действию. -
info: Дополнительно побочная информация , такая как лог действий вероятностей.
time_step_spec и action_spec являются спецификации для временного шага входного и выходного действия. Политики также имеют reset функцию , которая обычно используется для сброса в состояние с состоянием политики. update(new_policy) функция обновляет self по отношению к new_policy .
Теперь давайте посмотрим на пару примеров политик Python.
Пример 1: Случайная политика Python
Простой пример 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: Политика случайного 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 в качестве логитов.
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) и поэтому могут использоваться так же, как и любые другие политики.
Пример: жадная политика
Жадная обертка может быть использована , чтобы обернуть любую политику 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)