class_docstr: Base class to build policies based on reward predictions.
class_docstr_tok: ["Base","class","to","build","policies","based","on","reward","predictions","."]
docstr: Base class for building policies using a reward tf_agents.Network.
  
  This abstract base class of policies takes a tf_agents.Network predicting
  rewards and generates the action based on predicted rewards. Sub-classes
  need to implement the `_sample_action` abstract method, which defines the
  mapping from predicted rewards to sampled actions.
  
  Args:
    time_step_spec: A `TimeStep` spec of the expected time_steps.
    action_spec: A nest of BoundedTensorSpec representing the actions.
    reward_network: An instance of a `tf_agents.network.Network`, callable via
      `network(observation, step_type) -> (output, final_state)`.
    observation_and_action_constraint_splitter: A function used for masking
      valid/invalid actions with each state of the environment. The function
      takes in a full observation and returns a tuple consisting of 1) the
      part of the observation intended as input to the network and 2) the
      mask.  The mask should be a 0-1 `Tensor` of shape `[batch_size,
      num_actions]`. This function should also work with a `TensorSpec` as
      input, and should output `TensorSpec` objects for the observation and
      mask.
    accepts_per_arm_features: (bool) Whether the policy accepts per-arm
      features.
    constraints: Iterable of constraints objects that are instances of
      `tf_agents.bandits.agents.BaseConstraint`.
    emit_policy_info: (tuple of strings) what side information we want to get
      as part of the policy info. Allowed values can be found in
      `policy_utilities.PolicyInfo`.
    name: The name of this policy. All variables in this module will fall
      under that name. Defaults to the class name.
  
  Raises:
    NotImplementedError: If `action_spec` contains more than one
      `BoundedTensorSpec` or the `BoundedTensorSpec` is not valid.
docstr_tok: ["Base","class","for","building","policies","using","a","reward","tf_agents.Network",".","This","abstract","base","class","of","policies","takes","a","tf_agents.Network","predicting","rewards","and","generates","the","action","based","on","predicted","rewards",".","Sub-classes","need","to","implement","the","`","_sample_action","`","abstract","method",",","which","defines","the","mapping","from","predicted","rewards","to","sampled","actions",".","Args",":","time_step_spec",":","A","`","TimeStep","`","spec","of","the","expected","time_steps",".","action_spec",":","A","nest","of","BoundedTensorSpec","representing","the","actions",".","reward_network",":","An","instance","of","a","`","tf_agents.network.Network","`",",","callable","via","`","network","(","observation",",","step_type",")","-",">","(","output",",","final_state",")","`",".","observation_and_action_constraint_splitter",":","A","function","used","for","masking","valid\/invalid","actions","with","each","state","of","the","environment",".","The","function","takes","in","a","full","observation","and","returns","a","tuple","consisting","of","1",")","the","part","of","the","observation","intended","as","input","to","the","network","and","2",")","the","mask",".","The","mask","should","be","a","0-1","`","Tensor","`","of","shape","`","[","batch_size",",","num_actions","]","`",".","This","function","should","also","work","with","a","`","TensorSpec","`","as","input",",","and","should","output","`","TensorSpec","`","objects","for","the","observation","and","mask",".","accepts_per_arm_features",":","(","bool",")","Whether","the","policy","accepts","per-arm","features",".","constraints",":","Iterable","of","constraints","objects","that","are","instances","of","`","tf_agents.bandits.agents.BaseConstraint","`",".","emit_policy_info",":","(","tuple","of","strings",")","what","side","information","we","want","to","get","as","part","of","the","policy","info",".","Allowed","values","can","be","found","in","`","policy_utilities.PolicyInfo","`",".","name",":","The","name","of","this","policy",".","All","variables","in","this","module","will","fall","under","that","name",".","Defaults","to","the","class","name",".","Raises",":","NotImplementedError",":","If","`","action_spec","`","contains","more","than","one","`","BoundedTensorSpec","`","or","the","`","BoundedTensorSpec","`","is","not","valid","."]
code: def __init__(
      self,
      time_step_spec: types.TimeStep,
      action_spec: types.NestedTensorSpec,
      reward_network: types.Network,
      observation_and_action_constraint_splitter: Optional[
          types.Splitter
      ] = None,
      accepts_per_arm_features: bool = False,
      constraints: Iterable[constr.BaseConstraint] = (),
      emit_policy_info: Tuple[Text, ...] = (),
      name: Optional[Text] = None,
  ):
      """Base class for building policies using a reward tf_agents.Network.
  
      This abstract base class of policies takes a tf_agents.Network predicting
      rewards and generates the action based on predicted rewards. Sub-classes
      need to implement the `_sample_action` abstract method, which defines the
      mapping from predicted rewards to sampled actions.
  
      Args:
        time_step_spec: A `TimeStep` spec of the expected time_steps.
        action_spec: A nest of BoundedTensorSpec representing the actions.
        reward_network: An instance of a `tf_agents.network.Network`, callable via
          `network(observation, step_type) -> (output, final_state)`.
        observation_and_action_constraint_splitter: A function used for masking
          valid/invalid actions with each state of the environment. The function
          takes in a full observation and returns a tuple consisting of 1) the
          part of the observation intended as input to the network and 2) the
          mask.  The mask should be a 0-1 `Tensor` of shape `[batch_size,
          num_actions]`. This function should also work with a `TensorSpec` as
          input, and should output `TensorSpec` objects for the observation and
          mask.
        accepts_per_arm_features: (bool) Whether the policy accepts per-arm
          features.
        constraints: Iterable of constraints objects that are instances of
          `tf_agents.bandits.agents.BaseConstraint`.
        emit_policy_info: (tuple of strings) what side information we want to get
          as part of the policy info. Allowed values can be found in
          `policy_utilities.PolicyInfo`.
        name: The name of this policy. All variables in this module will fall
          under that name. Defaults to the class name.
  
      Raises:
        NotImplementedError: If `action_spec` contains more than one
          `BoundedTensorSpec` or the `BoundedTensorSpec` is not valid.
      """
      policy_utilities.check_no_mask_with_arm_features(
          accepts_per_arm_features,
          observation_and_action_constraint_splitter,
      )
      flat_action_spec = tf.nest.flatten(action_spec)
      if len(flat_action_spec) > 1:
          raise NotImplementedError(
              "action_spec can only contain a single BoundedTensorSpec."
          )
  
      action_spec = flat_action_spec[0]
      if (
          not tensor_spec.is_bounded(action_spec)
          or not tensor_spec.is_discrete(action_spec)
          or action_spec.shape.rank > 1
          or action_spec.shape.num_elements() != 1
      ):
          raise NotImplementedError(
              "action_spec must be a BoundedTensorSpec of type int32 and shape (). "
              "Found {}.".format(action_spec)
          )
      self._expected_num_actions = (
          action_spec.maximum - action_spec.minimum + 1
      )
      self._action_offset = action_spec.minimum
      reward_network.create_variables()
      self._reward_network = reward_network
      self._constraints = constraints
  
      self._emit_policy_info = emit_policy_info
      if (
          policy_utilities.InfoFields.PREDICTED_REWARDS_MEAN
          in emit_policy_info
      ):
          predicted_rewards_mean = tensor_spec.TensorSpec(
              [self._expected_num_actions]
          )
      else:
          predicted_rewards_mean = ()
  
      if (
          policy_utilities.InfoFields.BANDIT_POLICY_TYPE
          in emit_policy_info
      ):
          bandit_policy_type = (
              policy_utilities.create_bandit_policy_type_tensor_spec(
                  shape=[1]
              )
          )
      else:
          bandit_policy_type = ()
  
      if accepts_per_arm_features:
          # The features for the chosen arm is saved to policy_info.
          chosen_arm_features_info = (
              policy_utilities.create_chosen_arm_features_info_spec(
                  time_step_spec.observation
              )
          )
          info_spec = policy_utilities.PerArmPolicyInfo(
              predicted_rewards_mean=predicted_rewards_mean,
              bandit_policy_type=bandit_policy_type,
              chosen_arm_features=chosen_arm_features_info,
          )
      else:
          info_spec = policy_utilities.PolicyInfo(
              predicted_rewards_mean=predicted_rewards_mean,
              bandit_policy_type=bandit_policy_type,
          )
  
      self._accepts_per_arm_features = accepts_per_arm_features
  
      super(RewardPredictionBasePolicy, self).__init__(
          time_step_spec,
          action_spec,
          policy_state_spec=reward_network.state_spec,
          clip=False,
          info_spec=info_spec,
          emit_log_probability=policy_utilities.InfoFields.LOG_PROBABILITY
          in emit_policy_info,
          observation_and_action_constraint_splitter=(
              observation_and_action_constraint_splitter
          ),
          name=name,
      )
  
imports: import abc
  from typing import Iterable, Optional, Text, Tuple
  import gin
  import six
  import tensorflow
  import tensorflow_probability
  from tf_agents.bandits.networks import heteroscedastic_q_network
  from tf_agents.bandits.policies import constraints
  from tf_agents.bandits.specs import utils
  from tf_agents.policies import tf_policy
  from tf_agents.policies import utils
  from tf_agents.specs import tensor_spec
  from tf_agents.trajectories import policy_step
  from tf_agents.trajectories import time_step
  from tf_agents.typing import types
label_desc: Use image node_id 1 to create a new RewardPredictionBasePolicy object from inherited base classes: tf_policy with example: obj = RewardPredictionBasePolicy(time_step_spec, action_spec, reward_network, observation_and_action_constraint_splitter, accepts_per_arm_features, constraints, emit_policy_info, name)