Source code for ribs.emitters.opt._evolution_strategy_base
"""Provides EvolutionStrategyBase."""
from __future__ import annotations
from abc import ABC, abstractmethod
import numpy as np
from numpy.typing import DTypeLike
from ribs.typing import Float, Int
# Number of times solutions can be resampled before triggering a warning.
BOUNDS_SAMPLING_THRESHOLD = 100
# Warning for resampling solutions too many times.
BOUNDS_WARNING = (
"During bounds handling, this ES resampled at least "
f"{BOUNDS_SAMPLING_THRESHOLD} times. This may indicate that your solution "
"space bounds are too tight. When bounds are passed in, the ES resamples "
"until all solutions are within the bounds -- if the bounds are too tight "
"or the distribution is too large, the ES will resample forever."
)
[docs]
class EvolutionStrategyBase(ABC):
"""Base class for evolution strategy optimizers for use with emitters.
The basic usage is:
- Initialize the optimizer and reset it.
- Repeatedly:
- Request new solutions with ``ask()``
- Rank the solutions in the emitter (better solutions come first) and pass the
indices and values back with ``tell()``.
- Use ``check_stop()`` to see if the optimizer has reached a stopping condition,
and if so, call ``reset()``.
Args:
sigma0: Initial step size.
solution_dim: Size of the solution space.
batch_size: Number of solutions to evaluate at a time.
seed: Seed for the random number generator.
dtype: Data type of solutions.
lower_bounds: scalar or (solution_dim,) array indicating lower bounds of the
solution space. Scalars specify the same bound for the entire space, while
arrays specify a bound for each dimension. Pass -np.inf in the array or
scalar to indicated unbounded space.
upper_bounds: Same as above, but for upper bounds (and pass np.inf instead of
-np.inf).
"""
def __init__(
self,
sigma0: Float,
solution_dim: Int,
batch_size: Int | None = None,
seed: Int | None = None,
dtype: DTypeLike = np.float64,
lower_bounds: Float | np.ndarray = -np.inf,
upper_bounds: Float | np.ndarray = np.inf,
) -> None:
pass
[docs]
@abstractmethod
def reset(self, x0: np.ndarray) -> None:
"""Resets the ES to start at x0.
Args:
x0: Initial mean.
"""
[docs]
@abstractmethod
def check_stop(self, ranking_values: np.ndarray) -> bool:
"""Checks if the ES should stop and be reset.
Args:
ranking_values: Array of values that were used to rank the solutions. Shape
can be either ``(batch_size,)`` or (batch_size, n_values)``, where
``batch_size`` is the number of solutions and ``n_values`` is the number
of values that the ranker used. Note that unlike in :meth:`tell`, these
values must be sorted according to the ``ranking_indices`` passed to
:meth:`tell`.
Returns:
True if any of the stopping conditions are satisfied.
"""
[docs]
@abstractmethod
def ask(self, batch_size: Int | None = None) -> np.ndarray:
"""Samples new solutions.
Args:
batch_size: batch size of the sample. Defaults to ``self.batch_size``.
Returns:
Array of new solutions.
"""
[docs]
@abstractmethod
def tell(
self, ranking_indices: np.ndarray, ranking_values: np.ndarray, num_parents: Int
) -> None:
"""Passes the solutions back to the ES.
Args:
ranking_indices: Integer indices that are used to rank the solutions
returned in :meth:`ask`. Note that these are NOT the ranks of the
solutions. Rather, they are indices such that
``solutions[ranking_indices]`` will correctly rank the solutions (think
of an argsort).
ranking_values: Array of values that were used to rank the solutions. Shape
can be either ``(batch_size,)`` or (batch_size, n_values)``, where
``batch_size`` is the number of solutions and ``n_values`` is the number
of values that the ranker used. Note that we assume a descending sort,
i.e., higher values should come first.
num_parents: Number of top solutions to select from the ranked solutions.
"""