Ask the evaluator during an epoch, not only at its boundary

src/vectrify/cli.py

@@ -40,6 +40,21 @@
 # into an epoch -- all of it with no evaluator in the loop, which is where the
 # proxy runs away from what a viewer would call better.
 DEFAULT_EPOCH_MAX_TASKS = None
+# How often the evaluator is asked about the front mid-epoch, in tasks. The
+# cheap measures can be driven a long way without the drawing improving -- one
+# run took them 64% down while the evaluator saw no difference at all -- and
+# asking it only at the boundary means noticing that after the fact.
+#
+# A check costs one panel call over the front, about 13s measured, against a
+# throughput near 60 tasks/s. Nothing is asked twice: the evaluator's score is
+# absolute and cached per node, so a check re-prices only what is new.
+DEFAULT_EPOCH_EVAL_INTERVAL = 2000
+# Rounds without the evaluator seeing anything better before the epoch ends and
+# the model re-seeds. Rounds rather than checks, so the number means the same
+# thing whatever cadence the checks run at. Off until it is tuned: too low ends
+# epochs on the evaluator's noise, too high is the unsupervised drift it exists
+# to stop.
+DEFAULT_EPOCH_EVAL_PATIENCE = None
 # Tasks without improvement before an epoch is called converged. Measured over
 # eleven runs and 145 improvements, the gap between one improvement and the
 # next is 25 tasks at the median, 142 at the 95th percentile and 497 at the
@@ -227,6 +242,26 @@ def parse_args(args: list[str] | None = None) -> argparse.Namespace:
         "so this and --seeds together fix the run's entire LLM spend. "
         f"Default: {DEFAULT_EPOCHS}",
     )
+    g_epoch.add_argument(
+        "--epoch-eval-interval",
+        type=int,
+        default=DEFAULT_EPOCH_EVAL_INTERVAL,
+        dest="epoch_eval_interval",
+        metavar="N",
+        help="Ask the evaluator about the front every N local tasks, not only "
+        "at the epoch boundary. Scores are cached per candidate, so a check "
+        f"re-prices only what is new. 0 disables. Default: "
+        f"{DEFAULT_EPOCH_EVAL_INTERVAL}",
+    )
+    g_epoch.add_argument(
+        "--epoch-eval-patience",
+        type=int,
+        default=DEFAULT_EPOCH_EVAL_PATIENCE,
+        dest="epoch_eval_patience",
+        metavar="N",
+        help="End the epoch once the evaluator has gone this many rounds "
+        "without seeing a better candidate. Unset by default.",
+    )
     g_epoch.add_argument(
         "--epoch-max-tasks",
         type=int,

src/vectrify/main.py

@@ -137,6 +137,8 @@ def main():
             pool_size=args.pool_size,
             seeds=args.seeds,
             epoch_max_tasks=args.epoch_max_tasks,
+            epoch_eval_interval=args.epoch_eval_interval,
+            epoch_eval_patience=args.epoch_eval_patience,
             tournament_size=args.tournament_size,
             adaptive_operators=args.adaptive_operators,
             epochs=args.epochs,

src/vectrify/search/engine.py

@@ -120,6 +120,8 @@ def run(
         initial_seeds: int | None = None,
         epochs: int | None = None,
         epoch_max_tasks: int | None = None,
+        epoch_eval_interval: int | None = None,
+        epoch_eval_patience: int | None = None,
         operator_policy: OperatorPolicy | None = None,
         collector: StatCollector | None = None,
     ) -> None:
@@ -227,6 +229,13 @@ def _scorer_worker():
         epoch = 0
         epoch_no_improve = 0
         epoch_started_at = 0
+        # The evaluator's view of the epoch, kept between checks. Its score is
+        # a calibrated distance to the target, so a value from one check is
+        # comparable with the next -- which is the whole reason it can be
+        # tracked at all.
+        best_panel: float | None = None
+        last_eval_at = 0
+        rounds_since_panel_gain = 0
         # Reset at every transition, so each epoch is judged against the
         # pool it opened with rather than against the first one.
         pool_refilling = False  # True until a fresh epoch's pool reaches capacity
@@ -526,7 +535,7 @@ def _close_generation() -> None:
             full sort per result, which at pool size 20 costs about as much as
             producing the candidate did.
             """
-            nonlocal active_pool, epoch_no_improve
+            nonlocal active_pool, epoch_no_improve, rounds_since_panel_gain
 
             if not pending_children:
                 return
@@ -540,6 +549,8 @@ def _close_generation() -> None:
             # threshold on a magnitude -- an epoch goes stale when nothing new
             # can get to the front any more, whatever the numbers happen to be
             # denominated in.
+            rounds_since_panel_gain += 1
+
             new_ids = {n.id for n in pending_children}
             top_tier = self.strategy.top_tier_ids(combined)
             if new_ids & top_tier:
@@ -596,9 +607,13 @@ def _process_local_result(res: Result) -> None:
                 _close_generation()
 
         def _do_epoch_transition(reason: str) -> None:
-            nonlocal epoch, epoch_started_at
+            nonlocal epoch, epoch_started_at, rounds_since_panel_gain
 
             epoch_started_at = tasks_completed
+            # The evaluator's best carries across epochs -- it is an absolute
+            # score, and a later epoch has to beat what the run already has --
+            # but the patience counting restarts with the epoch.
+            rounds_since_panel_gain = 0
 
             # The next seed batch edits this pool's front, so the children that
             # arrived since the last generation have to land in it first.
@@ -614,6 +629,37 @@ def _do_epoch_transition(reason: str) -> None:
                 return
             _begin_seed_phase()
 
+        def _run_panel_check() -> None:
+            """Put the current front to the evaluator and record its verdict.
+
+            The field is the best-ranked distinct candidates, capped: the top
+            tier can be most of the pool, and evaluating near-clones spends the
+            expensive part of the run learning nothing. Whatever the evaluator
+            has already scored costs nothing to include, so the cap is about
+            new work, not about the size of the field.
+            """
+            nonlocal best_panel, last_eval_at, rounds_since_panel_gain, best_node
+
+            last_eval_at = tasks_completed
+            field = self.strategy.epoch_parents(active_pool, FRONT_EVAL_CAP)
+            if not field or self.rank_front is None:
+                return
+            try:
+                ranked = self.rank_front(field)
+            except Exception as exc:
+                log.warning(f"Evaluator check failed, continuing: {exc}")
+                return
+
+            top = next((n for n in ranked if FRONT_SCORE in n.metrics), None)
+            if top is None:
+                return
+            value = top.metrics[FRONT_SCORE]
+            if best_panel is None or value < best_panel:
+                best_panel = value
+                rounds_since_panel_gain = 0
+                best_node = top
+                log.info(f"Evaluator: node={top.id} score={value:.6f}")
+
         def _check_epoch_end():
             nonlocal pool_refilling
 
@@ -645,6 +691,27 @@ def _check_epoch_end():
             # without the evaluator seeing anything. The epoch boundary is where
             # the evaluator ranks the front and the model re-seeds from its
             # choice, so capping the epoch caps the drift.
+            # Ask the evaluator what it makes of the current front, now and
+            # then rather than only at the boundary. The cheap measures can be
+            # improved without the drawing getting better -- one run drove them
+            # 64% down while the evaluator saw no difference at all -- and the
+            # only way to notice is to ask the evaluator while it is happening.
+            if (
+                self.rank_front is not None
+                and epoch_eval_interval
+                and tasks_completed - last_eval_at >= epoch_eval_interval
+            ):
+                _run_panel_check()
+
+            # Rounds rather than checks, so the threshold means the same thing
+            # whatever cadence the checks are running at.
+            panel_stale = (
+                epoch_eval_patience is not None
+                and epoch_eval_patience > 0
+                and best_panel is not None
+                and rounds_since_panel_gain >= epoch_eval_patience
+            )
+
             over_budget = (
                 epoch_max_tasks is not None
                 and epoch_max_tasks > 0
@@ -670,6 +737,11 @@ def _check_epoch_end():
                     f"staleness ({epoch_no_improve} >="
                     f" {epoch_patience} tasks without improvement)"
                 )
+            elif panel_stale:
+                reason = (
+                    f"the evaluator has not seen a better candidate in "
+                    f"{rounds_since_panel_gain} rounds"
+                )
             elif over_budget:
                 reason = (
                     f"epoch budget ({tasks_completed - epoch_started_at} >="

src/vectrify/search/nsga.py

@@ -352,7 +352,11 @@ def epoch_parents(
                 if not any(self._is_duplicate(node, s) for s in pareto_nodes):
                     pareto_nodes.append(node)
 
-        pareto_nodes.sort(key=lambda n: n.score)
+        # Already in rank order: non_dominated_sort yields the best tier first
+        # and _is_duplicate has thinned each tier, so taking the head takes the
+        # best-ranked distinct candidates. There is nothing left to sort by --
+        # a score would have to be a blend of the measures, which is the thing
+        # dominance replaced.
         parents = pareto_nodes[:max_parents]
         return parents or valid[:max_parents]
 

src/vectrify/vector/runner.py

@@ -13,6 +13,8 @@
 from PIL import Image, UnidentifiedImageError
 
 from vectrify.cli import (
+    DEFAULT_EPOCH_EVAL_INTERVAL,
+    DEFAULT_EPOCH_EVAL_PATIENCE,
     DEFAULT_EPOCH_MAX_TASKS,
     DEFAULT_MAX_TOTAL_TASKS,
     DEFAULT_POOL_SIZE,
@@ -189,6 +191,8 @@ def run_vector_search(
     pool_size: int = DEFAULT_POOL_SIZE,
     seeds: int | None = None,
     epoch_max_tasks: int | None = DEFAULT_EPOCH_MAX_TASKS,
+    epoch_eval_interval: int | None = DEFAULT_EPOCH_EVAL_INTERVAL,
+    epoch_eval_patience: int | None = DEFAULT_EPOCH_EVAL_PATIENCE,
     tournament_size: int = DEFAULT_TOURNAMENT_SIZE,
     adaptive_operators: bool = True,
     epochs: int | None = None,
@@ -436,6 +440,8 @@ def score_fn(results):
             initial_seeds=first_batch,
             epochs=epochs,
             epoch_max_tasks=epoch_max_tasks,
+            epoch_eval_interval=epoch_eval_interval,
+            epoch_eval_patience=epoch_eval_patience,
             operator_policy=operator_policy,
             collector=collector,
         )

tests/search/test_engine.py

@@ -3,6 +3,7 @@
 
 import pytest
 
+from vectrify.score.metrics import FRONT_SCORE
 from vectrify.search import INVALID_SCORE, ChainState, Result, SearchNode
 from vectrify.search.engine import MultiprocessSearchEngine
 
@@ -1112,3 +1113,87 @@ def test_the_epoch_budget_ends_an_epoch_that_has_not_gone_stale():
     )
 
     collector.on_epoch_transition.assert_called()
+
+
+def test_the_evaluator_is_asked_during_an_epoch_not_only_at_its_boundary():
+    """The cheap measures can be driven a long way without the drawing getting
+    better, and the only way to notice is to ask the evaluator while it is
+    happening."""
+    seen: list[int] = []
+
+    def rank(nodes):
+        seen.append(len(nodes))
+        for i, node in enumerate(nodes):
+            node.metrics[FRONT_SCORE] = 0.5 - i * 0.1
+        return nodes
+
+    engine = MultiprocessSearchEngine(
+        workers=1,
+        strategy=FakeStrategy(),
+        storage=FakeStorage(),
+        max_total_tasks=4,
+        rank_front=rank,
+    )
+    for task_id in range(1, 5):
+        engine.unscored_q.put(
+            Result(task_id=task_id, parent_id=1, valid=True, score=0.0, payload="p")
+        )
+
+    engine.run(
+        initial_nodes=[
+            SearchNode(
+                score=0.0, id=1, parent_id=0, state=ChainState(score=0.0, payload=None)
+            )
+        ],
+        max_wall_seconds=None,
+        active_pool_size=3,
+        generation_size=1,
+        epoch_patience=10_000,
+        epoch_eval_interval=1,
+    )
+
+    assert seen, "the evaluator was never consulted mid-epoch"
+
+
+def test_the_epoch_ends_when_the_evaluator_stops_seeing_improvement():
+    """Rounds rather than checks, so the threshold means the same thing at any
+    cadence. The evaluator here always reports the same verdict, so nothing
+    ever improves on it and the epoch has to end on that."""
+    from unittest.mock import MagicMock
+
+    def rank(nodes):
+        for node in nodes:
+            node.metrics[FRONT_SCORE] = 0.5
+        return nodes
+
+    collector = MagicMock()
+    engine = MultiprocessSearchEngine(
+        workers=1,
+        strategy=FakeStrategy(),
+        storage=FakeStorage(),
+        max_total_tasks=8,
+        rank_front=rank,
+    )
+    for task_id in range(1, 9):
+        engine.unscored_q.put(
+            Result(task_id=task_id, parent_id=1, valid=True, score=0.0, payload="p")
+        )
+
+    engine.run(
+        initial_nodes=[
+            SearchNode(
+                score=0.0, id=1, parent_id=0, state=ChainState(score=0.0, payload=None)
+            )
+        ],
+        max_wall_seconds=None,
+        active_pool_size=3,
+        generation_size=1,
+        epochs=4,
+        # Neither of the other criteria may be what ends it.
+        epoch_patience=10_000,
+        epoch_eval_interval=1,
+        epoch_eval_patience=2,
+        collector=collector,
+    )
+
+    collector.on_epoch_transition.assert_called()

tests/search/test_nsga.py

@@ -335,18 +335,19 @@ def test_pool_size_one_always_returns_same_node():
     assert selected == {1}
 
 
-def test_epoch_parents_returns_pareto_front():
+def test_epoch_parents_returns_the_best_ranked_tier():
+    """Two candidates good at different things both survive; one that is worse
+    than a rival on every measure does not."""
     strategy = NsgaStrategy(pool_size=10)
     nodes = [
-        make_node(1, 0.1, edge=1000.0),  # good quality, complex
-        make_node(2, 0.5, edge=100.0),  # worse quality, simpler (dominates node 3)
-        make_node(3, 0.9, edge=900.0),  # dominated by node 2
+        make_node(1, 0.0, edge=100.0, colour=900.0),  # wins edge
+        make_node(2, 0.0, edge=900.0, colour=100.0),  # wins colour
+        make_node(3, 0.0, edge=950.0, colour=950.0),  # loses to both
     ]
+
     seeds = strategy.epoch_parents(nodes, max_parents=2)
-    seed_ids = {n.id for n in seeds}
-    assert 1 in seed_ids
-    assert 2 in seed_ids
-    assert 3 not in seed_ids
+
+    assert {n.id for n in seeds} == {1, 2}
 
 
 def test_epoch_parents_respects_max_parents():
@@ -375,17 +376,18 @@ def test_epoch_parents_empty_pool_returns_empty():
     assert seeds == []
 
 
-def test_epoch_parents_sorted_by_visual_score():
+def test_epoch_parents_come_back_in_rank_order():
+    """Best-ranked first, with nothing sorted afterwards: a score to sort by
+    would have to blend the measures, which is what dominance replaced."""
     strategy = NsgaStrategy(pool_size=10)
     nodes = [
-        make_node(1, 0.1, edge=800.0),
-        make_node(2, 0.3, edge=600.0),
-        make_node(3, 0.5, edge=400.0),
-        make_node(4, 0.7, edge=200.0),
+        make_node(1, 0.0, edge=200.0, colour=200.0),
+        make_node(2, 0.0, edge=400.0, colour=400.0),
+        make_node(3, 0.0, edge=600.0, colour=600.0),
+        make_node(4, 0.0, edge=800.0, colour=800.0),
     ]
     seeds = strategy.epoch_parents(nodes, max_parents=4)
-    scores = [n.score for n in seeds]
-    assert scores == sorted(scores)
+
     assert seeds[0].id == 1