Skip to main content

​
Overview

GEPA uses pluggable strategies to control two key aspects of optimization:
  1. Candidate Selection: Which candidate to mutate in each iteration
  2. Batch Sampling: Which training examples to use for reflection
These strategies allow you to customize the exploration-exploitation trade-off and data sampling behavior.

​
Candidate Selection Strategies

Candidate selectors implement the CandidateSelector protocol and determine which candidate program to select for mutation in each iteration.

​
CandidateSelector Protocol

from gepa.proposer.reflective_mutation.base import CandidateSelector
from gepa.core.state import GEPAState

class CandidateSelector(Protocol):
    def select_candidate_idx(self, state: GEPAState) -> int:
        """Select a candidate index given the current optimization state."""
        ...

​
Built-in Candidate Selectors

​
ParetoCandidateSelector

Selects candidates from the Pareto front, balancing exploration of diverse solutions. 
from gepa.strategies.candidate_selector import ParetoCandidateSelector
import random

selector = ParetoCandidateSelector(rng=random.Random(42))
​
rng
random.Random | None
default:"None"
Random number generator for reproducibility. If None, uses random.Random(0).
Behavior: Randomly selects a candidate from the current Pareto front based on per-instance dominance. This encourages diversity by exploring programs that excel on different subsets of the validation set. When to use: Best for multi-objective optimization or when you want to maintain diverse solutions.

​
CurrentBestCandidateSelector

Always selects the candidate with the highest aggregate validation score. 
from gepa.strategies.candidate_selector import CurrentBestCandidateSelector

selector = CurrentBestCandidateSelector()
Behavior: Greedily exploits the current best-performing candidate. When to use: When you want pure exploitation and rapid improvement of the best solution.

​
EpsilonGreedyCandidateSelector

Balances exploration and exploitation with epsilon-greedy selection. 
from gepa.strategies.candidate_selector import EpsilonGreedyCandidateSelector
import random

selector = EpsilonGreedyCandidateSelector(
    epsilon=0.1,
    rng=random.Random(42)
)
​
epsilon
float
required
Probability of random exploration (0.0 to 1.0)
​
rng
random.Random | None
default:"None"
Random number generator for reproducibility. If None, uses random.Random(0).
Behavior:
  • With probability epsilon: Randomly select any candidate
  • With probability 1 - epsilon: Select the current best candidate
When to use: When you want controlled exploration with mostly greedy exploitation.

​
Using Candidate Selectors

​
With optimize()

Use string shortcuts or custom instances: 
from gepa import optimize

# Using string shortcuts
result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    candidate_selection_strategy="pareto",  # or "current_best", "epsilon_greedy"
)

# Using custom instance
from gepa.strategies.candidate_selector import EpsilonGreedyCandidateSelector
import random

custom_selector = EpsilonGreedyCandidateSelector(
    epsilon=0.2,
    rng=random.Random(123)
)

result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    candidate_selection_strategy=custom_selector,
)

​
Custom Candidate Selector

Implement the protocol for custom selection logic: 
from gepa.core.state import GEPAState

class TemperatureBasedSelector:
    """Selects candidates with temperature-based sampling."""
    
    def __init__(self, temperature: float = 1.0):
        self.temperature = temperature
        self.rng = random.Random(42)
    
    def select_candidate_idx(self, state: GEPAState) -> int:
        scores = state.program_full_scores_val_set
        
        # Apply temperature scaling
        exp_scores = [math.exp(s / self.temperature) for s in scores]
        total = sum(exp_scores)
        probs = [e / total for e in exp_scores]
        
        # Sample based on probabilities
        return self.rng.choices(
            range(len(scores)),
            weights=probs
        )[0]

# Use it
result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    candidate_selection_strategy=TemperatureBasedSelector(temperature=0.5),
)

​
Batch Sampling Strategies

Batch samplers implement the BatchSampler protocol and determine which training examples to use for building reflective datasets.

​
BatchSampler Protocol

from gepa.strategies.batch_sampler import BatchSampler
from gepa.core.data_loader import DataLoader
from gepa.core.state import GEPAState

class BatchSampler(Protocol):
    def next_minibatch_ids(
        self,
        loader: DataLoader,
        state: GEPAState
    ) -> list[DataId]:
        """Sample the next minibatch of training example IDs."""
        ...

​
EpochShuffledBatchSampler

The default batch sampler that shuffles data each epoch and pads batches deterministically. 
from gepa.strategies.batch_sampler import EpochShuffledBatchSampler
import random

sampler = EpochShuffledBatchSampler(
    minibatch_size=8,
    rng=random.Random(42)
)
​
minibatch_size
int
required
Number of examples per minibatch
​
rng
random.Random | None
default:"None"
Random number generator for reproducibility. If None, uses random.Random(0).
Behavior:
  1. Shuffles all training IDs at the start of each epoch
  2. Pads the last batch with least-frequent examples to reach minibatch_size
  3. Cycles through shuffled batches deterministically based on iteration count
  4. Re-shuffles when:
    • Starting a new epoch
    • Training set size changes
    • First call
Padding Strategy: When the training set size is not divisible by minibatch_size, the sampler pads by repeating the least frequently sampled examples to maintain consistent batch sizes.

​
Using Batch Samplers

​
With optimize()

Use string shortcuts or custom instances: 
from gepa import optimize

# Using string shortcut
result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    batch_sampler="epoch_shuffled",
    reflection_minibatch_size=8,
)

# Using custom instance
from gepa.strategies.batch_sampler import EpochShuffledBatchSampler
import random

custom_sampler = EpochShuffledBatchSampler(
    minibatch_size=16,
    rng=random.Random(456)
)

result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    batch_sampler=custom_sampler,
)

​
Custom Batch Sampler

Implement the protocol for custom sampling logic: 
from gepa.strategies.batch_sampler import BatchSampler
from gepa.core.data_loader import DataLoader
from gepa.core.state import GEPAState
import random

class DynamicBatchSampler:
    """Dynamically adjusts batch size based on optimization progress."""
    
    def __init__(self, initial_size: int = 8, max_size: int = 32):
        self.initial_size = initial_size
        self.max_size = max_size
        self.rng = random.Random(0)
    
    def next_minibatch_ids(
        self,
        loader: DataLoader,
        state: GEPAState
    ) -> list[DataId]:
        # Increase batch size as optimization progresses
        progress_ratio = state.i / 100  # Assuming 100 iterations
        current_size = min(
            self.max_size,
            int(self.initial_size + progress_ratio * (self.max_size - self.initial_size))
        )
        
        all_ids = list(loader.all_ids())
        return self.rng.sample(all_ids, min(current_size, len(all_ids)))

# Use it
result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    batch_sampler=DynamicBatchSampler(initial_size=4, max_size=16),
)

​
Strategy Combinations

Combine different strategies to achieve specific optimization behaviors:

​
Aggressive Exploration

from gepa import optimize
from gepa.strategies.candidate_selector import EpsilonGreedyCandidateSelector
from gepa.strategies.batch_sampler import EpochShuffledBatchSampler
import random

result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    # High exploration
    candidate_selection_strategy=EpsilonGreedyCandidateSelector(
        epsilon=0.3,  # 30% random selection
        rng=random.Random(42)
    ),
    # Small batches for faster iteration
    batch_sampler=EpochShuffledBatchSampler(
        minibatch_size=4,
        rng=random.Random(42)
    ),
)

​
Conservative Exploitation

from gepa import optimize
from gepa.strategies.candidate_selector import CurrentBestCandidateSelector
from gepa.strategies.batch_sampler import EpochShuffledBatchSampler
import random

result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    # Pure exploitation
    candidate_selection_strategy=CurrentBestCandidateSelector(),
    # Larger batches for stability
    batch_sampler=EpochShuffledBatchSampler(
        minibatch_size=16,
        rng=random.Random(42)
    ),
)

​
Pareto-Based Multi-Objective

from gepa import optimize
from gepa.strategies.candidate_selector import ParetoCandidateSelector
import random

result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    # Select from Pareto front
    candidate_selection_strategy=ParetoCandidateSelector(
        rng=random.Random(42)
    ),
    # Instance-based frontier
    frontier_type="instance",
    reflection_minibatch_size=8,
)

​
Reproducibility

All strategies accept random number generators for reproducibility: 
import random
from gepa import optimize
from gepa.strategies.candidate_selector import EpsilonGreedyCandidateSelector
from gepa.strategies.batch_sampler import EpochShuffledBatchSampler

# Create RNG with fixed seed
rng = random.Random(42)

result = optimize(
    seed_candidate={"instructions": "..."},
    trainset=train_data,
    candidate_selection_strategy=EpsilonGreedyCandidateSelector(
        epsilon=0.1,
        rng=random.Random(42)  # Same seed
    ),
    batch_sampler=EpochShuffledBatchSampler(
        minibatch_size=8,
        rng=random.Random(42)  # Same seed
    ),
    seed=42,  # Overall optimization seed
)

​
Source Reference

The strategy implementations are in:

Build docs developers (and LLMs) love

Get started for freeTalk to us