A callback that delegates to multiple child callbacks.
This allows registering multiple callbacks and having them all
receive events.
Example
composite = CompositeCallback([callback1, callback2])
optimize(..., callbacks=[composite])
Initialize with a list of callbacks.
Parameters:
| Name |
Type |
Description |
Default |
callbacks
|
list[Any] | None
|
List of callback objects. Each should implement
some or all of the GEPACallback methods.
|
None
|
Source code in gepa/core/callbacks.py
| def __init__(self, callbacks: list[Any] | None = None):
"""Initialize with a list of callbacks.
Args:
callbacks: List of callback objects. Each should implement
some or all of the GEPACallback methods.
"""
self._callbacks: list[Any] = []
# Cache: method_name -> list of (callback, bound_method) tuples
self._method_cache: dict[str, list[tuple[Any, Any]]] = {}
if callbacks:
for cb in callbacks:
self.add(cb)
|
Attributes
callbacks: list[Any]
property
Return the list of registered callbacks.
Functions
add(callback: Any) -> None
Add a callback to the composite.
Parameters:
| Name |
Type |
Description |
Default |
callback
|
Any
|
A callback object to add.
|
required
|
Source code in gepa/core/callbacks.py
| def add(self, callback: Any) -> None:
"""Add a callback to the composite.
Args:
callback: A callback object to add.
"""
self._callbacks.append(callback)
# Update cache for all known method names
for method_name in self._method_cache:
method = getattr(callback, method_name, None)
if method is not None:
self._method_cache[method_name].append((callback, method))
|
on_optimization_start(event: OptimizationStartEvent) -> None
Source code in gepa/core/callbacks.py
| def on_optimization_start(self, event: OptimizationStartEvent) -> None:
self._notify("on_optimization_start", event)
|
on_optimization_end(event: OptimizationEndEvent) -> None
Source code in gepa/core/callbacks.py
| def on_optimization_end(self, event: OptimizationEndEvent) -> None:
self._notify("on_optimization_end", event)
|
on_iteration_start(event: IterationStartEvent) -> None
Source code in gepa/core/callbacks.py
| def on_iteration_start(self, event: IterationStartEvent) -> None:
self._notify("on_iteration_start", event)
|
on_iteration_end(event: IterationEndEvent) -> None
Source code in gepa/core/callbacks.py
| def on_iteration_end(self, event: IterationEndEvent) -> None:
self._notify("on_iteration_end", event)
|
on_candidate_selected(event: CandidateSelectedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_candidate_selected(self, event: CandidateSelectedEvent) -> None:
self._notify("on_candidate_selected", event)
|
on_minibatch_sampled(event: MinibatchSampledEvent) -> None
Source code in gepa/core/callbacks.py
| def on_minibatch_sampled(self, event: MinibatchSampledEvent) -> None:
self._notify("on_minibatch_sampled", event)
|
on_evaluation_start(event: EvaluationStartEvent) -> None
Source code in gepa/core/callbacks.py
| def on_evaluation_start(self, event: EvaluationStartEvent) -> None:
self._notify("on_evaluation_start", event)
|
on_evaluation_end(event: EvaluationEndEvent) -> None
Source code in gepa/core/callbacks.py
| def on_evaluation_end(self, event: EvaluationEndEvent) -> None:
self._notify("on_evaluation_end", event)
|
on_evaluation_skipped(event: EvaluationSkippedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_evaluation_skipped(self, event: EvaluationSkippedEvent) -> None:
self._notify("on_evaluation_skipped", event)
|
on_valset_evaluated(event: ValsetEvaluatedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_valset_evaluated(self, event: ValsetEvaluatedEvent) -> None:
self._notify("on_valset_evaluated", event)
|
on_reflective_dataset_built(event: ReflectiveDatasetBuiltEvent) -> None
Source code in gepa/core/callbacks.py
| def on_reflective_dataset_built(self, event: ReflectiveDatasetBuiltEvent) -> None:
self._notify("on_reflective_dataset_built", event)
|
on_proposal_start(event: ProposalStartEvent) -> None
Source code in gepa/core/callbacks.py
| def on_proposal_start(self, event: ProposalStartEvent) -> None:
self._notify("on_proposal_start", event)
|
on_proposal_end(event: ProposalEndEvent) -> None
Source code in gepa/core/callbacks.py
| def on_proposal_end(self, event: ProposalEndEvent) -> None:
self._notify("on_proposal_end", event)
|
on_candidate_accepted(event: CandidateAcceptedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_candidate_accepted(self, event: CandidateAcceptedEvent) -> None:
self._notify("on_candidate_accepted", event)
|
on_candidate_rejected(event: CandidateRejectedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_candidate_rejected(self, event: CandidateRejectedEvent) -> None:
self._notify("on_candidate_rejected", event)
|
on_merge_attempted(event: MergeAttemptedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_merge_attempted(self, event: MergeAttemptedEvent) -> None:
self._notify("on_merge_attempted", event)
|
on_merge_accepted(event: MergeAcceptedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_merge_accepted(self, event: MergeAcceptedEvent) -> None:
self._notify("on_merge_accepted", event)
|
on_merge_rejected(event: MergeRejectedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_merge_rejected(self, event: MergeRejectedEvent) -> None:
self._notify("on_merge_rejected", event)
|
on_pareto_front_updated(event: ParetoFrontUpdatedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_pareto_front_updated(self, event: ParetoFrontUpdatedEvent) -> None:
self._notify("on_pareto_front_updated", event)
|
on_state_saved(event: StateSavedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_state_saved(self, event: StateSavedEvent) -> None:
self._notify("on_state_saved", event)
|
on_budget_updated(event: BudgetUpdatedEvent) -> None
Source code in gepa/core/callbacks.py
| def on_budget_updated(self, event: BudgetUpdatedEvent) -> None:
self._notify("on_budget_updated", event)
|
on_error(event: ErrorEvent) -> None
Source code in gepa/core/callbacks.py
| def on_error(self, event: ErrorEvent) -> None:
self._notify("on_error", event)
|