Add surrogate optimizer (#82)

Surrogate optimizer, evaluator and model were added.
aimclub · Jun 14, 2023 · ce44e5a · ce44e5a
1 parent d4a655e
commit ce44e5a
Show file tree

Hide file tree

Showing 10 changed files with 254,077 additions and 34 deletions.
diff --git a/examples/surrogate/results/hist_2ring_n30_trial0.json b/examples/surrogate/results/hist_2ring_n30_trial0.json
diff --git a/examples/surrogate/results/hist_gnp_n30_trial0.json b/examples/surrogate/results/hist_gnp_n30_trial0.json
diff --git a/examples/surrogate/surrogate_example.py b/examples/surrogate/surrogate_example.py
@@ -0,0 +1,85 @@
+from datetime import timedelta
+from functools import partial
+from typing import Type, Optional, Sequence
+
+import networkx as nx
+
+from examples.synthetic_graph_evolution.experiment_setup import run_experiments
+from examples.synthetic_graph_evolution.generators import generate_labeled_graph
+from golem.core.adapter.nx_adapter import BaseNetworkxAdapter
+from golem.core.dag.verification_rules import has_no_self_cycled_nodes
+from golem.core.optimisers.genetic.gp_params import GPAlgorithmParameters
+from golem.core.optimisers.genetic.operators.base_mutations import MutationTypesEnum
+from golem.core.optimisers.genetic.operators.inheritance import GeneticSchemeTypesEnum
+from golem.core.optimisers.meta.surrogate_model import SurrogateModel, RandomValuesSurrogateModel
+from golem.core.optimisers.meta.surrogate_optimizer import SurrogateEachNgenOptimizer
+from golem.core.optimisers.objective import Objective
+from golem.core.optimisers.optimization_parameters import GraphRequirements
+from golem.core.optimisers.optimizer import GraphGenerationParams, GraphOptimizer
+from golem.metrics.graph_metrics import spectral_dist
+
+
+def surrogate_graph_search_setup(target_graph: nx.DiGraph,
+                                 optimizer_cls: Type[GraphOptimizer] = SurrogateEachNgenOptimizer,
+                                 surrogate_model: Type[SurrogateModel] = RandomValuesSurrogateModel(),
+                                 node_types: Sequence[str] = ('X',),
+                                 timeout: Optional[timedelta] = None,
+                                 num_iterations: Optional[int] = None):
+    # Setup parameters
+    num_nodes = target_graph.number_of_nodes()
+    requirements = GraphRequirements(
+        max_arity=num_nodes,
+        max_depth=num_nodes,
+        early_stopping_timeout=5,
+        early_stopping_iterations=1000,
+        keep_n_best=4,
+        timeout=timeout,
+        num_of_generations=num_iterations,
+        n_jobs=-1,
+        history_dir=None,
+    )
+    gp_params = GPAlgorithmParameters(
+        multi_objective=True,
+        genetic_scheme_type=GeneticSchemeTypesEnum.parameter_free,
+        mutation_types=[
+            MutationTypesEnum.single_add,
+            MutationTypesEnum.single_drop,
+            MutationTypesEnum.single_edge,
+        ]
+    )
+    graph_gen_params = GraphGenerationParams(
+        adapter=BaseNetworkxAdapter(),
+        rules_for_constraint=[has_no_self_cycled_nodes],
+        available_node_types=node_types,
+    )
+
+    # Setup objective that measures some graph-theoretic similarity measure
+    objective = Objective(
+        quality_metrics={
+            'sp_adj': partial(spectral_dist, target_graph, kind='adjacency')
+        }
+    )
+
+    # Generate simple initial population with line graphs
+    initial_graphs = [generate_labeled_graph('line', k + 3)
+                      for k in range(gp_params.pop_size)]
+    initial_graphs = graph_gen_params.adapter.adapt(initial_graphs)
+
+    # Build the optimizer
+    optimiser = optimizer_cls(objective, initial_graphs, requirements, graph_gen_params, gp_params,
+                              surrogate_model=surrogate_model)
+    return optimiser, objective
+
+
+if __name__ == '__main__':
+    results_log = run_experiments(
+        optimizer_setup=partial(surrogate_graph_search_setup,
+                                surrogate_model=RandomValuesSurrogateModel()),
+        optimizer_cls=SurrogateEachNgenOptimizer,
+        graph_names=['2ring', 'gnp'],
+        graph_sizes=[30, 100],
+        num_trials=1,
+        trial_timeout=5,
+        trial_iterations=2000,
+        visualize=True)
+    print(results_log)
diff --git a/examples/synthetic_graph_evolution/graph_search.py b/examples/synthetic_graph_evolution/graph_search.py
@@ -3,7 +3,6 @@
 from typing import Type, Optional, Sequence
 
 import networkx as nx
-
 from examples.synthetic_graph_evolution.experiment_setup import run_experiments
 from examples.synthetic_graph_evolution.generators import generate_labeled_graph
 from golem.core.adapter.nx_adapter import BaseNetworkxAdapter
@@ -68,7 +67,7 @@ def graph_search_setup(target_graph: Optional[nx.DiGraph] = None,
         mutation_types=[
             MutationTypesEnum.single_add,
             MutationTypesEnum.single_edge,
-            MutationTypesEnum.single_drop,
+            MutationTypesEnum.single_drop
         ],
         crossover_types=[CrossoverTypesEnum.none]
     )

diff --git a/golem/core/optimisers/genetic/evaluation.py b/golem/core/optimisers/genetic/evaluation.py
@@ -5,8 +5,7 @@
 from abc import ABC, abstractmethod
 from datetime import datetime
 from functools import partial
-from random import choice
-from typing import Dict, List, Optional, Sequence, Tuple, TypeVar
+from typing import List, Optional, Sequence, Tuple, TypeVar, Dict
 
 from joblib import Parallel, cpu_count, delayed
 
@@ -33,6 +32,7 @@
 
 class DelegateEvaluator:
     """Interface for delegate evaluator of graphs."""
+
     @property
     @abstractmethod
     def is_enabled(self) -> bool:
@@ -127,6 +127,7 @@ def __init__(self,
         self.logger = default_log(self)
         self._n_jobs = n_jobs
         self.evaluation_cache = None
+        self._reset_eval_cache()
 
     def dispatch(self, objective: ObjectiveFunction, timer: Optional[Timer] = None) -> EvaluationOperator:
         """Return handler to this object that hides all details
@@ -154,9 +155,12 @@ def population_evaluation_info(self, pop_size: int, evaluated_pop_size: int):
     def evaluate_population(self, individuals: PopulationT) -> Optional[PopulationT]:
         raise NotImplementedError()
 
-    def evaluate_single(self, graph: OptGraph, uid_of_individual: str, with_time_limit: bool = True, cache_key: Optional[str] = None,
+    def evaluate_single(self, graph: OptGraph, uid_of_individual: str, with_time_limit: bool = True,
+                        cache_key: Optional[str] = None,
                         logs_initializer: Optional[Tuple[int, pathlib.Path]] = None) -> OptionalEvalResult:
 
+        graph = self.evaluation_cache.get(cache_key, graph)
+
         if with_time_limit and self.timer.is_time_limit_reached():
             return None
         if logs_initializer is not None:
@@ -188,6 +192,24 @@ def _evaluate_graph(self, domain_graph: Graph) -> Tuple[Fitness, Graph]:
 
         return fitness, domain_graph
 
+    def evaluate_with_cache(self, population: PopulationT) -> Optional[PopulationT]:
+        reversed_population = list(reversed(population))
+        self._remote_compute_cache(reversed_population)
+        evaluated_population = self.evaluate_population(reversed_population)
+        self._reset_eval_cache()
+        return evaluated_population
+
+    def _reset_eval_cache(self):
+        self.evaluation_cache: Dict[str, Graph] = {}
+
+    def _remote_compute_cache(self, population: PopulationT):
+        self._reset_eval_cache()
+        if self._delegate_evaluator and self._delegate_evaluator.is_enabled:
+            self.logger.info('Remote fit used')
+            restored_graphs = self._adapter.restore(population)
+            computed_graphs = self._delegate_evaluator.compute_graphs(restored_graphs)
+            self.evaluation_cache = {ind.uid: graph for ind, graph in zip(population, computed_graphs)}
+
 
 class MultiprocessingDispatcher(BaseGraphEvaluationDispatcher):
     """Evaluates objective function on population using multiprocessing pool
@@ -210,21 +232,12 @@ def __init__(self,
 
         super().__init__(adapter, n_jobs, graph_cleanup_fn, delegate_evaluator)
 
-        self._reset_eval_cache()
-
     def dispatch(self, objective: ObjectiveFunction, timer: Optional[Timer] = None) -> EvaluationOperator:
         """Return handler to this object that hides all details
         and allows only to evaluate population with provided objective."""
         super().dispatch(objective, timer)
         return self.evaluate_with_cache
 
-    def evaluate_with_cache(self, population: PopulationT) -> Optional[PopulationT]:
-        reversed_population = list(reversed(population))
-        self._remote_compute_cache(reversed_population)
-        evaluated_population = self.evaluate_population(reversed_population)
-        self._reset_eval_cache()
-        return evaluated_population
-
     def evaluate_population(self, individuals: PopulationT) -> Optional[PopulationT]:
         individuals_to_evaluate, individuals_to_skip = self.split_individuals_to_evaluate(individuals)
         # Evaluate individuals without valid fitness in parallel.
@@ -250,25 +263,6 @@ def evaluate_population(self, individuals: PopulationT) -> Optional[PopulationT]
                             logging_level=logging.INFO)
         return successful_evals
 
-    def evaluate_single(self, graph: OptGraph, uid_of_individual: str, with_time_limit: bool = True,
-                        cache_key: Optional[str] = None,
-                        logs_initializer: Optional[Tuple[int, pathlib.Path]] = None) -> OptionalEvalResult:
-
-        graph = self.evaluation_cache.get(cache_key, graph)
-        eval_res = super().evaluate_single(graph, uid_of_individual, with_time_limit, cache_key, logs_initializer)
-        return eval_res
-
-    def _reset_eval_cache(self):
-        self.evaluation_cache: Dict[str, Graph] = {}
-
-    def _remote_compute_cache(self, population: PopulationT):
-        self._reset_eval_cache()
-        if self._delegate_evaluator and self._delegate_evaluator.is_enabled:
-            self.logger.info('Remote fit used')
-            restored_graphs = self._adapter.restore(population)
-            computed_graphs = self._delegate_evaluator.compute_graphs(restored_graphs)
-            self.evaluation_cache = {ind.uid: graph for ind, graph in zip(population, computed_graphs)}
-
 
 class SequentialDispatcher(BaseGraphEvaluationDispatcher):
     """Evaluates objective function on population in sequential way.

diff --git a/golem/core/optimisers/meta/__init__.py b/golem/core/optimisers/meta/__init__.py
diff --git a/golem/core/optimisers/meta/surrogate_evaluator.py b/golem/core/optimisers/meta/surrogate_evaluator.py
@@ -0,0 +1,50 @@
+import pathlib
+import timeit
+from datetime import datetime
+from typing import Optional, Tuple
+
+from golem.core.adapter import BaseOptimizationAdapter
+from golem.core.log import Log
+from golem.core.optimisers.genetic.evaluation import OptionalEvalResult, DelegateEvaluator, SequentialDispatcher
+from golem.core.optimisers.graph import OptGraph
+from golem.core.optimisers.meta.surrogate_model import SurrogateModel, RandomValuesSurrogateModel
+from golem.core.optimisers.objective.objective import to_fitness, GraphFunction
+from golem.core.optimisers.opt_history_objects.individual import GraphEvalResult
+
+
+class SurrogateDispatcher(SequentialDispatcher):
+    """Evaluates objective function with surrogate model.
+        Usage: call `dispatch(objective_function)` to get evaluation function.
+        Additionally, we need to pass surrogate_model object
+    """
+
+    def __init__(self,
+                 adapter: BaseOptimizationAdapter,
+                 n_jobs: int = 1,
+                 graph_cleanup_fn: Optional[GraphFunction] = None,
+                 delegate_evaluator: Optional[DelegateEvaluator] = None,
+                 surrogate_model: SurrogateModel = RandomValuesSurrogateModel()):
+        super().__init__(adapter, n_jobs, graph_cleanup_fn, delegate_evaluator)
+        self._n_jobs = 1
+        self.surrogate_model = surrogate_model
+
+    def evaluate_single(self, graph: OptGraph, uid_of_individual: str, with_time_limit: bool = True,
+                        cache_key: Optional[str] = None,
+                        logs_initializer: Optional[Tuple[int, pathlib.Path]] = None) -> OptionalEvalResult:
+        graph = self.evaluation_cache.get(cache_key, graph)
+        if logs_initializer is not None:
+            # in case of multiprocessing run
+            Log.setup_in_mp(*logs_initializer)
+
+        start_time = timeit.default_timer()
+        fitness = to_fitness(self.surrogate_model(graph, objective=self._objective_eval))
+        end_time = timeit.default_timer()
+
+        eval_res = GraphEvalResult(
+            uid_of_individual=uid_of_individual, fitness=fitness, graph=graph, metadata={
+                'computation_time_in_seconds': end_time - start_time,
+                'evaluation_time_iso': datetime.now().isoformat(),
+                'surrogate_evaluation': True
+            }
+        )
+        return eval_res
diff --git a/golem/core/optimisers/meta/surrogate_model.py b/golem/core/optimisers/meta/surrogate_model.py
@@ -0,0 +1,23 @@
+from abc import abstractmethod
+from typing import Any
+
+import numpy as np
+
+from golem.core.dag.graph import Graph
+
+
+class SurrogateModel:
+    """
+    Model for evaluating fitness function without time-consuming fitting pipeline
+    """
+    @abstractmethod
+    def __call__(self, graph: Graph, **kwargs: Any):
+        raise NotImplementedError()
+
+
+class RandomValuesSurrogateModel(SurrogateModel):
+    """
+        Model for evaluating fitness function based on returning random values for any model
+    """
+    def __call__(self, graph: Graph, **kwargs: Any):
+        return np.random.random(1)
diff --git a/golem/core/optimisers/meta/surrogate_optimizer.py b/golem/core/optimisers/meta/surrogate_optimizer.py
@@ -0,0 +1,58 @@
+from typing import Sequence
+
+from golem.core.optimisers.genetic.gp_optimizer import EvoGraphOptimizer
+from golem.core.optimisers.genetic.gp_params import GPAlgorithmParameters
+from golem.core.optimisers.graph import OptGraph
+from golem.core.optimisers.meta.surrogate_evaluator import SurrogateDispatcher
+from golem.core.optimisers.meta.surrogate_model import RandomValuesSurrogateModel
+from golem.core.optimisers.objective import Objective, ObjectiveFunction
+from golem.core.optimisers.optimization_parameters import GraphRequirements
+from golem.core.optimisers.optimizer import GraphGenerationParams
+from golem.core.optimisers.populational_optimizer import EvaluationAttemptsError, _try_unfit_graph
+
+
+class SurrogateEachNgenOptimizer(EvoGraphOptimizer):
+    """
+    Surrogate optimizer that uses surrogate model for evaluating part of individuals
+
+    Additionally, we need to pass surrogate_model object
+    """
+    def __init__(self,
+                 objective: Objective,
+                 initial_graphs: Sequence[OptGraph],
+                 requirements: GraphRequirements,
+                 graph_generation_params: GraphGenerationParams,
+                 graph_optimizer_params: GPAlgorithmParameters,
+                 surrogate_model=RandomValuesSurrogateModel(),
+                 surrogate_each_n_gen=5
+                 ):
+        super().__init__(objective, initial_graphs, requirements, graph_generation_params, graph_optimizer_params)
+        self.surrogate_model = surrogate_model
+        self.surrogate_each_n_gen = surrogate_each_n_gen
+        self.surrogate_dispatcher = SurrogateDispatcher(adapter=graph_generation_params.adapter,
+                                                        n_jobs=requirements.n_jobs,
+                                                        graph_cleanup_fn=_try_unfit_graph,
+                                                        delegate_evaluator=graph_generation_params.remote_evaluator,
+                                                        surrogate_model=surrogate_model)
+
+    def optimise(self, objective: ObjectiveFunction) -> Sequence[OptGraph]:
+        # eval_dispatcher defines how to evaluate objective on the whole population
+        evaluator = self.eval_dispatcher.dispatch(objective, self.timer)
+        # surrogate_dispatcher defines how to evaluate objective with surrogate model
+        surrogate_evaluator = self.surrogate_dispatcher.dispatch(objective, self.timer)
+
+        with self.timer, self._progressbar:
+            self._initial_population(evaluator)
+            while not self.stop_optimization():
+                try:
+                    if self.generations.generation_num % self.surrogate_each_n_gen == 0:
+                        new_population = self._evolve_population(surrogate_evaluator)
+                    else:
+                        new_population = self._evolve_population(evaluator)
+                except EvaluationAttemptsError as ex:
+                    self.log.warning(f'Composition process was stopped due to: {ex}')
+                    return [ind.graph for ind in self.best_individuals]
+                # Adding of new population to history
+                self._update_population(new_population)
+        self._update_population(self.best_individuals, 'final_choices')
+        return [ind.graph for ind in self.best_individuals]
diff --git a/test/unit/optimizers/test_evaluation.py b/test/unit/optimizers/test_evaluation.py
@@ -8,6 +8,7 @@
 from golem.core.optimisers.fitness import Fitness, null_fitness
 from golem.core.optimisers.genetic.evaluation import MultiprocessingDispatcher, SequentialDispatcher, \
     ObjectiveEvaluationDispatcher
+from golem.core.optimisers.meta.surrogate_evaluator import SurrogateDispatcher
 from golem.core.optimisers.objective import Objective
 from golem.core.optimisers.opt_history_objects.individual import Individual
 from golem.core.optimisers.timer import OptimisationTimer
@@ -65,6 +66,7 @@ def test_dispatchers_with_faulty_objectives(objective, dispatcher):
 @pytest.mark.parametrize('dispatcher', [
     MultiprocessingDispatcher(DirectAdapter()),
     SequentialDispatcher(DirectAdapter()),
+    SurrogateDispatcher(DirectAdapter()),
 ])
 def test_dispatcher_with_timeout(dispatcher: ObjectiveEvaluationDispatcher):
     adapter, population = set_up_tests()
@@ -76,7 +78,8 @@ def test_dispatcher_with_timeout(dispatcher: ObjectiveEvaluationDispatcher):
     fitness = [x.fitness for x in evaluated_population]
     assert all(x.valid for x in fitness), "At least one fitness value is invalid"
     assert len(evaluated_population) >= 1, "At least one graphs is evaluated"
-    assert len(evaluated_population) < len(population), "Not all graphs should be evaluated (not enough time)"
+    if type(dispatcher) != SurrogateDispatcher:
+        assert len(evaluated_population) < len(population), "Not all graphs should be evaluated (not enough time)"
 
     timeout = datetime.timedelta(minutes=5)
     with OptimisationTimer(timeout=timeout) as t: