ポリシー

TensorFlow.orgで表示 GoogleColabで実行 GitHubでソースを表示ノートブックをダウンロードする

前書き

強化学習の用語では、ポリシーは、環境からの観測をアクションまたはアクション全体の分布にマップします。 TF-Agentでは、環境からのTimeStep('step_type', 'discount', 'reward', 'observation')は名前付きタプルTimeStep('step_type', 'discount', 'reward', 'observation')に含まれ、ポリシーはタイムステップをアクションまたはアクション上の分布にマップします。ほとんどのポリシーはtimestep.observation使用timestep.observation 、一部のポリシーはtimestep.step_type使用timestep.step_type (たとえば、ステートフルポリシーのエピソードの開始時に状態をリセットするため)が、 timestep.discounttimestep.rewardは通常無視されます。

ポリシーは、次のようにTFエージェントの他のコンポーネントに関連付けられています。ほとんどのポリシーには、TimeStepsからのアクションやアクションの分布を計算するためのニューラルネットワークがあります。エージェントには、さまざまな目的のために1つ以上のポリシーを含めることができます。たとえば、展開用にトレーニングされているメインポリシーや、データ収集用のノイズの多いポリシーなどです。ポリシーは保存/復元でき、データの収集や評価などのためにエージェントとは独立して使用できます。

一部のポリシーはTensorflowで記述しやすい(たとえばニューラルネットワークを使用している)のに対し、他のポリシーはPythonで記述しやすい(たとえばアクションのスクリプトに従う)。そのため、TFエージェントでは、PythonポリシーとTensorflowポリシーの両方を許可します。さらに、TensorFlowで記述されたポリシーはPython環境で使用する必要がある場合があり、その逆もあります。たとえば、TensorFlowポリシーはトレーニングに使用されますが、後で本番Python環境にデプロイされます。これを簡単にするために、PythonポリシーとTensorFlowポリシーを変換するためのラッパーを提供しています。

別の興味深いクラスのポリシーは、特定のポリシーを特定の方法で変更するポリシーラッパーです。たとえば、特定の種類のノイズを追加したり、確率的ポリシーの欲張りまたはイプシロン欲張りバージョンを作成したり、複数のポリシーをランダムに混合したりします。

セットアップ

tf-agentsをまだインストールしていない場合は、次を実行します。

pip install -q 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を、次の属性を含むtupleという名前のtime_stepマップするaction(time_step)です。

  • action :環境に適用されるアクション。
  • state :次の召喚状に送られるポリシーの状態(RNN状態など)。
  • info :アクションログの確率などのオプションのサイド情報。

time_step_specaction_specは、入力タイムステップと出力アクションの仕様です。ポリシーには、ステートフルポリシーの状態をリセットするために通常使用されるreset機能もあります。 update(new_policy)関数は、 selfnew_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([-6,  2], dtype=int32), state=(), info=())
PolicyStep(action=array([-7,  7], 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.8538141  -0.95033884], 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_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.06422257  0.6671298   0.734995  ]
 [-0.06422257  0.6671298   0.734995  ]], 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(
[[ 1.         1.        -0.810334 ]
 [-0.5075199  1.         0.1654225]], 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()を選択するGreedyPolicyラッパー、およびその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([0 2], shape=(2,), dtype=int32)
Action distribution:
tfp.distributions.Categorical("Categorical", batch_shape=[2], event_shape=[], dtype=int32)

ポリシーラッパー

ポリシーラッパーを使用して、特定のポリシーをラップおよび変更できます(ノイズの追加など)。ポリシーラッパーはポリシー(Python / TensorFlow)のサブクラスであるため、他のポリシーと同じように使用できます。

例:欲張りポリシー

貪欲なラッパーを使用して、 distribution()を実装distribution() TensorFlowポリシーをラップできます。 GreedyPolicy.action()wrapped_policy.distribution().mode()GreedyPolicy.distribution()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)