Components

Components are the building blocks of the agent. They determine the agent’s behavior and can be composed to create complex agents.

Agent

If we look at the implementation of the agent from a high level, we can see that it is composed of several components, each with a specific role in the decision-making process. Notice that the agent holds an internal information database, comprised of a knowledge database and a situational awareness vector, as well as a potentially arbitrary number of components.

Generic component

Each colorful rectangle you see in the figure, with the exception of the agent’s state and the Entity - Model, is a subclass of the Component class. The base class provides a common interface for all components, allowing them to be easily composed and extended.

classDiagram class Component { +Identifier agent_id +bool is_initialised +initialise_component(Agent Agent, AwarenessVector initial_awd, KnowledgeDatabase initial_kwd) -_compute(...): Any$ -_update(...)$ +compute(...): Any +update(...) +compute_and_update(...): Any } class RiskComponent class UncertaintyEstimator class Controller class PerceptionSystem class CommunicationSender class CommunicationReceiver Component <|-- RiskComponent Component <|-- UncertaintyEstimator Component <|-- Controller Component <|-- PerceptionSystem Component <|-- CommunicationSender Component <|-- CommunicationReceiver

Compute and Update

Every specialized component must implement the abstract methods _compute and _update. No other alteration is needed to the base class, although it is always possible to introduce new utility methods or attributes to the specialized components, as well as to override the base class methods if needed. It is advisable the base class methods if you override them, to ensure the correct behavior of the component.

The compute method is responsible for processing the information available to the agent and producing a new value. While in principle it could accept any number of arguments, it is recommended to keep the signature as simple as possible, since most information can be collected from the agent’s internal state directly.

The update method is responsible for updating the agent’s internal state based on the computed value. Therefore it must accept the value computed by compute as an argument. How the value is used to update the agent’s internal state is up to the component’s implementation.

# In this example, CounterComponent just increments a counter
# in the agent's knowledge_database state at each step
from symaware.base.component import Component

class CounterComponent(Component):
    def _compute(self) -> int:
        return self.agent.self_knowledge['counter'] + 1

    def _update(self, value: int):
        self.agent.self_knowledge['counter'] = value

Component loop in synchronous mode

If the simulation is running the synchronous mode, at each step all components are called in sequence. More specifically, the compute_and_update method is called on each component. As the name suggests, this method runs the compute method first and then the update method. Both methods internally call the _compute and _update abstract methods, respectively.

sequenceDiagram autonumber loop For each component Agent ->>+ Component: compute_and_update Component ->> Component: compute -> Result Component ->> Component: update(Result) Component ->>- Agent: #10003; end

Component loop in asynchronous mode

If the simulation is running the asynchronous mode, each component starts a new awaitable routine that runs the async_compute_and_update method in a loop, with a lock between each iteration determining the frequency at which the component will run. The lock is an instance of the AsyncLoopLock class. Apart from this, the implementation can be the same as the one for the synchronous mode since, by default, the async methods will still end up calling the _compute and _update abstract methods. This said, there is the option to override the _async_compute and _async_update methods, which are the asynchronous counterparts of _compute and _update, if an asynchronous implementation of such operations is needed.

sequenceDiagram autonumber loop Asyncronously, for each component Agent ->>+ Component: await async_compute_and_update() activate Component Component ->> Component: await async_compute -> Result Component ->> Component: await async_update(Result) Component ->>- Agent: #10003; Agent ->>+ Component: await next_loop() Component ->>- Agent: #10003; end

Perception System

The Perception System can acts as a receiver for sensors or as an omniscient oracle, knowing the state of all other entities in the environment and collecting information that is then integrated with message the agent could receive from other agents, maybe running on different machines, through the Communication components.

Risk and Uncertainty Estimators

The knowledge database and situational awareness vector can be used by the risk and uncertainty estimators to compute new value for the risk and uncertainty, as the name suggests.

Controller

Usually the last step involves feeding the update agent’s state to the controller, which has the job to produce the control input to apply to the model, as well as an intent that will be stored in the situational awareness, and could play a role in successive steps. The control input applied to the model will then be played out in the environment, and the new state will be available to the Perception System for the next iteration, closing the loop.

Note

The controller is usually the most important component to implement, as it is the one that, in the end, will determine the agent’s behavior in the environment. The type of control input the controller will return at the end of the _compute method depends on the agent’s model, which in turn depends on the environment and the simulator used.

from symaware.base import Controller, TimeSeries
import numpy as np

class MyController(Controller):
    def _compute(self) -> tuple[np.ndarray, TimeSeries]:
        # Your implementation here
        # Example:
        # Get the state of the agent
        state = self._agent.self_state
        # Get the goal position from the knowledge database
        goal_pos = self._agent.self_knowledge["goal_pos"]
        # Compute the control input as the difference between the goal position and the current state
        control_input = goal_pos - state
        # Return the control input and an empty TimeSeries
        return control_input, TimeSeries()

    def _update(self, value: tuple[np.ndarray, TimeSeries]):
        # Your implementation here
        # Example:
        # Apply the control input to the agent's model
        control_input, intent = value
        self._agent.model.control_input = control_input
        self._agent.self_awareness.intent = intent

Communication

Optionally, the agent can use the communication components to share some information with other agents, helping them in their decisional process. The communication components can be used to send and receive messages, and can be used to implement a variety of communication protocols, such as gossip, flooding, or any other custom protocol, including internet-based TCP or UDP communication.