網路

在 TensorFlow.org 上檢視 在 Google Colab 中執行 在 GitHub 上檢視原始碼 下載筆記本

簡介

在本 Colab 中,我們將介紹如何為您的代理程式定義自訂網路。網路有助於我們定義由代理程式訓練的模型。在 TF-Agents 中,您會找到幾種不同類型的網路,這些網路在各種代理程式中都很有用

主要網路

  • QNetwork:用於 Q-learning 中,適用於具有離散動作的環境,此網路將觀察結果對應到每個可能動作的價值估計。
  • CriticNetworks:在文獻中也稱為 ValueNetworks,學習估計某個版本的價值函數,將某個狀態對應到策略預期報酬的估計值。這些網路會估計代理程式目前所處狀態的好壞。
  • ActorNetworks:學習從觀察結果到動作的對應。這些網路通常由我們的策略用來產生動作。
  • ActorDistributionNetworks:與 ActorNetworks 類似,但這些網路會產生分配,策略可以從中取樣以產生動作。

輔助網路

  • EncodingNetwork:讓使用者能夠輕鬆定義預先處理層的對應,以套用至網路的輸入。
  • DynamicUnrollLayer:當在時間序列上套用時,會在情節邊界自動重設網路的狀態。
  • ProjectionNetwork:諸如 CategoricalProjectionNetwork 或 NormalProjectionNetwork 等網路會接收輸入,並產生產生類別或常態分配所需的參數。

TF-Agents 中的所有範例都隨附預先設定的網路。但是,這些網路並未設定為處理複雜的觀察結果。

如果您有一個環境公開多個觀察/動作,並且您需要自訂網路,那麼本教學課程非常適合您!

設定

如果您尚未安裝 tf-agents,請執行

pip install tf-agents
pip install tf-keras
import os
# Keep using keras-2 (tf-keras) rather than keras-3 (keras).
os.environ['TF_USE_LEGACY_KERAS'] = '1'
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import abc
import tensorflow as tf
import numpy as np

from tf_agents.environments import random_py_environment
from tf_agents.environments import tf_py_environment
from tf_agents.networks import encoding_network
from tf_agents.networks import network
from tf_agents.networks import utils
from tf_agents.specs import array_spec
from tf_agents.utils import common as common_utils
from tf_agents.utils import nest_utils

定義網路

網路 API

在 TF-Agents 中,我們從 Keras Networks 繼承子類別。有了它,我們可以

  • 簡化建立目標網路時所需的複製操作。
  • 在呼叫 network.variables() 時執行自動變數建立。
  • 根據網路 input_specs 驗證輸入。

EncodingNetwork

如上所述,EncodingNetwork 讓我們能夠輕鬆定義預先處理層的對應,以套用至網路的輸入以產生一些編碼。

EncodingNetwork 由以下大多數為選用的層組成

  • 預先處理層
  • 預先處理組合器
  • Conv2D
  • Flatten
  • Dense

關於編碼網路的特別之處在於套用了輸入預先處理。輸入預先處理可透過 preprocessing_layers 和 preprocessing_combiner 層來實現。這些層中的每一個都可以指定為巢狀結構。如果 preprocessing_layers 巢狀結構比 input_tensor_spec 淺,則這些層將取得子巢狀結構。例如,如果

input_tensor_spec = ([TensorSpec(3)] * 2, [TensorSpec(3)] * 5)
preprocessing_layers = (Layer1(), Layer2())

則預先處理將呼叫

preprocessed = [preprocessing_layers[0](observations[0]),
                preprocessing_layers[1](observations[1])]

但是,如果

preprocessing_layers = ([Layer1() for _ in range(2)],
                        [Layer2() for _ in range(5)])

則預先處理將呼叫

preprocessed = [
  layer(obs) for layer, obs in zip(flatten(preprocessing_layers),
                                    flatten(observations))
]

自訂網路

若要建立您自己的網路,您只需要覆寫 __init__ 和呼叫方法。讓我們使用我們從 EncodingNetworks 學到的知識來建立一個自訂網路,以建立一個 ActorNetwork,該網路會接收包含影像和向量的觀察結果。

class ActorNetwork(network.Network):

  def __init__(self,
               observation_spec,
               action_spec,
               preprocessing_layers=None,
               preprocessing_combiner=None,
               conv_layer_params=None,
               fc_layer_params=(75, 40),
               dropout_layer_params=None,
               activation_fn=tf.keras.activations.relu,
               enable_last_layer_zero_initializer=False,
               name='ActorNetwork'):
    super(ActorNetwork, self).__init__(
        input_tensor_spec=observation_spec, state_spec=(), name=name)

    # For simplicity we will only support a single action float output.
    self._action_spec = action_spec
    flat_action_spec = tf.nest.flatten(action_spec)
    if len(flat_action_spec) > 1:
      raise ValueError('Only a single action is supported by this network')
    self._single_action_spec = flat_action_spec[0]
    if self._single_action_spec.dtype not in [tf.float32, tf.float64]:
      raise ValueError('Only float actions are supported by this network.')

    kernel_initializer = tf.keras.initializers.VarianceScaling(
        scale=1. / 3., mode='fan_in', distribution='uniform')
    self._encoder = encoding_network.EncodingNetwork(
        observation_spec,
        preprocessing_layers=preprocessing_layers,
        preprocessing_combiner=preprocessing_combiner,
        conv_layer_params=conv_layer_params,
        fc_layer_params=fc_layer_params,
        dropout_layer_params=dropout_layer_params,
        activation_fn=activation_fn,
        kernel_initializer=kernel_initializer,
        batch_squash=False)

    initializer = tf.keras.initializers.RandomUniform(
        minval=-0.003, maxval=0.003)

    self._action_projection_layer = tf.keras.layers.Dense(
        flat_action_spec[0].shape.num_elements(),
        activation=tf.keras.activations.tanh,
        kernel_initializer=initializer,
        name='action')

  def call(self, observations, step_type=(), network_state=()):
    outer_rank = nest_utils.get_outer_rank(observations, self.input_tensor_spec)
    # We use batch_squash here in case the observations have a time sequence
    # compoment.
    batch_squash = utils.BatchSquash(outer_rank)
    observations = tf.nest.map_structure(batch_squash.flatten, observations)

    state, network_state = self._encoder(
        observations, step_type=step_type, network_state=network_state)
    actions = self._action_projection_layer(state)
    actions = common_utils.scale_to_spec(actions, self._single_action_spec)
    actions = batch_squash.unflatten(actions)
    return tf.nest.pack_sequence_as(self._action_spec, [actions]), network_state

讓我們建立一個 RandomPyEnvironment 以產生結構化觀察結果並驗證我們的實作。

action_spec = array_spec.BoundedArraySpec((3,), np.float32, minimum=0, maximum=10)
observation_spec =  {
    'image': array_spec.BoundedArraySpec((16, 16, 3), np.float32, minimum=0,
                                        maximum=255),
    'vector': array_spec.BoundedArraySpec((5,), np.float32, minimum=-100,
                                          maximum=100)}

random_env = random_py_environment.RandomPyEnvironment(observation_spec, action_spec=action_spec)

# Convert the environment to a TFEnv to generate tensors.
tf_env = tf_py_environment.TFPyEnvironment(random_env)
/tmpfs/src/tf_docs_env/lib/python3.9/site-packages/tf_agents/specs/array_spec.py:352: RuntimeWarning: invalid value encountered in cast
  self._minimum[self._minimum == -np.inf] = low
/tmpfs/src/tf_docs_env/lib/python3.9/site-packages/tf_agents/specs/array_spec.py:353: RuntimeWarning: invalid value encountered in cast
  self._minimum[self._minimum == np.inf] = high
/tmpfs/src/tf_docs_env/lib/python3.9/site-packages/tf_agents/specs/array_spec.py:355: RuntimeWarning: invalid value encountered in cast
  self._maximum[self._maximum == -np.inf] = low
/tmpfs/src/tf_docs_env/lib/python3.9/site-packages/tf_agents/specs/array_spec.py:356: RuntimeWarning: invalid value encountered in cast
  self._maximum[self._maximum == np.inf] = high

由於我們已將觀察結果定義為 dict,因此我們需要建立預先處理層來處理這些結果。

preprocessing_layers = {
    'image': tf.keras.models.Sequential([tf.keras.layers.Conv2D(8, 4),
                                        tf.keras.layers.Flatten()]),
    'vector': tf.keras.layers.Dense(5)
    }
preprocessing_combiner = tf.keras.layers.Concatenate(axis=-1)
actor = ActorNetwork(tf_env.observation_spec(), 
                     tf_env.action_spec(),
                     preprocessing_layers=preprocessing_layers,
                     preprocessing_combiner=preprocessing_combiner)

現在我們有了演員網路,我們可以處理來自環境的觀察結果。

time_step = tf_env.reset()
actor(time_step.observation, time_step.step_type)
/tmpfs/src/tf_docs_env/lib/python3.9/site-packages/tf_keras/src/initializers/initializers.py:121: UserWarning: The initializer VarianceScaling is unseeded and being called multiple times, which will return identical values each time (even if the initializer is unseeded). Please update your code to provide a seed to the initializer, or avoid using the same initializer instance more than once.
  warnings.warn(
(<tf.Tensor: shape=(1, 3), dtype=float32, numpy=array([[5.8357787, 4.3249702, 3.6428978]], dtype=float32)>,
 ())

相同的策略可用於自訂代理程式使用的任何主要網路。您可以定義任何預先處理,並將其連接到網路的其餘部分。當您定義自己的自訂網路時,請確保網路的輸出層定義相符。