Deepen paper package and sync docs

Author: apartsin

README.md

@@ -44,7 +44,7 @@ StableSteering is built around that gap. It turns generation into a feedback loo
 
 That makes the project useful both as:
 
-- a research platform for studying human-in-the-loop steering
+- a research platform for iterative preference-guided steering
 - a concrete prototype for interactive generative workflows
 
 ## Current MVP

app/bootstrap/experiment_models.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from huggingface_hub.utils import LocalEntryNotFoundError
+
+from app.core.config import settings
+
+_CLIP_CACHE: dict[tuple[str, str], tuple[object, object]] = {}
+_DINO_CACHE: dict[tuple[str, str], tuple[object, object]] = {}
+
+
+def huggingface_cache_dir() -> Path:
+    path = settings.huggingface_cache_dir
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def get_clip_components(model_id: str, device: str, *, local_only: bool = True):
+    key = (model_id, device)
+    cached = _CLIP_CACHE.get(key)
+    if cached is not None:
+        return cached
+
+    from transformers import CLIPModel, CLIPProcessor
+
+    cache_dir = huggingface_cache_dir()
+    try:
+        model = CLIPModel.from_pretrained(
+            model_id,
+            cache_dir=str(cache_dir),
+            local_files_only=local_only,
+        ).to(device)
+        processor = CLIPProcessor.from_pretrained(
+            model_id,
+            cache_dir=str(cache_dir),
+            local_files_only=local_only,
+        )
+    except (OSError, LocalEntryNotFoundError) as exc:
+        if local_only:
+            raise RuntimeError(
+                f"CLIP model '{model_id}' is not available in the local cache. "
+                "Run scripts/preload_experiment_models.py first."
+            ) from exc
+        raise
+    model.eval()
+    _CLIP_CACHE[key] = (model, processor)
+    return model, processor
+
+
+def get_dino_components(model_id: str, device: str, *, local_only: bool = True):
+    key = (model_id, device)
+    cached = _DINO_CACHE.get(key)
+    if cached is not None:
+        return cached
+
+    from transformers import AutoImageProcessor, AutoModel
+
+    cache_dir = huggingface_cache_dir()
+    try:
+        processor = AutoImageProcessor.from_pretrained(
+            model_id,
+            cache_dir=str(cache_dir),
+            local_files_only=local_only,
+        )
+        model = AutoModel.from_pretrained(
+            model_id,
+            cache_dir=str(cache_dir),
+            local_files_only=local_only,
+        ).to(device)
+    except (OSError, LocalEntryNotFoundError) as exc:
+        if local_only:
+            raise RuntimeError(
+                f"DINO model '{model_id}' is not available in the local cache. "
+                "Run scripts/preload_experiment_models.py first."
+            ) from exc
+        raise
+    model.eval()
+    _DINO_CACHE[key] = (processor, model)
+    return processor, model

app/bootstrap/huggingface.py

@@ -79,6 +79,26 @@ def prepare_huggingface_model(
     model_dir.mkdir(parents=True, exist_ok=True)
 
     allow_patterns = build_allow_patterns(extra_patterns)
+    manifest_path = model_dir / "prepare_manifest.json"
+    model_index_path = model_dir / "model_index.json"
+    if manifest_path.exists() and model_index_path.exists():
+        try:
+            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+        except json.JSONDecodeError:
+            manifest = None
+        if (
+            isinstance(manifest, dict)
+            and str(manifest.get("model_id")) == model_id
+            and manifest.get("revision") == revision
+            and list(manifest.get("allow_patterns", [])) == allow_patterns
+        ):
+            return {
+                "model_id": model_id,
+                "model_dir": str(model_dir),
+                "snapshot_path": str(model_dir),
+                "manifest_path": str(manifest_path),
+            }
+
     snapshot_path = snapshot_download(
         repo_id=model_id,
         revision=revision,

app/core/config.py

@@ -14,6 +14,7 @@ class Settings(BaseSettings):
     data_dir: Path = Path("data")
     artifacts_dir_name: str = "artifacts"
     models_dir: Path = Path("models")
+    huggingface_cache_dir: Path = Path("models") / "hf_cache"
     traces_dir_name: str = "traces"
     default_candidate_count: int = 4
     default_image_size: str = "512x512"

app/core/config_yaml.py

@@ -14,8 +14,8 @@
     # 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 | 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
+    # sampler: random_local | exploit_orthogonal | uncertainty_guided | axis_sweep | incumbent_mix | diversity_shell | line_search | plateau_escape | annealed_shell | spherical_cover | two_scale_cover | quality_diversity_mix
+    # updater: winner_average | winner_copy | linear_preference | score_weighted_preference | contrastive_preference | softmax_preference | borda_preference | bradley_terry_preference | challenger_mixture_preference | plackett_luce_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

app/core/schema.py

@@ -70,6 +70,8 @@ class SamplerType(str, Enum):
     plateau_escape = "plateau_escape"
     annealed_shell = "annealed_shell"
     spherical_cover = "spherical_cover"
+    two_scale_cover = "two_scale_cover"
+    quality_diversity_mix = "quality_diversity_mix"
 
 
 class UpdaterType(str, Enum):
@@ -81,6 +83,8 @@ class UpdaterType(str, Enum):
     softmax_preference = "softmax_preference"
     borda_preference = "borda_preference"
     bradley_terry_preference = "bradley_terry_preference"
+    challenger_mixture_preference = "challenger_mixture_preference"
+    plackett_luce_preference = "plackett_luce_preference"
 
 
 class SteeringMode(str, Enum):

app/engine/orchestrator.py

@@ -33,14 +33,18 @@
 from app.samplers.incumbent_mix import IncumbentMixSampler
 from app.samplers.line_search import LineSearchSampler
 from app.samplers.plateau_escape import PlateauEscapeSampler
+from app.samplers.quality_diversity_mix import QualityDiversityMixSampler
 from app.samplers.random_local import RandomLocalSampler
 from app.samplers.spherical_cover import SphericalCoverSampler
+from app.samplers.two_scale_cover import TwoScaleCoverSampler
 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.challenger_mixture import ChallengerMixturePreferenceUpdater
 from app.updaters.linear_pref import LinearPreferenceUpdater
+from app.updaters.plackett_luce_pref import PlackettLucePreferenceUpdater
 from app.updaters.softmax_pref import SoftmaxPreferenceUpdater
 from app.updaters.score_weighted import ScoreWeightedPreferenceUpdater
 from app.updaters.winner_average import WinnerAverageUpdater
@@ -71,6 +75,8 @@ def __init__(
             "plateau_escape": PlateauEscapeSampler(),
             "annealed_shell": AnnealedShellSampler(),
             "spherical_cover": SphericalCoverSampler(),
+            "two_scale_cover": TwoScaleCoverSampler(),
+            "quality_diversity_mix": QualityDiversityMixSampler(),
         }
         self.updaters = {
             "winner_copy": WinnerCopyUpdater(),
@@ -81,6 +87,8 @@ def __init__(
             "softmax_preference": SoftmaxPreferenceUpdater(),
             "borda_preference": BordaPreferenceUpdater(),
             "bradley_terry_preference": BradleyTerryPreferenceUpdater(),
+            "challenger_mixture_preference": ChallengerMixturePreferenceUpdater(),
+            "plackett_luce_preference": PlackettLucePreferenceUpdater(),
         }
 
     @staticmethod

app/samplers/quality_diversity_mix.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import math
+
+from app.core.schema import Candidate, Session
+from app.samplers.base import clamp_vector, make_rng
+
+
+class QualityDiversityMixSampler:
+    """Sampler inspired by quality-diversity search with several complementary emitters."""
+
+    name = "quality_diversity_mix"
+
+    def propose(self, session: Session, seed: int) -> list[Candidate]:
+        rng = make_rng(seed + 991)
+        dimensions = max(1, len(session.current_z))
+        base_direction = self._base_direction(session.current_z, dimensions)
+        lateral_direction = self._orthogonal_direction(base_direction)
+        cover_pool = [self._unit_vector([rng.uniform(-1.0, 1.0) for _ in range(dimensions)]) for _ in range(28)]
+        far_directions = self._greedy_cover(cover_pool, max(2, session.config.candidate_count // 2))
+
+        medium = min(max(session.config.trust_radius * 0.42, 0.16), session.config.trust_radius)
+        far = min(max(session.config.trust_radius * 0.82, 0.28), session.config.trust_radius)
+        counter = min(max(session.config.trust_radius * 0.3, 0.12), session.config.trust_radius)
+
+        patterns: list[tuple[str, list[float], float]] = [
+            ("qd_refine", base_direction, medium * 0.62),
+            ("qd_forward", base_direction, medium),
+            ("qd_lateral_plus", lateral_direction, medium),
+            ("qd_far_cover_1", far_directions[0], far),
+            ("qd_lateral_minus", [-value for value in lateral_direction], medium),
+            ("qd_counter", [-value for value in base_direction], counter),
+        ]
+        for index, direction in enumerate(far_directions[1:], start=2):
+            patterns.append((f"qd_far_cover_{index + 1}", direction, far))
+
+        candidates: list[Candidate] = []
+        for index in range(session.config.candidate_count):
+            role, direction, radius = patterns[index % len(patterns)]
+            jitter_scale = 0.014 if "refine" in role else 0.024 if "far_cover" not in role else 0.03
+            jitter = [rng.uniform(-jitter_scale, jitter_scale) for _ in range(dimensions)]
+            z = clamp_vector(
+                [
+                    current + (axis * 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=role,
+                    predicted_score=sum(z) + (0.01 if "far_cover" in role else 0.0),
+                    predicted_uncertainty=0.16 + (0.02 * index),
+                    seed=seed,
+                    generation_params={
+                        "image_size": session.config.image_size,
+                        "qd_radius": round(radius, 4),
+                        "qd_direction": [round(value, 4) for value in direction],
+                        "qd_emitter_role": role,
+                    },
+                )
+            )
+        return candidates
+
+    @staticmethod
+    def _base_direction(current_z: list[float], dimensions: int) -> list[float]:
+        length = math.sqrt(sum(value * value for value in current_z))
+        if length > 1e-8:
+            return [value / length for value in current_z]
+        direction = [0.0 for _ in range(dimensions)]
+        direction[0] = 1.0
+        if dimensions > 1:
+            direction[1] = 0.35
+        norm = math.sqrt(sum(value * value for value in direction))
+        return [value / norm for value in direction]
+
+    @staticmethod
+    def _orthogonal_direction(base_direction: list[float]) -> list[float]:
+        dimensions = len(base_direction)
+        if dimensions == 1:
+            return [1.0]
+        lateral = [0.0 for _ in range(dimensions)]
+        lateral[0] = -base_direction[1]
+        lateral[1] = base_direction[0]
+        for index in range(2, dimensions):
+            lateral[index] = base_direction[index] * (-0.45 if index % 2 == 0 else 0.45)
+        length = math.sqrt(sum(value * value for value in lateral))
+        if length == 0.0:
+            lateral[1] = 1.0
+            return lateral
+        return [value / length for value in lateral]
+
+    @classmethod
+    def _greedy_cover(cls, pool: list[list[float]], count: int) -> list[list[float]]:
+        if not pool:
+            return []
+        selected = [pool[0]]
+        remaining = pool[1:]
+        while remaining and len(selected) < count:
+            best_direction = max(
+                remaining,
+                key=lambda candidate: min(cls._angular_distance(candidate, prior) for prior in selected),
+            )
+            selected.append(best_direction)
+            remaining = [candidate for candidate in remaining if candidate is not best_direction]
+        return selected[:count]
+
+    @staticmethod
+    def _angular_distance(left: list[float], right: list[float]) -> float:
+        cosine = sum(a * b for a, b in zip(left, right, strict=False))
+        cosine = max(-1.0, min(1.0, cosine))
+        return math.acos(cosine)
+
+    @staticmethod
+    def _unit_vector(values: list[float]) -> list[float]:
+        norm = math.sqrt(sum(value * value for value in values))
+        if norm == 0.0:
+            fallback = [0.0 for _ in values]
+            fallback[0] = 1.0
+            return fallback
+        return [value / norm for value in values]