Agents

Introduction

The client assumes that the agent would be present in a separate python package. Thus, we rely on Entry Points to register the agent. The entrypoint specification can be used by a python package to register an algorithm by specifying a key value pair in setup.py/pyproject.toml for the package where the key would be the name of the algorithm and the value is the path to the algorithm class relative to the package root. Refer to the registration section for an example.

Functions Used By the Protocol

The protocols assumes that the algorithm have the following functions during evaluation

1. FeatureExtraction: Primarily used for extracting features from the input modality so that it can be used by other stages of the algorithm.

2. WorldDetection: Encapsulates the functionality that is used by detecting change in distribution during an experiment. The logic in this function is used to determine the introduction of novelty in an experiment.

3. NoveltyClassification: Used for providing the probability of a sample being novel or known. The probability of known class is subdivided into a distribution over all the known classes. This results in a prediction of k+1 values for every sample, where k is the number of known classes and 1 value is reserved predicting that the sample is novel.

4. NoveltyCharacterization: Similar to NoveltyClassification function with one major difference in the output. The probability of unknown class is subdivided into a distribution over the unknown classes. This results in a prediction of k+d+1 values for every sample, where k is the number of known classes, d is the number of discovered classes and 1 value is reserved for the probability that the sample is novel.

5. NoveltyAdaption: Used by the algorithm to update its internal state after it had provided results for WorldDetection and NoveltyClassification/ NoveltyCharacterization depending on the protocol.

Adapters

Rather than enforcing a strict API for the functions required by the algorithms, sail on client provides adapters to fulfill input-output requirements of an algorithm. The adapters are provided a dictionary with the parameters obtained from the harness.

Sample Adapter

    def __init__(self, toolset: Dict) -> None:
        """
        Detector constructor.

        Args:
            toolset (dict): Dictionary containing parameters for the constructor
        """
        MockONDAgent.__init__(self)
        Checkpointer.__init__(self, toolset)
        self.detector = MockONDAgentWithAttributes()

    def get_config(self) -> Dict:
        """
        Get config for the plugin.

        Returns:
            Parameters for the agent
        """
        config = super().get_config()
        config.update(self.toolset)
        return config

    def execute(self, toolset: Dict, step_descriptor: str) -> Any:
        """
        Execute method used by the protocol to run different steps.

        Args:
            toolset (dict): Dictionary containing parameters for different steps
            step_descriptor (str): Name of the step
        """
        log.info(f"Executing {step_descriptor}")
        return self.detector.step_dict[step_descriptor](toolset)

    def __eq__(self, other: object) -> bool:
        """
        Overriden method to compare two mock adapters.

Sample Detector and Registration

Sample Detector

    def __init__(self) -> None:
        """Construct Mock OND Detector."""
        super().__init__()
        self.step_dict: Dict[str, Callable] = {
            "Initialize": self.initialize,
            "FeatureExtraction": self.feature_extraction,
            "WorldDetection": self.world_detection,
            "NoveltyClassification": self.novelty_classification,
            "NoveltyAdaption": self.novelty_adaptation,
            "NoveltyCharacterization": self.novelty_characterization,
        }

    def initialize(self, toolset: Dict) -> None:
        """
        Algorithm Initialization.

        Args:
            toolset (dict): Dictionary containing parameters for different steps

        Return:
            None
        """
        pass

    def feature_extraction(
        self, toolset: Dict
    ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
        """
        Feature extraction step for the algorithm.

        Args:
            toolset (dict): Dictionary containing parameters for different steps

        Return:
            Tuple of dictionary
        """
        self.dataset = toolset["dataset"]
        return {}, {}

    def world_detection(self, toolset: Dict) -> str:
        """
        Detect change in world ( Novelty has been introduced ).

        Args:
            toolset (dict): Dictionary containing parameters for different steps

        Return:
            path to csv file containing the results for change in world
        """
        dataset_dir = os.path.dirname(self.dataset)
        dst_file = os.path.join(dataset_dir, "wc.csv")
        shutil.copyfile(self.dataset, dst_file)
        return dst_file

    def novelty_classification(self, toolset: Dict) -> str:
        """
        Classify data provided in known classes and unknown class.

        Args:
            toolset (dict): Dictionary containing parameters for different steps

        Return:
            path to csv file containing the results for novelty classification step
        """
        dataset_dir = os.path.dirname(self.dataset)
        dst_file = os.path.join(dataset_dir, "ncl.csv")
        shutil.copyfile(self.dataset, dst_file)
        return dst_file

    def novelty_adaptation(self, toolset: Dict) -> None:
        """
        Update models based on novelty classification and characterization.

        Args:
            toolset (dict): Dictionary containing parameters for different steps

        Return:
            None
        """
        pass

    def novelty_characterization(self, toolset: Dict) -> str:
        """
        Characterize novelty by clustering different novel samples.

        Args:
            toolset (dict): Dictionary containing parameters for different steps

        Return:
            path to csv file containing the results for novelty characterization step
        """
        dataset_dir = os.path.dirname(self.dataset)
        dst_file = os.path.join(dataset_dir, "nc.csv")
        shutil.copyfile(self.dataset, dst_file)
        return dst_file

    def execute(self, toolset: Dict, step_descriptor: str) -> Any:
        """
        Execute method used by the protocol to run different steps.

        Args:
            toolset (dict): Dictionary containing parameters for different steps
            step_descriptor (str): Name of the step
        """
        log.info(f"Executing {step_descriptor}")
        return self.step_dict[step_descriptor](toolset)


class MockONDAgentWithAttributes(MockONDAgent):
    """Mock Detector for testing checkpointing."""

Registering the Detector

Note

The registration is highlighted below.

types-six = "^0.1.7"
types-toml = "^0.1.3"


[tool.poetry.plugins.smqtk_plugins]
ONDProtocol = "sail_on_client.protocol.ond_protocol"
CONDDAProtocol = "sail_on_client.protocol.condda_protocol"
LocalHarness = "sail_on_client.harness.local_harness"
ParHarness = "sail_on_client.harness.par_harness"
MockONDAgent = "sail_on_client.agent.mock_ond_agents"
MockCONDDAAgent = "sail_on_client.agent.mock_condda_agents"
PreComputedAgent = "sail_on_client.agent.pre_computed_detector"
PreComputedReactionAgent = "sail_on_client.agent.pre_computed_reaction_agent"