Source code for ribs.emitters.rankers

"""Rankers for use across emitters.

The rankers implemented in this file are intended to be used with emitters.
Specifically, a ranker object should be initialized or passed in the emitters.
The ``Ranker`` object will define the :meth:`~RankerBase.rank` method which
returns the result of a descending argsort of the solutions. It will also define
a :meth:`~RankerBase.reset` method which resets the internal state of the
object.

When specifying which ranker to use for each emitter, one could either pass in
the full name of a ranker, e.g. "ImprovementRanker", or the abbreviated name of
a ranker, e.g. "imp".
The supported abbreviations are:

* ``imp``: :class:`ImprovementRanker`
* ``2imp``: :class:`TwoStageImprovementRanker`
* ``rd``: :class:`RandomDirectionRanker`
* ``2rd``: :class:`TwoStageRandomDirectionRanker`
* ``obj``: :class:`ObjectiveRanker`
* ``2obj``: :class:`TwoStageObjectiveRanker`

.. autosummary::
    :toctree:

    ribs.emitters.rankers.ImprovementRanker
    ribs.emitters.rankers.TwoStageImprovementRanker
    ribs.emitters.rankers.RandomDirectionRanker
    ribs.emitters.rankers.TwoStageRandomDirectionRanker
    ribs.emitters.rankers.ObjectiveRanker
    ribs.emitters.rankers.TwoStageObjectiveRanker
    ribs.emitters.rankers.RankerBase
"""
from abc import ABC, abstractmethod

import numpy as np

__all__ = [
    "ImprovementRanker",
    "TwoStageImprovementRanker",
    "RandomDirectionRanker",
    "TwoStageRandomDirectionRanker",
    "ObjectiveRanker",
    "TwoStageObjectiveRanker",
    "RankerBase",
]

_RANK_ARGS = """
Args:
    emitter (ribs.emitters.EmitterBase): Emitter that this ``ranker``
        object belongs to.
    archive (ribs.archives.ArchiveBase): Archive used by ``emitter``
        when creating and inserting solutions.
    data (dict): Dict mapping from field names like ``solution`` and
        ``objective`` to arrays with data for the solutions. Rankers at least
        assume the presence of the ``solution`` key.
    add_info (dict): Information returned by an archive's add() method.

Returns:
    tuple(numpy.ndarray, numpy.ndarray): the first array (shape
    ``(batch_size,)``) is an array of indices representing a ranking of the
    solutions and the second array (shape ``(batch_size,)`` or (batch_size,
    n_values)``) is an array of values that this ranker used to rank the
    solutions. ``batch_size`` is the number of solutions and ``n_values`` is
    the number of values that the rank function used.
"""

_RESET_ARGS = """
Args:
    emitter (ribs.emitters.EmitterBase): Emitter that this ``ranker``
        object belongs to.
    archive (ribs.archives.ArchiveBase): Archive used by ``emitter``
        when creating and inserting solutions.
"""


[docs]class RankerBase(ABC): """Base class for rankers. Every ranker has a :meth:`rank` method that returns a list of indices that indicate how the solutions should be ranked and a :meth:`reset` method that resets the internal state of the ranker (e.g. in :class:`~ribs.emitters.rankers.RandomDirectionRanker`). Child classes are only required to override :meth:`rank`. """ def __init__(self, seed=None): self._rng = np.random.default_rng(seed)
[docs] @abstractmethod def rank(self, emitter, archive, data, add_info): # pylint: disable=missing-function-docstring pass
# Generates the docstring for rank rank.__doc__ = f""" Generates a batch of indices that represents an ordering of ``data["solution"]``. {_RANK_ARGS} """
[docs] def reset(self, emitter, archive): # pylint: disable=missing-function-docstring pass
reset.__doc__ = f""" Resets the internal state of the ranker. {_RESET_ARGS} """
[docs]class ImprovementRanker(RankerBase): """Ranks the solutions based on the improvement in the objective. This ranker ranks solutions in a single stage. The solutions are ranked by the improvement "value" described in :meth:`ribs.archives.ArchiveBase.add`. This ranker should not be used with CMA-ME. The improvement values for new solutions in CMA-ME are on a different scale from the improvement values for the other solutions, in that new solutions have an improvement value which is simply their objective, while other solutions have an improvement value which is the difference between their objective and the objective of their corresponding cell in the archive. """
[docs] def rank(self, emitter, archive, data, add_info): # Note that argsort sorts the values in ascending order, # so we use np.flip to reverse the sorted array. return np.flip(np.argsort(add_info["value"])), add_info["value"]
rank.__doc__ = f""" Generates a list of indices that represents an ordering of solutions. {_RANK_ARGS} """
[docs]class TwoStageImprovementRanker(RankerBase): """Ranks the solutions based on the improvement in the objective. This ranker originates in `Fontaine 2020 <https://arxiv.org/abs/1912.02400>`_ in which it was referred to as "Improvement Emitter". This ranker ranks solutions in two stages. First, solutions are ranked by "status" -- those that found a new cell in the archive rank above those that improved an existing cell, which rank above those that were not added to the archive. Second, solutions are ranked by the "value" described in :meth:`ribs.archives.ArchiveBase.add` """
[docs] def rank(self, emitter, archive, data, add_info): # To avoid using an array of tuples, ranking_values is a 2D array # [[status_0, value_0], ..., [status_n, value_n]] ranking_values = np.stack((add_info["status"], add_info["value"]), axis=-1) # New solutions sort ahead of improved ones, which sort ahead of ones # that were not added. # # Since lexsort uses the last column/row as the key, we flip the # ranking_values along the last axis so that we are sorting by status # first. # # Note that lexsort sorts the values in ascending order, # so we use np.flip to reverse the sorted array. return ( np.flip(np.lexsort(np.flip(ranking_values, axis=-1).T)), ranking_values, )
rank.__doc__ = f""" Generates a list of indices that represents an ordering of solutions. {_RANK_ARGS} """
[docs]class RandomDirectionRanker(RankerBase): """Ranks the solutions based on projection onto a direction in measure space. This ranker originates in `Fontaine 2020 <https://arxiv.org/abs/1912.02400>`_ as random direction emitter. The solutions are ranked solely based on their projection onto a random direction in the archive. To rank the solutions first by whether they were added, and then by the projection, refer to :class:`ribs.emitters.rankers.TwoStageRandomDirectionRanker`. """ def __init__(self, seed=None): super().__init__(seed) self._target_measure_dir = None @property def target_measure_dir(self): """numpy.ndarray: ``(measure_dim,)`` array with the target measure direction vector.""" return self._target_measure_dir @target_measure_dir.setter def target_measure_dir(self, value): self._target_measure_dir = value
[docs] def rank(self, emitter, archive, data, add_info): if self._target_measure_dir is None: raise RuntimeError("target measure direction not set") projections = np.dot(data["measures"], self._target_measure_dir) # Sort only by projection; use np.flip to reverse the order return np.flip(np.argsort(projections)), projections
rank.__doc__ = f""" Ranks the solutions based on projection onto a direction in the archive. {_RANK_ARGS} """
[docs] def reset(self, emitter, archive): ranges = archive.upper_bounds - archive.lower_bounds measure_dim = len(ranges) unscaled_dir = self._rng.standard_normal(measure_dim) self._target_measure_dir = unscaled_dir * ranges
# Generates the docstring for reset. reset.__doc__ = f""" Generates a random direction in the archive's measure space. A random direction is sampled from a standard Gaussian -- since the standard Gaussian is isotropic, there is equal probability for any direction. The direction is then scaled to the archive bounds so that it is a random archive direction. {_RESET_ARGS} """
[docs]class TwoStageRandomDirectionRanker(RankerBase): """Similar to :class:`ribs.emitters.rankers.RandomDirectionRanker`, but the solutions are first ranked by whether they are added, then by their projection onto a random direction in the archive space. This ranker originates from `Fontaine 2020 <https://arxiv.org/abs/1912.02400>`_ as RandomDirectionEmitter. """ def __init__(self, seed=None): super().__init__(seed) self._target_measure_dir = None @property def target_measure_dir(self): """numpy.ndarray: ``(measure_dim,)`` array with the target measure direction vector.""" return self._target_measure_dir @target_measure_dir.setter def target_measure_dir(self, value): self._target_measure_dir = value
[docs] def rank(self, emitter, archive, data, add_info): if self._target_measure_dir is None: raise RuntimeError("target measure direction not set") projections = np.dot(data["measures"], self._target_measure_dir) # To avoid using an array of tuples, ranking_values is a 2D array # [[status_0, projection_0], ..., [status_n, projection_n]] ranking_values = np.stack((add_info["status"], projections), axis=-1) # Sort by whether the solution was added into the archive, # followed by projection. return ( np.flip(np.lexsort(np.flip(ranking_values, axis=-1).T)), ranking_values, )
rank.__doc__ = f""" Ranks the solutions first by whether they are added, then by their projection onto a random direction in the archive. {_RANK_ARGS} """
[docs] def reset(self, emitter, archive): ranges = archive.upper_bounds - archive.lower_bounds measure_dim = len(ranges) unscaled_dir = self._rng.standard_normal(measure_dim) self._target_measure_dir = unscaled_dir * ranges
reset.__doc__ = RandomDirectionRanker.reset.__doc__
[docs]class ObjectiveRanker(RankerBase): """Ranks the solutions solely based on the objective values. This ranker originates in `Fontaine 2020 <https://arxiv.org/abs/1912.02400>`_ in which it was part of the optimizing emitter. """
[docs] def rank(self, emitter, archive, data, add_info): # Sort only by objective value. return np.flip(np.argsort(data["objective"])), data["objective"]
rank.__doc__ = f""" Ranks the solutions based on their objective values. {_RANK_ARGS} """
[docs]class TwoStageObjectiveRanker(RankerBase): """Similar to :class:`ribs.emitters.rankers.ObjectiveRanker`, but ranks newly added solutions before improved solutions. This ranker originates in `Fontaine 2020 <https://arxiv.org/abs/1912.02400>`_ as OptimizingEmitter. """
[docs] def rank(self, emitter, archive, data, add_info): # To avoid using an array of tuples, ranking_values is a 2D array # [[status_0, objective_0], ..., [status_0, objective_n]] ranking_values = np.stack((add_info["status"], data["objective"]), axis=-1) # Sort by whether the solution was added into the archive, followed # by the objective values. return ( np.flip(np.lexsort(np.flip(ranking_values, axis=-1).T)), ranking_values, )
rank.__doc__ = f""" Ranks the solutions based on their objective values, while prioritizing newly added solutions. {_RANK_ARGS} """
_NAME_TO_RANKER_MAP = { "ImprovementRanker": ImprovementRanker, "TwoStageImprovementRanker": TwoStageImprovementRanker, "RandomDirectionRanker": RandomDirectionRanker, "TwoStageRandomDirectionRanker": TwoStageRandomDirectionRanker, "ObjectiveRanker": ObjectiveRanker, "TwoStageObjectiveRanker": TwoStageObjectiveRanker, "imp": ImprovementRanker, "2imp": TwoStageImprovementRanker, "rd": RandomDirectionRanker, "2rd": TwoStageRandomDirectionRanker, "obj": ObjectiveRanker, "2obj": TwoStageObjectiveRanker } def _get_ranker(klass, seed): """Returns a ranker class based on its name. ``klass`` can be a reference to the class of the ranker, the full name of a ranker, e.g. "ImprovementRanker", or the abbreviated name for a ranker such as "imp". Args: klass (callable or str): This parameter may either be a callable (e.g. a class or a lambda function) that takes in no parameters and returns an instance of :class:`RankerBase`, or it may be a full or abbreviated ranker name. Returns: The corresponding ranker class instance. """ if isinstance(klass, str): if klass in _NAME_TO_RANKER_MAP: return _NAME_TO_RANKER_MAP[klass](seed) raise ValueError(f"`{klass}` is not the full or abbreviated " "name of a valid ranker") if callable(klass): ranker = klass(seed) if isinstance(ranker, RankerBase): return ranker raise ValueError(f"Callable `{klass}` did not return an instance " "of RankerBase.") raise ValueError(f"`{klass}` is neither a callable nor a string")