Skip to content

Candidate selector

candidate_selector

Candidate selector adapter implementations.

Defines adapter implementations of CandidateSelectorProtocol for Pareto, greedy, and epsilon-greedy selection strategies.

ATTRIBUTE DESCRIPTION
ParetoCandidateSelector

Pareto frontier sampling selector.

TYPE: class

CurrentBestCandidateSelector

Greedy best-average selector.

TYPE: class

EpsilonGreedyCandidateSelector

Epsilon-greedy exploration selector.

TYPE: class

create_candidate_selector

Selector factory by name.

TYPE: function

Note

This module provides selector adapters for Pareto-aware evolution workflows.

ParetoCandidateSelector

Sample from the Pareto front proportional to leadership frequency.

ATTRIBUTE DESCRIPTION
_rng

RNG for sampling candidates.

TYPE: Random

Note

A Pareto selector emphasizes candidates that lead more examples.

Examples:

selector = ParetoCandidateSelector(rng=random.Random(42))
candidate_idx = await selector.select_candidate(state)
Source code in src/gepa_adk/adapters/candidate_selector.py
class ParetoCandidateSelector:
    """Sample from the Pareto front proportional to leadership frequency.

    Attributes:
        _rng (random.Random): RNG for sampling candidates.

    Note:
        A Pareto selector emphasizes candidates that lead more examples.

    Examples:
        ```python
        selector = ParetoCandidateSelector(rng=random.Random(42))
        candidate_idx = await selector.select_candidate(state)
        ```
    """

    def __init__(self, rng: random.Random | None = None) -> None:
        """Initialize the selector.

        Args:
            rng: Optional random number generator for reproducibility.
        """
        self._rng = rng or random.Random()

    async def select_candidate(self, state: ParetoState) -> int:
        """Select a candidate index from the Pareto frontier.

        Args:
            state: Current evolution state with Pareto tracking.

        Returns:
            Selected candidate index.

        Raises:
            NoCandidateAvailableError: If no candidates or leaders exist.

        Examples:
            ```python
            candidate_idx = await selector.select_candidate(state)
            ```
        """
        if not state.candidates:
            raise NoCandidateAvailableError("No candidates available for selection")

        weights = state.frontier.get_selection_weights()
        if not weights:
            raise NoCandidateAvailableError("Pareto frontier is empty")

        sampling_list = [
            candidate_idx
            for candidate_idx, weight in weights.items()
            for _ in range(weight)
        ]
        if not sampling_list:
            raise NoCandidateAvailableError("Pareto frontier has no leaders")

        return self._rng.choice(sampling_list)

__init__

__init__(rng: Random | None = None) -> None

Initialize the selector.

PARAMETER DESCRIPTION
rng

Optional random number generator for reproducibility.

TYPE: Random | None DEFAULT: None

Source code in src/gepa_adk/adapters/candidate_selector.py
def __init__(self, rng: random.Random | None = None) -> None:
    """Initialize the selector.

    Args:
        rng: Optional random number generator for reproducibility.
    """
    self._rng = rng or random.Random()

select_candidate async

select_candidate(state: ParetoState) -> int

Select a candidate index from the Pareto frontier.

PARAMETER DESCRIPTION
state

Current evolution state with Pareto tracking.

TYPE: ParetoState

RETURNS DESCRIPTION
int

Selected candidate index.

RAISES DESCRIPTION
NoCandidateAvailableError

If no candidates or leaders exist.

Examples:

candidate_idx = await selector.select_candidate(state)
Source code in src/gepa_adk/adapters/candidate_selector.py
async def select_candidate(self, state: ParetoState) -> int:
    """Select a candidate index from the Pareto frontier.

    Args:
        state: Current evolution state with Pareto tracking.

    Returns:
        Selected candidate index.

    Raises:
        NoCandidateAvailableError: If no candidates or leaders exist.

    Examples:
        ```python
        candidate_idx = await selector.select_candidate(state)
        ```
    """
    if not state.candidates:
        raise NoCandidateAvailableError("No candidates available for selection")

    weights = state.frontier.get_selection_weights()
    if not weights:
        raise NoCandidateAvailableError("Pareto frontier is empty")

    sampling_list = [
        candidate_idx
        for candidate_idx, weight in weights.items()
        for _ in range(weight)
    ]
    if not sampling_list:
        raise NoCandidateAvailableError("Pareto frontier has no leaders")

    return self._rng.choice(sampling_list)

CurrentBestCandidateSelector

Always select the candidate with the highest average score.

Note

A greedy selector always exploits the best-average candidate.

Examples:

selector = CurrentBestCandidateSelector()
candidate_idx = await selector.select_candidate(state)
Source code in src/gepa_adk/adapters/candidate_selector.py
class CurrentBestCandidateSelector:
    """Always select the candidate with the highest average score.

    Note:
        A greedy selector always exploits the best-average candidate.

    Examples:
        ```python
        selector = CurrentBestCandidateSelector()
        candidate_idx = await selector.select_candidate(state)
        ```
    """

    async def select_candidate(self, state: ParetoState) -> int:
        """Return the best-average candidate index.

        Args:
            state: Current evolution state with Pareto tracking.

        Returns:
            Selected candidate index.

        Raises:
            NoCandidateAvailableError: If no candidates are available.

        Examples:
            ```python
            candidate_idx = await selector.select_candidate(state)
            ```
        """
        if state.best_average_idx is None:
            raise NoCandidateAvailableError("No candidates available for selection")
        return state.best_average_idx

select_candidate async

select_candidate(state: ParetoState) -> int

Return the best-average candidate index.

PARAMETER DESCRIPTION
state

Current evolution state with Pareto tracking.

TYPE: ParetoState

RETURNS DESCRIPTION
int

Selected candidate index.

RAISES DESCRIPTION
NoCandidateAvailableError

If no candidates are available.

Examples:

candidate_idx = await selector.select_candidate(state)
Source code in src/gepa_adk/adapters/candidate_selector.py
async def select_candidate(self, state: ParetoState) -> int:
    """Return the best-average candidate index.

    Args:
        state: Current evolution state with Pareto tracking.

    Returns:
        Selected candidate index.

    Raises:
        NoCandidateAvailableError: If no candidates are available.

    Examples:
        ```python
        candidate_idx = await selector.select_candidate(state)
        ```
    """
    if state.best_average_idx is None:
        raise NoCandidateAvailableError("No candidates available for selection")
    return state.best_average_idx

EpsilonGreedyCandidateSelector

Epsilon-greedy selection balancing exploration and exploitation.

ATTRIBUTE DESCRIPTION
_epsilon

Exploration probability.

TYPE: float

_rng

RNG for exploration decisions.

TYPE: Random

Note

A mixed strategy sometimes explores and otherwise exploits the best.

Examples:

selector = EpsilonGreedyCandidateSelector(epsilon=0.1, rng=random.Random(7))
candidate_idx = await selector.select_candidate(state)
Source code in src/gepa_adk/adapters/candidate_selector.py
class EpsilonGreedyCandidateSelector:
    """Epsilon-greedy selection balancing exploration and exploitation.

    Attributes:
        _epsilon (float): Exploration probability.
        _rng (random.Random): RNG for exploration decisions.

    Note:
        A mixed strategy sometimes explores and otherwise exploits the best.

    Examples:
        ```python
        selector = EpsilonGreedyCandidateSelector(epsilon=0.1, rng=random.Random(7))
        candidate_idx = await selector.select_candidate(state)
        ```
    """

    def __init__(self, epsilon: float, rng: random.Random | None = None) -> None:
        """Initialize the selector.

        Args:
            epsilon: Probability of random exploration.
            rng: Optional random number generator for reproducibility.

        Raises:
            ConfigurationError: If epsilon is outside [0.0, 1.0].
        """
        if not 0.0 <= epsilon <= 1.0:
            raise ConfigurationError(
                "epsilon must be between 0.0 and 1.0",
                field="epsilon",
                value=epsilon,
                constraint="0.0 <= epsilon <= 1.0",
            )
        self._epsilon = epsilon
        self._rng = rng or random.Random()

    async def select_candidate(self, state: ParetoState) -> int:
        """Select a candidate using epsilon-greedy strategy.

        Args:
            state: Current evolution state with Pareto tracking.

        Returns:
            Selected candidate index.

        Raises:
            NoCandidateAvailableError: If no candidates are available.

        Examples:
            ```python
            candidate_idx = await selector.select_candidate(state)
            ```
        """
        if not state.candidates:
            raise NoCandidateAvailableError("No candidates available for selection")

        if self._rng.random() < self._epsilon:
            return self._rng.randint(0, len(state.candidates) - 1)

        if state.best_average_idx is None:
            raise NoCandidateAvailableError("No candidates available for selection")

        return state.best_average_idx

__init__

__init__(epsilon: float, rng: Random | None = None) -> None

Initialize the selector.

PARAMETER DESCRIPTION
epsilon

Probability of random exploration.

TYPE: float

rng

Optional random number generator for reproducibility.

TYPE: Random | None DEFAULT: None

RAISES DESCRIPTION
ConfigurationError

If epsilon is outside [0.0, 1.0].

Source code in src/gepa_adk/adapters/candidate_selector.py
def __init__(self, epsilon: float, rng: random.Random | None = None) -> None:
    """Initialize the selector.

    Args:
        epsilon: Probability of random exploration.
        rng: Optional random number generator for reproducibility.

    Raises:
        ConfigurationError: If epsilon is outside [0.0, 1.0].
    """
    if not 0.0 <= epsilon <= 1.0:
        raise ConfigurationError(
            "epsilon must be between 0.0 and 1.0",
            field="epsilon",
            value=epsilon,
            constraint="0.0 <= epsilon <= 1.0",
        )
    self._epsilon = epsilon
    self._rng = rng or random.Random()

select_candidate async

select_candidate(state: ParetoState) -> int

Select a candidate using epsilon-greedy strategy.

PARAMETER DESCRIPTION
state

Current evolution state with Pareto tracking.

TYPE: ParetoState

RETURNS DESCRIPTION
int

Selected candidate index.

RAISES DESCRIPTION
NoCandidateAvailableError

If no candidates are available.

Examples:

candidate_idx = await selector.select_candidate(state)
Source code in src/gepa_adk/adapters/candidate_selector.py
async def select_candidate(self, state: ParetoState) -> int:
    """Select a candidate using epsilon-greedy strategy.

    Args:
        state: Current evolution state with Pareto tracking.

    Returns:
        Selected candidate index.

    Raises:
        NoCandidateAvailableError: If no candidates are available.

    Examples:
        ```python
        candidate_idx = await selector.select_candidate(state)
        ```
    """
    if not state.candidates:
        raise NoCandidateAvailableError("No candidates available for selection")

    if self._rng.random() < self._epsilon:
        return self._rng.randint(0, len(state.candidates) - 1)

    if state.best_average_idx is None:
        raise NoCandidateAvailableError("No candidates available for selection")

    return state.best_average_idx

create_candidate_selector

create_candidate_selector(
    selector_type: str,
    *,
    epsilon: float = 0.1,
    rng: Random | None = None,
) -> CandidateSelectorProtocol

Create a candidate selector by name.

PARAMETER DESCRIPTION
selector_type

Selector identifier (pareto, greedy, epsilon_greedy).

TYPE: str

epsilon

Exploration rate for epsilon-greedy selector.

TYPE: float DEFAULT: 0.1

rng

Optional RNG for selectors using randomness.

TYPE: Random | None DEFAULT: None

RETURNS DESCRIPTION
CandidateSelectorProtocol

CandidateSelectorProtocol implementation.

RAISES DESCRIPTION
ConfigurationError

If selector_type is unsupported.

Examples:

selector = create_candidate_selector("pareto")
candidate_idx = await selector.select_candidate(state)
Source code in src/gepa_adk/adapters/candidate_selector.py
def create_candidate_selector(
    selector_type: str,
    *,
    epsilon: float = 0.1,
    rng: random.Random | None = None,
) -> CandidateSelectorProtocol:
    """Create a candidate selector by name.

    Args:
        selector_type: Selector identifier (pareto, greedy, epsilon_greedy).
        epsilon: Exploration rate for epsilon-greedy selector.
        rng: Optional RNG for selectors using randomness.

    Returns:
        CandidateSelectorProtocol implementation.

    Raises:
        ConfigurationError: If selector_type is unsupported.

    Examples:
        ```python
        selector = create_candidate_selector("pareto")
        candidate_idx = await selector.select_candidate(state)
        ```
    """
    normalized = selector_type.strip().lower()
    if normalized in {"pareto"}:
        return ParetoCandidateSelector(rng=rng)
    if normalized in {"greedy", "current_best", "current-best"}:
        return CurrentBestCandidateSelector()
    if normalized in {"epsilon_greedy", "epsilon-greedy"}:
        return EpsilonGreedyCandidateSelector(epsilon=epsilon, rng=rng)
    raise ConfigurationError(
        "selector_type must be one of pareto, greedy, epsilon_greedy",
        field="selector_type",
        value=selector_type,
        constraint="pareto|greedy|epsilon_greedy",
    )