Documentation Index Fetch the complete documentation index at: https://mintlify.com/gepa-ai/gepa/llms.txt
Use this file to discover all available pages before exploring further.
Overview GEPA provides a comprehensive callback system for observing and instrumenting optimization runs. Callbacks are synchronous, observational (cannot modify state), and receive full GEPAState access for maximum flexibility.
All callbacks implement the GEPACallback protocol and receive typed event objects containing relevant parameters.
GEPACallback Protocol The GEPACallback protocol defines optional callback methods for different optimization events. Implement only the methods you need.
from gepa.core.callbacks import GEPACallback class MyCallback : def on_optimization_start ( self , event : OptimizationStartEvent) -> None : print ( f "Starting optimization with { event[ 'trainset_size' ] } training examples" ) def on_iteration_end ( self , event : IterationEndEvent) -> None : status = 'accepted' if event[ 'proposal_accepted' ] else 'rejected' print ( f "Iteration { event[ 'iteration' ] } : { status } " )
Using Callbacks Pass callbacks to the optimize() function:
from gepa import optimize result = optimize( seed_candidate = { "instructions" : "..." }, trainset = data, callbacks = [MyCallback()], # ... other parameters )
Callback Events Each callback method receives a single event TypedDict containing all relevant parameters. This design allows easy extension without breaking changes.
Optimization Lifecycle on_optimization_start Called when optimization begins.
The initial candidate program
Number of training examples
Number of validation examples
Optimization configuration
on_optimization_end Called when optimization completes.
Index of the best candidate
Total number of iterations completed
Total evaluation calls made
Iteration Events on_iteration_start Called at the start of each iteration.
Current optimization state
on_iteration_end Called at the end of each iteration.
Current optimization state
Whether the proposed candidate was accepted
Evaluation Events on_evaluation_start Called before evaluating a candidate.
Index of candidate being evaluated
Number of examples in the evaluation batch
Whether execution traces are being captured
Whether this is the seed candidate
on_evaluation_end Called after evaluating a candidate.
Index of evaluated candidate
Whether trajectories were captured
Execution trajectories if captured
objective_scores
list[dict[str, float]] | None
Multi-objective scores if available
Whether this was the seed candidate
on_evaluation_skipped Called when an evaluation is skipped.
Index of skipped candidate
Cached scores if available
Whether this is the seed candidate
on_valset_evaluated Called after evaluating on the validation set.
Index of evaluated candidate
Validation scores by example ID
Number of validation examples evaluated
Total validation set size
Whether this is the current best program
Validation outputs by example ID
Candidate Selection Events on_candidate_selected Called when a candidate is selected for mutation.
Index of selected candidate
on_minibatch_sampled Called when a training minibatch is sampled.
Proposal Events on_reflective_dataset_built Called after building the reflective dataset.
event
ReflectiveDatasetBuiltEvent
Component names in dataset
dataset
dict[str, list[dict[str, Any]]]
Reflective dataset by component
on_proposal_start Called before proposing new instructions.
The parent candidate being mutated
reflective_dataset
dict[str, list[dict[str, Any]]]
Reflective dataset for proposal
on_proposal_end Called after proposing new instructions.
Newly proposed instructions
Acceptance Events on_candidate_accepted Called when a new candidate is accepted.
Index of accepted candidate
Score of accepted candidate
on_candidate_rejected Called when a candidate is rejected.
Rejected candidate’s score
Merge Events on_merge_attempted Called when a merge is attempted.
Indices of programs to merge
on_merge_accepted Called when a merge is accepted.
Index of merged candidate
on_merge_rejected Called when a merge is rejected.
State Events on_pareto_front_updated Called when the Pareto front is updated.
New Pareto front candidate indices
Candidates removed from front
on_state_saved Called after state is saved to disk.
Directory where state was saved
on_budget_updated Called when the evaluation budget is updated.
Metric calls used in this update
Remaining metric calls (if budget set)
Error Events on_error Called when an error occurs during optimization.
Iteration where error occurred
The exception that occurred
Whether optimization will continue
CompositeCallback The CompositeCallback class allows you to register multiple callbacks that all receive events.
from gepa.core.callbacks import CompositeCallback composite = CompositeCallback([callback1, callback2, callback3]) result = optimize( seed_candidate = { "instructions" : "..." }, trainset = data, callbacks = [composite], )
You can also add callbacks dynamically:
composite = CompositeCallback() composite.add(my_callback) composite.add(another_callback)
Example: Progress Tracking Callback from gepa.core.callbacks import ( OptimizationStartEvent, IterationEndEvent, OptimizationEndEvent, ) class ProgressCallback : def __init__ ( self ): self .best_score = float ( '-inf' ) self .iterations = 0 def on_optimization_start ( self , event : OptimizationStartEvent) -> None : print ( f "Starting optimization with:" ) print ( f " Training examples: { event[ 'trainset_size' ] } " ) print ( f " Validation examples: { event[ 'valset_size' ] } " ) def on_iteration_end ( self , event : IterationEndEvent) -> None : self .iterations += 1 if event[ 'proposal_accepted' ]: print ( f "✓ Iteration { event[ 'iteration' ] } : New candidate accepted" ) else : print ( f "✗ Iteration { event[ 'iteration' ] } : Candidate rejected" ) def on_optimization_end ( self , event : OptimizationEndEvent) -> None : print ( f " \n Optimization complete after { event[ 'total_iterations' ] } iterations" ) print ( f "Best candidate: { event[ 'best_candidate_idx' ] } " ) print ( f "Total evaluations: { event[ 'total_metric_calls' ] } " )
Source Reference The callback system is implemented in src/gepa/core/callbacks.py