Copyright 2021 The TF-Agents Authors.
 TensorFlow.org에서 보기 |
 Google Colab에서 실행하기 |
 GitHub에서 소스 보기 |
 노트북 다운로드하기 |
소개
강화 학습 용어에서 정책은 환경의 관찰 값을 행동 또는 행동에 대한 분포로 매핑합니다. TF-Agents에서 환경의 관찰 값은 명명된 튜플 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 정책 간 변환을 위한 래퍼를 제공합니다.
또 다른 흥미로운 정책 클래스는 특정 유형의 노이즈를 추가하거나 확률적 정책의 최대(greedy) 또는 엡실론 최대(epsilon-greedy) 버전을 만들고 여러 정책을 무작위로 혼합하는 등 특정 방식으로 주어진 정책을 수정하는 정책 래퍼입니다.
설정
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
tf.compat.v1.enable_v2_behavior()
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
가장 중요한 메서드는 환경의 관찰 값을 포함하는 time_step을 다음 속성이 포함된 PolicyStep 명명된 튜플에 매핑하는 action(time_step)입니다.
action: 환경에 적용할 행동state: 다음 행동 호출에 제공될 정책의 상태(예: RNN 상태)info: 행동 로그 확률과 같은 선택적 보조 정보
time_step_spec 및 action_spec은 입력 타임스텝 및 출력 행동의 사양입니다. 정책에는 일반적으로 상태 저장 정책에서 상태를 재설정하는 데 사용되는 reset 함수가 있습니다. update(new_policy) 함수는 new_policy 쪽으로 self를 업데이트합니다.
이제 Python 정책의 몇 가지 예제를 살펴보겠습니다.
예제 1: 임의 Python 정책
PyPolicy의 간단한 예제는 주어진 불연속/연속 action_spec에 대한 무작위 행동을 생성하는 RandomPyPolicy입니다. 입력 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([ -2, -10], dtype=int32), state=(), info=()) PolicyStep(action=array([-7, 8], 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([ 1.3286481 -0.61121655], shape=(2,), dtype=float32)
예제 2: Actor 정책
Actor 정책은 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()).
이제 위의 네트워크를 사용하여 actor 정책을 만들 수 있습니다.
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 다음에 오는 모든 time_steps 배치에 적용할 수 있습니다.
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.47939157 0.98388875 0.7276587 ]
[0.47939157 0.98388875 0.7276587 ]], 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)의 출력 주위에서의 결정적(델타) 분포입니다. 확률적 정책을 생성하는 한 가지 방법은 행동에 노이즈를 추가하는 정책 래퍼로 actor 정책을 래핑하는 것입니다. 또 다른 방법은 아래와 같이 행동 네트워크 대신 행동 분포 네트워크를 사용하여 actor 정책을 만드는 것입니다.
행동 분포 네트워크 사용하기
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.11090657 -0.96768284 0.14140609]
[-0.7287691 1. -0.7402514 ]], 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로 설정하면 네트워크에서 생성된 잘리지 않은 행동이 반환됩니다.
예를 들어, stochastic_policy.distribution().mode()를 행동으로 선택하는, 그리고 이 최대 행동 주변의 결정론적/델타 분포를 distribution()으로 선택하는 GreedyPolicy 래퍼를 사용하여 확률적 정책을 결정론적 정책으로 변환할 수 있습니다.
예제 3: Q 정책
Q 정책은 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([0 0], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.Categorical("Categorical", batch_shape=[2], event_shape=[], dtype=int32)
정책 래퍼
정책 래퍼를 사용하여 주어진 정책을 래핑하고 수정할 수 있습니다(예: 노이즈 추가). 정책 래퍼는 정책 (Python/TensorFlow)의 서브 클래스이므로 다른 정책과 마찬가지로 사용할 수 있습니다.
예제: 최대 정책(Greedy Policy)
최대 래퍼(greedy wrapper)를 사용하여 distribution()을 구현하는 모든 TensorFlow 정책을 래핑할 수 있습니다. 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([0 0], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.DeterministicWithLogProb("Deterministic", batch_shape=[2], event_shape=[], dtype=int32)