models/moa_model.py
def compute_influence_reward(self, input_dict, prev_action_logits, counterfactual_logits):
        """
        Compute influence of this agent on other agents.
        :param input_dict: The model input tensors.
        :param prev_action_logits: Logits for the agent's own policy/actions at t-1
        :param counterfactual_logits: The counterfactual action logits for actions made by other
        agents at t.
        """
        # Probability of the next action for all other agents. Shape is [B, N, A].
        # This is the predicted probability given the actions that we DID take.
        # extract out the probability under the actions we actually did take

        # We don't have the current action yet, so the reward for the previous step is calculated.
        # This is corrected for in the function weigh_and_add_influence_reward
        prev_agent_actions = tf.cast(tf.reshape(input_dict["prev_actions"], [-1, 1]), tf.int32)
        # Use the agent's actions as indices to select the predicted logits of other agents for
        # actions that the agent did take, discard the rest.
        predicted_logits = tf.gather_nd(
            params=counterfactual_logits, indices=prev_agent_actions, batch_dims=1
        )

        predicted_logits = tf.reshape(
            predicted_logits, [-1, self.num_other_agents, self.num_outputs]
        )
        predicted_logits = tf.nn.softmax(predicted_logits)
        predicted_logits = predicted_logits / tf.reduce_sum(
            predicted_logits, axis=-1, keepdims=True
        )  # reduce numerical inaccuracies

        # Get marginal predictions where effect of self is marginalized out
        marginal_logits = self.marginalize_predictions_over_own_actions(
            prev_action_logits, counterfactual_logits
        )  # [B, Num agents, Num actions]

        # Compute influence per agent/step ([B, N]) using different metrics
        if self.influence_divergence_measure == "kl":
            influence_reward = self.kl_div(predicted_logits, marginal_logits)
        elif self.influence_divergence_measure == "jsd":
            mean_probs = 0.5 * (predicted_logits + marginal_logits)
            influence_reward = 0.5 * self.kl_div(predicted_logits, mean_probs) + 0.5 * self.kl_div(
                marginal_logits, mean_probs
            )
        else:
            sys.exit("Please specify an influence divergence measure from [kl, jsd]")

        # Zero out influence for steps where the other agent isn't visible.
        if self.influence_only_when_visible:
            visibility = tf.cast(input_dict["obs"]["prev_visible_agents"], tf.float32)
            influence_reward *= visibility
        influence_reward = tf.reduce_sum(influence_reward, axis=-1)
        self._social_influence_reward = influence_reward

        Compute influence of this agent on other agents.
        :param input_dict: The model input tensors.
        :param prev_action_logits: Logits for the agent's own policy/actions at t-1
        :param counterfactual_logits: The counterfactual action logits for actions made by other
        agents at t.
        
Compute influence of this agent on other agents.
Compute influence of this agent on other agents .
def compute_influence_reward(self, input_dict, prev_action_logits, counterfactual_logits):
        prev_agent_actions = tf.cast(tf.reshape(input_dict["prev_actions"], [-1, 1]), tf.int32)
        predicted_logits = tf.gather_nd(
            params=counterfactual_logits, indices=prev_agent_actions, batch_dims=1
        )
        predicted_logits = tf.reshape(
            predicted_logits, [-1, self.num_other_agents, self.num_outputs]
        )
        predicted_logits = tf.nn.softmax(predicted_logits)
        predicted_logits = predicted_logits / tf.reduce_sum(
            predicted_logits, axis=-1, keepdims=True
        )  
        marginal_logits = self.marginalize_predictions_over_own_actions(
            prev_action_logits, counterfactual_logits
        )  
        if self.influence_divergence_measure == "kl":
            influence_reward = self.kl_div(predicted_logits, marginal_logits)
        elif self.influence_divergence_measure == "jsd":
            mean_probs = 0.5 * (predicted_logits + marginal_logits)
            influence_reward = 0.5 * self.kl_div(predicted_logits, mean_probs) + 0.5 * self.kl_div(
                marginal_logits, mean_probs
            )
        else:
            sys.exit("Please specify an influence divergence measure from [kl, jsd]")
        if self.influence_only_when_visible:
            visibility = tf.cast(input_dict["obs"]["prev_visible_agents"], tf.float32)
            influence_reward *= visibility
        influence_reward = tf.reduce_sum(influence_reward, axis=-1)
        self._social_influence_reward = influence_reward
def compute_influence_reward ( self , input_dict , prev_action_logits , counterfactual_logits ) : prev_agent_actions = tf . cast ( tf . reshape ( input_dict [ "prev_actions" ] , [ - 1 , 1 ] ) , tf . int32 ) predicted_logits = tf . gather_nd ( params = counterfactual_logits , indices = prev_agent_actions , batch_dims = 1 ) predicted_logits = tf . reshape ( predicted_logits , [ - 1 , self . num_other_agents , self . num_outputs ] ) predicted_logits = tf . nn . softmax ( predicted_logits ) predicted_logits = predicted_logits / tf . reduce_sum ( predicted_logits , axis = - 1 , keepdims = True ) marginal_logits = self . marginalize_predictions_over_own_actions ( prev_action_logits , counterfactual_logits ) if self . influence_divergence_measure == "kl" : influence_reward = self . kl_div ( predicted_logits , marginal_logits ) elif self . influence_divergence_measure == "jsd" : mean_probs = 0.5 * ( predicted_logits + marginal_logits ) influence_reward = 0.5 * self . kl_div ( predicted_logits , mean_probs ) + 0.5 * self . kl_div ( marginal_logits , mean_probs ) else : sys . exit ( "Please specify an influence divergence measure from [kl, jsd]" ) if self . influence_only_when_visible : visibility = tf . cast ( input_dict [ "obs" ] [ "prev_visible_agents" ] , tf . float32 ) influence_reward *= visibility influence_reward = tf . reduce_sum ( influence_reward , axis = - 1 ) self . _social_influence_reward = influence_reward
Compute influence of this agent on other agents.
Compute influence of this agent on other agents .
"""
        Compute influence of this agent on other agents.
        :param input_dict: The model input tensors.
        :param prev_action_logits: Logits for the agent's own policy/actions at t-1
        :param counterfactual_logits: The counterfactual action logits for actions made by other
        agents at t.
        """ # Probability of the next action for all other agents. Shape is [B, N, A]. # This is the predicted probability given the actions that we DID take. # extract out the probability under the actions we actually did take # We don't have the current action yet, so the reward for the previous step is calculated. # This is corrected for in the function weigh_and_add_influence_reward # Use the agent's actions as indices to select the predicted logits of other agents for # actions that the agent did take, discard the rest. # reduce numerical inaccuracies # Get marginal predictions where effect of self is marginalized out # [B, Num agents, Num actions] # Compute influence per agent/step ([B, N]) using different metrics # Zero out influence for steps where the other agent isn't visible.
self
input_dict
The model input tensors.
The model input tensors .
prev_action_logits
Logits for the agent's own policy/actions at t-1
Logits for the agent ' s own policy / actions at t - 1
counterfactual_logits
The counterfactual action logits for actions made by other
agents at t.
The counterfactual action logits for actions made by other agents at t .