Agents

Phantom has two base agent classes, Agent and StrategicAgent. All user implemented agent classes should be derived from one of these two classes.

Depending on which methods are implemented / properties set, agents can have differing levels of functionality:

The most basic type of agent is one that only responds to messages sent from other agents. This agent does not take actions by itself and as such does not have a defined Policy or reward function.

If the generate_messages() method is also implemented the agent can now also pro-actively create messages at the start of each step.

Finally there is the StrategicAgent that has a Policy, reward function, observation encoder and action decoder. The generate_messages() should not be implemented here as the action decoder provides the message generation functionality.

Each of the reward function, observation encoder and action decoder components can be provided in two ways:

By setting an Encoder/Decoder/RewardFunction object to this class’s
observation_encoder / action_decoder / reward_function properties.
By subclassing this class and providing a custom implementation to this class’s
encode_observation() / decode_action() / compute_reward() methods.

These two options can be mixed between the three elements.

If the encode_observation() method is provided with a custom implementation, the observation_space property of the agent must be set. Otherwise the Encoder will provide the observation space.

If the decode_action() method is provided with a custom implementation, the action_space property of the agent must be set. Otherwise the Decoder will provide the action space.

Agent

class phantom.agents.Agent(agent_id, supertype=None)[source]

Representation of an agent in the network.

Instances of phantom.Agent occupy the nodes on the network graph. They are resonsible for storing and monitoring internal state, constructing View instances and handling messages.

Parameters:

agent_id (Hashable) – Unique identifier for the agent.
supertype (Optional[Supertype]) – Optional Supertype instance. When the agent’s reset function is called the supertype will be sampled from and the values set as the agent’s type property.

Implementations can make use of the msg_handler function decorator:

class SomeAgent(ph.Agent):
    ...

    @ph.agents.msg_handler(RequestMessage)
    def handle_request_msg(self, ctx: ph.Context, message: ph.Message):
        response_msgs = do_something_with_msg(message)

        return [response_msgs]

handle_batch(ctx, batch)[source]

Handle a batch of messages from multiple potential senders.

Parameters:

ctx (Context) – A Context object representing agent’s the local view of the environment.
batch (Sequence[Message]) – The incoming batch of messages to handle.

Return type:

List[Tuple[Hashable, MsgPayload]]

Returns:

A list of receiver ID / message payload pairs to form into messages in response to further resolve.

handle_message(ctx, message)[source]

Handle a messages sent from other agents. The default implementation is the use the msg_handler function decorators.

Parameters:

ctx (Context) – A Context object representing agent’s the local view of the environment.
message (Message) – The contents of the message.

Return type:

List[Tuple[Hashable, MsgPayload]]

Returns:

A list of receiver ID / message payload pairs to form into messages in response to further resolve.

property id: Hashable: The unique ID of the agent.

post_message_resolution(ctx)[source]

Perform internal, post-message resolution updates to the agent.

Return type:: None

pre_message_resolution(ctx)[source]

Perform internal, pre-message resolution updates to the agent.

Return type:: None

reset()[source]

Resets the Agent.

Can be extended by subclasses to provide additional functionality.

Return type:: None

view(neighbour_id=None)[source]

Return an immutable view to the agent’s public state.

Return type:: Optional[AgentView]

StrategicAgent

class phantom.agents.StrategicAgent(agent_id, observation_encoder=None, action_decoder=None, reward_function=None, supertype=None)[source]

Representation of a behavioural agent in the network.

Instances of phantom.Agent occupy the nodes on the network graph. They are resonsible for storing and monitoring internal state, constructing View instances and handling messages.

Parameters:

agent_id (Hashable) – Unique identifier for the agent.
observation_encoder (Optional[Encoder]) – Optional Encoder instance, otherwise define an encode_observation() method on the Agent sub-class.
action_decoder (Optional[Decoder]) – Optional Decoder instance, otherwise define an decode_action() method on the Agent sub-class.
reward_function (Optional[RewardFunction]) – Optional RewardFunction instance, otherwise define an compute_reward() method on the Agent sub-class.
supertype (Optional[Supertype]) – Optional Supertype instance. When the agent’s reset function is called the supertype will be sampled from and the values set as the agent’s type property.

collect_infos(ctx)[source]

Provides diagnostic information about the agent, usefult for debugging.

Note

This method may be extended by sub-classes to provide additional functionality.

Parameters:: ctx (Context) – A Context object representing the agent’s local view of the environment.
Return type:: Dict[str, Any]
Returns:: A dictionary containing informations about the agent

compute_reward(ctx)[source]

Computes a reward value based on an agents current state.

Note

This method may be extended by sub-classes to provide additional functionality.

Parameters:: ctx (Context) – A Context object representing the agent’s local view of the environment.
Return type:: float
Returns:: A float representing the present reward value.

decode_action(ctx, action)[source]

Decodes an action taken by the agent policy into a set of messages to be sent to other agents in the network.

Note

This method may be extended by sub-classes to provide additional functionality.

Parameters:

ctx (Context) – A Context object representing the agent’s local view of the environment.
action (TypeVar(Action)) – The action taken by the agent.

Return type:

Optional[List[Tuple[Hashable, MsgPayload]]]

Returns:

A list of receiver ID / message payload pairs to form into messages in response to further resolve.

encode_observation(ctx)[source]

Encodes a local view of the environment state into a set of observations.

Note

This method may be extended by sub-classes to provide additional functionality.

Parameters:: ctx (Context) – A Context object representing agent’s the local view of the environment.
Return type:: TypeVar(Observation)
Returns:: A numpy array encoding the observations.

handle_batch(ctx, batch)

Handle a batch of messages from multiple potential senders.

Parameters:

ctx (Context) – A Context object representing agent’s the local view of the environment.
batch (Sequence[Message]) – The incoming batch of messages to handle.

Return type:

List[Tuple[Hashable, MsgPayload]]

Returns:

A list of receiver ID / message payload pairs to form into messages in response to further resolve.

handle_message(ctx, message)

Handle a messages sent from other agents. The default implementation is the use the msg_handler function decorators.

Parameters:

ctx (Context) – A Context object representing agent’s the local view of the environment.
message (Message) – The contents of the message.

Return type:

List[Tuple[Hashable, MsgPayload]]

Returns:

A list of receiver ID / message payload pairs to form into messages in response to further resolve.

property id: Hashable: The unique ID of the agent.

is_terminated(ctx)[source]

Indicates whether ‘a terminal state (as defined under the MDP of the task) is reached’ for the agent. The default logic is for the agent to be done only once all the timesteps have been executed.

Note

This method may be extended by sub-classes to provide additional functionality.

Parameters:: ctx (Context) – A Context object representing the agent’s local view of the environment.
Return type:: bool
Returns:: A boolean representing the terminal status of the agent.

is_truncated(ctx)[source]

Indicates whether ‘a truncation condition outside the scope of the MDP is satisfied’ for the agent.

Note

This method may be extended by sub-classes to provide additional functionality.

Parameters:: ctx (Context) – A Context object representing the agent’s local view of the environment.
Return type:: bool
Returns:: A boolean representing the truncated status of the agent.

post_message_resolution(ctx)

Perform internal, post-message resolution updates to the agent.

Return type:: None

pre_message_resolution(ctx)

Perform internal, pre-message resolution updates to the agent.

Return type:: None

reset()

Resets the Agent.

Can be extended by subclasses to provide additional functionality.

Return type:: None

view(neighbour_id=None)

Return an immutable view to the agent’s public state.

Return type:: Optional[AgentView]