Methods/moeh

iohblade/fitness.py

@@ -45,32 +45,32 @@ def _dominates(self, other: "Fitness") -> tuple[bool, bool]:
         strictly_better = any(self[k] < other[k] for k in self.keys())
         return better_or_equal, strictly_better
 
-    def __eq__(self, other: object) -> bool:
+    def __eq__(self, other) -> bool:
         if not isinstance(other, Fitness):
             return False
         return self._fitness == other._fitness
 
-    def __lt__(self, other: "Fitness") -> bool:
+    def __lt__(self, other) -> bool:
         if not isinstance(other, Fitness):
-            return NotImplemented
+            return False
         be, sb = self._dominates(other)
         return be and sb
 
-    def __gt__(self, other: "Fitness") -> bool:
+    def __gt__(self, other) -> bool:
         if not isinstance(other, Fitness):
-            return NotImplemented
+            return False
         be, sb = other._dominates(self)
         return be and sb
 
-    def __le__(self, other: "Fitness") -> bool:
+    def __le__(self, other) -> bool:
         if not isinstance(other, Fitness):
-            return NotImplemented
+            return False
         be, sb = self._dominates(other)
         return be or sb
 
-    def __ge__(self, other: "Fitness") -> bool:
+    def __ge__(self, other) -> bool:
         if not isinstance(other, Fitness):
-            return NotImplemented
+            return False
         be, sb = other._dominates(self)
         return be or sb
 

iohblade/methods/__init__.py

@@ -3,6 +3,8 @@
 from .random_search import RandomSearch
 from .lhns import LHNS_Method
 from .mcts_ahd import MCTS_Method
+from .moeh_method.mutationtype import MutationType
+from .moeh_method.moeh import MoEH_Method
 
 try:
     from .eoh import EoH
@@ -20,4 +22,6 @@
     "ReEvo",
     "LHNS_Method",
     "MCTS_Method",
+    "MoEH_Method",
+    "MutationType",
 ]

iohblade/methods/mcts_ahd/prompts.py

@@ -119,7 +119,7 @@ def get_prompt_m1(
 "
             + indiv.code
             + "\n\
-Please create a new algorithm that has a different form but can be a modified version of the provided algorithm. Attempt to introduce more novel mechanisms and new equations or programme segments.\n"
+Please create a new algorithm that is a modified version of the provided algorithm. Attempt to introduce more novel mechanisms and new equations or programme segments.\n"
             "Respond in adherance to following contract:"
         )
         prompt_content += "\n".join([task_prompt, example_prompt, format_prompt])

iohblade/methods/moeh_method/__init__.py

@@ -0,0 +1,4 @@
+from .mutationtype import MutationType
+from .moeh import MoEH_Method
+
+__all__ = ["MutationType", "MoEH_Method"]

iohblade/methods/moeh_method/moeh.py

@@ -0,0 +1,320 @@
+import random
+import inspect
+from typing import Any
+
+from iohblade.llm import LLM
+from iohblade.method import Method
+from iohblade.problem import Problem
+from iohblade.solution import Solution
+
+from iohblade.methods.moeh_method.prompts import MoEH_Prompts
+from iohblade.methods.moeh_method.population import Population
+from iohblade.methods.moeh_method.mutationtype import MutationType
+
+
+class MoEH:
+    def __init__(
+        self,
+        llm: LLM,
+        problem: Problem,
+        iterations: int,
+        max_sample_nums: int,
+        population_size: int,
+        minimisation: bool,
+        use_e2_operator: bool = True,
+        use_m1_operator: bool = True,
+        use_m2_operator: bool = True,
+        max_workers: int = 2,
+    ) -> None:
+        """
+        A `Multi-Objective Evolution of Heuristic Using LLLM` implementation for BLADE, as defined in paper: https://arxiv.org/pdf/2409.16867
+
+        ## Params:
+        - `llm: LLM`: Object of one of the final class in `iohblade.llm`.
+        - `problem: Problem`: Object instance of a class that inherits from `iohblade.problem`.
+        - `iterations: int` Max number of iterations for which the algorithm is to be run.
+        - `max_sample_nums: int`: Max number sample that can be generated by the LLM.
+        - `population_size: int`: Population size (λ / µ) for the evolutionary algorithm.
+        - `minimisation: int`: Optimisation direction for the evolutionary algorithm.
+        - `use_e2_operator: bool(True)`: Use e2 operation as stated in the `moeh.prompts`; multi-individuals -> 1 individual mutation.
+        - `use_m1_operator: bool(True)`: Use m1 operation as stated in the `moeh.prompts`; 1 -> 1 mutation (adds novel methods.)
+        - `use_m2_operator: bool(True)`: Use m1 operation as stated in the `moeh.prompts`; 1 -> 1 mutation (used different approach.)
+        - `max_workers: int(2)`: Declaration of max `loky` workers for `joblib.Parallel`.
+
+        ## Note:
+
+        - The algorithm quits running when either of the conditons are met (||):
+            - `iterations` count of generations are iterated over.
+            - `max_sample_nums` count of samples are generated.
+        """
+
+        assert max_sample_nums > 1, "Max sample number must be positive integer."
+        assert (
+            type(max_sample_nums) == int
+        ), "Max sample number must be positive integer."
+        assert population_size >= 1, "Population size must be positive integer."
+        assert type(population_size) == int, "Population size must be positive integer."
+        assert iterations >= 1, "Iteration count must be positive integer."
+        assert type(iterations) == int, "Iteration count must be positive integer."
+        assert max_workers >= 1, "Max Workers must be positive integer."
+        assert type(max_workers) == int, "Max Workers must be positive integer."
+
+        # Accepted parameters.
+        self.llm = llm
+        self.problem = problem
+        self.iterations = iterations
+        self.max_sample_nums = max_sample_nums
+
+        self.max_sample_nums = max_sample_nums
+        self._remianing_budget = max_sample_nums
+        self.population_size = population_size
+        self.iterations = iterations
+
+        self.minimisation = minimisation
+        config = self.problem.get_config()
+        assert (
+            self.minimisation == config["minimisation"]
+        ), "Problem.minimisation and minimisation parameter passed must match."
+
+        # Runtime parameters:
+        self._current_iteration = 0
+        self.population = Population(population_size, self.minimisation)
+
+        # Mutation Types
+        self.allowed_mutation_types: list[MutationType] = []
+        if use_e2_operator:
+            self.allowed_mutation_types.append(MutationType.E2)
+        if use_m1_operator:
+            self.allowed_mutation_types.append(MutationType.M1)
+        if use_m2_operator:
+            self.allowed_mutation_types.append(MutationType.M2)
+
+        assert (
+            len(self.allowed_mutation_types) >= 1
+        ), "Atleast one mutation type is expected, use_x_operator all set false in invalid state."
+
+    def initialise(self):
+        """
+        Initialises the evolutionary algorithm.
+
+        # Params:
+            - `None`
+
+        # Returns:
+            - Inline function, doesn't return anything, updates `self.population`.
+        """
+        prompt = MoEH_Prompts.get_prompt_i1(
+            self.problem.task_prompt,
+            self.problem.example_prompt,
+            self.problem.format_prompt,
+        )
+
+        # TODO: Parallel this.....
+        while (
+            len(self.population.population) < self.population_size
+            and self._remianing_budget > 0
+        ):
+            individual = self._sample_solution(prompt, [])
+            individual = self.get_fitness(individual)
+            self.population.append(individual)
+            self._remianing_budget -= 1
+
+    def get_fitness(self, individual: Solution):
+        individual = self.problem(individual)
+        return individual
+
+    def _sample_solution(self, prompt: str, parents: list[Solution]) -> Solution:
+        message = [{"role": "user", "content": prompt}]
+        parent_ids = list(map(lambda x: x.id, parents))
+        solution = self.llm.sample_solution(message, parent_ids)
+        return solution
+
+    def query_individual(self, type: MutationType, pop: list[Solution]) -> Solution:
+        """
+        Queries and returns a new LLM written solution using one of the MutationTypes.
+
+        ## Parameters:
+        - `type: MutationType`: Expects an enum typed `MutationType` to decide which prompt to use for an evolution.
+        - `pop: list[Solution]`: Expects an array of solution of size 1 for `M1` and `M2` types, and of arbitrary size [1, n] for `E2`.
+
+        ## Expectations:
+        - For each `MutationType` following size of population is expected:
+            - `M1`: 1
+            - `M2`: 1
+            - `E2`: [1, n]
+
+        ## Returns:
+        - `Solution`: an evolved solution generated by LLM using given pop and `MutationType`.
+        """
+        match (type):
+            case MutationType.M1:
+                prompt = MoEH_Prompts.get_prompt_m1(
+                    self.problem.example_prompt, self.problem.format_prompt, pop[0]
+                )
+                solution = self._sample_solution(prompt, [pop[0]])
+                return solution
+            case MutationType.M2:
+                prompt = MoEH_Prompts.get_prompt_m2(
+                    self.problem.example_prompt, self.problem.format_prompt, pop[0]
+                )
+                solution = self._sample_solution(prompt, [pop[0]])
+                return solution
+
+            case MutationType.E2:
+                prompt = MoEH_Prompts.get_prompt_e2(
+                    self.problem.example_prompt, self.problem.format_prompt, pop
+                )
+                solution = self._sample_solution(prompt, pop)
+                return solution
+
+    def evolve_solution(self, population_index: int, pop: list[Solution]) -> Solution:
+        mutation_type = random.choice(self.allowed_mutation_types)
+        match (mutation_type):
+            case MutationType.E2:
+                solution = self.query_individual(mutation_type, pop)
+            case _:
+                solution = self.query_individual(mutation_type, [pop[population_index]])
+        solution = self.get_fitness(solution)
+        self.max_sample_nums -= 1
+        return solution
+
+    def run(self) -> list[Solution]:
+        new_population: list[Solution] = []
+
+        self.initialise()
+        print(f"Started MoEH with {len(self.population._population)} individuals.")
+
+        while (self._current_iteration < self.iterations) and (
+            self.max_sample_nums > 0
+        ):
+            print(
+                f"Generation {self._current_iteration + 1} / {self.iterations}: {len(self.population.population)} individuals. Best solution = {[individual.fitness for individual in self.population.get_best()]} Attempts remaining {self.max_sample_nums}."
+            )
+            new_population = []
+            old_population = self.population.parent_selection(self.population_size)
+            for index in range(len(old_population)):
+                new_individual = self.evolve_solution(index, old_population)
+                new_population.append(new_individual)
+
+            for individual in new_population:
+                self.population.append(individual)
+
+            _ = self.population.population_management()
+            self._current_iteration += 1
+        return self.population.get_best()
+
+
+class MoEH_Method(Method):
+    def __init__(
+        self,
+        llm: LLM,
+        budget: int,
+        iterations: int,
+        population_size: int,
+        minimisation: bool,
+        use_e2_operator: bool = True,
+        use_m1_operator: bool = True,
+        use_m2_operator: bool = True,
+        name="MoEH",
+    ):
+        """A `Multi-Objective Evolution of Heuristic Using LLLM` implementation for BLADE, as defined in paper: https://arxiv.org/pdf/2409.16867
+
+        ## Params:
+        - `llm: LLM`: Object of one of the final class in `iohblade.llm`.
+        - `problem: Problem`: Object instance of a class that inherits from `iohblade.problem`.
+        - `budget: int`: Max number sample that can be generated by the LLM.
+        - `iterations: int` Max number of iterations for which the algorithm is to be run.
+        - `population_size: int`: Population size (λ / µ) for the evolutionary algorithm.
+        - `use_e2_operator: bool(True)`: Use e2 operation as stated in the `moeh.prompts`; multi-individuals -> 1 individual mutation.
+        - `use_m1_operator: bool(True)`: Use m1 operation as stated in the `moeh.prompts`; 1 -> 1 mutation (adds novel methods.)
+        - `use_m2_operator: bool(True)`: Use m1 operation as stated in the `moeh.prompts`; 1 -> 1 mutation (used different approach.)
+        - `max_workers: int(2)`: Declaration of max `loky` workers for `joblib.Parallel`.
+
+        ## Note:
+
+        - The algorithm quits running when either of the conditons are met (||):
+            - `iterations` count of generations are iterated over.
+            - `max_sample_nums` count of samples are generated.
+        """
+        super().__init__(llm, budget, name=name)
+        self.iterations = iterations
+        self.population_size = population_size
+        self.use_e2_operator = use_e2_operator
+        self.use_m1_operator = use_m1_operator
+        self.use_m2_operator = use_m2_operator
+        self.minimisation = minimisation
+        sig = inspect.signature(self.__init__)
+        self.init_params = {
+            k: getattr(self, k)
+            for k in sig.parameters
+            if k not in ("self", "name", "budget", "llm")
+        }
+
+    def __call__(self, problem: Problem):
+        self.moeh_instance = MoEH(
+            self.llm,
+            problem,
+            self.iterations,
+            self.budget,
+            self.population_size,
+            self.minimisation,
+            self.use_e2_operator,
+            self.use_m1_operator,
+            self.use_m2_operator,
+            1,
+        )
+        try:
+            return self.moeh_instance.run()
+        except Exception as e:
+            import traceback
+
+            print("Some error occured:")
+            print(traceback.print_exc(), e, "-----------", sep="\n")
+            return self.moeh_instance.population.get_best()
+
+    def to_dict(self):
+        """
+        Returns a dictionary representation of the method including all parameters.
+
+        Returns:
+            dict: Dictionary representation of the method.
+        """
+        kwargs = dict(self.init_params)
+        return {
+            "method_name": self.name if self.name != None else "MoEH",
+            "budget": self.budget,
+            "kwargs": kwargs,
+        }
+
+    # region Database Logger:
+    def get_config(self) -> dict[str, Any]:
+        config = {
+            "iterations": self.iterations,
+            "max_sample_nums": self.budget,
+            "population_size": self.population_size,
+            "use_e2_operator": self.use_e2_operator,
+            "use_m1_operator": self.use_m1_operator,
+            "use_m2_operator": self.use_m2_operator,
+            "minimisation": self.minimisation,
+        }
+
+        return {
+            "name": self.name,
+            "source": "https://github.com/XAI-liacs/BLADE/tree/main/iohblade/methods/moeh",
+            "config": config,
+        }
+
+
+# endregion
+
+if __name__ == "__main__":
+    from iohblade.benchmarks.packing.circles import CirclePacking
+    from iohblade.llm import Dummy_LLM
+
+    problem = CirclePacking()
+    llm = Dummy_LLM()
+    moeh = MoEH_Method(llm, 10, 10, 2, False)
+
+    for key, value in moeh.get_config().items():
+        print(f"==========================={key}===========================")
+        print(value)