Skip to content

Composite

composite

Composite stopper for combining multiple stop conditions.

This module provides a CompositeStopper that combines multiple stoppers with configurable AND/OR logic, enabling complex stopping rules like "stop after 5 minutes OR when score >= 0.95".

ATTRIBUTE DESCRIPTION
CompositeStopper

Combine multiple stoppers with AND/OR logic.

TYPE: class

Examples:

Combining stoppers with OR logic (stop if any fires):

from gepa_adk.adapters.stoppers import (
    CompositeStopper,
    ScoreThresholdStopper,
    TimeoutStopper,
)

# Stop after 5 minutes OR when score >= 0.95
composite = CompositeStopper(
    [TimeoutStopper(300), ScoreThresholdStopper(0.95)],
    mode="any",
)

Combining stoppers with AND logic (stop only if all fire):

# Stop only when BOTH conditions met
composite = CompositeStopper(
    [TimeoutStopper(60), ScoreThresholdStopper(0.8)],
    mode="all",
)
Note

This composite stopper is useful for building complex termination policies from simpler building blocks.

CompositeStopper

Combine multiple stoppers with AND/OR logic.

A meta-stopper that composes multiple stoppers into a single stopping condition with configurable logic. Use mode='any' for OR semantics (stop if any stopper fires) or mode='all' for AND semantics (stop only if all stoppers fire).

ATTRIBUTE DESCRIPTION
stoppers

The sequence of stoppers to combine.

TYPE: list[StopperProtocol]

mode

Combination mode - 'any' for OR, 'all' for AND.

TYPE: Literal['any', 'all']

Examples:

Stop after 5 minutes OR when score reaches 95%:

from gepa_adk.adapters.stoppers import (
    CompositeStopper,
    ScoreThresholdStopper,
    TimeoutStopper,
)
from gepa_adk.domain.stopper import StopperState

composite = CompositeStopper(
    [TimeoutStopper(300), ScoreThresholdStopper(0.95)],
    mode="any",
)

state = StopperState(
    iteration=10,
    best_score=0.97,  # Exceeds threshold
    stagnation_counter=2,
    total_evaluations=50,
    candidates_count=3,
    elapsed_seconds=120.0,  # Below timeout
)

composite(state)  # Returns True (score threshold met)

Stop only when minimum time AND score threshold are both met:

composite = CompositeStopper(
    [TimeoutStopper(60), ScoreThresholdStopper(0.8)],
    mode="all",
)
Note

A composite stopper can contain other composite stoppers for arbitrarily complex stopping conditions like "(A OR B) AND (C OR D)".

Source code in src/gepa_adk/adapters/stoppers/composite.py
class CompositeStopper:
    """Combine multiple stoppers with AND/OR logic.

    A meta-stopper that composes multiple stoppers into a single stopping
    condition with configurable logic. Use mode='any' for OR semantics
    (stop if any stopper fires) or mode='all' for AND semantics (stop
    only if all stoppers fire).

    Attributes:
        stoppers (list[StopperProtocol]): The sequence of stoppers to combine.
        mode (Literal["any", "all"]): Combination mode - 'any' for OR, 'all' for AND.

    Examples:
        Stop after 5 minutes OR when score reaches 95%:

        ```python
        from gepa_adk.adapters.stoppers import (
            CompositeStopper,
            ScoreThresholdStopper,
            TimeoutStopper,
        )
        from gepa_adk.domain.stopper import StopperState

        composite = CompositeStopper(
            [TimeoutStopper(300), ScoreThresholdStopper(0.95)],
            mode="any",
        )

        state = StopperState(
            iteration=10,
            best_score=0.97,  # Exceeds threshold
            stagnation_counter=2,
            total_evaluations=50,
            candidates_count=3,
            elapsed_seconds=120.0,  # Below timeout
        )

        composite(state)  # Returns True (score threshold met)
        ```

        Stop only when minimum time AND score threshold are both met:

        ```python
        composite = CompositeStopper(
            [TimeoutStopper(60), ScoreThresholdStopper(0.8)],
            mode="all",
        )
        ```

    Note:
        A composite stopper can contain other composite stoppers for arbitrarily
        complex stopping conditions like "(A OR B) AND (C OR D)".
    """

    def __init__(
        self,
        stoppers: Sequence[StopperProtocol],
        mode: Literal["any", "all"] = "any",
    ) -> None:
        """Initialize composite stopper with child stoppers and mode.

        Args:
            stoppers: Sequence of stoppers to combine. Must contain at least
                one stopper. Each stopper must implement StopperProtocol.
            mode: Combination logic - 'any' (OR) or 'all' (AND). Defaults
                to 'any'.

        Raises:
            ValueError: If stoppers sequence is empty.
            ValueError: If mode is not 'any' or 'all'.

        Examples:
            ```python
            # OR logic - stop if either condition met
            composite = CompositeStopper(
                [TimeoutStopper(300), ScoreThresholdStopper(0.95)],
                mode="any",
            )

            # AND logic - stop only if both conditions met
            composite = CompositeStopper(
                [TimeoutStopper(60), ScoreThresholdStopper(0.8)],
                mode="all",
            )
            ```

        Note:
            Consider using 'any' mode for fail-safe conditions (timeout OR
            resource limit) and 'all' mode for minimum requirements.
        """
        if not stoppers:
            raise ValueError("At least one stopper required")
        if mode not in ("any", "all"):
            raise ValueError(f"mode must be 'any' or 'all', got {mode!r}")

        self.stoppers: list[StopperProtocol] = list(stoppers)
        self.mode: Literal["any", "all"] = mode

    def __call__(self, state: StopperState) -> bool:
        """Check if evolution should stop based on combined stopper logic.

        Args:
            state: Current evolution state snapshot passed to all child stoppers.

        Returns:
            For mode='any': True if any child stopper returns True.
            For mode='all': True only if all child stoppers return True.

        Examples:
            ```python
            composite = CompositeStopper(
                [TimeoutStopper(60), ScoreThresholdStopper(0.9)],
                mode="any",
            )

            # Below both thresholds
            state1 = StopperState(
                iteration=5,
                best_score=0.5,
                stagnation_counter=0,
                total_evaluations=25,
                candidates_count=1,
                elapsed_seconds=30.0,
            )
            composite(state1)  # False

            # Score threshold met
            state2 = StopperState(
                iteration=10,
                best_score=0.95,
                stagnation_counter=1,
                total_evaluations=50,
                candidates_count=2,
                elapsed_seconds=45.0,
            )
            composite(state2)  # True (any mode - score threshold met)
            ```

        Note:
            Often called after each iteration. For 'any' mode, evaluation
            short-circuits on first True. For 'all' mode, short-circuits on
            first False.
        """
        if self.mode == "any":
            return any(stopper(state) for stopper in self.stoppers)
        # mode == "all"
        return all(stopper(state) for stopper in self.stoppers)

    def __repr__(self) -> str:
        """Return string representation of the composite stopper.

        Returns:
            String showing the stoppers list and mode.

        Examples:
            ```python
            composite = CompositeStopper(
                [TimeoutStopper(60), ScoreThresholdStopper(0.9)],
                mode="any",
            )
            repr(composite)
            # "CompositeStopper([TimeoutStopper(...), ScoreThresholdStopper(...)], mode='any')"
            ```
        """
        return f"CompositeStopper({self.stoppers!r}, mode={self.mode!r})"

__init__

__init__(
    stoppers: Sequence[StopperProtocol],
    mode: Literal["any", "all"] = "any",
) -> None

Initialize composite stopper with child stoppers and mode.

PARAMETER DESCRIPTION
stoppers

Sequence of stoppers to combine. Must contain at least one stopper. Each stopper must implement StopperProtocol.

TYPE: Sequence[StopperProtocol]

mode

Combination logic - 'any' (OR) or 'all' (AND). Defaults to 'any'.

TYPE: Literal['any', 'all'] DEFAULT: 'any'

RAISES DESCRIPTION
ValueError

If stoppers sequence is empty.

ValueError

If mode is not 'any' or 'all'.

Examples:

# OR logic - stop if either condition met
composite = CompositeStopper(
    [TimeoutStopper(300), ScoreThresholdStopper(0.95)],
    mode="any",
)

# AND logic - stop only if both conditions met
composite = CompositeStopper(
    [TimeoutStopper(60), ScoreThresholdStopper(0.8)],
    mode="all",
)
Note

Consider using 'any' mode for fail-safe conditions (timeout OR resource limit) and 'all' mode for minimum requirements.

Source code in src/gepa_adk/adapters/stoppers/composite.py
def __init__(
    self,
    stoppers: Sequence[StopperProtocol],
    mode: Literal["any", "all"] = "any",
) -> None:
    """Initialize composite stopper with child stoppers and mode.

    Args:
        stoppers: Sequence of stoppers to combine. Must contain at least
            one stopper. Each stopper must implement StopperProtocol.
        mode: Combination logic - 'any' (OR) or 'all' (AND). Defaults
            to 'any'.

    Raises:
        ValueError: If stoppers sequence is empty.
        ValueError: If mode is not 'any' or 'all'.

    Examples:
        ```python
        # OR logic - stop if either condition met
        composite = CompositeStopper(
            [TimeoutStopper(300), ScoreThresholdStopper(0.95)],
            mode="any",
        )

        # AND logic - stop only if both conditions met
        composite = CompositeStopper(
            [TimeoutStopper(60), ScoreThresholdStopper(0.8)],
            mode="all",
        )
        ```

    Note:
        Consider using 'any' mode for fail-safe conditions (timeout OR
        resource limit) and 'all' mode for minimum requirements.
    """
    if not stoppers:
        raise ValueError("At least one stopper required")
    if mode not in ("any", "all"):
        raise ValueError(f"mode must be 'any' or 'all', got {mode!r}")

    self.stoppers: list[StopperProtocol] = list(stoppers)
    self.mode: Literal["any", "all"] = mode

__call__

__call__(state: StopperState) -> bool

Check if evolution should stop based on combined stopper logic.

PARAMETER DESCRIPTION
state

Current evolution state snapshot passed to all child stoppers.

TYPE: StopperState

RETURNS DESCRIPTION
bool

For mode='any': True if any child stopper returns True.

bool

For mode='all': True only if all child stoppers return True.

Examples:

composite = CompositeStopper(
    [TimeoutStopper(60), ScoreThresholdStopper(0.9)],
    mode="any",
)

# Below both thresholds
state1 = StopperState(
    iteration=5,
    best_score=0.5,
    stagnation_counter=0,
    total_evaluations=25,
    candidates_count=1,
    elapsed_seconds=30.0,
)
composite(state1)  # False

# Score threshold met
state2 = StopperState(
    iteration=10,
    best_score=0.95,
    stagnation_counter=1,
    total_evaluations=50,
    candidates_count=2,
    elapsed_seconds=45.0,
)
composite(state2)  # True (any mode - score threshold met)
Note

Often called after each iteration. For 'any' mode, evaluation short-circuits on first True. For 'all' mode, short-circuits on first False.

Source code in src/gepa_adk/adapters/stoppers/composite.py
def __call__(self, state: StopperState) -> bool:
    """Check if evolution should stop based on combined stopper logic.

    Args:
        state: Current evolution state snapshot passed to all child stoppers.

    Returns:
        For mode='any': True if any child stopper returns True.
        For mode='all': True only if all child stoppers return True.

    Examples:
        ```python
        composite = CompositeStopper(
            [TimeoutStopper(60), ScoreThresholdStopper(0.9)],
            mode="any",
        )

        # Below both thresholds
        state1 = StopperState(
            iteration=5,
            best_score=0.5,
            stagnation_counter=0,
            total_evaluations=25,
            candidates_count=1,
            elapsed_seconds=30.0,
        )
        composite(state1)  # False

        # Score threshold met
        state2 = StopperState(
            iteration=10,
            best_score=0.95,
            stagnation_counter=1,
            total_evaluations=50,
            candidates_count=2,
            elapsed_seconds=45.0,
        )
        composite(state2)  # True (any mode - score threshold met)
        ```

    Note:
        Often called after each iteration. For 'any' mode, evaluation
        short-circuits on first True. For 'all' mode, short-circuits on
        first False.
    """
    if self.mode == "any":
        return any(stopper(state) for stopper in self.stoppers)
    # mode == "all"
    return all(stopper(state) for stopper in self.stoppers)

__repr__

__repr__() -> str

Return string representation of the composite stopper.

RETURNS DESCRIPTION
str

String showing the stoppers list and mode.

Examples:

composite = CompositeStopper(
    [TimeoutStopper(60), ScoreThresholdStopper(0.9)],
    mode="any",
)
repr(composite)
# "CompositeStopper([TimeoutStopper(...), ScoreThresholdStopper(...)], mode='any')"
Source code in src/gepa_adk/adapters/stoppers/composite.py
def __repr__(self) -> str:
    """Return string representation of the composite stopper.

    Returns:
        String showing the stoppers list and mode.

    Examples:
        ```python
        composite = CompositeStopper(
            [TimeoutStopper(60), ScoreThresholdStopper(0.9)],
            mode="any",
        )
        repr(composite)
        # "CompositeStopper([TimeoutStopper(...), ScoreThresholdStopper(...)], mode='any')"
        ```
    """
    return f"CompositeStopper({self.stoppers!r}, mode={self.mode!r})"