nip.protocols.protocol_base.ProtocolHandler#

class nip.protocols.protocol_base.ProtocolHandler(hyper_params: HyperParameters, settings: ExperimentSettings)[source]#

Base class for protocol handlers.

A protocol handler gives the implementation of an exchange protocol, specifying what agents are present, how they interact, and how the environment is updated.

To implement a new protocol, subclass this class and implement the following properties and methods:

  • agent_names (property): The names of the agents in the protocol.

  • max_message_rounds (property): The maximum number of rounds in the protocol.

  • min_message_rounds (property): The minimum number of rounds in the protocol.

  • max_verifier_questions (property): The maximum number of questions the verifier can make to each prover.

  • message_channel_names (property): The names of the message channels in the protocol.

  • agent_channel_visibility (property): A specification of which agents can see

    which message channels.

  • get_active_agents_mask_from_rounds_and_seed (method): Get a boolean mask of active agents for a batch of rounds.

  • can_agent_be_active (method): Specify whether an agent can be active in a given round and channel.

  • get_verifier_guess_mask_from_rounds_and_seed (method): Get a boolean mask of the verifier’s guesses for a batch of rounds.

  • step_interaction_protocol (method): Take a step in the interaction protocol.

  • reward_mid_point_estimate (method): Get an estimate of the expected reward if all agents play randomly.

Parameters:

hyper_params (HyperParameters) – The parameters of the experiment.

Methods Summary

__init__(hyper_params, settings)

_get_agent_decision_made_mask(round_id, y, ...)

Get a mask indicating whether an agent has made a decision.

can_agent_be_active(agent_name, round_id, ...)

Specify whether an agent can be active in a given round and channel.

can_agent_be_active_any_channel(agent_name, ...)

Specify whether an agent can be active in any channel in a given round.

can_agent_see_channel(agent_name, channel_name)

Determine whether an agent can see a channel.

get_active_agents_mask_from_rounds_and_seed(...)

Get a boolean mask of active agents for a batch of rounds.

get_agent_visible_channels(agent_name)

Get the names of the channels visible to an agent.

get_verifier_guess_mask_from_rounds_and_seed(...)

Get a boolean mask indicating when the verifiers can make a guess/decision.

reward_mid_point_estimate(agent_name)

Get an estimate of the expected reward if all agents play randomly.

step_interaction_protocol(env_td)

Take a step in the interaction protocol.

Attributes

agent_channel_visibility

A specification of which agents can see which message channels.

agent_channel_visibility_mask

A boolean mask indicating which agents can see which message channels.

agent_first_active_round

The first round in which each agent is or can be active.

agent_names

The names of the agents in the protocol.

can_be_zero_knowledge

default_stackelberg_sequence

The default Stackelberg sequence for the protocol.

max_message_rounds

The maximum number of rounds in the protocol.

max_verifier_questions

The maximum number of questions the verifier can make to each prover.

message_channel_names

The names of the message channels in the protocol.

min_message_rounds

The minimum number of rounds in the protocol.

num_agents

The number of agents in the protocol.

num_message_channels

The number of message channels in the protocol.

prover_indices

The indices of the provers in the list of agent names.

prover_names

The names of the provers in the protocol.

stackelberg_sequence

The actual Stackelberg sequence used in this experiment.

verifier_names

The names of the verifiers in the protocol.

Methods

__init__(hyper_params: HyperParameters, settings: ExperimentSettings)[source]#
_get_agent_decision_made_mask(round_id: Int[Tensor, '...'], y: Int[Tensor, '... 1'], guess_mask: Bool[Tensor, '...'], decision: Int[Tensor, '...'], *, follow_force_guess: bool = True) Bool[Tensor, '...'][source]#

Get a mask indicating whether an agent has made a decision.

Parameters:
  • round_id (Int[Tensor, "..."]) – The round number.

  • y (Int[Tensor, "... 1"]) – The target value.

  • guess_mask (Bool[Tensor, "..."]) – A mask indicating whether the agent is allowed to make a guess.

  • decision (Int[Tensor, "..."]) – The decision output of the agent.

  • follow_force_guess (bool, default=True) – Whether to follow the force_guess parameter, which forces the agent to make a certain decision.

abstract can_agent_be_active(agent_name: str, round_id: int, channel_name: str) bool[source]#

Specify whether an agent can be active in a given round and channel.

For non-deterministic protocols, this is true if the agent has some probability of being active.

Returns:

can_be_active (bool) – Whether the agent can be active in the given round and channel.

can_agent_be_active_any_channel(agent_name: str, round_id: int) bool[source]#

Specify whether an agent can be active in any channel in a given round.

For non-deterministic protocols, this is true if the agent has some probability of being active.

Returns:

can_be_active (bool) – Whether the agent can be active in the given round.

can_agent_see_channel(agent_name: str, channel_name: str) bool[source]#

Determine whether an agent can see a channel.

Returns:

can_see_channel (bool) – Whether the agent can see the channel.

abstract get_active_agents_mask_from_rounds_and_seed(round_id: Int[Tensor, '...'], seed: Int[Tensor, '...']) Bool[Tensor, '... agent channel'][source]#

Get a boolean mask of active agents for a batch of rounds.

Given a batch or rounds, returns a boolean mask indicating which agents are sending messages in each round and channel.

Parameters:
  • round_id (Int[Tensor, "..."]) – The round of the protocol.

  • seed (Int[Tensor, "..."]) – The per-environment seed.

Returns:

active_agents (Bool[Tensor, “… agent channel”]) – The boolean mask. active_agents[*batch, agent, channel] is True if the agent sends a message in the channel in round round[*batch].

get_agent_visible_channels(agent_name: str) list[str][source]#

Get the names of the channels visible to an agent.

Parameters:

agent_name (str) – The name of the agent.

Returns:

visible_channels (list[str]) – The names of the channels visible to the agent.

abstract get_verifier_guess_mask_from_rounds_and_seed(round_id: Int[Tensor, '...'], seed: Int[Tensor, '...']) Bool[Tensor, '...'][source]#

Get a boolean mask indicating when the verifiers can make a guess/decision.

Takes as input a tensor of rounds and returns a boolean mask indicating when the verifiers can make a guess for each element in the batch.

Parameters:
  • round_id (Int[Tensor, "..."]) – The batch of rounds.

  • seed (Int[Tensor, "..."]) – The per-environment seed.

Returns:

verifier_turn (Bool[Tensor, “…”]) – Which batch items the verifiers can make a guess in.

abstract reward_mid_point_estimate(agent_name: str) float[source]#

Get an estimate of the expected reward if all agents play randomly.

This is used to compute the mid-point of the reward range for the agent.

For example, if the agent gets reward -1 for a wrong guess and 1 for a correct guess, the mid-point estimate could be 0.

Parameters:

agent_name (str) – The name of the agent to get the reward mid-point for.

Returns:

reward_mid_point (float) – The expected reward for the agent if all agents play randomly.

abstract step_interaction_protocol(env_td: TensorDictBase | NestedArrayDict) tuple[Bool[Tensor, '...'], Bool[Tensor, '... agent'], Bool[Tensor, '...'], Float[Tensor, '... agent']][source]#

Take a step in the interaction protocol.

Computes the done signals and reward.

Used in the _step method of the environment.

Parameters:

env_td (TensorDictBase | NestedArrayDict) – The current observation and state. If a NestedArrayDict, it is converted to a TensorDictBase.

Returns:

  • shared_done (Bool[Tensor, “…”]) – A boolean mask indicating whether the episode is done because all relevant agents have made a decision.

  • agent_done (Bool[Tensor, “… agent”]) – A boolean mask indicating whether each agent is done, because they have made a decision. This is the same as shared_done for agents which don’t make decisions.

  • terminated (Bool[Tensor, “…”]) – A boolean mask indicating whether the episode has been terminated because the max number of rounds has been reached and the verifier has not guessed.

  • reward (Float[Tensor, “… agent”]) – The reward for the agents.