Interrupt optimization through IncorporateRunResultCallback

changelog.md

@@ -1,5 +1,8 @@
 # 1.0.2
 
+## Features
+* Option to use an own stopping strategy using `IncorporateRunResultCallback`.
+
 ## Minor Changes
 * Made smac-validate.py consistent with runhistory and tae.
 

doc/callbacks.rst

@@ -0,0 +1,19 @@
+Callbacks
+---------
+
+Callbacks allow customizing the behavior of SMAC to ones needs. Currently, the list of implemented callbacks is
+very limited, but they can easily be added.
+
+If you want to create a new callback, please check `smac.callbacks` and create a new pull request.
+
+
+IncorporateRunResultCallback 
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Callback to react on a new run result.
+
+Called after the finished run is added to the runhistory.
+Optionally return `False` to (gracefully) stop the optimization.
+
+
+

smac/callbacks.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 if TYPE_CHECKING:
     from smac.optimizer.smbo import SMBO
@@ -31,12 +31,16 @@
 
 class IncorporateRunResultCallback:
 
-    """Callback to react on a new run result. Called after the finished run is added to the runhistory."""
+    """Callback to react on a new run result.
+
+    Called after the finished run is added to the runhistory.
+    Optionally return `False` to (gracefully) stop the optimization.
+    """
 
     def __call__(
             self, smbo: 'SMBO',
             run_info: RunInfo,
             result: RunValue,
             time_left: float,
-    ) -> None:
+    ) -> Optional[bool]:
         pass

smac/optimizer/smbo.py

@@ -301,7 +301,7 @@ def run(self) -> Configuration:
                 if self.stats.is_budget_exhausted():
                     self.logger.debug("Exhausted configuration budget")
                 else:
-                    self.logger.debug("Shutting down because a configuration returned status STOP")
+                    self.logger.debug("Shutting down because a configuration or callback returned status STOP")
 
                 # The budget can be exhausted  for 2 reasons: number of ta runs or
                 # time. If the number of ta runs is reached, but there is still budget,
@@ -489,7 +489,12 @@ def _incorporate_run_results(self, run_info: RunInfo, result: RunValue,
         )
 
         for callback in self._callbacks['_incorporate_run_results']:
-            callback(smbo=self, run_info=run_info, result=result, time_left=time_left)
+            response = callback(smbo=self, run_info=run_info, result=result, time_left=time_left)
+            # If a callback returns False, the optimization loop should be interrupted
+            # the other callbacks are still being called
+            if response is False:
+                self.logger.debug("An IncorporateRunResultCallback returned False, requesting abort.")
+                self._stop = True
 
         if self.scenario.save_results_instantly:  # type: ignore[attr-defined] # noqa F821
             self.save()

test/test_smbo/test_smbo.py

@@ -50,7 +50,8 @@ class TestSMBO(unittest.TestCase):
     def setUp(self):
         self.scenario = Scenario({'cs': test_helpers.get_branin_config_space(),
                                   'run_obj': 'quality',
-                                  'output_dir': 'data-test_smbo'})
+                                  'output_dir': 'data-test_smbo',
+                                  "runcount-limit": 5})
         self.output_dirs = []
         self.output_dirs.append(self.scenario.output_dir)
 
@@ -411,6 +412,35 @@ def __call__(self, smbo, run_info, result, time_left) -> None:
         self.assertEqual(callback.num_call, 1)
         self.assertEqual(callback.config, config)
 
+    @unittest.mock.patch.object(smac.facade.smac_ac_facade.Intensifier, 'process_results')
+    def test_incorporate_run_results_callback_stop_loop(self, process_results_mock):
+
+        def target(x):
+            return 5
+
+        process_results_mock.return_value = None, None
+
+        class TestCallback(IncorporateRunResultCallback):
+            def __init__(self):
+                self.num_call = 0
+
+            def __call__(self, smbo, run_info, result, time_left) -> None:
+                self.num_call += 1
+                if self.num_call > 2:
+                    return False
+
+        callback = TestCallback()
+
+        self.scenario.output_dir = None
+        smac = SMAC4AC(self.scenario, tae_runner=target, rng=1)
+        smac.register_callback(callback)
+
+        self.output_dirs.append(smac.output_dir)
+
+        smac.optimize()
+
+        self.assertEqual(callback.num_call, 3)
+
 
 if __name__ == "__main__":
     unittest.main()