Make the panel score absolute so two evaluations can be compared

src/vectrify/score/ensemble.py

@@ -45,6 +45,7 @@
 
 import io
 import logging
+import statistics
 from dataclasses import dataclass
 from typing import Any
 
@@ -75,6 +76,13 @@
 class PanelReference:
     image: Image.Image
     tiles: list[Any]
+    # Each member's distance from the target to a blank canvas, measured once.
+    # It is what makes a member's distance mean something on its own: raw
+    # cosine distances come from three different embedding spaces and span
+    # different widths, so an uncalibrated average is decided by whichever
+    # member happens to spread widest -- one model steering the run, which is
+    # what a panel exists to prevent.
+    blank: list[float]
 
 
 def _tiles(image: Image.Image) -> list[Image.Image]:
@@ -106,12 +114,18 @@ def validate_environment(self) -> None:
 
     def prepare_reference(self, original_rgb: Image.Image) -> PanelReference:
         cells = _tiles(original_rgb)
-        return PanelReference(
+        reference = PanelReference(
             image=original_rgb,
             tiles=[m.embed_images(cells) for m in self._members],
+            blank=[],
         )
+        empty = Image.new("RGB", original_rgb.size, (255, 255, 255))
+        reference.blank.extend(
+            max(d[0], 1e-6) for d in self._raw_distances(reference, [empty])
+        )
+        return reference
 
-    def _distances(self, reference: PanelReference, images: list[Image.Image]):
+    def _raw_distances(self, reference: PanelReference, images: list[Image.Image]):
         """Each member's distance to every image, cell by cell then averaged."""
         per_member = []
         for member, reference_tiles in zip(self._members, reference.tiles, strict=True):
@@ -124,40 +138,70 @@ def _distances(self, reference: PanelReference, images: list[Image.Image]):
             )
         return per_member
 
-    def score(self, reference: PanelReference, candidate_png: bytes) -> float:
-        """Mean distance across the panel.
+    def _distances(self, reference: PanelReference, images: list[Image.Image]):
+        """Per-member distances, each as a fraction of that member's distance
+        from the target to a blank canvas. 0 is the target itself and about 1
+        is as wrong as an empty drawing, on every member and every target."""
+        raw = self._raw_distances(reference, images)
+        return [
+            [value / scale for value in member]
+            for member, scale in zip(raw, reference.blank, strict=True)
+        ]
 
-        A single candidate has nothing to be compared against, so there is no
-        vote to take. This exists for callers that need a scalar per candidate;
-        the panel's actual judgement is ``rank``, which is what decides
-        direction.
+    def score(self, reference: PanelReference, candidate_png: bytes) -> float:
+        """The panel's verdict on one candidate: the median calibrated distance.
+
+        The median rather than the mean, and that is the whole panel argument
+        in absolute form. With three members the median is the majority
+        position: for any standard you might hold a candidate to, "the panel
+        says it meets this" is true exactly when the median says so, and a
+        member that is idiosyncratic about this particular drawing cannot move
+        it. The pairwise vote said the same thing about pairs; this says it
+        about candidates, which is what lets two scores be compared at all.
+
+        Absolute, so it means the same thing in every call and in every run --
+        the property `rank` could not have, since counting rivals beaten only
+        describes the field a candidate was ranked against.
         """
         try:
             image = Image.open(io.BytesIO(candidate_png)).convert("RGB")
         except Exception:
             return MAX_SCORE
         values = [d[0] for d in self._distances(reference, [image])]
-        return sum(values) / len(values) if values else MAX_SCORE
+        return statistics.median(values) if values else MAX_SCORE
 
     def rank(
         self, reference: PanelReference, candidate_pngs: list[bytes]
     ) -> list[float]:
-        """Score candidates by how many rivals the panel puts them ahead of.
-
-        Every pair is put to the panel and the majority wins, then candidates
-        are ordered by wins less losses. That ordering step is not decoration:
-        a majority relation is a tournament and cycles, so it cannot be sorted
-        by pairwise comparison -- three candidates can each beat the next.
-        Counting wins ranks a tournament without needing it to be transitive.
-
-        Returns one value per candidate, lower being better, so it drops into
-        a caller that expects a distance. Values span [0, 1] up to the width of
-        the tie-break term, which can carry the extremes a little past either
-        end; nothing compares them against an absolute threshold.
+        """Score every candidate, lower being better.
+
+        One absolute number each, so the caller may compare them with anything
+        else the panel has scored -- a candidate from an earlier check, from an
+        earlier epoch, or from another run entirely.
+
+        This used to put every pair to the panel and count rivals beaten, which
+        ranked a field correctly and said nothing outside it: the same drawing
+        scored differently depending on who it was ranked against, so two calls
+        could not be compared and nothing could be cached between them.
         """
         if not candidate_pngs:
             return []
 
+        images = []
+        for png in candidate_pngs:
+            try:
+                images.append(Image.open(io.BytesIO(png)).convert("RGB"))
+            except Exception:
+                # A candidate that will not open is scored worst rather than
+                # failing the field it was ranked with.
+                images.append(Image.new("RGB", reference.image.size, (255, 255, 255)))
+
+        per_member = self._distances(reference, images)
+        return [
+            statistics.median([member[i] for member in per_member])
+            for i in range(len(images))
+        ]
+
         images = []
         for png in candidate_pngs:
             try:

src/vectrify/vector/runner.py

@@ -1,6 +1,7 @@
 import io
 import logging
 import os
+from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
 from typing import TYPE_CHECKING, Any
 
@@ -99,6 +100,73 @@ def initial_seed_tasks(epoch_seeds: int, initial_nodes: list[SearchNode]) -> int
     return max(0, epoch_seeds - seeded)
 
 
+def evaluate_front(
+    nodes: list[SearchNode],
+    *,
+    front_scorer: Callable[[], tuple[Any, Any]],
+    format_plugin: Any,
+    out_w: int,
+    out_h: int,
+) -> list[SearchNode]:
+    """Order *nodes* by the evaluator, best first, scoring only what is new.
+
+    *front_scorer* is called for (scorer, reference) and only when there is
+    something to score, so a call the cache answers in full never builds a
+    model.
+
+    Re-rasterises rather than reading a node's stored render, which is only
+    kept when --write-lineage or --save-raster is on.
+    """
+    renders: list[tuple[bytes, SearchNode]] = []
+    for node in nodes:
+        # Already judged, and the judgement travels: the panel's score is a
+        # calibrated distance to the target, so it means the same thing in
+        # every call. Re-rasterising and re-embedding a node the evaluator has
+        # already seen would buy an identical number at full price -- and a run
+        # asks about the same pool members repeatedly.
+        if FRONT_SCORE in node.metrics:
+            continue
+        content = getattr(node.state.payload, "content", None)
+        if not content:
+            continue
+        try:
+            renders.append(
+                (format_plugin.rasterize(content, out_w=out_w, out_h=out_h), node)
+            )
+        except Exception as exc:
+            log.debug(f"Front evaluation skipped node {node.id}: {exc}")
+
+    if renders:
+        scorer, ref = front_scorer()
+        pngs = [png for png, _ in renders]
+        try:
+            values = scorer.rank(ref, pngs)
+        except AttributeError:
+            values = [scorer.score(ref, png) for png in pngs]
+        except Exception as exc:
+            log.warning(f"Front evaluation failed, keeping rank order: {exc}")
+            return nodes
+
+        for value, (_png, node) in zip(values, renders, strict=True):
+            node.metrics[FRONT_SCORE] = value
+
+    # Every node the panel has ever scored, freshly measured or recalled.
+    scored = [
+        (node.metrics[FRONT_SCORE], node)
+        for node in nodes
+        if FRONT_SCORE in node.metrics
+    ]
+    if not scored:
+        return nodes
+    scored.sort(key=lambda pair: pair[0])
+    log.info(
+        f"Front evaluated: {len(scored)} candidate(s) "
+        f"({len(renders)} newly scored), "
+        f"best {scored[0][0]:.6f}, worst {scored[-1][0]:.6f}"
+    )
+    return [node for _value, node in scored]
+
+
 def run_vector_search(
     image_path: str,
     storage: StorageAdapter,
@@ -256,56 +324,13 @@ def _front_scorer() -> tuple[Any, Any]:
         return _front[0], _front[1]
 
     def rank_front(nodes: list[SearchNode]) -> list[SearchNode]:
-        """Order a converged front by the run's real objective.
-
-        Re-rasterises rather than reading the node's stored render, which is
-        only kept when --write-lineage or --save-raster is on.
-        """
-        scorer, ref = _front_scorer()
-        renders: list[tuple[bytes, SearchNode]] = []
-        for node in nodes:
-            content = getattr(node.state.payload, "content", None)
-            if not content:
-                continue
-            try:
-                renders.append(
-                    (
-                        format_plugin.rasterize(
-                            content, out_w=original_w, out_h=original_h
-                        ),
-                        node,
-                    )
-                )
-            except Exception as exc:
-                log.debug(f"Front evaluation skipped node {node.id}: {exc}")
-
-        if not renders:
-            return nodes
-
-        pngs = [png for png, _ in renders]
-        try:
-            # A panel ranks the field as a whole, because a majority vote needs
-            # candidates to compare; a single scorer just scores each one.
-            values = scorer.rank(ref, pngs)
-        except AttributeError:
-            values = [scorer.score(ref, png) for png in pngs]
-        except Exception as exc:
-            log.warning(f"Front evaluation failed, keeping round order: {exc}")
-            return nodes
-
-        scored: list[tuple[float, SearchNode]] = []
-        for value, (_png, node) in zip(values, renders, strict=True):
-            node.metrics[FRONT_SCORE] = value
-            scored.append((value, node))
-
-        if not scored:
-            return nodes
-        scored.sort(key=lambda pair: pair[0])
-        log.info(
-            f"Front evaluated: {len(scored)} candidate(s), "
-            f"best {scored[0][0]:.6f}, worst {scored[-1][0]:.6f}"
+        return evaluate_front(
+            nodes,
+            front_scorer=_front_scorer,
+            format_plugin=format_plugin,
+            out_w=original_w,
+            out_h=original_h,
         )
-        return [node for _value, node in scored]
 
     engine = MultiprocessSearchEngine(
         workers=workers,