Skip to content

Stopper

stopper

Stopper protocol for evolution stop conditions.

This module defines the protocol interface that all stopper implementations must follow. Stoppers are callables that receive evolution state and return whether evolution should terminate.

ATTRIBUTE DESCRIPTION
StopperProtocol

Protocol for stop condition objects.

TYPE: protocol

Examples:

Implementing a custom stopper:

from gepa_adk.ports.stopper import StopperProtocol
from gepa_adk.domain.stopper import StopperState


class MaxIterationsStopper:
    def __init__(self, max_iterations: int) -> None:
        self.max_iterations = max_iterations

    def __call__(self, state: StopperState) -> bool:
        return state.iteration >= self.max_iterations


# Runtime type checking works
stopper = MaxIterationsStopper(100)
assert isinstance(stopper, StopperProtocol)
Note

This protocol is runtime_checkable, allowing isinstance() checks to work without explicit inheritance and follows the structural typing pattern used throughout gepa-adk's ports layer.

StopperProtocol

Bases: Protocol


              flowchart TD
              gepa_adk.ports.stopper.StopperProtocol[StopperProtocol]

              

              click gepa_adk.ports.stopper.StopperProtocol href "" "gepa_adk.ports.stopper.StopperProtocol"

Protocol for stop condition objects.

A stopper is a callable that returns True when evolution should stop. Stoppers receive a StopperState snapshot of current evolution progress.

The protocol uses structural typing - any callable with the correct signature satisfies it, no explicit inheritance required.

Examples:

Class-based stopper:

from gepa_adk.ports.stopper import StopperProtocol
from gepa_adk.domain.stopper import StopperState


class TimeoutStopper:
    def __init__(self, max_seconds: float) -> None:
        self.max_seconds = max_seconds

    def __call__(self, state: StopperState) -> bool:
        return state.elapsed_seconds >= self.max_seconds


stopper = TimeoutStopper(3600.0)  # 1 hour
isinstance(stopper, StopperProtocol)  # True

Function-based stopper:

def score_threshold_stopper(state: StopperState) -> bool:
    return state.best_score >= 0.95


isinstance(score_threshold_stopper, StopperProtocol)  # True
Note

All stoppers should be pure functions of their input state - they ought not have side effects or depend on external mutable state for deterministic behavior.

Source code in src/gepa_adk/ports/stopper.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@runtime_checkable
class StopperProtocol(Protocol):
    """Protocol for stop condition objects.

    A stopper is a callable that returns True when evolution should stop.
    Stoppers receive a StopperState snapshot of current evolution progress.

    The protocol uses structural typing - any callable with the correct
    signature satisfies it, no explicit inheritance required.

    Examples:
        Class-based stopper:

        ```python
        from gepa_adk.ports.stopper import StopperProtocol
        from gepa_adk.domain.stopper import StopperState


        class TimeoutStopper:
            def __init__(self, max_seconds: float) -> None:
                self.max_seconds = max_seconds

            def __call__(self, state: StopperState) -> bool:
                return state.elapsed_seconds >= self.max_seconds


        stopper = TimeoutStopper(3600.0)  # 1 hour
        isinstance(stopper, StopperProtocol)  # True
        ```

        Function-based stopper:

        ```python
        def score_threshold_stopper(state: StopperState) -> bool:
            return state.best_score >= 0.95


        isinstance(score_threshold_stopper, StopperProtocol)  # True
        ```

    Note:
        All stoppers should be pure functions of their input state - they
        ought not have side effects or depend on external mutable state for
        deterministic behavior.
    """

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

        Args:
            state: Current evolution state snapshot containing iteration count,
                best score, stagnation counter, and other metrics.

        Returns:
            True if evolution should stop, False to continue.

        Note:
            Often this method is called after each iteration. Return True as
            soon as the stop condition is met to avoid unnecessary computation.
        """
        ...

__call__ 

__call__(state: StopperState) -> bool

Check if evolution should stop.

PARAMETER DESCRIPTION
state

Current evolution state snapshot containing iteration count, best score, stagnation counter, and other metrics.

TYPE: StopperState

RETURNS DESCRIPTION
bool

True if evolution should stop, False to continue.
Note

Often this method is called after each iteration. Return True as soon as the stop condition is met to avoid unnecessary computation.

Source code in src/gepa_adk/ports/stopper.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def __call__(self, state: StopperState) -> bool:
    """Check if evolution should stop.

    Args:
        state: Current evolution state snapshot containing iteration count,
            best score, stagnation counter, and other metrics.

    Returns:
        True if evolution should stop, False to continue.

    Note:
        Often this method is called after each iteration. Return True as
        soon as the stop condition is met to avoid unnecessary computation.
    """
    ...