Add method extensions and paper experiment updates

Author: apartsin

app/core/config_yaml.py

@@ -14,15 +14,17 @@
     # Edit any of these values before creating a new session.
     # This YAML is reloaded fresh for each setup page visit or reset action.
     #
-    # sampler: random_local | exploit_orthogonal | uncertainty_guided | axis_sweep | incumbent_mix
-    # updater: winner_average | winner_copy | linear_preference
+    # sampler: random_local | exploit_orthogonal | uncertainty_guided | axis_sweep | incumbent_mix | diversity_shell | line_search | plateau_escape | annealed_shell | spherical_cover
+    # updater: winner_average | winner_copy | linear_preference | score_weighted_preference | contrastive_preference | softmax_preference | borda_preference | bradley_terry_preference
     # feedback_mode: scalar_rating | pairwise | top_k | winner_only | approve_reject
     # seed_policy: fixed-per-round | fixed-per-candidate | fixed-per-candidate-role
     # steering_mode: currently low_dimensional
     # steering_dimension: low-dimensional steering vector size, for example 3 or 5
     # candidate_count: visible candidates per round
     # image_size: WIDTHxHEIGHT, for example 512x512
     # trust_radius: steering search radius around the current state
+    # stagnation_patience: rounds of identical selected image before challenger search widens
+    # stagnation_trust_radius_scale: multiplier applied to trust_radius during stagnation escape
     # anchor_strength: strength of the steering offset applied to prompt embeddings
     # guidance_scale: classifier-free guidance strength, for example 7.5
     # num_inference_steps: diffusion denoising steps, for example 15 or 30

app/core/schema.py

@@ -65,12 +65,22 @@ class SamplerType(str, Enum):
     uncertainty_guided = "uncertainty_guided"
     axis_sweep = "axis_sweep"
     incumbent_mix = "incumbent_mix"
+    diversity_shell = "diversity_shell"
+    line_search = "line_search"
+    plateau_escape = "plateau_escape"
+    annealed_shell = "annealed_shell"
+    spherical_cover = "spherical_cover"
 
 
 class UpdaterType(str, Enum):
     winner_average = "winner_average"
     winner_copy = "winner_copy"
     linear_preference = "linear_preference"
+    score_weighted_preference = "score_weighted_preference"
+    contrastive_preference = "contrastive_preference"
+    softmax_preference = "softmax_preference"
+    borda_preference = "borda_preference"
+    bradley_terry_preference = "bradley_terry_preference"
 
 
 class SteeringMode(str, Enum):
@@ -89,6 +99,8 @@ class StrategyConfig(BaseModel):
     candidate_count: int = Field(default=5, ge=1, le=12)
     image_size: str = "512x512"
     trust_radius: float = Field(default=0.55, gt=0.0, le=1.0)
+    stagnation_patience: int = Field(default=0, ge=0, le=10)
+    stagnation_trust_radius_scale: float = Field(default=1.0, ge=1.0, le=3.0)
     anchor_strength: float = Field(default=0.7, ge=0.0, le=2.0)
     guidance_scale: float = Field(default=7.5, gt=0.0, le=20.0)
     num_inference_steps: int = Field(default=15, ge=1, le=100)

app/engine/orchestrator.py

@@ -26,13 +26,23 @@
 from app.core.tracing import TraceRecorder
 from app.feedback.normalization import normalize_feedback
 from app.samplers.axis_sweep import AxisSweepSampler
+from app.samplers.annealed_shell import AnnealedShellSampler
 from app.samplers.base import clamp_vector
+from app.samplers.diversity_shell import DiversityShellSampler
 from app.samplers.exploit_orthogonal import ExploitOrthogonalSampler
 from app.samplers.incumbent_mix import IncumbentMixSampler
+from app.samplers.line_search import LineSearchSampler
+from app.samplers.plateau_escape import PlateauEscapeSampler
 from app.samplers.random_local import RandomLocalSampler
+from app.samplers.spherical_cover import SphericalCoverSampler
 from app.samplers.uncertainty import UncertaintyGuidedSampler
 from app.storage.repository import JsonRepository
+from app.updaters.contrastive_pref import ContrastivePreferenceUpdater
+from app.updaters.borda_pref import BordaPreferenceUpdater
+from app.updaters.bradley_terry_pref import BradleyTerryPreferenceUpdater
 from app.updaters.linear_pref import LinearPreferenceUpdater
+from app.updaters.softmax_pref import SoftmaxPreferenceUpdater
+from app.updaters.score_weighted import ScoreWeightedPreferenceUpdater
 from app.updaters.winner_average import WinnerAverageUpdater
 from app.updaters.winner_copy import WinnerCopyUpdater
 from app.engine.generation import GenerationEngine, build_generation_engine
@@ -56,11 +66,21 @@ def __init__(
             "uncertainty_guided": UncertaintyGuidedSampler(),
             "axis_sweep": AxisSweepSampler(),
             "incumbent_mix": IncumbentMixSampler(),
+            "diversity_shell": DiversityShellSampler(),
+            "line_search": LineSearchSampler(),
+            "plateau_escape": PlateauEscapeSampler(),
+            "annealed_shell": AnnealedShellSampler(),
+            "spherical_cover": SphericalCoverSampler(),
         }
         self.updaters = {
             "winner_copy": WinnerCopyUpdater(),
             "winner_average": WinnerAverageUpdater(),
             "linear_preference": LinearPreferenceUpdater(),
+            "score_weighted_preference": ScoreWeightedPreferenceUpdater(),
+            "contrastive_preference": ContrastivePreferenceUpdater(),
+            "softmax_preference": SoftmaxPreferenceUpdater(),
+            "borda_preference": BordaPreferenceUpdater(),
+            "bradley_terry_preference": BradleyTerryPreferenceUpdater(),
         }
 
     @staticmethod
@@ -164,8 +184,14 @@ def generate_round(
         carried_forward = self._build_carried_forward_candidate(session)
         baseline_candidate = self._build_baseline_prompt_candidate(session)
         sampler_seed = self._seed_token(session.id, round_index, "sampler")
+        sampler_session, stagnation_streak, boosted_radius = self._sampler_session_for_round(session)
         self._report_progress(progress_callback, 36, f"Sampling {session.config.candidate_count} candidate directions")
-        proposed_candidates = sampler.propose(session, sampler_seed)
+        proposed_candidates = sampler.propose(sampler_session, sampler_seed)
+        if boosted_radius is not None:
+            for candidate in proposed_candidates:
+                candidate.generation_params["stagnation_escape_active"] = True
+                candidate.generation_params["stagnation_streak"] = stagnation_streak
+                candidate.generation_params["stagnation_boosted_trust_radius"] = round(boosted_radius, 4)
         proposed_candidates = self._widen_first_round_candidates(session, proposed_candidates)
         candidates = self._compose_round_candidates(
             pinned_candidate=carried_forward or baseline_candidate,
@@ -573,6 +599,51 @@ def _assign_candidate_seeds(self, session: Session, round_index: int, candidates
             candidate.generation_params["seed_group"] = seed_group
             candidate.generation_params["round_seed"] = round_seed
 
+    def _sampler_session_for_round(self, session: Session) -> tuple[Session, int, float | None]:
+        """Return the sampling view of the session, optionally widened after repeated no-change wins."""
+
+        patience = int(session.config.stagnation_patience)
+        if patience <= 0 or session.current_round <= 0:
+            return session, 0, None
+
+        streak = self._trailing_selected_image_streak(session)
+        if streak < patience:
+            return session, streak, None
+
+        boosted_radius = min(1.0, session.config.trust_radius * session.config.stagnation_trust_radius_scale)
+        if boosted_radius <= session.config.trust_radius + 1e-9:
+            return session, streak, None
+
+        boosted_config = session.config.model_copy(update={"trust_radius": boosted_radius})
+        boosted_session = session.model_copy(deep=True)
+        boosted_session.config = boosted_config
+        return boosted_session, streak, boosted_radius
+
+    def _trailing_selected_image_streak(self, session: Session) -> int:
+        """Count trailing completed rounds that ended with the same selected image artifact."""
+
+        rounds = self.repository.list_rounds_for_session(session.id)
+        streak = 0
+        image_key: str | None = None
+        for round_obj in reversed(rounds):
+            if not round_obj.update_summary:
+                break
+            winner_id = round_obj.update_summary.get("winner_candidate_id")
+            if not winner_id:
+                break
+            winner = next((candidate for candidate in round_obj.candidates if candidate.id == winner_id), None)
+            if winner is None:
+                break
+            current_key = winner.image_path or repr([round(value, 6) for value in winner.z])
+            if image_key is None:
+                image_key = current_key
+                streak = 1
+                continue
+            if current_key != image_key:
+                break
+            streak += 1
+        return streak
+
     @staticmethod
     def _seed_token(*parts: object) -> int:
         """Create one stable positive seed from arbitrary deterministic inputs."""

app/samplers/annealed_shell.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import math
+
+from app.core.schema import Candidate, Session
+from app.samplers.base import clamp_vector, make_rng
+
+
+class AnnealedShellSampler:
+    """Sampler that shrinks from broad shell exploration toward local refinement over rounds."""
+
+    name = "annealed_shell"
+
+    def propose(self, session: Session, seed: int) -> list[Candidate]:
+        """Generate shell probes whose radius anneals with session progress."""
+
+        rng = make_rng(seed + 613)
+        dimensions = max(1, len(session.current_z))
+        progress = min(max(session.current_round, 0), 8) / 8.0
+        shell_fraction = 0.95 - (0.42 * progress)
+        shell_radius = min(max(session.config.trust_radius * shell_fraction, 0.2), session.config.trust_radius)
+        jitter_scale = 0.04 - (0.018 * progress)
+        candidates: list[Candidate] = []
+
+        for index in range(session.config.candidate_count):
+            direction = self._spread_direction(index, dimensions)
+            jitter = [rng.uniform(-jitter_scale, jitter_scale) for _ in range(dimensions)]
+            z = clamp_vector(
+                [
+                    current + (axis * shell_radius) + noise
+                    for current, axis, noise in zip(session.current_z, direction, jitter, strict=False)
+                ],
+                session.config.trust_radius,
+            )
+            candidates.append(
+                Candidate(
+                    round_id="",
+                    candidate_index=index,
+                    z=z,
+                    sampler_role="annealed_probe" if index % 2 == 0 else "annealed_counterprobe",
+                    predicted_score=sum(z) - (0.008 * index),
+                    predicted_uncertainty=max(0.05, 0.2 - (0.08 * progress) + (0.01 * index)),
+                    seed=seed,
+                    generation_params={
+                        "image_size": session.config.image_size,
+                        "annealed_progress": round(progress, 4),
+                        "shell_radius": round(shell_radius, 4),
+                        "jitter_scale": round(jitter_scale, 4),
+                        "spread_direction": [round(value, 4) for value in direction],
+                    },
+                )
+            )
+        return candidates
+
+    @staticmethod
+    def _spread_direction(index: int, dimensions: int) -> list[float]:
+        vector = [0.0 for _ in range(dimensions)]
+        primary_axis = index % dimensions
+        secondary_axis = (index + 1) % dimensions
+        tertiary_axis = (index + 2) % dimensions
+
+        vector[primary_axis] = 1.0 if index % 2 == 0 else -1.0
+        if dimensions > 1:
+            vector[secondary_axis] += 0.58 if index % 3 != 1 else -0.58
+        if dimensions > 2:
+            vector[tertiary_axis] += 0.26 if index % 4 < 2 else -0.26
+        if dimensions > 3:
+            vector[(index + 3) % dimensions] += 0.15 if index % 2 == 0 else -0.15
+
+        norm = math.sqrt(sum(value * value for value in vector))
+        if norm == 0.0:
+            vector[0] = 1.0
+            return vector
+        return [value / norm for value in vector]

app/samplers/diversity_shell.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import math
+
+from app.core.schema import Candidate, Session
+from app.samplers.base import clamp_vector, make_rng
+
+
+class DiversityShellSampler:
+    """Sampler that spreads challengers across a high-radius shell."""
+
+    name = "diversity_shell"
+
+    def propose(self, session: Session, seed: int) -> list[Candidate]:
+        """Generate deliberately separated shell probes around the current state."""
+
+        rng = make_rng(seed + 401)
+        dimensions = max(1, len(session.current_z))
+        shell_radius = min(max(session.config.trust_radius * 0.92, 0.28), session.config.trust_radius)
+        candidates: list[Candidate] = []
+
+        for index in range(session.config.candidate_count):
+            direction = self._spread_direction(index, dimensions)
+            jitter = [rng.uniform(-0.035, 0.035) for _ in range(dimensions)]
+            z = clamp_vector(
+                [
+                    current + (axis * shell_radius) + noise
+                    for current, axis, noise in zip(session.current_z, direction, jitter, strict=False)
+                ],
+                session.config.trust_radius,
+            )
+            candidates.append(
+                Candidate(
+                    round_id="",
+                    candidate_index=index,
+                    z=z,
+                    sampler_role="shell_probe" if index % 2 == 0 else "shell_counterprobe",
+                    predicted_score=sum(z) - (0.01 * index),
+                    predicted_uncertainty=0.18 + (0.03 * index),
+                    seed=seed,
+                    generation_params={
+                        "image_size": session.config.image_size,
+                        "shell_radius": round(shell_radius, 4),
+                        "spread_direction": [round(value, 4) for value in direction],
+                    },
+                )
+            )
+        return candidates
+
+    @staticmethod
+    def _spread_direction(index: int, dimensions: int) -> list[float]:
+        """Return a deterministic spread direction for one shell position."""
+
+        vector = [0.0 for _ in range(dimensions)]
+        primary_axis = index % dimensions
+        secondary_axis = (index + 1) % dimensions
+        tertiary_axis = (index + 2) % dimensions
+
+        primary_sign = 1.0 if index % 2 == 0 else -1.0
+        secondary_sign = -1.0 if index % 4 in {1, 2} else 1.0
+        tertiary_sign = -1.0 if index % 3 == 2 else 1.0
+
+        vector[primary_axis] = 1.0 * primary_sign
+        if dimensions > 1:
+            vector[secondary_axis] += 0.62 * secondary_sign
+        if dimensions > 2:
+            vector[tertiary_axis] += 0.28 * tertiary_sign
+        if dimensions > 3:
+            extra_axis = (index + 3) % dimensions
+            vector[extra_axis] += 0.18 if index % 2 == 0 else -0.18
+
+        length = math.sqrt(sum(value * value for value in vector))
+        if length == 0.0:
+            vector[0] = 1.0
+            return vector
+        return [value / length for value in vector]