This method returns the next configuration to evaluate. It ignores already processed configurations, i.e.,
the configurations from the runhistory, if the runhistory is not empty.
The method (after yielding the initial design configurations) trains the surrogate model, maximizes the
acquisition function and yields n configurations. After the n configurations, the surrogate model is
trained again, etc. The program stops if retries was reached within each iteration. A configuration
is ignored, if it was used already before.
Note
When SMAC continues a run, processed configurations from the runhistory are ignored. For example, if the
intitial design configurations already have been processed, they are ignored here. After the run is
continued, however, the surrogate model is trained based on the runhistory in all cases.
Returns
next_config : Iterator[Configuration]
The next configuration to evaluate.
Source code in smac/main/config_selector.py
| def __iter__(self) -> Iterator[Configuration]:
"""This method returns the next configuration to evaluate. It ignores already processed configurations, i.e.,
the configurations from the runhistory, if the runhistory is not empty.
The method (after yielding the initial design configurations) trains the surrogate model, maximizes the
acquisition function and yields ``n`` configurations. After the ``n`` configurations, the surrogate model is
trained again, etc. The program stops if ``retries`` was reached within each iteration. A configuration
is ignored, if it was used already before.
Note
----
When SMAC continues a run, processed configurations from the runhistory are ignored. For example, if the
intitial design configurations already have been processed, they are ignored here. After the run is
continued, however, the surrogate model is trained based on the runhistory in all cases.
Returns
-------
next_config : Iterator[Configuration]
The next configuration to evaluate.
"""
assert self._runhistory is not None
assert self._runhistory_encoder is not None
assert self._model is not None
assert self._acquisition_maximizer is not None
assert self._acquisition_function is not None
assert self._random_design is not None
self._processed_configs = self._runhistory.get_configs()
# We add more retries because there could be a case in which the processed configs are sampled again
self._retries += len(self._processed_configs)
logger.debug("Search for the next configuration...")
self._call_callbacks_on_start()
# First: We return the initial configurations
for config in self._initial_design_configs:
if config not in self._processed_configs:
self._processed_configs.append(config)
self._call_callbacks_on_end(config)
yield config
self._call_callbacks_on_start()
# We want to generate configurations endlessly
while True:
# Cost value of incumbent configuration (required for acquisition function).
# If not given, it will be inferred from runhistory or predicted.
# If not given and runhistory is empty, it will raise a ValueError.
incumbent_value: float | None = None
# Everytime we re-train the surrogate model, we also update our multi-objective algorithm
if (mo := self._runhistory_encoder.multi_objective_algorithm) is not None:
mo.update_on_iteration_start()
X, Y, X_configurations = self._collect_data()
previous_configs = self._runhistory.get_configs()
if X.shape[0] == 0:
# Only return a single point to avoid an overly high number of random search iterations.
# We got rid of random search here and replaced it with a simple configuration sampling from
# the configspace.
logger.debug("No data available to train the model. Sample a random configuration.")
config = self._scenario.configspace.sample_configuration()
self._call_callbacks_on_end(config)
yield config
self._call_callbacks_on_start()
# Important to continue here because we still don't have data available
continue
# Check if X/Y differs from the last run, otherwise use cached results
if self._previous_entries != Y.shape[0]:
self._model.train(X, Y)
x_best_array: np.ndarray | None = None
if incumbent_value is not None:
best_observation = incumbent_value
else:
if self._runhistory.empty():
raise ValueError("Runhistory is empty and the cost value of the incumbent is unknown.")
x_best_array, best_observation = self._get_x_best(X_configurations)
self._acquisition_function.update(
model=self._model,
eta=best_observation,
incumbent_array=x_best_array,
num_data=len(self._get_evaluated_configs()),
X=X_configurations,
)
# We want to cache how many entries we used because if we have the same number of entries
# we don't need to train the next time
self._previous_entries = Y.shape[0]
# Now we maximize the acquisition function
challengers = self._acquisition_maximizer.maximize(
previous_configs,
random_design=self._random_design,
)
counter = 0
failed_counter = 0
for config in challengers:
if config not in self._processed_configs:
counter += 1
self._processed_configs.append(config)
self._call_callbacks_on_end(config)
yield config
retrain = counter == self._retrain_after
self._call_callbacks_on_start()
# We break to enforce a new iteration of the while loop (i.e. we retrain the surrogate model)
if retrain:
logger.debug(
f"Yielded {counter} configurations. Start new iteration and retrain surrogate model."
)
break
else:
failed_counter += 1
# We exit the loop if we have tried to add the same configuration too often
if failed_counter == self._retries:
logger.warning(f"Could not return a new configuration after {self._retries} retries." "")
return
|