Skip to content

Regression

regression

Regression detection stopper for evolution termination on score decline.

This module provides a RegressionStopper that terminates evolution when the current best score declines compared to the score from a configurable lookback window ago, preventing wasted compute once performance regresses relative to a previous baseline.

ATTRIBUTE DESCRIPTION
RegressionStopper

Stop evolution when score declines relative to the score from N iterations ago, using a bounded deque for O(1) memory regardless of run length.

TYPE: class

Examples:

Basic usage with default window:

from gepa_adk import RegressionStopper

stopper = RegressionStopper()  # window=3
# Use stopper in evolution config

Custom lookback window:

from gepa_adk import RegressionStopper

stopper = RegressionStopper(window=5)  # Stop if score drops vs 5 iters ago

Composing with other stoppers:

from gepa_adk import RegressionStopper
from gepa_adk.adapters.stoppers import CompositeStopper, ScoreThresholdStopper

composite = CompositeStopper(
    [RegressionStopper(window=3), ScoreThresholdStopper(0.99)],
    mode="any",
)
See Also

RegressionStopper

Stops evolution when best score declines over a lookback window.

Detects regression by comparing the current best score to the best score from window iterations ago. Returns True (stop) when the current score is strictly lower than the score window steps prior.

Requires at least window + 1 calls before any regression can be detected. Call setup() (or let the engine call it) to reset history between runs.

ATTRIBUTE DESCRIPTION
window

Number of iterations to look back for comparison.

TYPE: int

PARAMETER DESCRIPTION
window

Number of iterations to look back for comparison. Must be >= 1. Default is 3.

TYPE: int DEFAULT: 3

Examples:

Detecting degrading runs:

from gepa_adk import RegressionStopper
from gepa_adk.domain.stopper import StopperState

stopper = RegressionStopper(window=3)

# Simulated evolution calls (StopperState requires all 6 fields)
stopper(
    StopperState(
        iteration=0,
        best_score=0.5,
        stagnation_counter=0,
        total_evaluations=0,
        candidates_count=1,
        elapsed_seconds=0.0,
    )
)  # False (cold start)
stopper(
    StopperState(
        iteration=1,
        best_score=0.6,
        stagnation_counter=0,
        total_evaluations=1,
        candidates_count=1,
        elapsed_seconds=1.0,
    )
)  # False
stopper(
    StopperState(
        iteration=2,
        best_score=0.7,
        stagnation_counter=0,
        total_evaluations=2,
        candidates_count=1,
        elapsed_seconds=2.0,
    )
)  # False
stopper(
    StopperState(
        iteration=3,
        best_score=0.4,
        stagnation_counter=0,
        total_evaluations=3,
        candidates_count=1,
        elapsed_seconds=3.0,
    )
)  # True (0.4 < 0.5, the baseline from window=3 ago)
Notes

Equal scores (plateau) are NOT considered regression. Only strictly lower scores trigger a stop.

Source code in src/gepa_adk/adapters/stoppers/regression.py
class RegressionStopper:
    """Stops evolution when best score declines over a lookback window.

    Detects regression by comparing the current best score to the best score
    from ``window`` iterations ago. Returns ``True`` (stop) when the current
    score is strictly lower than the score ``window`` steps prior.

    Requires at least ``window + 1`` calls before any regression can be detected.
    Call ``setup()`` (or let the engine call it) to reset history between runs.

    Attributes:
        window (int): Number of iterations to look back for comparison.

    Args:
        window: Number of iterations to look back for comparison. Must be >= 1.
            Default is 3.

    Examples:
        Detecting degrading runs:

        ```python
        from gepa_adk import RegressionStopper
        from gepa_adk.domain.stopper import StopperState

        stopper = RegressionStopper(window=3)

        # Simulated evolution calls (StopperState requires all 6 fields)
        stopper(
            StopperState(
                iteration=0,
                best_score=0.5,
                stagnation_counter=0,
                total_evaluations=0,
                candidates_count=1,
                elapsed_seconds=0.0,
            )
        )  # False (cold start)
        stopper(
            StopperState(
                iteration=1,
                best_score=0.6,
                stagnation_counter=0,
                total_evaluations=1,
                candidates_count=1,
                elapsed_seconds=1.0,
            )
        )  # False
        stopper(
            StopperState(
                iteration=2,
                best_score=0.7,
                stagnation_counter=0,
                total_evaluations=2,
                candidates_count=1,
                elapsed_seconds=2.0,
            )
        )  # False
        stopper(
            StopperState(
                iteration=3,
                best_score=0.4,
                stagnation_counter=0,
                total_evaluations=3,
                candidates_count=1,
                elapsed_seconds=3.0,
            )
        )  # True (0.4 < 0.5, the baseline from window=3 ago)
        ```

    Notes:
        Equal scores (plateau) are NOT considered regression. Only strictly
        lower scores trigger a stop.
    """

    def __init__(self, *, window: int = 3) -> None:
        """Initialize RegressionStopper with lookback window.

        Args:
            window: Number of iterations to look back for score comparison.
                Must be >= 1. Default is 3.

        Raises:
            ConfigurationError: If window < 1.

        Examples:
            ```python
            stopper = RegressionStopper()  # window=3
            stopper = RegressionStopper(window=5)  # window=5
            ```

        Notes:
            Score history is stored in a ``deque(maxlen=window+1)`` so memory
            usage is bounded to exactly ``window + 1`` floats regardless of
            run length.
        """
        if window < 1:
            raise ConfigurationError(
                f"RegressionStopper window must be >= 1, got {window}",
                field="window",
                value=window,
                constraint="Must be >= 1",
            )
        self.window = window
        self._score_history: deque[float] = deque(maxlen=self.window + 1)

    def setup(self) -> None:
        """Reset score history. Called by engine at start of each run.

        Clears the internal score history by replacing the bounded deque with
        a fresh one, so the stopper can be safely reused across multiple
        ``evolve()`` calls with the same instance. Without this reset, history
        from one run would bleed into the next.

        Examples:
            ```python
            stopper = RegressionStopper()
            # ... run evolution ...
            stopper.setup()  # reset for next run
            ```
        """
        self._score_history = deque(maxlen=self.window + 1)

    def __call__(self, state: StopperState) -> bool:
        """Check if evolution should stop due to score regression.

        Appends the current best score to history, then compares the latest
        score against the score from ``window`` iterations ago. Returns
        ``False`` during the cold-start phase (fewer than ``window + 1`` calls).

        Args:
            state: Current evolution state snapshot containing best_score.

        Returns:
            True if current best score is strictly lower than the score
            ``window`` iterations ago, False otherwise.

        Examples:
            ```python
            stopper = RegressionStopper(window=3)
            # Prime with 3 cold-start calls
            for score in [0.5, 0.6, 0.7]:
                stopper(
                    StopperState(
                        iteration=0,
                        best_score=score,
                        stagnation_counter=0,
                        total_evaluations=0,
                        candidates_count=1,
                        elapsed_seconds=0.0,
                    )
                )
            state = StopperState(
                iteration=3,
                best_score=0.4,
                stagnation_counter=0,
                total_evaluations=3,
                candidates_count=1,
                elapsed_seconds=3.0,
            )
            stopper(state)  # True (0.4 < 0.5, the baseline from window=3 ago)
            ```
        """
        self._score_history.append(state.best_score)
        if len(self._score_history) <= self.window:
            return False
        current = self._score_history[-1]
        baseline = self._score_history[-(self.window + 1)]
        if current < baseline:
            logger.info(
                "stopper.regression.triggered",
                window=self.window,
                current_score=current,
                baseline_score=baseline,
            )
            return True
        return False

__init__

__init__(*, window: int = 3) -> None

Initialize RegressionStopper with lookback window.

PARAMETER DESCRIPTION
window

Number of iterations to look back for score comparison. Must be >= 1. Default is 3.

TYPE: int DEFAULT: 3

RAISES DESCRIPTION
ConfigurationError

If window < 1.

Examples:

stopper = RegressionStopper()  # window=3
stopper = RegressionStopper(window=5)  # window=5
Notes

Score history is stored in a deque(maxlen=window+1) so memory usage is bounded to exactly window + 1 floats regardless of run length.

Source code in src/gepa_adk/adapters/stoppers/regression.py
def __init__(self, *, window: int = 3) -> None:
    """Initialize RegressionStopper with lookback window.

    Args:
        window: Number of iterations to look back for score comparison.
            Must be >= 1. Default is 3.

    Raises:
        ConfigurationError: If window < 1.

    Examples:
        ```python
        stopper = RegressionStopper()  # window=3
        stopper = RegressionStopper(window=5)  # window=5
        ```

    Notes:
        Score history is stored in a ``deque(maxlen=window+1)`` so memory
        usage is bounded to exactly ``window + 1`` floats regardless of
        run length.
    """
    if window < 1:
        raise ConfigurationError(
            f"RegressionStopper window must be >= 1, got {window}",
            field="window",
            value=window,
            constraint="Must be >= 1",
        )
    self.window = window
    self._score_history: deque[float] = deque(maxlen=self.window + 1)

setup

setup() -> None

Reset score history. Called by engine at start of each run.

Clears the internal score history by replacing the bounded deque with a fresh one, so the stopper can be safely reused across multiple evolve() calls with the same instance. Without this reset, history from one run would bleed into the next.

Examples:

stopper = RegressionStopper()
# ... run evolution ...
stopper.setup()  # reset for next run
Source code in src/gepa_adk/adapters/stoppers/regression.py
def setup(self) -> None:
    """Reset score history. Called by engine at start of each run.

    Clears the internal score history by replacing the bounded deque with
    a fresh one, so the stopper can be safely reused across multiple
    ``evolve()`` calls with the same instance. Without this reset, history
    from one run would bleed into the next.

    Examples:
        ```python
        stopper = RegressionStopper()
        # ... run evolution ...
        stopper.setup()  # reset for next run
        ```
    """
    self._score_history = deque(maxlen=self.window + 1)

__call__

__call__(state: StopperState) -> bool

Check if evolution should stop due to score regression.

Appends the current best score to history, then compares the latest score against the score from window iterations ago. Returns False during the cold-start phase (fewer than window + 1 calls).

PARAMETER DESCRIPTION
state

Current evolution state snapshot containing best_score.

TYPE: StopperState

RETURNS DESCRIPTION
bool

True if current best score is strictly lower than the score

bool

window iterations ago, False otherwise.

Examples:

stopper = RegressionStopper(window=3)
# Prime with 3 cold-start calls
for score in [0.5, 0.6, 0.7]:
    stopper(
        StopperState(
            iteration=0,
            best_score=score,
            stagnation_counter=0,
            total_evaluations=0,
            candidates_count=1,
            elapsed_seconds=0.0,
        )
    )
state = StopperState(
    iteration=3,
    best_score=0.4,
    stagnation_counter=0,
    total_evaluations=3,
    candidates_count=1,
    elapsed_seconds=3.0,
)
stopper(state)  # True (0.4 < 0.5, the baseline from window=3 ago)
Source code in src/gepa_adk/adapters/stoppers/regression.py
def __call__(self, state: StopperState) -> bool:
    """Check if evolution should stop due to score regression.

    Appends the current best score to history, then compares the latest
    score against the score from ``window`` iterations ago. Returns
    ``False`` during the cold-start phase (fewer than ``window + 1`` calls).

    Args:
        state: Current evolution state snapshot containing best_score.

    Returns:
        True if current best score is strictly lower than the score
        ``window`` iterations ago, False otherwise.

    Examples:
        ```python
        stopper = RegressionStopper(window=3)
        # Prime with 3 cold-start calls
        for score in [0.5, 0.6, 0.7]:
            stopper(
                StopperState(
                    iteration=0,
                    best_score=score,
                    stagnation_counter=0,
                    total_evaluations=0,
                    candidates_count=1,
                    elapsed_seconds=0.0,
                )
            )
        state = StopperState(
            iteration=3,
            best_score=0.4,
            stagnation_counter=0,
            total_evaluations=3,
            candidates_count=1,
            elapsed_seconds=3.0,
        )
        stopper(state)  # True (0.4 < 0.5, the baseline from window=3 ago)
        ```
    """
    self._score_history.append(state.best_score)
    if len(self._score_history) <= self.window:
        return False
    current = self._score_history[-1]
    baseline = self._score_history[-(self.window + 1)]
    if current < baseline:
        logger.info(
            "stopper.regression.triggered",
            window=self.window,
            current_score=current,
            baseline_score=baseline,
        )
        return True
    return False