class AsyncGEPAEngine(Generic[DataInst, Trajectory, RolloutOutput]): # noqa: UP046 # TypeVars are exported names shared across ports and engine
"""Async evolution engine orchestrating the GEPA loop.
This engine executes the core evolution algorithm:
1. Evaluate baseline candidate
2. For each iteration until max_iterations or convergence:
a. Generate reflective dataset from traces
b. Propose new candidate text (an empty reflection after one retry,
or a reflection that times out, records a skipped iteration and
moves on)
c. Evaluate proposal
d. Accept if improves above threshold
e. Record iteration
3. Return frozen EvolutionResult
The engine is generic over the adapter's ``DataInst``, ``Trajectory``
and ``RolloutOutput`` types.
Attributes:
adapter (AsyncGEPAAdapter): Implementation of AsyncGEPAAdapter protocol.
config (EvolutionConfig): Evolution parameters.
Examples:
Basic usage:
```python
from gepa_adk.engine import AsyncGEPAEngine
from gepa_adk.domain.models import EvolutionConfig, Candidate
engine = AsyncGEPAEngine(
adapter=my_adapter,
config=EvolutionConfig(max_iterations=50),
initial_candidate=Candidate(components={"instruction": "Be helpful"}),
batch=training_data,
)
result = await engine.run()
print(f"Final score: {result.final_score}")
```
Notes:
Avoid reusing engine instances after run() completes; a repeated
run() still starts fresh through ``_reset_run_state``.
When a seeded ``rng`` is provided, it is used for the auto-created
merge proposer. The API layer passes the same ``rng`` to candidate
selectors for full determinism across stochastic components.
Evaluation, acceptance, reuse, Pareto and merge log events carry
``candidate_id`` (``Candidate.id``) as the per-candidate correlation
key. Rows named in ``EvaluationBatch.failed_indices`` are counted per
iteration and reported on the result, and so is the token usage the
evaluated rows' trajectories report and the reflection calls the
adapter's proposer reports (``TokenRollup`` with ``evaluation`` and
``reflection`` splits). A proposal whose
``Candidate.id`` was already scored in the run is not evaluated
again; its iteration is recorded with ``skip_reason="duplicate"``.
A proposal whose ``output_schema`` text fails validation is not
evaluated; its iteration is recorded with
``skip_reason="schema_validation_failed"`` and counts toward
stagnation, so ``max_iterations``, ``patience`` and stoppers still
end the run. A retryable ``ReflectionError`` (the proposer already
retried once) is recorded the same way with
``skip_reason="reflection_error"``; a non-retryable one aborts the
run with the partial result attached to the error, and so do
``config.max_consecutive_reflection_errors`` consecutive ones with
the same cause type and text. An
``IncompleteProposalError`` (truncated reflection output) is recorded
with ``skip_reason="incomplete_proposal"`` and not evaluated.
A proposal that ``config.proposal_validator`` rejects is recorded
with ``skip_reason="proposal_rejected"`` and the validator's reason,
counts toward stagnation and is not evaluated.
A merge candidate that is already scored or has an invalid schema is
not evaluated, and the iteration that scheduled it is still recorded.
An evaluation policy takes effect only through the Pareto state that
a candidate selector creates. No policy and no selector is full
evaluation; a selector alone is full evaluation over the Pareto
state; a selector with a policy uses the policy; a policy without a
selector raises ``ConfigurationError`` at construction.
With ``config.reflection_minibatch_size`` smaller than the trainset,
a proposal runs first on a fresh seeded sample of the trainset rows
its parent has scores for, and goes on only when its mean there
beats the parent's cached scores on the same rows; otherwise its
iteration is recorded with ``skip_reason="minibatch_rejected"``.
When the valset is the trainset, a passing proposal is evaluated on
the full trainset and that batch is reused for scoring. When the
valset is separate, no full trainset pass runs: the minibatch batch
is the proposal's reflection batch, the valset is scored, and on
acceptance that batch and its trainset rows become the candidate's
cached batch, so ``trainset_score`` is the mean over those rows.
With ``config.checkpoint_path`` set, state is checkpointed after the
baseline and after every recorded iteration, and ``config.resume``
continues a run from that file (see ``_resume_from_checkpoint``).
"""
def __init__( # noqa: PLR0913, PLR0915 # public constructor; decomposition tracked in #507
self,
adapter: AsyncGEPAAdapter[DataInst, Trajectory, RolloutOutput],
config: EvolutionConfig,
initial_candidate: Candidate,
batch: list[DataInst],
valset: list[DataInst] | None = None,
candidate_selector: CandidateSelectorProtocol | None = None,
component_selector: ComponentSelectorProtocol | None = None,
evaluation_policy: EvaluationPolicyProtocol | None = None,
merge_proposer: ProposerProtocol | None = None,
rng: random.Random | None = None,
) -> None:
"""Initialize the evolution engine.
Args:
adapter (AsyncGEPAAdapter[DataInst, Trajectory, RolloutOutput]):
Implementation of AsyncGEPAAdapter protocol for evaluation and proposal
generation.
config (EvolutionConfig): Evolution parameters controlling iterations, thresholds,
and early stopping.
initial_candidate (Candidate): Starting candidate with at least one component.
batch (list[DataInst]): Trainset data instances for reflection and mutation.
valset (list[DataInst] | None): Optional validation data for scoring
candidates. Defaults to trainset when omitted. Must be non-empty if
provided.
candidate_selector (CandidateSelectorProtocol | None): Optional selector
strategy for Pareto-aware candidate sampling. When provided, initializes
ParetoState for multi-objective tracking.
component_selector (ComponentSelectorProtocol | None): Optional selector
strategy for choosing which components to update. Defaults to
RoundRobinComponentSelector.
evaluation_policy (EvaluationPolicyProtocol | None): Optional policy for
selecting which validation examples to evaluate per iteration. Defaults
to FullEvaluationPolicy. Requires ``candidate_selector``, because the
policy acts only through the Pareto state the selector creates.
merge_proposer (ProposerProtocol | None): Optional proposer for merge
operations. If provided and config.use_merge is True, merge proposals
will be attempted after successful mutations.
rng (random.Random | None): Optional seeded random.Random instance for
deterministic engine decisions. When provided, used for the auto-created
merge proposer and for drawing reflection minibatch rows. None preserves
current random behavior, and minibatch rows are then drawn from a
``random.Random(config.seed)``.
Raises:
ValueError: If batch is empty, valset is provided but empty,
or initial_candidate has no components.
ConfigurationError: If evaluation_policy is provided without a
candidate_selector, including an explicit FullEvaluationPolicy,
or if ``config.checkpoint_path`` is set together with a
candidate_selector (Pareto state is not checkpointed).
Examples:
Creating an engine:
```python
engine = AsyncGEPAEngine(
adapter=my_adapter,
config=EvolutionConfig(max_iterations=50),
initial_candidate=Candidate(components={"instruction": "Be helpful"}),
batch=training_data,
candidate_selector=selector,
)
```
Notes:
Configures trainset and valset routing for reflection and scoring.
A valset that is the same object as the batch (or omitted) marks
the engine to reuse reflection batches for scoring.
Initializes stopper lifecycle tracking for custom stop callbacks.
Initializes the evaluation and pending failure counters, the
pending and run token rollups, the pending and run reflection
rollups (None until the proposer reports one), the last proposer
rollup consumed (so an unchanged one is not recounted), the map
from each scored
candidate's id to its acceptance score, the random source for
reflection minibatch rows, the slots holding the last mutation
parent's trainset batch and the trainset rows it covers, the
slots for the batch and rows the last minibatch gate evaluated,
and the per-candidate map of cached batch rows (None for the
full trainset).
Initializes an in-memory streak of identical consecutive
reflection errors (the cause's type name and text, the count, and
the history index of the latest one), which ``run()`` resets and
``_record_reflection_error`` uses to enforce
``config.max_consecutive_reflection_errors``.
Valid selector and policy combinations: neither is full
evaluation; a selector alone is full evaluation over the Pareto
state; a selector with a policy uses that policy; a policy alone
is rejected. A checkpoint path with a candidate selector is
rejected too, and the restored-state flag starts False.
"""
# Validation
if len(batch) == 0:
raise ValueError("batch must contain at least one data instance")
if valset is not None and len(valset) == 0:
raise ValueError(
"valset must contain at least one validation data instance"
)
if not initial_candidate.components:
raise ValueError("initial_candidate must have at least one component")
if evaluation_policy is not None and candidate_selector is None:
raise ConfigurationError(
"evaluation_policy requires a candidate_selector: a policy "
"takes effect only through the Pareto state a candidate "
"selector creates",
field="evaluation_policy",
value=type(evaluation_policy).__name__,
constraint="candidate_selector is not None",
)
if config.checkpoint_path is not None and candidate_selector is not None:
raise ConfigurationError(
"checkpoint_path cannot be combined with a candidate_selector: "
"Pareto state is not checkpointed yet",
field="candidate_selector",
value=type(candidate_selector).__name__,
constraint="candidate_selector is None when checkpoint_path is set",
)
# Store dependencies
self.adapter = adapter
self.config = config
self._initial_candidate = initial_candidate
self._trainset = batch
self._valset = valset if valset is not None else batch
# Identity, not equality: an equal-but-separate valset is scored apart.
self._valset_is_trainset = self._valset is self._trainset
self._state: _EngineState | None = None
self._rng = rng
# Draws each iteration's reflection minibatch rows
self._minibatch_rng = rng if rng is not None else random.Random(config.seed) # noqa: S311 # evolution sampling, not security
# Trainset batch of the parent the last mutation reflected on, and
# the trainset rows it covers (None: the full trainset in order)
self._mutation_parent_batch: EvaluationBatch | None = None
self._mutation_parent_rows: list[int] | None = None
# The batch and rows the last minibatch gate evaluated the proposal on
self._gate_batch: EvaluationBatch | None = None
self._gate_rows: list[int] | None = None
self._candidate_selector = candidate_selector
self._component_selector = component_selector or RoundRobinComponentSelector()
self._pareto_state: ParetoState | None = None
self._candidate_eval_batches: dict[int, EvaluationBatch] = {}
# Trainset rows each cached batch covers (None: the full trainset)
self._candidate_eval_rows: dict[int, list[int] | None] = {}
# Acceptance score of every candidate scored this run, keyed by id
self._scored: dict[str, float] = {}
if merge_proposer is not None:
self._merge_proposer = merge_proposer
elif config.use_merge:
self._merge_proposer = MergeProposer(rng=rng or random.Random()) # noqa: S311 # evolution sampling, not security
else:
self._merge_proposer = None
self._merges_due: int = 0
self._merge_invocations: int = 0
# Stopper state tracking (T001, T002)
self._start_time: float | None = None
self._total_evaluations: int = 0
# Failed rows evaluated since the last record (see _count_batch)
self._pending_failed_evaluations: int = 0
# Token usage since the last record and over the run (see _count_batch)
self._pending_token_usage: TokenRollup = _ZERO_TOKENS
self._run_token_usage: TokenRollup = _ZERO_TOKENS
# Reflection usage of this iteration and of the run; None = not observed
self._pending_reflection_usage: TokenRollup | None = None
self._run_reflection_usage: TokenRollup | None = None
# Last proposer rollup consumed, so an unchanged one is not recounted
self._consumed_proposer_usage: TokenRollup | None = None
self._active_stoppers: list[object] = []
# Run of identical consecutive reflection errors (in memory only):
# the cause's (type name, text), how many in a row, and the history
# index of the latest one
self._reflection_error_signature: tuple[str, str] | None = None
self._reflection_error_streak: int = 0
self._reflection_error_index: int = -1
# True once run() restored state from a checkpoint (skips the baseline)
self._restored: bool = False
if evaluation_policy is None:
self._evaluation_policy: EvaluationPolicyProtocol = FullEvaluationPolicy()
else:
self._evaluation_policy = evaluation_policy
def _take_pending_failed_evaluations(self) -> int:
"""Return the pending failure count and reset it to zero.
Returns:
Failed rows counted by ``_count_batch`` since the last call.
"""
count = self._pending_failed_evaluations
self._pending_failed_evaluations = 0
return count
def _take_pending_token_usage(self) -> TokenRollup:
"""Return the pending token rollup and reset it to zero.
Returns:
Token usage folded in by ``_count_batch`` since the last call.
"""
usage = self._pending_token_usage
self._pending_token_usage = _ZERO_TOKENS
return usage
def _take_proposer_usage(self) -> None:
"""Store the reflection usage the adapter's proposer reported.
Reads ``last_token_usage`` from the adapter's ``_proposer`` into
``_pending_reflection_usage``, so the next record carries it. An
adapter without a proposer, or a proposer without the attribute,
leaves it None (not observed). The proposer builds a new rollup
object on every ``propose()`` and every reflection call, so a rollup
that is the same object (``is``) as the last one consumed means
``propose()`` did not run this iteration; it is treated as not
observed, and the previous iteration's reflection is not counted
again. A rollup with no counted and no unknown row means
``propose()`` ran but made no reflection call (an empty reflective
dataset), which is not observed reflection either.
"""
usage = getattr(
getattr(self.adapter, "_proposer", None), "last_token_usage", None
)
if not isinstance(usage, TokenRollup) or usage is self._consumed_proposer_usage:
self._pending_reflection_usage = None
return
if usage.rows_counted == 0 and usage.rows_unknown == 0:
self._consumed_proposer_usage = usage
self._pending_reflection_usage = None
return
self._consumed_proposer_usage = usage
self._pending_reflection_usage = usage
def _take_record_token_usage(self) -> TokenRollup:
"""Build the next record's token rollup and reset the pending usage.
Returns:
A split rollup whose ``evaluation`` is the pending evaluation
rollup and whose ``reflection`` is the pending reflection usage,
or None when none was observed. The reflection usage is also
folded into the run's reflection rollup.
"""
evaluation = self._take_pending_token_usage()
reflection = self._pending_reflection_usage
self._pending_reflection_usage = None
if reflection is not None:
self._run_reflection_usage = (
reflection
if self._run_reflection_usage is None
else self._run_reflection_usage.combine(reflection)
)
return TokenRollup.split(evaluation=evaluation, reflection=reflection)
def _run_token_rollup(self) -> TokenRollup:
"""Return the run's token rollup with its evaluation and reflection splits.
Returns:
``TokenRollup.split`` of the run's evaluation rollup and its
reflection rollup, which is None when no reflection usage was
observed.
"""
return TokenRollup.split(
evaluation=self._run_token_usage, reflection=self._run_reflection_usage
)
def _aggregate_acceptance_score(self, scores: list[float]) -> float:
"""Aggregate scores for acceptance decisions based on acceptance_metric.
Args:
scores: List of per-example scores from evaluation batch.
Returns:
Aggregated acceptance score (sum or mean based on config).
Raises:
InvalidScoreListError: If scores list is empty or contains
non-finite values.
Notes:
Sums or averages acceptance scores after validating they are
non-empty and finite. Uses sum or mean based on config.acceptance_metric.
"""
# Validate scores are non-empty
if not scores:
raise InvalidScoreListError(
"Cannot aggregate acceptance score from empty score list",
scores=scores,
reason="empty",
)
# Validate scores are finite
if not all(math.isfinite(score) for score in scores):
raise InvalidScoreListError(
"Cannot aggregate acceptance score from non-finite values (NaN/inf)",
scores=scores,
reason="non-finite",
)
# Aggregate based on acceptance_metric
if self.config.acceptance_metric == "sum":
return sum(scores)
else: # acceptance_metric == "mean"
return sum(scores) / len(scores)
def _setup_stoppers(self) -> list[object]:
"""Call setup() on stoppers that have lifecycle methods.
Returns:
List of stoppers that had setup() called (for cleanup in reverse order).
Notes:
Only invokes setup() on stoppers implementing the lifecycle method.
Stoppers that fail setup() are excluded from stop_callbacks for the
remainder of execution to prevent inconsistent state.
"""
setup_stoppers: list[object] = []
active_stoppers: list[object] = []
stop_callbacks = self.config.stop_callbacks
if stop_callbacks:
for stopper in stop_callbacks:
setup_method = getattr(stopper, "setup", None)
if setup_method is not None and callable(setup_method):
try:
setup_method()
setup_stoppers.append(stopper)
active_stoppers.append(stopper)
except Exception:
logger.exception(
"stopper.setup_error",
stopper=type(stopper).__name__,
)
# Stopper excluded from active list due to setup failure
else:
# Stopper has no setup() method, still active
active_stoppers.append(stopper)
# Store active stoppers for _should_stop() to use
self._active_stoppers = active_stoppers
return setup_stoppers
def _cleanup_stoppers(self, setup_stoppers: list[object]) -> None:
"""Call cleanup() on stoppers in reverse order of setup.
Args:
setup_stoppers: List of stoppers that had setup() called.
Notes:
Follows the reverse-order cleanup contract (T025).
When cleanup() raises, logs the error and continues (T026).
"""
for stopper in reversed(setup_stoppers):
cleanup_method = getattr(stopper, "cleanup", None)
if cleanup_method is not None and callable(cleanup_method):
try:
cleanup_method()
except Exception:
logger.exception(
"stopper.cleanup_error",
stopper=type(stopper).__name__,
)
def _build_stopper_state(self) -> StopperState:
"""Build a StopperState snapshot from current engine state.
Constructs an immutable snapshot of evolution state for stopper
callbacks to evaluate. Captures all metrics needed by stoppers
including elapsed time, total evaluations and the latest score.
Returns:
Frozen StopperState containing current iteration, best score,
stagnation counter, total evaluations, candidates count, elapsed
time, and the latest iteration's score.
Notes:
Obtains elapsed_seconds from monotonic time since run() started.
Uses zero if _start_time has not yet been set. Takes latest_score
from the last ``iteration_history`` record, accepted or not, and
leaves it ``None`` while the history is empty. Must run after
run() created the engine state, which the method asserts.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
start, history = self._start_time, self._state.iteration_history
elapsed = time.monotonic() - start if start is not None else 0.0
candidates_count = (
len(self._pareto_state.candidates) if self._pareto_state is not None else 0
)
return StopperState(
iteration=self._state.iteration,
best_score=self._state.best_score,
stagnation_counter=self._state.stagnation_counter,
total_evaluations=self._total_evaluations,
candidates_count=candidates_count,
elapsed_seconds=elapsed,
latest_score=history[-1].score if history else None,
)
@property
def pareto_state(self) -> ParetoState | None:
"""Return the current Pareto state, if initialized."""
return self._pareto_state
def _build_component_list(self, candidate: Candidate) -> list[str]:
"""Build list of available component keys from candidate.
Excludes generic 'instruction' alias if agent-specific keys exist
(e.g., 'agent1_instruction').
Args:
candidate: Candidate to extract component keys from.
Returns:
List of component keys to consider for update.
Notes:
Selects component keys, filtering out the default component name when
more specific per-agent component keys are present.
"""
keys = list(candidate.components.keys())
if len(keys) > 1 and DEFAULT_COMPONENT_NAME in keys:
# If multiple keys exist, assume default component might be an alias/proxy
# or simply one of many.
# For now, simplistic rule: if other keys exist, exclude default.
return [k for k in keys if k != DEFAULT_COMPONENT_NAME]
return keys
def _count_batch(
self, candidate: Candidate, batch: EvaluationBatch, phase: str
) -> None:
"""Count one adapter evaluation and log it under the candidate's id.
Adds the batch's rows to the evaluation counter that stoppers read
and its ``failed_indices`` to the failures pending for the next
record (the baseline or the current iteration), folds the batch's
``TokenRollup`` into the pending and the run token rollups, then logs
``evaluation.completed`` with the candidate's id, the phase, the row
count and that same failure count.
Args:
candidate: Candidate whose components were evaluated.
batch: Evaluation batch the adapter just returned.
phase: ``"reflection"`` for a trainset evaluation with traces,
``"scoring"`` for a valset evaluation.
Notes:
Called once per adapter ``evaluate()`` call, so a reused batch
is neither counted nor logged, because no row was evaluated
again. An adapter that leaves ``failed_indices`` as None reports
zero failures.
"""
failed = len(batch.failed_indices or [])
self._total_evaluations += len(batch.scores)
self._pending_failed_evaluations += failed
usage = TokenRollup.from_batch(batch)
self._pending_token_usage = self._pending_token_usage.combine(usage)
self._run_token_usage = self._run_token_usage.combine(usage)
logger.debug(
"evaluation.completed",
candidate_id=candidate.id,
phase=phase,
n=len(batch.scores),
failed=failed,
iteration=self._state.iteration if self._state is not None else 0,
)
async def _initialize_baseline(self) -> None:
"""Initialize baseline evaluation.
Evaluates the initial candidate on trainset for reflection and
on valset for scoring. Caches the reflection batch for use in
the first mutation proposal. When the valset is the trainset, the
reflection batch also serves scoring, so the baseline is evaluated
once.
Notes:
Sets up both reflection and scoring baselines up front. With a
candidate selector and objective scores, the baseline's Pareto
entry gets per-objective means (``statistics.fmean``) for the
OBJECTIVE, HYBRID and CARTESIAN frontiers, and per-example
objective scores as well for CARTESIAN. The
baseline evaluation is counted and logged via ``_count_batch``.
Rows that failed during these evaluations are stored on the state
as ``baseline_failed_evaluations``. The baseline's token usage
stays in the run rollup and is cleared from the pending one, so
the first iteration's record does not include it. The baseline's
cached batch covers the full trainset, so its rows are None on
the state and in the Pareto row map. Ends by writing
a checkpoint through ``_write_checkpoint`` when
``config.checkpoint_path`` is set.
"""
# Create pareto_state before evaluation if candidate_selector exists
# so that _evaluate_scoring can use evaluation_policy
if self._candidate_selector is not None:
self._pareto_state = ParetoState(frontier_type=self.config.frontier_type)
reflection_batch = await self.adapter.evaluate(
self._trainset,
self._initial_candidate.components,
capture_traces=True,
)
self._count_batch(self._initial_candidate, reflection_batch, "reflection")
# Use _evaluate_scoring for baseline to get eval_indices
(
baseline_score,
scoring_batch,
baseline_eval_indices,
) = await self._evaluate_scoring(
self._initial_candidate, reflection_batch=reflection_batch
)
baseline_reflection_score = sum(reflection_batch.scores) / len(
reflection_batch.scores
)
baseline_valset_mean = (
sum(scoring_batch.scores) / len(scoring_batch.scores)
if scoring_batch.scores
else 0.0
)
self._state = _EngineState(
best_candidate=self._initial_candidate,
best_score=baseline_score,
original_score=baseline_score,
iteration=0,
stagnation_counter=0,
iteration_history=[],
last_eval_batch=reflection_batch,
best_reflection_score=baseline_reflection_score,
best_valset_mean=baseline_valset_mean,
best_objective_scores=scoring_batch.objective_scores,
baseline_failed_evaluations=self._take_pending_failed_evaluations(),
)
self._take_pending_token_usage()
if self._candidate_selector is not None:
# Prepare objective scores for baseline if needed
objective_scores: dict[str, float] | None = None
per_example_objective_scores: dict[int, dict[str, float]] | None = None
if scoring_batch.objective_scores is not None:
if self.config.frontier_type in (
FrontierType.OBJECTIVE,
FrontierType.HYBRID,
):
objective_scores_by_name: dict[str, list[float]] = {}
for obj_scores in scoring_batch.objective_scores:
for obj_name, obj_score in obj_scores.items():
objective_scores_by_name.setdefault(obj_name, []).append(
obj_score
)
objective_scores = {
obj_name: fmean(scores)
for obj_name, scores in objective_scores_by_name.items()
}
if self.config.frontier_type == FrontierType.CARTESIAN:
per_example_objective_scores = {
baseline_eval_indices[i]: scoring_batch.objective_scores[i]
for i in range(len(baseline_eval_indices))
}
objective_scores_by_name: dict[str, list[float]] = {}
for obj_scores in scoring_batch.objective_scores:
for obj_name, obj_score in obj_scores.items():
objective_scores_by_name.setdefault(obj_name, []).append(
obj_score
)
objective_scores = {
obj_name: fmean(scores)
for obj_name, scores in objective_scores_by_name.items()
}
assert self._pareto_state is not None, "Pareto state not initialized" # noqa: S101 # narrows state that run() sets before this call
candidate_idx = self._pareto_state.add_candidate(
self._initial_candidate,
scoring_batch.scores,
score_indices=baseline_eval_indices,
objective_scores=objective_scores,
per_example_objective_scores=per_example_objective_scores,
logger=logger,
)
self._candidate_eval_batches[candidate_idx] = reflection_batch
self._candidate_eval_rows[candidate_idx] = None
self._write_checkpoint()
async def _evaluate_reflection(
self, candidate: Candidate
) -> tuple[float, EvaluationBatch]:
"""Evaluate a candidate on the trainset for reflection.
Args:
candidate: Candidate to evaluate.
Returns:
Tuple of (mean score across trainset examples, evaluation batch).
Notes:
Supplies trajectories for reflective dataset construction.
The evaluation is counted and logged via ``_count_batch``.
"""
eval_batch = await self.adapter.evaluate(
self._trainset,
candidate.components,
capture_traces=True,
)
self._count_batch(candidate, eval_batch, "reflection")
score = sum(eval_batch.scores) / len(eval_batch.scores)
return score, eval_batch
async def _evaluate_scoring(
self,
candidate: Candidate,
reflection_batch: EvaluationBatch | None = None,
) -> tuple[float, EvaluationBatch, list[int]]:
"""Evaluate a candidate on the valset for scoring decisions.
Args:
candidate: Candidate to evaluate on the validation set.
reflection_batch: The candidate's trainset reflection batch. When
the valset is the trainset, scoring reuses this batch (or its
entries at the selected indices) instead of calling the
adapter again.
Returns:
Tuple of (aggregated acceptance score, evaluation batch, eval_indices).
Score is aggregated using acceptance_metric (sum or mean).
eval_indices are the valset indices that were actually evaluated.
Notes:
Supplies scores without traces for acceptance decisions.
Aggregation method (sum/mean) is determined by config.acceptance_metric.
Uses evaluation_policy to determine which examples to evaluate.
Only the canonical ordered index list counts as a full evaluation;
any other selection, including a permutation, is built row by row
in the policy's order.
A reused batch adds nothing to the evaluation or failure
counters, because no example is evaluated again; its reuse log
names the ``candidate_id``. The acceptance score is stored under
the candidate's id so a later duplicate proposal is not
evaluated again.
"""
# Get indices to evaluate from evaluation policy
valset_ids = list(range(len(self._valset)))
if self._pareto_state is not None:
eval_indices = self._evaluation_policy.get_eval_batch(
valset_ids, self._pareto_state
)
else:
# Fallback to all indices if no pareto state yet
eval_indices = valset_ids
# Filter valset to only include selected indices
# Only the canonical ordered list counts as a full evaluation, so a
# policy that permutes the indices gets a batch built in its order.
is_full_eval = list(eval_indices) == valset_ids
if reflection_batch is not None and self._valset_is_trainset:
logger.debug(
"evaluation.reuse_trainset_batch",
candidate_id=candidate.id,
components=sorted(candidate.components),
reason="valset_is_trainset",
full=is_full_eval,
n=len(eval_indices),
)
eval_batch = (
reflection_batch
if is_full_eval
else _select_batch_rows(reflection_batch, eval_indices)
)
else:
eval_valset = (
self._valset
if is_full_eval
else [self._valset[i] for i in eval_indices]
)
eval_batch = await self.adapter.evaluate(
eval_valset,
candidate.components,
capture_traces=False,
)
self._count_batch(candidate, eval_batch, "scoring")
score = self._aggregate_acceptance_score(eval_batch.scores)
self._scored[candidate.id] = score
return score, eval_batch, eval_indices
async def _propose_mutation(
self,
) -> tuple[Candidate, list[str], int | None]:
"""Propose a new candidate via reflective mutation.
Uses the cached evaluation batch from the most recent best candidate
evaluation to generate the reflective dataset, avoiding redundant
adapter calls.
Returns:
Tuple of (new candidate with proposed component updates,
list of component names that were updated, Pareto index of the
parent the reflector rewrote). The index is the selector's draw;
without a selector (or when the selector found no candidate) it
is the index of the current best candidate in the Pareto state,
searched from the end, and None when there is no Pareto state or
the parent is absent from it.
Raises:
ConfigurationError: If the candidate selector returns an index
with no cached reflection batch, such as a negative index.
An empty frontier (``NoCandidateAvailableError``) is not an
error: the most recent cached batch is used instead.
Notes:
Spawns a new candidate with updated components based on reflective
dataset analysis and component selector strategy. The selected
parent is logged with its ``candidate_id``. Every candidate's
reflection batch is cached when it is added to the Pareto state,
so the selected parent is never re-evaluated. The parent's
trainset batch it reflected on, and the trainset rows that batch
covers (None for the full trainset), are kept on the engine as
``_mutation_parent_batch`` and ``_mutation_parent_rows`` for the
reflection minibatch gate. The reflective dataset is built from
whatever batch the parent has, which is its sampled minibatch
rows when it was accepted with a separate valset. The parent is drawn
once per iteration: the proposal carries ``parent_id`` set to the
parent's ``Candidate.id`` and ``generation`` set to the parent's
plus one, and the returned index is the one the loop passes to
``ParetoState.add_candidate`` as ``parent_indices``.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
assert self._state.last_eval_batch is not None, "No eval batch cached" # noqa: S101 # narrows state that run() sets before this call
selected_candidate = self._state.best_candidate
selected_idx: int | None = None
eval_batch = self._state.last_eval_batch
eval_rows = self._state.last_eval_rows
if self._candidate_selector is not None and self._pareto_state is not None:
try:
selected_idx = await self._candidate_selector.select_candidate(
self._pareto_state
)
selected_candidate = self._pareto_state.candidates[selected_idx]
eval_batch = self._candidate_eval_batches.get(selected_idx)
eval_rows = self._candidate_eval_rows.get(selected_idx)
logger.info(
"pareto_selection.mutation_parent_selected",
candidate_idx=selected_idx,
candidate_id=selected_candidate.id,
iteration=self._state.iteration,
selector_type=type(self._candidate_selector).__name__,
)
except NoCandidateAvailableError as exc:
logger.info(
"pareto_selection.empty_frontier_fallback",
iteration=self._state.iteration,
selector_type=type(self._candidate_selector).__name__,
error=str(exc),
)
eval_batch = self._state.last_eval_batch
eval_rows = self._state.last_eval_rows
if eval_batch is None:
# Every added candidate is cached at once; a miss is a bad index
raise ConfigurationError(
f"candidate selector returned index {selected_idx}, which has "
"no cached reflection batch",
field="candidate_selector",
value=selected_idx,
constraint="0 <= index < len(pareto_state.candidates)",
)
# The minibatch gate compares the proposal with this parent batch
self._mutation_parent_batch = eval_batch
self._mutation_parent_rows = eval_rows
# Build component list
available_components = self._build_component_list(selected_candidate)
# Select components to update
components_to_update = await self._component_selector.select_components(
components=available_components,
iteration=self._state.iteration,
candidate_idx=selected_idx if selected_idx is not None else 0,
)
logger.info(
"mutation.components_selected",
iteration=self._state.iteration,
components=components_to_update,
selector=type(self._component_selector).__name__,
)
# Build reflective dataset
reflective_dataset = await self.adapter.make_reflective_dataset(
selected_candidate.components,
eval_batch,
components_to_update,
)
# Propose new texts
proposed_components = await self.adapter.propose_new_texts(
selected_candidate.components,
reflective_dataset,
components_to_update,
)
# The parent's Pareto index, for the genealogy edge
parent_idx = selected_idx
if parent_idx is None and self._pareto_state is not None:
parent_idx = next(
(
idx
for idx in range(len(self._pareto_state.candidates) - 1, -1, -1)
if self._pareto_state.candidates[idx].id == selected_candidate.id
),
None,
)
# Create new candidate with proposed components and its lineage
new_components = dict(selected_candidate.components)
new_components.update(proposed_components)
return (
Candidate(
components=new_components,
generation=selected_candidate.generation + 1,
parent_id=selected_candidate.id,
),
components_to_update,
parent_idx,
)
async def _record_iteration( # noqa: PLR0913 # one argument per recorded field; private helper
self,
score: float,
component_text: str,
evolved_component: str,
accepted: bool,
objective_scores: list[dict[str, float]] | None = None,
reflection_reasoning: str | None = None,
skip_reason: str | None = None,
candidate_id: str | None = None,
parent_ids: list[str] | None = None,
rejection_reason: str | None = None,
*,
rows_evaluated: int | None = None,
) -> None:
"""Record iteration outcome and notify the ``on_iteration`` callback.
Args:
score (float): Score achieved in this iteration.
component_text (str): The text of the component that was evaluated.
evolved_component (str): The name of the component that was evolved
(e.g., "instruction", "output_schema").
accepted (bool): Whether proposal was accepted.
objective_scores (list[dict[str, float]] | None): Optional objective scores
from this iteration's evaluation. None when adapter does not provide
objective scores.
reflection_reasoning (str | None): Optional natural language reasoning from
the reflection agent explaining the mutation. None when
reasoning is not available.
skip_reason (str | None): Why the iteration produced no evaluated proposal
(e.g., ``"empty_proposal"``). None for ordinary iterations.
candidate_id (str | None): Id of the candidate the record concerns, stored on
the record and passed to the callback. None when nothing was
proposed.
parent_ids (list[str] | None): Ids of the candidates the proposal was made from,
stored on the record. None when nothing was proposed.
rejection_reason (str | None): The reason ``config.proposal_validator``
returned for a ``"proposal_rejected"`` skip, stored on the
record. None for every other iteration.
rows_evaluated (int | None): Number of rows ``score`` aggregates over, stored
on the record: the scoring batch's row count on an evaluated
record, the minibatch's on a ``"minibatch_rejected"`` skip,
0 on a skip that evaluated nothing and None on a
``"duplicate"`` skip, whose score is not this iteration's.
Defaults to None (not recorded); every engine path passes it.
Notes:
Appends an IterationRecord to ``state.iteration_history`` so the
chronological evolution trace is preserved for analysis. The
record's ``failed_evaluations`` takes the failures counted since
the iteration began, including any merge evaluation, and its
``token_usage`` the token rollup of the same evaluations (zeros
when the iteration evaluated nothing), split into ``evaluation``
and ``reflection``; ``reflection`` is None unless the proposer
reported the iteration's reflection usage. Every record
path goes through here, so ``config.on_iteration``, when set, is
called with the record and ``candidate_id`` right after the
append; an awaitable return value is awaited. Exceptions from the
callback are not caught. Ends by writing a checkpoint through
``_write_checkpoint`` when ``config.checkpoint_path`` is set.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
record = IterationRecord(
iteration_number=self._state.iteration,
score=score,
component_text=component_text,
evolved_component=evolved_component,
accepted=accepted,
objective_scores=objective_scores,
reflection_reasoning=reflection_reasoning,
skip_reason=skip_reason,
failed_evaluations=self._take_pending_failed_evaluations(),
token_usage=self._take_record_token_usage(),
candidate_id=candidate_id,
parent_ids=parent_ids,
rejection_reason=rejection_reason,
rows_evaluated=rows_evaluated,
)
self._state.iteration_history.append(record)
callback = self.config.on_iteration
if callback is not None:
outcome = callback(record, candidate_id)
if inspect.isawaitable(outcome):
await outcome
self._write_checkpoint()
async def _record_empty_proposal(self, error: EmptyProposalError) -> None:
"""Record an iteration whose reflection returned an empty proposal.
Args:
error (EmptyProposalError): The error raised while proposing; its ``component`` names
the component the reflection was working on.
Notes:
Logs ``evolution.proposal_skipped`` with
``reason="empty_proposal"``, counts the iteration toward
stagnation and appends a not-accepted IterationRecord with
``score=0.0``, empty ``component_text`` and
``skip_reason="empty_proposal"``. Nothing is evaluated, so the
record's ``rows_evaluated`` is 0. The
``on_iteration`` callback receives ``None`` as the candidate id.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
logger.debug(
"evolution.proposal_skipped",
iteration=self._state.iteration,
reason="empty_proposal",
component=error.component,
)
self._state.stagnation_counter += 1
await self._record_iteration(
score=0.0,
component_text="",
evolved_component=error.component,
accepted=False,
skip_reason="empty_proposal",
rows_evaluated=0,
)
async def _record_reflection_timeout(self, error: ReflectionTimeoutError) -> None:
"""Record an iteration whose reflection agent timed out.
Args:
error (ReflectionTimeoutError): The error raised while proposing; its
``component`` names the component the reflection was working on and its
``timeout_seconds`` the timeout it ran under.
Notes:
Logs ``evolution.proposal_skipped`` with
``reason="reflection_timeout"`` and ``timeout_seconds``, counts
the iteration toward stagnation and appends a not-accepted
IterationRecord with ``score=0.0``, empty ``component_text`` and
``skip_reason="reflection_timeout"``. Nothing is evaluated
(``rows_evaluated=0``), and accepted candidates and the Pareto state are untouched.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
logger.debug(
"evolution.proposal_skipped",
iteration=self._state.iteration,
reason="reflection_timeout",
component=error.component,
timeout_seconds=error.timeout_seconds,
)
self._state.stagnation_counter += 1
await self._record_iteration(
score=0.0,
component_text="",
evolved_component=error.component,
accepted=False,
skip_reason="reflection_timeout",
rows_evaluated=0,
)
async def _record_reflection_error(self, error: ReflectionError) -> None:
"""Record an iteration whose reflection function kept raising.
Args:
error (ReflectionError): The retryable error the proposer raised after its retry;
its ``component`` names the component the reflection was
working on, ``cause`` the provider exception and
``attempts`` the number of reflection calls made.
Notes:
Logs ``evolution.proposal_skipped`` at warning level with
``reason="reflection_error"``, the cause's ``error_type`` and
text and ``attempts``, counts the iteration toward stagnation
and appends a not-accepted IterationRecord with ``score=0.0``,
empty ``component_text`` and ``skip_reason="reflection_error"``.
Nothing is evaluated (``rows_evaluated=0``), and accepted candidates and the Pareto
state are untouched.
When the record just appended is the
``config.max_consecutive_reflection_errors``-th in a row whose
cause has the same exception type name and text, raises a
non-retryable ``ReflectionError``; ``run()`` attaches the
partial result. Any other record in between, from a successful
reflection or another skip, or a different cause, restarts the
count.
Raises:
ReflectionError: With ``retryable=False``, the last cause and its
attempts, once the identical errors reach the limit.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
error_type = type(error.cause).__name__
logger.warning(
"evolution.proposal_skipped",
iteration=self._state.iteration,
reason="reflection_error",
component=error.component,
error_type=error_type,
error=str(error.cause),
attempts=error.attempts,
)
self._state.stagnation_counter += 1
signature = (error_type, str(error.cause))
index = len(self._state.iteration_history)
if (
signature == self._reflection_error_signature
and self._reflection_error_index == index - 1
):
self._reflection_error_streak += 1
else:
self._reflection_error_signature = signature
self._reflection_error_streak = 1
self._reflection_error_index = index
await self._record_iteration(
score=0.0,
component_text="",
evolved_component=error.component,
accepted=False,
skip_reason="reflection_error",
rows_evaluated=0,
)
limit = self.config.max_consecutive_reflection_errors
if self._reflection_error_streak >= limit:
logger.error(
"evolution.reflection_errors_exhausted",
component=error.component,
error_type=error_type,
error=str(error.cause),
consecutive=self._reflection_error_streak,
)
raise ReflectionError(
error.component,
cause=error.cause,
retryable=False,
attempts=error.attempts,
) from error
async def _record_incomplete_proposal(self, error: IncompleteProposalError) -> None:
"""Record an iteration whose reflection output was incomplete.
Args:
error (IncompleteProposalError): The error the reflection function raised; its
``component`` names the component the reflection was working
on, ``finish_reason`` why the output is incomplete and
``raw_text`` the truncated text.
Notes:
Logs ``evolution.proposal_skipped`` at warning level with
``reason="incomplete_proposal"``, ``finish_reason`` and
``response_length``, counts the iteration toward stagnation and
appends a not-accepted IterationRecord with ``score=0.0``, the
truncated text as ``component_text`` and
``skip_reason="incomplete_proposal"``. Nothing is evaluated
(``rows_evaluated=0``), and accepted candidates and the Pareto state are untouched.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
logger.warning(
"evolution.proposal_skipped",
iteration=self._state.iteration,
reason="incomplete_proposal",
component=error.component,
finish_reason=error.finish_reason,
response_length=error.output_length,
)
self._state.stagnation_counter += 1
await self._record_iteration(
score=0.0,
component_text=error.raw_text,
evolved_component=error.component,
accepted=False,
skip_reason="incomplete_proposal",
rows_evaluated=0,
)
async def _record_schema_validation_skip(
self, proposal: Candidate, evolved_components: list[str]
) -> None:
"""Record an iteration whose proposed output schema failed validation.
Args:
proposal (Candidate): The proposal whose ``output_schema`` text is invalid.
evolved_components (list[str]): Names of the components the proposal evolved,
logged with the skip.
Notes:
Logs ``evolution.proposal_skipped`` with
``reason="schema_validation_failed"``, counts the iteration toward
stagnation and appends a not-accepted IterationRecord with
``score=0.0``, the invalid schema text as ``component_text``,
``evolved_component="output_schema"`` and
``skip_reason="schema_validation_failed"``. Nothing is evaluated
(``rows_evaluated=0``).
The ``on_iteration`` callback receives the proposal's id, and the
record carries it as ``candidate_id`` with ``parent_ids`` naming
the proposal's parent.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
logger.debug(
"evolution.proposal_skipped",
iteration=self._state.iteration,
reason="schema_validation_failed",
candidate_id=proposal.id,
components=evolved_components,
)
self._state.stagnation_counter += 1
await self._record_iteration(
score=0.0,
component_text=proposal.components["output_schema"],
evolved_component="output_schema",
accepted=False,
skip_reason="schema_validation_failed",
rows_evaluated=0,
candidate_id=proposal.id,
parent_ids=_parent_ids(proposal),
)
async def _record_if_rejected(
self, proposal: Candidate, evolved_components: list[str]
) -> bool:
"""Record a skipped iteration when the proposal validator rejects it.
Args:
proposal (Candidate): The candidate proposed this iteration.
evolved_components (list[str]): Names of the components evolved this
iteration; each is passed to the validator in order. When
empty, every component of the proposal is checked.
Returns:
True when ``config.proposal_validator`` returned a reason for a
component and the iteration was recorded as rejected; False when
no validator is set or it accepted every component.
Notes:
The validator is called with the component name and the proposed
text; the first non-None return is the reason. A rejection logs
``evolution.proposal_skipped`` at warning level with
``reason="proposal_rejected"`` and the ``rejection_reason``,
counts toward stagnation and appends a not-accepted
IterationRecord with ``score=0.0``, the rejected text, the
proposal's ``candidate_id`` and ``parent_ids`` and the reason.
No adapter call is made, so the record's ``rows_evaluated`` is
0. Exceptions from the validator are not caught and propagate
out of ``run()``.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
validator = self.config.proposal_validator
if validator is None:
return False
names = evolved_components or list(proposal.components)
for name in names:
text = proposal.components[name]
reason = validator(name, text)
if reason is None:
continue
logger.warning(
"evolution.proposal_skipped",
iteration=self._state.iteration,
reason="proposal_rejected",
rejection_reason=reason,
component=name,
candidate_id=proposal.id,
)
self._state.stagnation_counter += 1
await self._record_iteration(
score=0.0,
component_text=text,
evolved_component=name,
accepted=False,
skip_reason="proposal_rejected",
rows_evaluated=0,
candidate_id=proposal.id,
parent_ids=_parent_ids(proposal),
rejection_reason=reason,
)
return True
return False
async def _record_if_duplicate(
self, proposal: Candidate, evolved_components: list[str]
) -> bool:
"""Record a skipped iteration when the proposal was already scored.
Args:
proposal (Candidate): The candidate proposed this iteration.
evolved_components (list[str]): Names of the components evolved this
iteration; the first one names the record's component.
Returns:
True when the proposal's id was already scored and the iteration
was recorded as a duplicate; False when it must be evaluated.
Notes:
A duplicate logs ``proposal.duplicate`` with the stored
acceptance score, counts toward stagnation and appends a
not-accepted IterationRecord carrying that score, the proposal's
text for the evolved component and ``skip_reason="duplicate"``.
No adapter call is made. The record's ``rows_evaluated`` is None
because its score comes from an earlier evaluation. The
``on_iteration`` callback receives the duplicate proposal's id,
and the record carries it as
``candidate_id`` with ``parent_ids`` naming the proposal's parent.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
score = self._scored.get(proposal.id)
if score is None:
return False
component = (
evolved_components[0]
if evolved_components
else next(iter(proposal.components))
)
logger.info(
"proposal.duplicate",
iteration=self._state.iteration,
candidate_id=proposal.id,
score=score,
)
self._state.stagnation_counter += 1
await self._record_iteration(
score=score,
component_text=proposal.components.get(component, ""),
evolved_component=component,
accepted=False,
skip_reason="duplicate",
candidate_id=proposal.id,
parent_ids=_parent_ids(proposal),
rows_evaluated=None,
)
return True
def _effective_minibatch_size(self) -> int | None:
"""Return the reflection minibatch size in effect for this run.
Returns:
``min(reflection_minibatch_size, len(trainset))`` when that is
smaller than the trainset, otherwise None, meaning every proposal
is evaluated on the full trainset.
"""
size = self.config.reflection_minibatch_size
if size is None:
return None
k = min(size, len(self._trainset))
return k if k < len(self._trainset) else None
async def _gate_on_minibatch(
self, proposal: Candidate, evolved_components: list[str]
) -> bool:
"""Decide whether a proposal earns its full evaluation.
Samples trainset rows from the rows the mutation parent has scores
for, evaluates the proposal on them with traces, and compares the
mean of its scores with the mean of the parent's cached scores on the
same rows.
Args:
proposal (Candidate): The candidate proposed this iteration.
evolved_components (list[str]): Names of the components evolved this
iteration; the first one names the record's component.
Returns:
True when the minibatch is disabled or the proposal's mean is
strictly greater than the parent's; False when the proposal was
rejected and its iteration recorded.
Notes:
The parent's known rows are ``_mutation_parent_rows``, or every
trainset row when that is None. When the parent has at most
``k`` known rows, all of them are used (sorted) and the seeded
random source is not drawn from; otherwise ``k`` of them are
drawn with it and sorted. Each sampled trainset index is mapped
to its position in the parent's batch, so the comparison is
always on rows both sides scored. The evaluated batch and its
trainset indices are stored as ``_gate_batch`` and
``_gate_rows`` (both None when the minibatch is disabled), so the
loop can reuse them as the proposal's reflection batch.
The minibatch evaluation is counted through ``_count_batch`` as a
``"reflection"`` evaluation. A rejection logs
``proposal.minibatch_rejected``, counts toward stagnation and
appends a not-accepted IterationRecord whose score is the
acceptance aggregate over the minibatch rows, whose
``rows_evaluated`` is the number of minibatch rows and whose
``skip_reason`` is ``"minibatch_rejected"``, with the proposal's
``candidate_id`` and its parent in ``parent_ids``. A rejected proposal
is not stored as scored, so an identical later proposal draws a
fresh sample. A pass logs ``proposal.minibatch_passed``. Both
log events carry ``parent_rows``, the number of rows the parent
has scores for. A full parent batch is sampled from ``range(n)``
and read by trainset row, so no per-iteration row list or
position map is built for it; only a sampled parent batch maps
row to position.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
self._gate_batch = None
self._gate_rows = None
k = self._effective_minibatch_size()
if k is None:
return True
parent_batch = self._mutation_parent_batch
assert parent_batch is not None, "No parent batch cached" # noqa: S101 # set by the mutation step that precedes the gate
# A full batch is indexed by trainset row, so no position map or
# row list is built for it; a sampled batch maps row to position.
known: list[int] | range = (
self._mutation_parent_rows
if self._mutation_parent_rows is not None
else range(len(self._trainset))
)
position = (
{row: pos for pos, row in enumerate(known)}
if self._mutation_parent_rows is not None
else None
)
indices = (
sorted(known)
if len(known) <= k
else sorted(self._minibatch_rng.sample(known, k))
)
minibatch = await self.adapter.evaluate(
[self._trainset[i] for i in indices],
proposal.components,
capture_traces=True,
)
self._count_batch(proposal, minibatch, "reflection")
self._gate_batch = minibatch
self._gate_rows = indices
proposal_mean = sum(minibatch.scores) / len(minibatch.scores)
parent_mean = sum(
parent_batch.scores[i if position is None else position[i]] for i in indices
) / len(indices)
fields = {
"iteration": self._state.iteration,
"candidate_id": proposal.id,
"indices": indices,
"parent_rows": len(known),
"proposal_mean": proposal_mean,
"parent_mean": parent_mean,
}
if proposal_mean > parent_mean:
logger.info("proposal.minibatch_passed", **fields)
return True
logger.info("proposal.minibatch_rejected", **fields)
component = (
evolved_components[0]
if evolved_components
else next(iter(proposal.components))
)
self._state.stagnation_counter += 1
await self._record_iteration(
score=self._aggregate_acceptance_score(minibatch.scores),
component_text=proposal.components.get(component, ""),
evolved_component=component,
accepted=False,
skip_reason="minibatch_rejected",
candidate_id=proposal.id,
parent_ids=_parent_ids(proposal),
rows_evaluated=len(minibatch.scores),
)
return False
async def _evaluate_passed_proposal(
self, proposal: Candidate
) -> tuple[float, EvaluationBatch, list[int] | None]:
"""Build the reflection batch of a proposal that passed the gate.
Args:
proposal: The candidate that passed the minibatch gate (or ran
with the minibatch disabled).
Returns:
Tuple of (mean score over the batch, the batch, the trainset
indices its rows cover or None for the full trainset in order).
Notes:
With a reflection minibatch in effect and a valset separate from
the trainset, the gate's batch (``_gate_batch``, evaluated with
traces) is the reflection batch and no full trainset pass runs;
its rows are ``_gate_rows``. Otherwise the proposal is evaluated
on the full trainset through ``_evaluate_reflection``, so a
valset that is the trainset can reuse that batch for scoring.
"""
if (
self._effective_minibatch_size() is not None
and not self._valset_is_trainset
and self._gate_batch is not None
):
batch = self._gate_batch
return sum(batch.scores) / len(batch.scores), batch, self._gate_rows
score, batch = await self._evaluate_reflection(proposal)
return score, batch, None
def _start_merge(self, merge_result: ProposalResult) -> bool:
"""Count a merge attempt and decide whether to evaluate it.
Args:
merge_result (ProposalResult): The merge proposer's result for this iteration.
Returns:
True when the merge candidate should be evaluated; False when it
is skipped for an invalid schema or an already scored id.
Notes:
Decrements ``merges_due``, increments the invocation count and
logs ``merge_scheduling.merge_attempted`` for every attempt,
skipped or not.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
self._merges_due -= 1
self._merge_invocations += 1
logger.info(
"merge_scheduling.merge_attempted",
iteration=self._state.iteration,
candidate_id=merge_result.candidate.id,
parent_indices=merge_result.parent_indices,
ancestor_idx=merge_result.metadata.get("ancestor_idx"),
merges_due=self._merges_due,
total_invocations=self._merge_invocations,
)
return not self._skip_merge_candidate(merge_result.candidate)
def _skip_merge_candidate(self, candidate: Candidate) -> bool:
"""Decide whether a merge candidate must not be evaluated.
Args:
candidate (Candidate): The candidate produced by the merge proposer.
Returns:
True when the candidate has an invalid output schema or its id
was already scored; False when it should be evaluated.
Notes:
Logs ``merge.proposal_skipped`` with
``reason="schema_validation_failed"`` or ``reason="duplicate"``.
No iteration record is written for a skipped merge; the
iteration's own proposal record is unaffected.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
if not self._validate_schema_component(candidate):
reason = "schema_validation_failed"
elif candidate.id in self._scored:
reason = "duplicate"
else:
return False
logger.debug(
"merge.proposal_skipped",
iteration=self._state.iteration,
candidate_id=candidate.id,
reason=reason,
)
return True
def _should_stop(self) -> StopReason | None:
"""Check if evolution should terminate.
Returns:
The ``StopReason`` that triggered termination, or ``None`` if
evolution should continue. Conditions are checked in priority
order: max iterations, early stopping patience, custom stoppers.
Notes:
Only active stoppers (those that passed setup or have no setup
method) are invoked. Patience-based early stopping returns
``PATIENCE`` and is off when ``config.patience`` is 0;
``MAX_ITERATIONS`` means only the iteration cap.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
# Condition 1: Max iterations reached (built-in, fast path)
if self._state.iteration >= self.config.max_iterations:
return StopReason.MAX_ITERATIONS
# Condition 2: Early stopping (patience exhausted, built-in)
if (
self.config.patience > 0
and self._state.stagnation_counter >= self.config.patience
):
return StopReason.PATIENCE
# Condition 3: Custom stoppers (T010-T013)
# Use _active_stoppers which excludes stoppers that failed setup
active_stoppers = getattr(self, "_active_stoppers", None)
if active_stoppers:
stopper_state = self._build_stopper_state()
for stopper in active_stoppers:
try:
if stopper(stopper_state):
# Log stopper trigger (T013)
logger.info(
"stopper.triggered",
stopper=type(stopper).__name__,
iteration=self._state.iteration,
)
return StopReason.STOPPER_TRIGGERED
except Exception:
# T032: Handle stopper exception gracefully
logger.exception(
"stopper.error",
stopper=type(stopper).__name__,
iteration=self._state.iteration,
)
# Continue checking other stoppers
return None
def _should_accept(self, proposal_score: float, best_score: float) -> bool:
"""Check if proposal should be accepted.
Args:
proposal_score: Score of the proposed candidate.
best_score: Current best score.
Returns:
True if proposal_score > best_score + min_improvement_threshold.
Notes:
Signals True when proposal exceeds best score by the configured
improvement threshold, enabling configurable acceptance sensitivity.
"""
threshold = self.config.min_improvement_threshold
return proposal_score > best_score + threshold
def _validate_schema_component(self, proposal: Candidate) -> bool:
"""Validate output_schema component if present.
Validates that proposed schema text is syntactically correct and
structurally valid (inherits from BaseModel, no imports/functions).
Invalid schemas are rejected to prevent evolution from accepting
non-functional schema proposals.
Args:
proposal (Candidate): Candidate containing components to validate.
Returns:
True if valid or no output_schema component present.
False if output_schema validation fails.
Notes:
This validation runs before expensive evaluation to reject
invalid schemas early. Security checks (no imports, no functions)
are enforced to prevent code injection.
"""
if "output_schema" not in proposal.components:
return True
schema_text = proposal.components["output_schema"]
try:
validate_schema_text(schema_text)
except SchemaValidationError as e:
logger.warning(
"schema_validation.rejected",
iteration=self._state.iteration if self._state else None,
validation_stage=e.validation_stage,
line_number=e.line_number,
error=e.validation_error,
)
return False
else:
logger.debug(
"schema_validation.passed",
iteration=self._state.iteration if self._state else None,
)
return True
def _accept_proposal( # noqa: PLR0913 # optional fields are keyword-only; private helper
self,
proposal: Candidate,
score: float,
eval_batch: EvaluationBatch,
*,
candidate_idx: int | None = None,
eval_rows: list[int] | None = None,
reflection_score: float | None = None,
valset_mean: float | None = None,
objective_scores: list[dict[str, float]] | None = None,
) -> None:
"""Accept a proposal and update state.
Args:
proposal (Candidate): Proposed candidate to accept.
score (float): Acceptance score of the proposed candidate (sum or mean).
eval_batch (EvaluationBatch): Reflection batch from proposal evaluation (cached for
next iteration's reflective dataset generation).
candidate_idx (int | None): Optional ParetoState candidate index to update with
lineage metadata.
eval_rows (list[int] | None): Trainset indices the rows of ``eval_batch`` cover, in
batch order; None means the full trainset in order.
reflection_score (float | None): Optional trainset score to store with best
candidate metadata: the mean over ``eval_batch``.
valset_mean (float | None): Optional valset mean score to track separately from
acceptance score.
objective_scores (list[dict[str, float]] | None): Optional objective scores
from scoring batch. None when adapter does not provide objective scores.
Notes:
Replaces the cached reflection batch and its trainset rows for
the next proposal iteration and tracks acceptance score and valset mean separately.
The proposal is stored as it is: a mutation already carries its
parent's id and generation from ``_propose_mutation``, and a
merge candidate keeps the lineage the merge proposer set.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
if candidate_idx is not None and self._pareto_state is not None:
self._pareto_state.candidates[candidate_idx] = proposal
self._state.best_candidate = proposal
self._state.best_score = score
self._state.stagnation_counter = 0
self._state.last_eval_batch = eval_batch
self._state.last_eval_rows = eval_rows
if reflection_score is not None:
self._state.best_reflection_score = reflection_score
if valset_mean is not None:
self._state.best_valset_mean = valset_mean
self._state.best_objective_scores = objective_scores
def _build_result(
self, stop_reason: StopReason = StopReason.COMPLETED
) -> EvolutionResult:
"""Build final result from current state.
Args:
stop_reason (StopReason): Why the evolution run terminated.
Returns:
Frozen EvolutionResult with all metrics and original_components.
Notes:
Synthesizes a frozen EvolutionResult containing all evolution metrics,
history, and original_components snapshot, suitable for immutable
result reporting. The evolved_components dict contains all component
values from the best candidate. ``total_failed_evaluations`` is
the baseline count plus the sum over ``iteration_history``, and
``token_usage`` is the run token rollup, baseline included,
with its ``evaluation`` split and a ``reflection`` split that is
None when no reflection usage was observed.
"""
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
baseline_failed = self._state.baseline_failed_evaluations
return EvolutionResult(
stop_reason=stop_reason,
original_score=self._state.original_score,
final_score=self._state.best_score,
evolved_components=dict(self._state.best_candidate.components),
iteration_history=self._state.iteration_history,
total_iterations=self._state.iteration,
valset_score=self._state.best_valset_mean,
trainset_score=self._state.best_reflection_score,
objective_scores=self._state.best_objective_scores,
original_components=dict(self._initial_candidate.components),
baseline_failed_evaluations=baseline_failed,
total_failed_evaluations=baseline_failed
+ sum(r.failed_evaluations for r in self._state.iteration_history),
token_usage=self._run_token_rollup(),
)
def _write_checkpoint(self) -> None:
"""Write the engine's state to ``config.checkpoint_path``.
Does nothing when no checkpoint path is configured. Otherwise
builds the checkpoint dict (version, iteration, stagnation counter,
best candidate and scores, history, scored map, evaluation counter,
best reflection batch, random states and run identity) and writes it
atomically through ``write_checkpoint``.
Raises:
TypeError: If a batch field holds a value JSON cannot encode.
OSError: If the file cannot be written. Nothing is swallowed.
Notes:
``rng_state`` is the engine ``rng``'s state when one was given,
else null; ``minibatch_rng_state`` is always written.
``valset_size`` is null when the valset is the trainset.
``last_eval_rows`` holds the trainset indices the cached batch
covers, null for the full trainset.
``run_token_usage`` is the run's token rollup so far, with its
``evaluation`` and ``reflection`` splits, so a resumed run
reports the whole run's usage.
"""
if self.config.checkpoint_path is None or self._state is None:
return
path = Path(self.config.checkpoint_path)
state = self._state
data = {
"checkpoint_version": CHECKPOINT_VERSION,
"iteration": state.iteration,
"stagnation_counter": state.stagnation_counter,
"best_candidate": state.best_candidate.to_dict(),
"best_score": state.best_score,
"original_score": state.original_score,
"best_reflection_score": state.best_reflection_score,
"best_valset_mean": state.best_valset_mean,
"best_objective_scores": state.best_objective_scores,
"baseline_failed_evaluations": state.baseline_failed_evaluations,
"iteration_history": [r.to_dict() for r in state.iteration_history],
"scored": dict(self._scored),
"total_evaluations": self._total_evaluations,
"last_eval_batch": batch_to_dict(state.last_eval_batch),
"last_eval_rows": state.last_eval_rows,
"rng_state": (
None if self._rng is None else rng_state_to_json(self._rng.getstate())
),
"minibatch_rng_state": rng_state_to_json(self._minibatch_rng.getstate()),
"initial_candidate_id": self._initial_candidate.id,
"trainset_size": len(self._trainset),
"valset_size": None if self._valset_is_trainset else len(self._valset),
"seed": self.config.seed,
"run_token_usage": self._run_token_rollup().to_dict(),
"written_at": datetime.now(UTC).isoformat(),
}
write_checkpoint(path, data)
logger.debug("checkpoint.written", iteration=state.iteration, path=str(path))
def _load_checkpoint(self) -> dict[str, Any]:
"""Read the checkpoint and refuse one from a different run.
Returns:
The checkpoint dict.
Raises:
ConfigurationError: If ``config.checkpoint_path`` is unset, the
file is missing, its version is not ``CHECKPOINT_VERSION``,
or its initial candidate, trainset size or valset size
differs from this engine's.
"""
path = self.config.checkpoint_path
if path is None:
raise ConfigurationError(
"resume=True requires checkpoint_path",
field="checkpoint_path",
value=None,
constraint="not None when resume=True",
)
data = read_checkpoint(Path(path))
check_run_matches(
data,
initial_candidate_id=self._initial_candidate.id,
trainset_size=len(self._trainset),
valset_size=None if self._valset_is_trainset else len(self._valset),
)
return data
def _resume_from_checkpoint(self) -> None:
"""Restore engine state from ``config.checkpoint_path``.
Rebuilds ``_EngineState``, the scored map, the evaluation counter,
the run token rollup, the mutation parent batch and the random
states from the file, and marks the engine restored so the loop
skips the baseline.
Raises:
ConfigurationError: If ``_load_checkpoint`` refuses the file.
Notes:
The engine ``rng`` is restored only when one was given and a
state was stored; the minibatch random source always is. A
checkpoint without ``run_token_usage`` restores an unknown
rollup whose ``rows_unknown`` is the checkpointed evaluation
count, so the pre-resume rows read as unknown, not free. A
checkpoint without ``last_eval_rows`` restores the cached batch
as covering the full trainset; ``_propose_mutation`` reads the
parent's rows from the restored state. A
stored rollup without splits restores as evaluation usage with
reflection not observed; a split one restores its ``evaluation``
split (zeros when absent) as evaluation and its ``reflection``
split as reflection, so reflection usage never enters the
evaluation total. The
``on_iteration`` callback is not called for restored history.
Logs ``checkpoint.resumed`` with the iteration, the stagnation
counter, the evaluation counter and the path.
"""
data = self._load_checkpoint()
last_eval_batch = batch_from_dict(data["last_eval_batch"])
# Absent in older files: the cached batch is the full trainset
last_eval_rows = data.get("last_eval_rows")
self._state = _EngineState(
best_candidate=Candidate.from_dict(data["best_candidate"]),
best_score=data["best_score"],
original_score=data["original_score"],
iteration=data["iteration"],
stagnation_counter=data["stagnation_counter"],
iteration_history=[
IterationRecord.from_dict(r) for r in data["iteration_history"]
],
last_eval_batch=last_eval_batch,
last_eval_rows=last_eval_rows,
best_reflection_score=data["best_reflection_score"],
best_valset_mean=data["best_valset_mean"],
best_objective_scores=data["best_objective_scores"],
baseline_failed_evaluations=data["baseline_failed_evaluations"],
)
self._scored = dict(data["scored"])
self._total_evaluations = data["total_evaluations"]
stored_usage = data.get("run_token_usage")
loaded = (
TokenRollup.from_dict(stored_usage)
if stored_usage is not None
else _unknown_tokens(data["total_evaluations"])
)
has_split = loaded.evaluation is not None or loaded.reflection is not None
self._run_token_usage = (
(loaded.evaluation or _ZERO_TOKENS) if has_split else loaded
)
self._run_reflection_usage = loaded.reflection
self._mutation_parent_batch = last_eval_batch
if self._rng is not None and data["rng_state"] is not None:
self._rng.setstate(rng_state_from_json(data["rng_state"]))
self._minibatch_rng.setstate(rng_state_from_json(data["minibatch_rng_state"]))
self._restored = True
logger.info(
"checkpoint.resumed",
iteration=self._state.iteration,
stagnation_counter=self._state.stagnation_counter,
total_evaluations=self._total_evaluations,
path=str(self.config.checkpoint_path),
)
def _reset_run_state(self) -> None:
"""Clear per-run state so a repeated ``run()`` starts fresh.
Resets the stopper timer, the evaluation and pending failure
counters, the pending and run token and reflection rollups, the
reflection error streak, the scored-candidate map, the engine state
and the restored flag.
"""
# Initialize stopper state tracking (T004)
self._start_time = time.monotonic()
self._total_evaluations = 0
self._pending_failed_evaluations = 0
self._pending_token_usage = self._run_token_usage = _ZERO_TOKENS
self._pending_reflection_usage = self._run_reflection_usage = None
self._reflection_error_signature = None
self._reflection_error_streak = 0
self._reflection_error_index = -1
# A fresh run scores every candidate again; stale ids would read
# first-iteration proposals as duplicates.
self._scored.clear()
# State from an earlier run must not leak into this run's partial
# result; the baseline or the checkpoint rebuilds it.
self._state = None
self._restored = False
async def run(self) -> EvolutionResult:
"""Execute the evolution loop.
Runs the core evolution loop:
1. Evaluate baseline candidate
2. For each iteration until max_iterations or convergence:
a. Generate reflective dataset from traces
b. Propose new candidate text
c. Evaluate proposal
d. Accept if improves above threshold
e. Record iteration
3. Return frozen EvolutionResult
4. On ``KeyboardInterrupt`` or ``asyncio.CancelledError``, return
partial result with best-so-far components
5. On an ``EvolutionError`` after the baseline, attach a partial
result to the error and re-raise it
Returns:
EvolutionResult containing evolution metrics and components.
On normal completion, ``stop_reason`` reflects the termination
condition. On interrupt, a partial result is returned with
``StopReason.KEYBOARD_INTERRUPT`` or ``StopReason.CANCELLED``.
Raises:
KeyboardInterrupt: Re-raised if interrupt occurs before baseline
evaluation completes (no meaningful partial result possible).
asyncio.CancelledError: Re-raised if cancellation occurs before
baseline evaluation completes.
EvolutionError: Re-raised when it aborts the run, for example a
non-retryable ``ReflectionError`` (also raised after
``config.max_consecutive_reflection_errors`` identical
reflection errors in a row). When the baseline was
scored, its ``partial_result`` holds an ``EvolutionResult``
with ``StopReason.ERROR``, the recorded iterations and the
best candidate so far; otherwise it stays ``None``.
Exception: Adapter ``Exception`` subclasses propagate unchanged.
Examples:
Running evolution:
```python
result = await engine.run()
print(f"Improved: {result.improved}")
print(f"Best score: {result.final_score}")
```
Notes:
Outputs a frozen EvolutionResult after completing the evolution
loop. Engine instance should not be reused after run() completes.
Method is idempotent if called multiple times (restarts fresh):
the evaluation counters and the scored-candidate map used for
duplicate detection are cleared before the baseline runs.
Fail-fast behavior: adapter ``Exception`` subclasses propagate
unchanged. ``KeyboardInterrupt`` and ``asyncio.CancelledError``
(``BaseException`` subclasses) are caught and converted to partial
results with appropriate ``StopReason``. An ``EvolutionError`` is
logged as ``evolution.aborted`` at error level, without a
traceback, and re-raised; after the
baseline it carries ``partial_result`` whose
``total_iterations`` counts recorded iterations only, since the
aborted iteration was never recorded. Logs seed value at start
for reproducibility tracking, and logs
``reflection.minibatch.enabled`` when a reflection minibatch
smaller than the trainset is in effect; with a separate valset a
proposal that passes that gate is scored on the valset without a
full trainset pass, and ``trainset_score`` is then the mean over
the accepted candidate's sampled rows. ``_reset_run_state``
resets the evaluation counter, the pending failure counter, the
engine state, the pending and run token rollups and the pending
and run reflection rollups at start, so a reused engine
never attaches an earlier run's state to ``partial_result``.
With ``config.resume`` it then restores
the checkpoint through ``_resume_from_checkpoint``, which sets
the counters and scored map from the file, and the loop skips
the baseline. An exception raised by
``config.proposal_validator`` is not an ``EvolutionError``, so it
propagates unchanged without a partial result; a rejection it
returns is recorded with ``skip_reason="proposal_rejected"``.
"""
logger.info("engine.start", seed=self.config.seed)
if self._effective_minibatch_size() is not None:
logger.info(
"reflection.minibatch.enabled",
size=self.config.reflection_minibatch_size,
trainset_size=len(self._trainset),
)
self._reset_run_state()
if self.config.resume:
self._resume_from_checkpoint()
# Setup stopper lifecycle (T023)
setup_stoppers = self._setup_stoppers()
try:
return await self._run_evolution_loop()
except EvolutionError as error:
logger.error( # noqa: TRY400 # re-raised below with partial_result attached
"evolution.aborted",
iteration=self._state.iteration if self._state else 0,
error_type=type(error).__name__,
error=str(error),
)
if self._state is not None:
# The aborted iteration was never recorded
self._state.iteration = len(self._state.iteration_history)
error.partial_result = self._build_result(stop_reason=StopReason.ERROR)
raise
except KeyboardInterrupt:
logger.info(
"evolution.interrupted",
iteration=self._state.iteration if self._state else 0,
)
if self._state is None:
raise
return self._build_result(stop_reason=StopReason.KEYBOARD_INTERRUPT)
except asyncio.CancelledError:
logger.info(
"evolution.cancelled",
iteration=self._state.iteration if self._state else 0,
)
if self._state is None:
raise
return self._build_result(stop_reason=StopReason.CANCELLED)
finally:
# Cleanup stopper lifecycle (T024)
self._cleanup_stoppers(setup_stoppers)
async def _run_evolution_loop(self) -> EvolutionResult: # noqa: C901, PLR0912, PLR0915 # decomposition tracked in #507
"""Run propose-evaluate-accept iterations until a stop condition holds.
This method contains the actual evolution loop logic, separated
from lifecycle management for clean try/finally handling.
Returns:
EvolutionResult with evolution outcomes.
Raises:
ReflectionError: If proposing raises a non-retryable
``ReflectionError``, or after
``config.max_consecutive_reflection_errors`` identical
retryable ones in a row; ``run()`` attaches the partial result.
Notes:
Only called from run(). Handles the evolution loop body
while run() manages stopper lifecycle. The loop tracks
``StopReason`` to report why evolution terminated.
The parent index ``_propose_mutation`` returns is the proposal's
only genealogy edge in the Pareto state; the selector is not
asked a second time.
Each iteration records ``reflection_reasoning`` from the
adapter's proposer via a ``getattr`` chain when available.
Each proposal's reflection batch is handed to scoring so a
valset that is the trainset is not evaluated twice.
An ``EmptyProposalError`` from proposing is recorded as a
skipped iteration (``skip_reason="empty_proposal"``) that counts
toward stagnation; the loop then checks stop conditions and
continues. A ``ReflectionTimeoutError`` is handled the same way
with ``skip_reason="reflection_timeout"``, and a retryable
``ReflectionError`` with ``skip_reason="reflection_error"``; a
non-retryable ``ReflectionError`` propagates to ``run()``,
which attaches the partial result. An
``IncompleteProposalError`` is recorded with
``skip_reason="incomplete_proposal"``, counts toward stagnation
and is followed by the stop check. A proposal that
``config.proposal_validator`` rejects is recorded, before schema
validation, with ``skip_reason="proposal_rejected"`` and the
validator's reason, counts toward stagnation and is followed by
the stop check; an exception from the validator propagates. A
proposal whose
``output_schema`` text fails validation is recorded the same way
with ``skip_reason="schema_validation_failed"``, counts toward
stagnation and is followed by the stop check. A proposal whose
``Candidate.id`` was already scored (baseline, proposal or merge)
is recorded with
``skip_reason="duplicate"`` and its stored score instead of being
evaluated. With a reflection minibatch in effect, a proposal
that does not beat its parent on the iteration's sampled rows is
recorded with ``skip_reason="minibatch_rejected"`` and not
evaluated further. An evaluated proposal's record carries the
scoring batch's row count as ``rows_evaluated``. A proposal that
passes is evaluated on the
full trainset when the valset is the trainset (that batch is
reused for scoring); with a separate valset the gate's batch is
its reflection batch and only the valset is evaluated (see
``_evaluate_passed_proposal``). The trainset rows of each cached
batch are stored beside it for the next gate and reflection. A
merge candidate is always evaluated on the full trainset, so
its rows are None. A merge candidate that is a duplicate or has an
invalid schema is not evaluated, and the iteration's own record
and stop check still follow. Each iteration starts with no
pending failures, a zero pending token rollup and no pending
reflection usage, so an unrecorded iteration's failures and tokens
do not carry into the next record. After the proposal returns,
or raises one of the skip errors, the adapter proposer's
``last_token_usage`` is read through ``_take_proposer_usage`` so
the iteration's record carries its reflection cost.
Each accept decision is logged as ``proposal.accepted`` or
``proposal.rejected`` with the proposal's ``candidate_id``.
Every recorded iteration, skipped or not, reaches
``config.on_iteration`` before the stop check.
State restored from a checkpoint replaces the baseline
evaluation.
"""
# A restored checkpoint already holds the baseline
if not self._restored:
await self._initialize_baseline()
assert self._state is not None, "Engine state not initialized" # noqa: S101 # narrows state that run() sets before this call
# Evolution loop
stop_reason = self._should_stop()
while stop_reason is None:
self._state.iteration += 1
# Failures from an unrecorded previous iteration do not carry over
self._pending_failed_evaluations = 0
self._pending_token_usage = _ZERO_TOKENS
self._pending_reflection_usage = None
# Propose mutation (returns candidate and list of components evolved)
try:
(
proposal,
evolved_components_list,
parent_idx,
) = await self._propose_mutation()
except EmptyProposalError as error:
# Empty reflection after retry: failed iteration, not fatal
self._take_proposer_usage()
await self._record_empty_proposal(error)
stop_reason = self._should_stop()
continue
except ReflectionTimeoutError as error:
# Reflection timed out: skipped iteration, not fatal
self._take_proposer_usage()
await self._record_reflection_timeout(error)
stop_reason = self._should_stop()
continue
except ReflectionError as error:
# A programming or client error will not clear on its own
if not error.retryable:
raise
# Provider error survived the proposer's retry: skip it
self._take_proposer_usage()
await self._record_reflection_error(error)
stop_reason = self._should_stop()
continue
except IncompleteProposalError as error:
# Truncated reflection output: skipped iteration, not fatal
self._take_proposer_usage()
await self._record_incomplete_proposal(error)
stop_reason = self._should_stop()
continue
self._take_proposer_usage()
# Caller's validator rejected the text: record it, no evaluation
if await self._record_if_rejected(proposal, evolved_components_list):
stop_reason = self._should_stop()
continue
# Validate schema component if present (reject invalid early)
if not self._validate_schema_component(proposal):
# Invalid schema: record the skip, no evaluation
await self._record_schema_validation_skip(
proposal, evolved_components_list
)
stop_reason = self._should_stop()
continue
# Already scored this run: reuse its score, no evaluation
if await self._record_if_duplicate(proposal, evolved_components_list):
stop_reason = self._should_stop()
continue
# Beat the parent on a trainset minibatch before the full pass
if not await self._gate_on_minibatch(proposal, evolved_components_list):
stop_reason = self._should_stop()
continue
# Reflection batch: the gate's rows with a separate valset,
# otherwise the full trainset pass
(
reflection_score,
reflection_batch,
reflection_rows,
) = await self._evaluate_passed_proposal(proposal)
proposal_score, scoring_batch, eval_indices = await self._evaluate_scoring(
proposal,
reflection_batch=reflection_batch if reflection_rows is None else None,
)
candidate_idx = None
if self._pareto_state is not None:
# Use eval_indices returned from _evaluate_scoring (T066)
# Prepare objective scores if available (T068)
objective_scores: dict[str, float] | None = None
per_example_objective_scores: dict[int, dict[str, float]] | None = None
if scoring_batch.objective_scores is not None:
if self.config.frontier_type in (
FrontierType.OBJECTIVE,
FrontierType.HYBRID,
):
# Aggregate objective scores across evaluated examples
objective_scores_accum: dict[str, list[float]] = {}
for obj_scores in scoring_batch.objective_scores:
for obj_name, obj_score in obj_scores.items():
objective_scores_accum.setdefault(obj_name, []).append(
obj_score
)
# Take mean per objective
objective_scores = {
obj_name: fmean(scores)
for obj_name, scores in objective_scores_accum.items()
}
if self.config.frontier_type == FrontierType.CARTESIAN:
# For CARTESIAN, need per-example objective scores mapped to valset indices
per_example_objective_scores = {
eval_indices[i]: scoring_batch.objective_scores[i]
for i in range(len(eval_indices))
}
# Also need aggregated for validation
objective_scores_by_name: dict[str, list[float]] = {}
for obj_scores in scoring_batch.objective_scores:
for obj_name, obj_score in obj_scores.items():
objective_scores_by_name.setdefault(
obj_name, []
).append(obj_score)
objective_scores = {
obj_name: fmean(scores)
for obj_name, scores in objective_scores_by_name.items()
}
# The genealogy edge is the parent the reflector rewrote
parent_indices = [parent_idx] if parent_idx is not None else None
# Pass scores with correct index mapping (T066)
candidate_idx = self._pareto_state.add_candidate(
proposal,
scoring_batch.scores,
score_indices=eval_indices,
objective_scores=objective_scores,
per_example_objective_scores=per_example_objective_scores,
parent_indices=parent_indices,
logger=logger,
)
self._candidate_eval_batches[candidate_idx] = reflection_batch
self._candidate_eval_rows[candidate_idx] = reflection_rows
logger.info(
"pareto_frontier.candidate_added",
candidate_idx=candidate_idx,
candidate_id=proposal.id,
iteration=self._state.iteration,
)
# Calculate valset mean using only evaluated scores (T067)
valset_mean = (
sum(scoring_batch.scores) / len(scoring_batch.scores)
if scoring_batch.scores
else 0.0
)
# Accept if improves above threshold
accepted = self._should_accept(proposal_score, self._state.best_score)
if accepted:
logger.info(
"proposal.accepted",
iteration=self._state.iteration,
candidate_id=proposal.id,
score=proposal_score,
best_score=self._state.best_score,
)
self._accept_proposal(
proposal,
proposal_score,
reflection_batch,
candidate_idx=candidate_idx,
eval_rows=reflection_rows,
reflection_score=reflection_score,
valset_mean=valset_mean,
objective_scores=scoring_batch.objective_scores,
)
# Schedule merge if enabled
if (
self.config.use_merge
and self._merge_proposer is not None
and self._merge_invocations < self.config.max_merge_invocations
):
self._merges_due += 1
logger.debug(
"merge_scheduling.merge_scheduled",
iteration=self._state.iteration,
candidate_id=proposal.id,
merges_due=self._merges_due,
)
else:
logger.debug(
"proposal.rejected",
iteration=self._state.iteration,
candidate_id=proposal.id,
score=proposal_score,
best_score=self._state.best_score,
)
# Increment stagnation counter on rejection
self._state.stagnation_counter += 1
# Attempt merge if scheduled
if (
self._merges_due > 0
and self._merge_proposer is not None
and self._pareto_state is not None
and self._merge_invocations < self.config.max_merge_invocations
):
merge_result = await self._merge_proposer.propose(self._pareto_state)
# A skipped merge (invalid schema or already scored) is not
# evaluated; the iteration is still recorded below
if merge_result is not None and self._start_merge(merge_result):
# Evaluate merge proposal
(
merge_reflection_score,
merge_reflection_batch,
) = await self._evaluate_reflection(merge_result.candidate)
(
merge_proposal_score,
merge_scoring_batch,
merge_eval_indices,
) = await self._evaluate_scoring(
merge_result.candidate,
reflection_batch=merge_reflection_batch,
)
# Add merge candidate to ParetoState
merge_candidate_idx = None
assert self._pareto_state is not None, ( # noqa: S101 # narrows state that run() sets before this call
"Pareto state not initialized"
)
merge_objective_scores: dict[str, float] | None = None
merge_per_example_objective_scores: (
dict[int, dict[str, float]] | None
) = None
if merge_scoring_batch.objective_scores is not None:
if self.config.frontier_type in (
FrontierType.OBJECTIVE,
FrontierType.HYBRID,
):
merge_objective_scores_accum: dict[str, list[float]] = {}
for obj_scores in merge_scoring_batch.objective_scores:
for obj_name, obj_score in obj_scores.items():
merge_objective_scores_accum.setdefault(
obj_name, []
).append(obj_score)
merge_objective_scores = {
obj_name: fmean(scores)
for obj_name, scores in merge_objective_scores_accum.items()
}
if self.config.frontier_type == FrontierType.CARTESIAN:
merge_per_example_objective_scores = {
merge_eval_indices[
i
]: merge_scoring_batch.objective_scores[i]
for i in range(len(merge_eval_indices))
}
merge_objective_scores_by_name: dict[str, list[float]] = {}
for obj_scores in merge_scoring_batch.objective_scores:
for obj_name, obj_score in obj_scores.items():
merge_objective_scores_by_name.setdefault(
obj_name, []
).append(obj_score)
merge_objective_scores = {
obj_name: fmean(scores)
for obj_name, scores in merge_objective_scores_by_name.items()
}
merge_candidate_idx = self._pareto_state.add_candidate(
merge_result.candidate,
merge_scoring_batch.scores,
score_indices=merge_eval_indices,
objective_scores=merge_objective_scores,
per_example_objective_scores=merge_per_example_objective_scores,
parent_indices=merge_result.parent_indices,
logger=logger,
)
self._candidate_eval_batches[merge_candidate_idx] = (
merge_reflection_batch
)
self._candidate_eval_rows[merge_candidate_idx] = None
merge_valset_mean = (
sum(merge_scoring_batch.scores)
/ len(merge_scoring_batch.scores)
if merge_scoring_batch.scores
else 0.0
)
# Accept merge if improves
merge_accepted = self._should_accept(
merge_proposal_score, self._state.best_score
)
if merge_accepted:
self._accept_proposal(
merge_result.candidate,
merge_proposal_score,
merge_reflection_batch,
candidate_idx=merge_candidate_idx,
reflection_score=merge_reflection_score,
valset_mean=merge_valset_mean,
objective_scores=merge_scoring_batch.objective_scores,
)
logger.info(
"merge_scheduling.merge_accepted",
iteration=self._state.iteration,
candidate_id=merge_result.candidate.id,
merge_score=merge_proposal_score,
)
else:
logger.debug(
"merge_scheduling.merge_rejected",
iteration=self._state.iteration,
candidate_id=merge_result.candidate.id,
merge_score=merge_proposal_score,
best_score=self._state.best_score,
)
elif merge_result is None:
# Merge not possible, decrement counter
if self._merges_due > 0:
self._merges_due -= 1
# Record iteration with actual evolved component name (T033)
# For single-component evolution, use the first (and only) component.
# For multi-component round-robin, this tracks which component was
# evolved in this iteration.
if evolved_components_list:
evolved_component_name = evolved_components_list[0]
else:
# Empty list indicates logic error - use first key from proposal
logger.warning(
"engine.empty_evolved_components_list",
iteration=self._state.iteration,
proposal_keys=list(proposal.components.keys()),
)
evolved_component_name = next(iter(proposal.components.keys()))
await self._record_iteration(
score=proposal_score,
component_text=proposal.components.get(evolved_component_name, ""),
evolved_component=evolved_component_name,
accepted=accepted,
objective_scores=scoring_batch.objective_scores,
# TODO: Surface last_reasoning through adapter protocol instead
# of reaching into private _proposer attribute. Safe (defaults
# to None) but couples engine to adapter internals.
reflection_reasoning=getattr(
getattr(self.adapter, "_proposer", None),
"last_reasoning",
None,
),
candidate_id=proposal.id,
parent_ids=_parent_ids(proposal),
rows_evaluated=len(scoring_batch.scores),
)
stop_reason = self._should_stop()
# Build and return result
return self._build_result(stop_reason=stop_reason)