Copyright 2021 TF-AgentsAuthors。
TensorFlow.orgで表示
GoogleColabで実行
GitHubでソースを表示
ノートブックをダウンロード序章
強化学習の用語では、ポリシーは、環境からの観測をアクションまたはアクション全体の分布にマップします。 TF-エージェントでは、環境からの観測は、名前付きタプルに含まれているTimeStep('step_type', 'discount', 'reward', 'observation')およびポリシーがアクションオーバーアクションやディストリビューションへのタイムステップをマップします。ほとんどのポリシーが使用timestep.observation 、いくつかのポリシーが使用timestep.step_type （ステートフル方針のエピソードの初めに状態をリセットするために例えば）が、 timestep.discountとtimestep.reward通常無視されます。
ポリシーは、次のようにTF-Agentの他のコンポーネントに関連付けられています。ほとんどのポリシーには、TimeStepsからのアクションやアクションの分布を計算するためのニューラルネットワークがあります。エージェントには、さまざまな目的のために1つ以上のポリシーを含めることができます。たとえば、展開用にトレーニングされているメインポリシーや、データ収集用のノイズの多いポリシーなどです。ポリシーは保存/復元でき、データの収集や評価などのためにエージェントとは独立して使用できます。
一部のポリシーはTensorflowで記述しやすく（たとえばニューラルネットワークを使用している）、他のポリシーはPythonで記述しやすい（たとえばアクションのスクリプトに従う）。そのため、TFエージェントでは、PythonとTensorflowの両方のポリシーを許可します。さらに、TensorFlowで記述されたポリシーは、Python環境で使用する必要がある場合があります。その逆の場合もあります。たとえば、TensorFlowポリシーはトレーニングに使用されますが、後で本番Python環境にデプロイされます。これを簡単にするために、PythonポリシーとTensorFlowポリシーの間で変換するためのラッパーを提供しています。
別の興味深いクラスのポリシーは、特定のポリシーを特定の方法で変更するポリシーラッパーです。たとえば、特定の種類のノイズを追加したり、確率的ポリシーの貪欲またはイプシロン貪欲なバージョンを作成したり、複数のポリシーをランダムに混合したりします。
設定
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
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：ポリシー（例えばRNNの状態）の状態は、アクションの次の呼び出しに供給されます。 -
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ポリシー
A 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の正しい範囲でアクションを生成する責任があります。これは、従来、例えば、A 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_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.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)確率的ポリシーを作成する1つの方法は、アクションにノイズを追加するポリシーラッパーでアクターポリシーをラップすることです。もう1つの方法は、以下に示すように、アクションネットワークの代わりにアクション配布ネットワークを使用してアクターポリシーを作成することです。
アクション配信ネットワークの使用
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)