From 0f472b2f9e8406d6a84f46c0aa2bc098846d5785 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 30 Jun 2026 09:01:01 -0400 Subject: [PATCH 01/10] fix(explore): diversify "more like this" so it stops getting stuck (#1188) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pure nearest-cosine piled near-identical images into the neighbour grid — a reposted banner filled all 24 slots, and once you wandered into a B&W / comic-panel cluster every neighbour was more of the same with no way back to colour without the Random button (operator-reported, with screenshot). similar() now over-fetches a wide candidate pool (5x the requested limit, cap 200), then diversifies down to `limit`: - pHash near-duplicate collapse: drop candidates within 6 Hamming bits of the anchor or an already-kept candidate, so a repost (and the anchor's own clones) appears at most once. - MMR re-rank: greedily pick for closeness-to-anchor minus similarity-to-already -picked (lambda 0.55), so the result SPANS clusters instead of returning 40 variations of one image. Falls back to nearest-order on any failure / small pool, so existing nearest-first behaviour is unchanged when there's nothing to diversify. Frontend forwardTarget drops the now-redundant skip-nearest-third hack (the list is already diversified server-side) — plain random-over-unvisited gives the variance now. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa --- backend/app/services/gallery_service.py | 95 ++++++++++++++++++++++--- frontend/src/stores/explore.js | 12 ++-- tests/test_gallery_similar.py | 23 ++++++ 3 files changed, 115 insertions(+), 15 deletions(-) diff --git a/backend/app/services/gallery_service.py b/backend/app/services/gallery_service.py index cd93804..ca2a7df 100644 --- a/backend/app/services/gallery_service.py +++ b/backend/app/services/gallery_service.py @@ -289,6 +289,75 @@ def _gallery_images(rows, artists: dict[int, dict]) -> list[GalleryImage]: ] +def _diversify_similar(src, rows, limit, *, dup_threshold=6, lam=0.55): + """Trim a nearest-cosine candidate pool down to `limit` diverse picks. + + 1. pHash collapse: drop any candidate whose perceptual hash is within + `dup_threshold` Hamming bits of the anchor or an already-kept candidate — + so a reposted banner (and the anchor's own clones) appears at most once. + 2. MMR (Maximal Marginal Relevance): greedily pick the candidate maximising + `lam * sim_to_anchor - (1 - lam) * max_sim_to_already_picked`. This keeps + the most relevant up top but pushes the selection to SPAN clusters + instead of returning 40 variations of one image. + + Falls back to nearest-order (`rows[:limit]`) on any failure or a small pool. + """ + if len(rows) <= 1: + return rows[:limit] + try: + import imagehash + import numpy as np + except Exception: + return rows[:limit] + + # --- 1. pHash near-duplicate collapse (videos/NULL phash pass through) --- + kept = [] + seen = [] + if src.phash: + try: + seen.append(imagehash.hex_to_hash(src.phash)) + except Exception: + pass + for row in rows: + ph = row[0].phash + if ph: + try: + h = imagehash.hex_to_hash(ph) + if any((h - k) <= dup_threshold for k in seen): + continue + seen.append(h) + except Exception: + pass + kept.append(row) + if len(kept) <= limit: + return kept + + # --- 2. MMR re-rank on the L2-normalised SigLIP embeddings --- + try: + a = np.asarray(src.siglip_embedding, dtype=np.float32) + a = a / (np.linalg.norm(a) or 1.0) + V = np.vstack([ + np.asarray(row[0].siglip_embedding, dtype=np.float32) for row in kept + ]) + V = V / np.clip(np.linalg.norm(V, axis=1, keepdims=True), 1e-8, None) + except Exception: + return kept[:limit] + + rel = V @ a # (N,) cosine to the anchor + n = len(kept) + picked_mask = np.zeros(n, dtype=bool) + max_sim = np.zeros(n, dtype=np.float32) # max sim to anything picked yet + order = [] + for _ in range(min(limit, n)): + scores = lam * rel - (1.0 - lam) * max_sim + scores[picked_mask] = -np.inf + i = int(np.argmax(scores)) + order.append(i) + picked_mask[i] = True + max_sim = np.maximum(max_sim, V @ V[i]) + return [kept[i] for i in order] + + async def _artists_for(session, image_ids: list[int]) -> dict[int, dict]: """Map image_id -> {"name","slug"} via the canonical image_record.artist_id (FC-2d-vii-c). Bounded by page size.""" @@ -565,14 +634,20 @@ class GalleryService: untagged: bool = False, no_artist: bool = False, date_from: datetime | None = None, date_to: datetime | None = None, ) -> list[GalleryImage] | None: - """Visual "more like this": images ranked by cosine distance to - `image_id`'s SigLIP embedding (pgvector, HNSW-indexed — alembic 0036). - No ML inference here; the embedding was computed at import. + """Visual "more like this": images near `image_id`'s SigLIP embedding + (pgvector, HNSW-indexed — alembic 0036), then DIVERSIFIED so the result + doesn't collapse into one cluster. No ML inference here. - Returns None if the source image doesn't exist (→ 404), [] if it has - no embedding (a video / not-yet-embedded). Composes with the Phase-1/2 - scope filters (AND) but REPLACES the date sort — always nearest-first, - bounded to `limit` (no cursor; distance-ranking has no date cursor). + Pure nearest-cosine piles up near-identical images — a reposted banner + fills the whole grid, and once you wander into a B&W / comic-panel + cluster every neighbour is more of the same with no way back to colour + (operator-reported 2026-06-30). So we pull a WIDER candidate pool, then: + 1. collapse near-duplicate pHashes (and drop clones of the anchor), + 2. MMR re-rank — pick for closeness-to-anchor but penalise similarity + to what's already picked, so the result SPANS clusters. + + Returns None if the source doesn't exist (→ 404), [] if it has no + embedding. Composes with the scope filters (AND); REPLACES the date sort. """ if limit < 1 or limit > 200: raise ValueError("limit must be between 1 and 200") @@ -582,6 +657,9 @@ class GalleryService: if src.siglip_embedding is None: return [] + # Over-fetch so diversification has clusters to spread across — without a + # wide pool there's nothing but the near-dupes to choose from. + pool_n = min(200, max(limit * 5, 60)) distance = ImageRecord.siglip_embedding.cosine_distance(src.siglip_embedding) eff = _effective_date_col() stmt = select(ImageRecord, Post.post_date, eff.label("eff")) @@ -597,8 +675,9 @@ class GalleryService: platform=platform, untagged=untagged, no_artist=no_artist, date_from=date_from, date_to=date_to, ) - stmt = stmt.order_by(distance.asc()).limit(limit) + stmt = stmt.order_by(distance.asc()).limit(pool_n) rows = (await self.session.execute(stmt)).all() + rows = _diversify_similar(src, rows, limit) artists = await _artists_for(self.session, [r[0].id for r in rows]) return _gallery_images(rows, artists) diff --git a/frontend/src/stores/explore.js b/frontend/src/stores/explore.js index 1bb269f..8b5c45f 100644 --- a/frontend/src/stores/explore.js +++ b/frontend/src/stores/explore.js @@ -93,13 +93,11 @@ export const useExploreStore = defineStore('explore', () => { // a crumb (which snaps the cursor back into the trail — the "loops back" // report). Fall back to the full set only if every neighbour's been seen. const seen = new Set(breadcrumb.value.map((c) => c.id)) - let pool = neighbors.value.filter((n) => !seen.has(n.id)) - if (!pool.length) pool = neighbors.value - // neighbors come similarity-sorted (nearest first). Skip the closest slice — - // those near-duplicates are exactly what you get stuck cycling through — and - // pick from the more-varied remainder, for real variance in the walk. - const skip = pool.length >= 6 ? Math.floor(pool.length / 3) : 0 - const cands = pool.slice(skip) + const pool = neighbors.value.filter((n) => !seen.has(n.id)) + const cands = pool.length ? pool : neighbors.value + // The list is already pHash-deduped + MMR-diversified server-side (it spans + // clusters, not 40 near-dupes), so a plain random pick gives real variance — + // no need to skip the nearest slice the way the raw nearest-list required. return cands[Math.floor(Math.random() * cands.length)].id } diff --git a/tests/test_gallery_similar.py b/tests/test_gallery_similar.py index 07ae149..ea9b355 100644 --- a/tests/test_gallery_similar.py +++ b/tests/test_gallery_similar.py @@ -87,6 +87,29 @@ async def test_similar_composes_with_tag_filter(db): assert [i.id for i in res] == [tagged.id] # scope AND-narrows the ranked set +@pytest.mark.asyncio +async def test_similar_collapses_near_duplicate_phashes(db): + # The reported failure: a reposted image fills the whole neighbour grid. + # A wall of same-pHash reposts must collapse to at most one, and the + # genuinely distinct images must still come through. + src = await _img(db, 1, _vec(1, 0)) + dupes = [] + for n in range(2, 7): # 5 near-identical reposts + r = await _img(db, n, _vec(1, 0.01 * n)) + r.phash = "ffffffffffffffff" # identical perceptual hash + dupes.append(r) + distinct_a = await _img(db, 7, _vec(1, 1)) + distinct_a.phash = "0000000000000000" + distinct_b = await _img(db, 8, _vec(0, 1)) + distinct_b.phash = "0f0f0f0f0f0f0f0f" + await db.flush() + + res = await GalleryService(db).similar(src.id, limit=10) + ids = [i.id for i in res] + assert sum(1 for d in dupes if d.id in ids) <= 1 # repost wall collapsed + assert distinct_a.id in ids and distinct_b.id in ids + + @pytest.mark.asyncio async def test_similar_respects_limit(db): src = await _img(db, 1, _vec(1, 0)) From 4daa3f27903838a5be98d8a0bd35ba83d2597f96 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 30 Jun 2026 10:24:30 -0400 Subject: [PATCH 02/10] =?UTF-8?q?feat(ml):=20operator=20model=20swap=20?= =?UTF-8?q?=E2=80=94=20GPU=20re-embed=20+=20embedder=20as=20a=20setting=20?= =?UTF-8?q?(#1190)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make the SigLIP embedder an operator choice (drop-in to SigLIP 2: google/siglip2-so400m-patch16-512 is a verified 1152-d model at 512px → no schema change, better small-cue fidelity). A swap = set model + re-embed + retrain, all operator-driven; the GPU agent does the re-embed so it's fast. - settings: embedder_model_name is now a setting (migration 0065) alongside the existing embedder_model_version; both editable + validated (non-empty) in the ml admin API. The server embedder loads by HF name (AutoImageProcessor/Model, model-agnostic), preferring the pre-downloaded local dir for the default so existing deploys don't re-download; rebuilds on a name change. - agent: new 'embed' job = whole-image SigLIP embedding (mean-pool video frames) under the lease-announced model → POST /jobs/submit_embedding writes image_record.siglip_embedding + siglip_model_version. The lease now announces the model FROM THE SETTING (not a constant). - re-embed routing: enqueue_gpu_backfill('embed') selects unembedded + stale- version images; 'siglip' now re-embeds concept crops whose version != current (so a swap re-triggers crops, not just the never-embedded back-catalogue). The CPU ml-worker backfill no longer re-embeds on a version mismatch (it can't churn the library at 512px) — the GPU agent owns version re-embeds. Daily 'embed' + 'siglip' beats self-heal. - scoring: score_image only bags embeddings in the CURRENT model's space (whole- image gated by siglip_model_version, concept regions by embedding_version) so a mid-swap stale vector isn't scored by new-space heads; legacy NULL = current. - UI: GpuAgentCard "Embedding model (advanced)" — edit name/version, Save, and "Re-embed library (GPU)" (queues embed + siglip); points at SigLIP 2. Tests: lease announces model + submit_embedding round-trip; enqueue 'embed' selects stale/unembedded; stale-version excluded from scoring; embedder model settable + empty rejected; siglip gate updated to current-version concept. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa --- agent/fc_agent/client.py | 13 ++++ agent/fc_agent/worker.py | 37 +++++++--- alembic/versions/0065_embedder_model_name.py | 35 ++++++++++ backend/app/api/gpu.py | 40 +++++++++-- backend/app/api/ml_admin.py | 8 +++ backend/app/celery_app.py | 5 ++ backend/app/models/ml_settings.py | 6 ++ backend/app/services/ml/embedder.py | 53 +++++++++------ backend/app/services/ml/heads.py | 19 ++++-- backend/app/tasks/ml.py | 43 +++++++++--- .../src/components/settings/GpuAgentCard.vue | 68 +++++++++++++++++++ tests/test_api_gpu.py | 33 +++++++++ tests/test_api_ml_admin.py | 20 ++++++ tests/test_gpu_jobs.py | 38 ++++++++++- tests/test_ml_suggestions.py | 14 ++++ 15 files changed, 379 insertions(+), 53 deletions(-) create mode 100644 alembic/versions/0065_embedder_model_name.py diff --git a/agent/fc_agent/client.py b/agent/fc_agent/client.py index 7003bcc..1c297c9 100644 --- a/agent/fc_agent/client.py +++ b/agent/fc_agent/client.py @@ -40,6 +40,19 @@ class FcClient: r.raise_for_status() return r.json() + def submit_embedding(self, job_id: int, embedding: list, version: str) -> dict: + """Post a whole-image SigLIP embedding (the 'embed' task) → image_record.""" + r = self.s.post( + f"{self.base}/api/gpu/jobs/submit_embedding", + json={ + "agent_id": self.agent_id, "job_id": job_id, + "embedding": embedding, "embedding_version": version, + }, + timeout=120, + ) + r.raise_for_status() + return r.json() + def heartbeat(self, job_ids: list[int]) -> None: try: self.s.post( diff --git a/agent/fc_agent/worker.py b/agent/fc_agent/worker.py index 8b42277..dd5b206 100644 --- a/agent/fc_agent/worker.py +++ b/agent/fc_agent/worker.py @@ -11,6 +11,7 @@ orphaned work is re-picked at once rather than waiting out the lease. """ import threading +import numpy as np import requests from . import media, models @@ -193,28 +194,42 @@ class Worker: else: frames = [(None, media.load_image(data))] + task = job.get("task") or "ccip" + embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION + model_name = ( + self.cfg.embed_model_override + or job.get("embed_model_name") + or DEFAULT_EMBED_MODEL + ) + + # 'embed' = WHOLE-IMAGE SigLIP embedding (re-embed the library under a + # new model, #1190) → image_record.siglip_embedding. Mean-pool video + # frames, matching the server's tag_and_embed. No regions. + if task == "embed": + embedder = self._ensure_embedder(model_name) + vecs = [embedder.embed(frame) for _, frame in frames] + if len(vecs) > 1: + vec = np.mean( + np.asarray(vecs, dtype=np.float32), axis=0 + ).tolist() + else: + vec = vecs[0] + self.client.submit_embedding(job["job_id"], vec, embed_version) + self._bump(processed=1) + return True + # task picks what to produce per crop: # 'siglip' (backfill existing images) → concept (SigLIP) regions # ONLY, so it never churns their figure/CCIP regions or the # character-reference cache. # 'ccip' / 'both' (a new image's first pass) → figure (CCIP) AND # concept (SigLIP) in one go, off the same crop. - task = job.get("task") or "ccip" want_ccip = task in ("ccip", "both") want_siglip = task in ("ccip", "siglip", "both") replace_kinds = ( ["concept"] if task == "siglip" else ["figure", "face", "concept"] ) - - embed_version = job.get("embed_version") or DEFAULT_EMBED_VERSION - embedder = None - if want_siglip: - model_name = ( - self.cfg.embed_model_override - or job.get("embed_model_name") - or DEFAULT_EMBED_MODEL - ) - embedder = self._ensure_embedder(model_name) + embedder = self._ensure_embedder(model_name) if want_siglip else None regions = [] ccip_ev = self.cfg.ccip_model or "ccip-default" diff --git a/alembic/versions/0065_embedder_model_name.py b/alembic/versions/0065_embedder_model_name.py new file mode 100644 index 0000000..0a986b3 --- /dev/null +++ b/alembic/versions/0065_embedder_model_name.py @@ -0,0 +1,35 @@ +"""ml_settings: embedder_model_name (#1190 operator model swap) + +The embedder MODEL VERSION was already a setting (and stamps image_record. +siglip_model_version); the HF model NAME was env-only, so an operator couldn't +actually point the pipeline at a different embedder. Storing the name as a +setting makes the model an operator choice: set name + version → re-embed (the +GPU agent) → retrain heads. Default = the current SigLIP so400m. + +Revision ID: 0065 +Revises: 0064 +Create Date: 2026-06-30 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0065" +down_revision: Union[str, None] = "0064" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "ml_settings", + sa.Column( + "embedder_model_name", sa.String(length=128), nullable=False, + server_default="google/siglip-so400m-patch14-384", + ), + ) + + +def downgrade() -> None: + op.drop_column("ml_settings", "embedder_model_name") diff --git a/backend/app/api/gpu.py b/backend/app/api/gpu.py index cf8fc73..09a488f 100644 --- a/backend/app/api/gpu.py +++ b/backend/app/api/gpu.py @@ -17,7 +17,6 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from ..extensions import get_session from ..models import AppSetting, GpuJob, ImageRecord, MLSettings from ..services.gallery_service import image_url -from ..services.ml.embedder import MODEL_NAME as EMBED_MODEL_NAME from ..services.ml.gpu_jobs import GpuJobService from ..services.ml.regions import RegionService @@ -138,11 +137,12 @@ async def lease(): # For video/animated: the agent samples at this cadence. "frame_interval_seconds": ml.video_frame_interval_seconds, "max_frames": ml.video_max_frames, - # The embedding model the agent must use for concept crops, so - # its region vectors land in the SAME space the heads trained in. - # Server-announced → the agent stays model-agnostic; a swap is a - # server setting + a re-embed migration, never an agent change. - "embed_model_name": EMBED_MODEL_NAME, + # The embedding model the agent must use for concept crops + the + # whole-image 'embed' task, so its vectors land in the SAME space + # the heads trained in. Server-announced FROM THE SETTING → the + # agent stays model-agnostic; an operator swap is a setting + a + # re-embed, never an agent change. + "embed_model_name": ml.embedder_model_name, "embed_version": ml.embedder_model_version, }) return jsonify({"jobs": out}) @@ -188,6 +188,34 @@ async def submit(): return jsonify({"ok": True, "stored": len(regions)}) +@gpu_bp.route("/jobs/submit_embedding", methods=["POST"]) +async def submit_embedding(): + """Store a whole-image SigLIP embedding (the 'embed' task) on image_record + + close the job. Body: {agent_id, job_id, embedding:[...], embedding_version}. + This is how the GPU agent re-embeds the library under a new model (#1190) — + much faster than the CPU ml-worker at higher resolutions.""" + body = await request.get_json(silent=True) or {} + agent_id = str(body.get("agent_id") or "agent") + job_id = body.get("job_id") + embedding = body.get("embedding") + version = body.get("embedding_version") + if job_id is None or not embedding or not version: + return jsonify({"error": "job_id, embedding, embedding_version required"}), 400 + async with get_session() as session: + if not await _agent_authed(session): + return jsonify({"error": "unauthorized"}), 401 + job = await session.get(GpuJob, int(job_id)) + if job is None or job.status != "leased" or job.lease_token != agent_id: + return jsonify({"error": "lease_invalid"}), 409 + img = await session.get(ImageRecord, job.image_record_id) + if img is not None: + img.siglip_embedding = embedding + img.siglip_model_version = version + await GpuJobService(session).complete(agent_id, int(job_id)) + await session.commit() + return jsonify({"ok": True}) + + @gpu_bp.route("/jobs/fail", methods=["POST"]) async def fail(): body = await request.get_json(silent=True) or {} diff --git a/backend/app/api/ml_admin.py b/backend/app/api/ml_admin.py index 1472770..102c004 100644 --- a/backend/app/api/ml_admin.py +++ b/backend/app/api/ml_admin.py @@ -24,6 +24,8 @@ _EDITABLE = ( "ccip_match_threshold", "ccip_auto_apply_enabled", "ccip_auto_apply_threshold", + "embedder_model_name", + "embedder_model_version", ) @@ -54,6 +56,7 @@ async def get_settings(): "ccip_match_threshold": s.ccip_match_threshold, "ccip_auto_apply_enabled": s.ccip_auto_apply_enabled, "ccip_auto_apply_threshold": s.ccip_auto_apply_threshold, + "embedder_model_name": s.embedder_model_name, } ) @@ -125,6 +128,11 @@ def _validate(p: dict) -> str | None: return "ccip_match_threshold must be between 0.5 and 0.999" if not (0.5 <= float(p["ccip_auto_apply_threshold"]) <= 0.999): return "ccip_auto_apply_threshold must be between 0.5 and 0.999" + # Embedder model swap (#1190): both must be non-empty. Changing them means a + # different embedding space — the operator must re-embed + retrain after. + for key in ("embedder_model_name", "embedder_model_version"): + if not str(p[key]).strip(): + return f"{key} must not be empty" return None diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 5778a1e..7b7a4e3 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -131,6 +131,11 @@ def make_celery() -> Celery: "schedule": 86400.0, # drain the concept-crop back-catalogue + "args": ("siglip",), # retry failed embeds, no button needed }, + "enqueue-embed-backfill-daily": { + "task": "backend.app.tasks.ml.enqueue_gpu_backfill", + "schedule": 86400.0, # whole-image re-embed under the current + "args": ("embed",), # model (an operator swap) drains via agent + }, "ccip-auto-apply-daily": { "task": "backend.app.tasks.ml.scheduled_ccip_auto_apply", "schedule": 86400.0, # no-op unless ccip_auto_apply_enabled diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index 9825568..0e84940 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -107,6 +107,12 @@ class MLSettings(Base): embedder_model_version: Mapped[str] = mapped_column( String(128), nullable=False, default="siglip-so400m-patch14-384" ) + # The HF model NAME the embedder loads (server CPU embed + announced to the + # GPU agent in the lease). Operator-settable so the embedder is a choice, not + # a hardcode (#1190): set name + version together, then re-embed + retrain. + embedder_model_name: Mapped[str] = mapped_column( + String(128), nullable=False, default="google/siglip-so400m-patch14-384" + ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) diff --git a/backend/app/services/ml/embedder.py b/backend/app/services/ml/embedder.py index d94f71e..d55646c 100644 --- a/backend/app/services/ml/embedder.py +++ b/backend/app/services/ml/embedder.py @@ -18,9 +18,11 @@ ImageFile.LOAD_TRUNCATED_IMAGES = True # N_replicas × this within the cores allotted to ML to avoid oversubscription. _INTRA_OP_THREADS = 4 -MODEL_NAME = os.environ.get( +DEFAULT_MODEL_NAME = os.environ.get( "SIGLIP_MODEL_NAME", "google/siglip-so400m-patch14-384" ) +# Back-compat alias (api/gpu imported this name as the fallback embedder id). +MODEL_NAME = DEFAULT_MODEL_NAME MODEL_VERSION = os.environ.get( "SIGLIP_MODEL_VERSION", "siglip-so400m-patch14-384" ) @@ -29,35 +31,42 @@ _LOCAL_DIR = Path(os.environ.get("ML_MODEL_DIR", "/models")) / "siglip" class Embedder: - def __init__(self, model_dir: Path | None = None): - self._model_dir = model_dir or _LOCAL_DIR + """Loads whatever SigLIP-family model it's given by HF NAME. For the default + model it prefers the pre-downloaded local dir (no re-download on existing + deploys); any other name resolves as an HF repo id (downloaded + cached on + first use), so an operator model swap (#1190) just works server-side.""" + + def __init__(self, model_name: str | None = None, model_dir: Path | None = None): + self.model_name = model_name or DEFAULT_MODEL_NAME + self._explicit_dir = model_dir self._model = None self._processor = None self._torch = None + def _source(self) -> str: + if self._explicit_dir is not None: + return str(self._explicit_dir) + if self.model_name == DEFAULT_MODEL_NAME and _LOCAL_DIR.exists(): + return str(_LOCAL_DIR) + return self.model_name + def load(self) -> None: if self._model is not None: return import torch - from transformers import AutoModel, SiglipImageProcessor + from transformers import AutoImageProcessor, AutoModel self._torch = torch # Bound torch's CPU thread pool (see _INTRA_OP_THREADS) so each replica # stays a predictable core consumer on a shared node. torch.set_num_threads(_INTRA_OP_THREADS) - # FC's embedder only does IMAGE inference — never text. AutoProcessor - # loads the full processor including SiglipTokenizer, which requires - # the sentencepiece library at import time even if we never call it. - # SiglipImageProcessor loads ONLY preprocessor_config.json (image - # side) and skips the tokenizer config entirely. Operator hit the - # ImportError 2026-05-25 once the ml-worker started actually running - # tag_and_embed; switching to the image-only loader avoids the - # tokenizer dep without adding ~30 MB of unused C++ build to the - # lean ml-worker image. - self._processor = SiglipImageProcessor.from_pretrained( - str(self._model_dir) - ) - self._model = AutoModel.from_pretrained(str(self._model_dir)) + # IMAGE inference only — AutoImageProcessor loads just the image side + # (preprocessor_config.json), skipping the SigLIP tokenizer + its + # sentencepiece dep (operator hit that ImportError 2026-05-25). Works + # for any SigLIP-family model, keeping the embedder model-agnostic. + src = self._source() + self._processor = AutoImageProcessor.from_pretrained(src) + self._model = AutoModel.from_pretrained(src) self._model.eval() def infer(self, image_path: Path) -> np.ndarray: @@ -74,8 +83,12 @@ class Embedder: _default_embedder: Embedder | None = None -def get_embedder() -> Embedder: +def get_embedder(model_name: str | None = None) -> Embedder: + """Cached embedder for `model_name` (default if None). Rebuilds the singleton + when the requested name changes, so an operator model swap takes effect + without restarting the worker.""" global _default_embedder - if _default_embedder is None: - _default_embedder = Embedder() + name = model_name or DEFAULT_MODEL_NAME + if _default_embedder is None or _default_embedder.model_name != name: + _default_embedder = Embedder(model_name=name) return _default_embedder diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py index 879b922..b234897 100644 --- a/backend/app/services/ml/heads.py +++ b/backend/app/services/ml/heads.py @@ -308,25 +308,36 @@ async def score_image( import numpy as np img = await session.get(ImageRecord, image_id) - if img is None or img.siglip_embedding is None: + if img is None: return [] settings = await _settings_async(session) - heads = await _current_heads(session, settings.embedder_model_version) + cur_version = settings.embedder_model_version + heads = await _current_heads(session, cur_version) if heads["W"] is None: return [] - bag = [np.asarray(img.siglip_embedding, dtype=np.float32)] + # Only embeddings in the CURRENT model's space enter the bag. Mid model-swap + # (#1190), an image still carrying the OLD-version whole-image vector is + # skipped rather than scored by heads trained in a different space; a legacy + # NULL version is treated as current (those predate per-row stamping). + bag = [] + if img.siglip_embedding is not None and img.siglip_model_version in ( + cur_version, None, + ): + bag.append(np.asarray(img.siglip_embedding, dtype=np.float32)) region_vecs = ( await session.execute( select(ImageRegion.siglip_embedding) .where(ImageRegion.image_record_id == image_id) .where(ImageRegion.siglip_embedding.is_not(None)) - .where(ImageRegion.embedding_version == settings.embedder_model_version) + .where(ImageRegion.embedding_version == cur_version) ) ).all() for (vec,) in region_vecs: if vec is not None: bag.append(np.asarray(vec, dtype=np.float32)) + if not bag: + return [] X = np.vstack(bag) # (B, D) norms = np.linalg.norm(X, axis=1, keepdims=True) diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 84b40a2..f2aa6a7 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -95,7 +95,7 @@ def tag_and_embed(self, image_id: int) -> dict: phase = "load_models" tagger = get_tagger() - embedder = get_embedder() + embedder = get_embedder(settings.embedder_model_name) if is_vid: # Layer-3 isolation: ffprobe (a separate process) validates @@ -330,10 +330,10 @@ def backfill(self) -> int: != settings.tagger_model_version ) | (ImageRecord.siglip_embedding.is_(None)) - | ( - ImageRecord.siglip_model_version - != settings.embedder_model_version - ) + # NB: a siglip MODEL-VERSION mismatch (an operator model swap, + # #1190) is intentionally NOT re-embedded here — the CPU + # ml-worker can't churn the whole library at 384/512px. The + # GPU agent owns version re-embeds via the 'embed' job. ) .order_by(ImageRecord.id.asc()) .limit(500) @@ -750,17 +750,40 @@ def enqueue_gpu_backfill(task_name: str) -> int: job, so it picks up the back-catalogue of images that were CCIP-embedded before concept crops existed, and retries images whose concept embed failed — without re-touching their figure/CCIP regions.""" - from sqlalchemy import exists, insert, literal + from sqlalchemy import exists, insert, literal, or_ from sqlalchemy import select as sa_select - from ..models import GpuJob, ImageRecord, ImageRegion + from ..models import GpuJob, ImageRecord, ImageRegion, MLSettings SessionLocal = _sync_session_factory() with SessionLocal() as session: - if task_name == "siglip": - has_concept = exists().where( + cur_version = session.execute( + select(MLSettings.embedder_model_version).where(MLSettings.id == 1) + ).scalar_one() + if task_name == "embed": + # Whole-image GPU re-embed (#1190): images with no embedding, or one + # stamped under a DIFFERENT model version (an operator model swap). + stale = or_( + ImageRecord.siglip_embedding.is_(None), + ImageRecord.siglip_model_version.is_(None), + ImageRecord.siglip_model_version != cur_version, + ) + queued = exists().where( + GpuJob.image_record_id == ImageRecord.id, + GpuJob.task == "embed", + GpuJob.status.in_(["pending", "leased"]), + ) + sel = sa_select( + ImageRecord.id, literal("embed"), literal("pending") + ).where(stale).where(~queued) + elif task_name == "siglip": + # Concept-crop re-embed: enqueue when there's no concept region AT THE + # CURRENT model version — so a model swap re-triggers crops too, not + # only the never-embedded back-catalogue. + has_current_concept = exists().where( ImageRegion.image_record_id == ImageRecord.id, ImageRegion.kind == "concept", + ImageRegion.embedding_version == cur_version, ) queued = exists().where( GpuJob.image_record_id == ImageRecord.id, @@ -769,7 +792,7 @@ def enqueue_gpu_backfill(task_name: str) -> int: ) sel = sa_select( ImageRecord.id, literal("siglip"), literal("pending") - ).where(~has_concept).where(~queued) + ).where(~has_current_concept).where(~queued) else: already = exists().where( GpuJob.image_record_id == ImageRecord.id, diff --git a/frontend/src/components/settings/GpuAgentCard.vue b/frontend/src/components/settings/GpuAgentCard.vue index dc0ed24..b0da818 100644 --- a/frontend/src/components/settings/GpuAgentCard.vue +++ b/frontend/src/components/settings/GpuAgentCard.vue @@ -106,6 +106,37 @@ reversible) — so identity tags keep flowing without review. Stricter than the suggest cut; 0.92 recommended.

+ + +
Embedding model (advanced)
+
+ + +
+ Save model + Re-embed library (GPU) +
+

+ Changing the model means a DIFFERENT embedding space. After saving a new + model + version, run Re-embed library (the GPU agent re-embeds + whole images + concept crops), then Retrain heads. Suggestions + degrade until both finish. SigLIP 2 (google/siglip2-so400m-patch16-512, + version siglip2-so400m-patch16-512) is a 1152-d drop-in at + 512px — no schema change. +

+
@@ -131,6 +162,10 @@ const savingThreshold = ref(false) const autoApply = ref(true) const autoThreshold = ref(0.92) const savingAuto = ref(false) +const modelName = ref('') +const modelVersion = ref('') +const savingModel = ref(false) +const reembedding = ref(false) const queue = ref({ pending: 0, leased: 0, done: 0, error: 0 }) let pollTimer = null @@ -157,9 +192,42 @@ onMounted(async () => { autoApply.value = ml.settings.ccip_auto_apply_enabled autoThreshold.value = ml.settings.ccip_auto_apply_threshold } + if (ml.settings?.embedder_model_name != null) { + modelName.value = ml.settings.embedder_model_name + modelVersion.value = ml.settings.embedder_model_version + } } catch { /* non-fatal */ } }) +async function onSaveModel() { + savingModel.value = true + try { + await ml.patchSettings({ + embedder_model_name: modelName.value.trim(), + embedder_model_version: modelVersion.value.trim(), + }) + toast({ text: 'Embedding model saved — now Re-embed library, then Retrain heads', type: 'success' }) + } catch (e) { + toast({ text: `Could not save model: ${e.message}`, type: 'error' }) + } finally { + savingModel.value = false + } +} + +async function onReembed() { + reembedding.value = true + try { + await store.backfill('embed') + await store.backfill('siglip') + toast({ text: 'Queued whole-image + concept re-embed — run the agent, then Retrain heads', type: 'success' }) + await refreshQueue() + } catch (e) { + toast({ text: `Could not queue re-embed: ${e.message}`, type: 'error' }) + } finally { + reembedding.value = false + } +} + async function onSaveAuto() { savingAuto.value = true try { diff --git a/tests/test_api_gpu.py b/tests/test_api_gpu.py index 40345d8..08fbc5e 100644 --- a/tests/test_api_gpu.py +++ b/tests/test_api_gpu.py @@ -69,6 +69,39 @@ async def test_lease_submit_round_trip(client, db): assert len(regs) == 1 and len(list(regs[0].ccip_embedding)) == 768 +@pytest.mark.asyncio +async def test_lease_announces_embed_model_then_submit_embedding(client, db): + # Whole-image GPU re-embed (#1190): the lease announces the embedder model so + # the agent loads the right one, and submit_embedding writes it back onto + # image_record with its version stamp. + img = await _img(db, "b" * 64) + await GpuJobService(db).enqueue(img.id, "embed") + await db.commit() + + token = (await (await client.post("/api/gpu/token/rotate")).get_json())["token"] + hdr = {"Authorization": f"Bearer {token}"} + + leased = await client.post( + "/api/gpu/jobs/lease", json={"agent_id": "a1", "batch_size": 5}, headers=hdr, + ) + j = (await leased.get_json())["jobs"][0] + assert j["task"] == "embed" + assert j["embed_model_name"] and j["embed_version"] # server-announced model + + submitted = await client.post("/api/gpu/jobs/submit_embedding", json={ + "agent_id": "a1", "job_id": j["job_id"], + "embedding": [0.2] * 1152, "embedding_version": "siglip2-test-v9", + }, headers=hdr) + assert submitted.status_code == 200 + + st = await (await client.get("/api/gpu/status")).get_json() + assert st["done"] == 1 and st["leased"] == 0 + + await db.refresh(img) + assert img.siglip_model_version == "siglip2-test-v9" + assert img.siglip_embedding is not None and len(list(img.siglip_embedding)) == 1152 + + @pytest.mark.asyncio async def test_submit_with_stale_lease_is_409(client, db): img = await _img(db, "b" * 64) diff --git a/tests/test_api_ml_admin.py b/tests/test_api_ml_admin.py index e46be48..ebc944d 100644 --- a/tests/test_api_ml_admin.py +++ b/tests/test_api_ml_admin.py @@ -34,6 +34,26 @@ async def test_get_and_patch_settings(client): assert (await resp.get_json())["suggestion_threshold_general"] == pytest.approx(0.90) +@pytest.mark.asyncio +async def test_embedder_model_settable_and_empty_rejected(client): + # #1190: the embedder model name + version are operator-settable (a swap), + # and neither may be blanked. + body = await (await client.get("/api/ml/settings")).get_json() + assert body["embedder_model_name"] == "google/siglip-so400m-patch14-384" + + ok = await client.patch("/api/ml/settings", json={ + "embedder_model_name": "google/siglip2-so400m-patch16-512", + "embedder_model_version": "siglip2-so400m-patch16-512", + }) + assert ok.status_code == 200 + out = await ok.get_json() + assert out["embedder_model_name"] == "google/siglip2-so400m-patch16-512" + assert out["embedder_model_version"] == "siglip2-so400m-patch16-512" + + bad = await client.patch("/api/ml/settings", json={"embedder_model_name": " "}) + assert bad.status_code == 400 + + @pytest.mark.asyncio async def test_tagger_store_floor_default_and_patch(client): body = await (await client.get("/api/ml/settings")).get_json() diff --git a/tests/test_gpu_jobs.py b/tests/test_gpu_jobs.py index 3606755..38d23ca 100644 --- a/tests/test_gpu_jobs.py +++ b/tests/test_gpu_jobs.py @@ -25,13 +25,17 @@ async def test_enqueue_siglip_backfill_gates_on_concept_region(db): # 'siglip' backfill enqueues images that lack a concept region (the # back-catalogue) and skips ones that already have one — and never double- # enqueues an image that already has a pending siglip job. + from backend.app.models import MLSettings from backend.app.tasks.ml import enqueue_gpu_backfill + cur = (await db.execute( + select(MLSettings.embedder_model_version).where(MLSettings.id == 1) + )).scalar_one() need = await _img(db, "e1" * 32) # no concept region → wants one - have = await _img(db, "e2" * 32) # already embedded → skip + have = await _img(db, "e2" * 32) # concept @ current version → skip db.add(ImageRegion( image_record_id=have.id, kind="concept", rx=0.0, ry=0.0, rw=1.0, rh=1.0, - siglip_embedding=[0.0] * 1152, embedding_version="siglip-test", + siglip_embedding=[0.0] * 1152, embedding_version=cur, )) await db.commit() @@ -57,6 +61,36 @@ async def test_enqueue_siglip_backfill_gates_on_concept_region(db): assert n_for_need == 1 +@pytest.mark.asyncio +async def test_enqueue_embed_backfill_selects_stale_and_unembedded(db): + # Whole-image GPU re-embed (#1190): enqueue images with no embedding or one + # stamped under a DIFFERENT model version (an operator swap); skip ones + # already at the current version. + from backend.app.models import MLSettings + from backend.app.tasks.ml import enqueue_gpu_backfill + + cur = (await db.execute( + select(MLSettings.embedder_model_version).where(MLSettings.id == 1) + )).scalar_one() + current = await _img(db, "f1" * 32) + current.siglip_embedding = [0.0] * 1152 + current.siglip_model_version = cur # up to date → skip + stale = await _img(db, "f2" * 32) + stale.siglip_embedding = [0.0] * 1152 + stale.siglip_model_version = "old-embedder-v0" # wrong space → re-embed + unembedded = await _img(db, "f3" * 32) # never embedded → embed + await db.commit() + + assert enqueue_gpu_backfill("embed") >= 2 + queued = { + j.image_record_id for j in ( + await db.execute(select(GpuJob).where(GpuJob.task == "embed")) + ).scalars() + } + assert stale.id in queued and unembedded.id in queued + assert current.id not in queued + + @pytest.mark.asyncio async def test_enqueue_dedupes_same_pair(db): img = await _img(db, "a" * 64) diff --git a/tests/test_ml_suggestions.py b/tests/test_ml_suggestions.py index 40d938b..a797013 100644 --- a/tests/test_ml_suggestions.py +++ b/tests/test_ml_suggestions.py @@ -145,6 +145,20 @@ async def test_concept_region_surfaces_via_max_over_bag(db): assert any(s.canonical_tag_id == tag.id and s.score > 0.7 for s in general) +@pytest.mark.asyncio +async def test_stale_embedding_version_excluded_from_scoring(db): + # Mid model-swap (#1190): an image still carrying an OLD-version whole-image + # embedding must NOT be scored by heads trained in the new model's space — + # even though the vector aligns with the head, it's the wrong coordinate + # system, so nothing surfaces until it's re-embedded. + tag = await TagService(db).find_or_create("glasses", TagKind.general) + img = await _img(db, "c1" * 32, _emb(0)) + img.siglip_model_version = "some-old-model-v0" # != current embedder + await _head(db, tag.id, slot=0, suggest_threshold=0.5) + await db.commit() + assert not (await SuggestionService(db).for_image(img.id)).by_category.get("general") + + @pytest.mark.asyncio async def test_rejected_tag_surfaced_flagged_then_reversible(db): # A dismissed suggestion is NOT dropped: it stays flagged rejected so the From 3d77a38a25c5bd94beb07a33042827849d40203f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 30 Jun 2026 11:48:09 -0400 Subject: [PATCH 03/10] refactor(ml): remove the dead per-tag centroid subsystem (#1189) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The v2 pivot replaced per-tag SigLIP centroids with learned heads + CCIP. Centroids were still recomputed (on every tag merge + a daily beat) but NOTHING read them — suggestions come from heads+CCIP and apply_allowlist_tags applies via Camie predictions, not centroids. Pure dead wiring; remove it. Removed: CentroidService, recompute_centroid/recompute_centroids tasks, the daily beat, POST /api/ml/recompute-centroids, the recompute-on-merge trigger, the tag_reference_embedding table + model, the centroid_similarity_threshold + min_reference_images settings (migration 0066), the CentroidRecomputeCard + its store action + MaintenancePanel tile, and the centroid slider in MLThresholdSliders. _keep_as_alias drops its vestigial has-centroid branch (the allowlist branch already covers "could re-emit"); tag merge no longer clears a table that no longer exists. NOT touched (still live, parallel to heads): the Camie tagger, ImagePrediction, and the allowlist bulk-apply — accepting a suggestion still allowlists + applies it across the library. The tag-eval "centroid" baseline metric is unrelated (in-memory) and stays. (image_record.centroid_scores JSON column also remains — separate legacy field, its own micro-cleanup.) Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa --- alembic/versions/0066_drop_centroids.py | 57 ++++++ backend/app/api/ml_admin.py | 14 +- backend/app/api/tags.py | 6 - backend/app/celery_app.py | 4 - backend/app/models/__init__.py | 2 - backend/app/models/ml_settings.py | 15 +- backend/app/models/tag_reference_embedding.py | 23 --- backend/app/services/ml/centroids.py | 163 ------------------ backend/app/services/tag_service.py | 26 +-- backend/app/tasks/ml.py | 63 +------ .../settings/CentroidRecomputeCard.vue | 36 ---- .../settings/MLThresholdSliders.vue | 6 +- .../components/settings/MaintenancePanel.vue | 7 +- frontend/src/stores/ml.js | 6 +- tests/test_api_ml_admin.py | 4 +- tests/test_migration_0003.py | 8 - tests/test_ml_artist_retired.py | 6 - tests/test_ml_centroids.py | 112 ------------ tests/test_tag_merge.py | 28 --- 19 files changed, 78 insertions(+), 508 deletions(-) create mode 100644 alembic/versions/0066_drop_centroids.py delete mode 100644 backend/app/models/tag_reference_embedding.py delete mode 100644 backend/app/services/ml/centroids.py delete mode 100644 frontend/src/components/settings/CentroidRecomputeCard.vue delete mode 100644 tests/test_ml_centroids.py diff --git a/alembic/versions/0066_drop_centroids.py b/alembic/versions/0066_drop_centroids.py new file mode 100644 index 0000000..d75a334 --- /dev/null +++ b/alembic/versions/0066_drop_centroids.py @@ -0,0 +1,57 @@ +"""drop the dead per-tag centroid subsystem (#1189 cleanup) + +The v2 pivot replaced per-tag SigLIP centroids with learned heads + CCIP. +Nothing read the centroids anymore — they were recomputed (on merge + a daily +beat) but never consumed for suggestions or auto-apply. Remove the storage + +its two now-unused settings columns. (The recompute tasks, beat, endpoint, +service, and UI card are removed in the same change.) + +Revision ID: 0066 +Revises: 0065 +Create Date: 2026-06-30 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0066" +down_revision: Union[str, None] = "0065" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.drop_table("tag_reference_embedding") + op.drop_column("ml_settings", "centroid_similarity_threshold") + op.drop_column("ml_settings", "min_reference_images") + + +def downgrade() -> None: + op.add_column( + "ml_settings", + sa.Column( + "min_reference_images", sa.Integer(), nullable=False, + server_default="5", + ), + ) + op.add_column( + "ml_settings", + sa.Column( + "centroid_similarity_threshold", sa.Float(), nullable=False, + server_default="0.55", + ), + ) + op.create_table( + "tag_reference_embedding", + sa.Column("tag_id", sa.Integer(), nullable=False), + sa.Column("embedding", sa.LargeBinary(), nullable=False), + sa.Column("reference_count", sa.Integer(), nullable=False), + sa.Column("model_version", sa.String(length=128), nullable=False), + sa.Column( + "updated_at", sa.DateTime(timezone=True), + server_default=sa.func.now(), nullable=False, + ), + sa.ForeignKeyConstraint(["tag_id"], ["tag.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("tag_id"), + ) diff --git a/backend/app/api/ml_admin.py b/backend/app/api/ml_admin.py index 102c004..cd0cf56 100644 --- a/backend/app/api/ml_admin.py +++ b/backend/app/api/ml_admin.py @@ -1,4 +1,4 @@ -"""ML admin API: settings, backfill trigger, centroid recompute trigger.""" +"""ML admin API: settings + backfill trigger.""" from quart import Blueprint, jsonify, request @@ -11,8 +11,6 @@ ml_admin_bp = Blueprint("ml_admin", __name__, url_prefix="/api/ml") _EDITABLE = ( "suggestion_threshold_character", "suggestion_threshold_general", - "centroid_similarity_threshold", - "min_reference_images", "tagger_store_floor", "video_frame_interval_seconds", "video_max_frames", @@ -41,8 +39,6 @@ async def get_settings(): { "suggestion_threshold_character": s.suggestion_threshold_character, "suggestion_threshold_general": s.suggestion_threshold_general, - "centroid_similarity_threshold": s.centroid_similarity_threshold, - "min_reference_images": s.min_reference_images, "tagger_store_floor": s.tagger_store_floor, "video_frame_interval_seconds": s.video_frame_interval_seconds, "video_max_frames": s.video_max_frames, @@ -142,11 +138,3 @@ async def trigger_backfill(): r = backfill.delay() return jsonify({"celery_task_id": r.id}), 202 - - -@ml_admin_bp.route("/recompute-centroids", methods=["POST"]) -async def trigger_recompute(): - from ..tasks.ml import recompute_centroids - - r = recompute_centroids.delay() - return jsonify({"celery_task_id": r.id}), 202 diff --git a/backend/app/api/tags.py b/backend/app/api/tags.py index 23b0817..59a1031 100644 --- a/backend/app/api/tags.py +++ b/backend/app/api/tags.py @@ -304,12 +304,6 @@ async def merge_tag(source_id: int): from ..tasks.ml import apply_allowlist_tags apply_allowlist_tags.delay(tag_id=result.target_id) - # Tag merge invalidates the target's centroid (the merged-in source - # tag's images now contribute to it). Daily list_drifted catches it - # within 24h, but eager recompute closes the suggestion-quality dip - # in the meantime. Audit 2026-06-02. - from ..tasks.ml import recompute_centroid - recompute_centroid.delay(result.target_id) return jsonify( { "target": { diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 7b7a4e3..d022717 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -101,10 +101,6 @@ def make_celery() -> Celery: "task": "backend.app.tasks.ml.backfill", "schedule": 86400.0, }, - "recompute-centroids-daily": { - "task": "backend.app.tasks.ml.recompute_centroids", - "schedule": 86400.0, - }, "apply-allowlist-sweep-daily": { "task": "backend.app.tasks.ml.apply_allowlist_tags", "schedule": 86400.0, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 087ff0b..5cd3c7e 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -38,7 +38,6 @@ from .tag_allowlist import TagAllowlist from .tag_eval_run import TagEvalRun from .tag_head import TagHead from .tag_positive_confirmation import TagPositiveConfirmation -from .tag_reference_embedding import TagReferenceEmbedding from .tag_suggestion_rejection import TagSuggestionRejection from .task_run import TaskRun @@ -83,7 +82,6 @@ __all__ = [ "TagEvalRun", "TagHead", "TagPositiveConfirmation", - "TagReferenceEmbedding", "TagSuggestionRejection", "TaskRun", ] diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index 0e84940..75db5b2 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -33,21 +33,14 @@ class MLSettings(Base): suggestion_threshold_general: Mapped[float] = mapped_column( Float, nullable=False, default=0.70 ) - centroid_similarity_threshold: Mapped[float] = mapped_column( - Float, nullable=False, default=0.55 - ) # Ingest floor: tagger predictions below this confidence are not stored - # (tagger.Tagger.infer). Default 0.70 — the suggestion path already - # filters at 0.70 and the centroid/learned path covers low-confidence - # preferred tags, so the sub-0.70 tail is redundant weight (it had - # bloated image_record's TOAST to ~100 GB; plan-task #764). Operator- - # tunable via Settings → ML; must stay ≤ the suggestion thresholds. + # (tagger.Tagger.infer). Default 0.70 — the suggestion path already filters + # there, so the sub-0.70 tail is redundant weight (it had bloated + # image_record's TOAST to ~100 GB; plan-task #764). Operator-tunable via + # Settings → ML; must stay ≤ the suggestion thresholds. tagger_store_floor: Mapped[float] = mapped_column( Float, nullable=False, default=0.70 ) - min_reference_images: Mapped[int] = mapped_column( - Integer, nullable=False, default=5 - ) # Video tagging (#747). Sample one frame every N seconds (fixed CADENCE, not a # fixed count) so a tag's frame-presence reflects real screen time regardless # of video length; cap the total so a long video can't explode into hundreds diff --git a/backend/app/models/tag_reference_embedding.py b/backend/app/models/tag_reference_embedding.py deleted file mode 100644 index 2dc841b..0000000 --- a/backend/app/models/tag_reference_embedding.py +++ /dev/null @@ -1,23 +0,0 @@ -"""TagReferenceEmbedding — per-tag centroid (mean SigLIP embedding of members).""" - -from datetime import datetime - -from pgvector.sqlalchemy import Vector -from sqlalchemy import DateTime, ForeignKey, Integer, String, func -from sqlalchemy.orm import Mapped, mapped_column - -from .base import Base - - -class TagReferenceEmbedding(Base): - __tablename__ = "tag_reference_embedding" - - tag_id: Mapped[int] = mapped_column( - ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True - ) - embedding: Mapped[list[float]] = mapped_column(Vector(1152), nullable=False) - reference_count: Mapped[int] = mapped_column(Integer, nullable=False) - model_version: Mapped[str] = mapped_column(String(128), nullable=False) - updated_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), nullable=False, server_default=func.now() - ) diff --git a/backend/app/services/ml/centroids.py b/backend/app/services/ml/centroids.py deleted file mode 100644 index e18907f..0000000 --- a/backend/app/services/ml/centroids.py +++ /dev/null @@ -1,163 +0,0 @@ -"""Tag centroids: the mean SigLIP embedding of a tag's member images. - -Powers centroid-augmented suggestions (a tag whose centroid is close to an -image's embedding becomes a suggestion even if Camie didn't predict it). -""" - -from dataclasses import dataclass - -import numpy as np -from sqlalchemy import func, select -from sqlalchemy.dialects.postgresql import insert -from sqlalchemy.ext.asyncio import AsyncSession - -from ...models import ( - ImageRecord, - MLSettings, - Tag, - TagKind, - TagReferenceEmbedding, -) -from ...models.tag import image_tag - -ELIGIBLE_KINDS = { - TagKind.character, - TagKind.fandom, - TagKind.general, - TagKind.series, -} - - -@dataclass(frozen=True) -class CentroidHit: - tag_id: int - similarity: float - - -class CentroidService: - def __init__(self, session: AsyncSession): - self.session = session - - async def _min_reference_images(self) -> int: - return ( - await self.session.execute( - select(MLSettings.min_reference_images).where(MLSettings.id == 1) - ) - ).scalar_one() - - async def _model_version(self) -> str: - """Audit 2026-06-02: SigLIP model-version stamp comes from the - DB row, not the env constant. tag_and_embed (tasks/ml.py:110) - already reads from MLSettings.embedder_model_version, so by - sourcing centroid stamps + drift checks from the same row, we - eliminate the silent-drift case the audit flagged. env - SIGLIP_MODEL_VERSION still drives which model embedder.py - loads at runtime; the version stamp is purely the operator- - controlled identifier.""" - return ( - await self.session.execute( - select(MLSettings.embedder_model_version).where(MLSettings.id == 1) - ) - ).scalar_one() - - async def recompute_for_tag(self, tag_id: int) -> bool: - """Recompute one tag's centroid. Returns True if a centroid was - written, False if skipped (ineligible kind or too few members).""" - tag = await self.session.get(Tag, tag_id) - if tag is None or tag.kind not in ELIGIBLE_KINDS: - return False - - min_refs = await self._min_reference_images() - - stmt = ( - select(ImageRecord.siglip_embedding) - .join(image_tag, image_tag.c.image_record_id == ImageRecord.id) - .where(image_tag.c.tag_id == tag_id) - .where(ImageRecord.siglip_embedding.is_not(None)) - ) - embeddings = [ - np.array(e, dtype=np.float32) - for e in (await self.session.execute(stmt)).scalars().all() - ] - if len(embeddings) < min_refs: - return False - - centroid = np.mean(np.stack(embeddings), axis=0).astype(np.float32) - model_version = await self._model_version() - - stmt = insert(TagReferenceEmbedding).values( - tag_id=tag_id, - embedding=centroid.tolist(), - reference_count=len(embeddings), - model_version=model_version, - ) - stmt = stmt.on_conflict_do_update( - index_elements=["tag_id"], - set_={ - "embedding": centroid.tolist(), - "reference_count": len(embeddings), - "model_version": model_version, - "updated_at": func.now(), - }, - ) - await self.session.execute(stmt) - return True - - async def list_drifted(self) -> list[int]: - """Tag ids whose centroid is stale: member count != reference_count, - OR no centroid row, OR centroid built on a different SigLIP version. - Only considers eligible-kind tags with embeddings present.""" - current_model_version = await self._model_version() - member_counts = ( - select( - image_tag.c.tag_id.label("tag_id"), - func.count(image_tag.c.image_record_id).label("members"), - ) - .join(ImageRecord, ImageRecord.id == image_tag.c.image_record_id) - .where(ImageRecord.siglip_embedding.is_not(None)) - .group_by(image_tag.c.tag_id) - .subquery() - ) - stmt = ( - select(Tag.id) - .join(member_counts, member_counts.c.tag_id == Tag.id) - .outerjoin( - TagReferenceEmbedding, - TagReferenceEmbedding.tag_id == Tag.id, - ) - .where(Tag.kind.in_(ELIGIBLE_KINDS)) - .where( - (TagReferenceEmbedding.tag_id.is_(None)) - | ( - TagReferenceEmbedding.reference_count - != member_counts.c.members - ) - | (TagReferenceEmbedding.model_version != current_model_version) - ) - ) - return list((await self.session.execute(stmt)).scalars().all()) - - async def find_similar_tags( - self, image_id: int, limit: int = 20 - ) -> list[CentroidHit]: - """Cosine similarity between an image's embedding and stored - centroids. Returns top-`limit` by similarity DESC. pgvector's - cosine_distance gives 1 - cosine_similarity.""" - img = await self.session.get(ImageRecord, image_id) - if img is None or img.siglip_embedding is None: - return [] - emb = img.siglip_embedding - distance = TagReferenceEmbedding.embedding.cosine_distance(emb) - stmt = ( - select( - TagReferenceEmbedding.tag_id, - (1 - distance).label("similarity"), - ) - .order_by(distance.asc()) - .limit(limit) - ) - rows = (await self.session.execute(stmt)).all() - return [ - CentroidHit(tag_id=r.tag_id, similarity=float(r.similarity)) - for r in rows - ] diff --git a/backend/app/services/tag_service.py b/backend/app/services/tag_service.py index 839071c..66e174f 100644 --- a/backend/app/services/tag_service.py +++ b/backend/app/services/tag_service.py @@ -11,7 +11,6 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..models import HeadMetric, Tag, TagHead, TagKind, image_tag from ..models.tag_allowlist import TagAllowlist -from ..models.tag_reference_embedding import TagReferenceEmbedding from .db_helpers import get_or_create from .tag_query import fandom_join_alias, tag_columns @@ -304,10 +303,10 @@ class TagService: async def _keep_as_alias(self, tag_id: int) -> bool: """A merged-away tag's old name must survive as an alias iff the ML - pipeline has ever applied it OR could re-emit it (allowlisted / has - a centroid) — otherwise the proactive apply_allowlist_tags worker - would silently regenerate it. Purely-manual, ML-unknown tags are - deleted outright (no DB bloat).""" + pipeline has ever applied it OR could re-emit it (allowlisted) — + otherwise the proactive apply_allowlist_tags worker would silently + regenerate it. Purely-manual, ML-unknown tags are deleted outright (no + DB bloat).""" is_machine = await self.session.scalar( select( exists().where( @@ -325,14 +324,7 @@ class TagService: allowlisted = await self.session.scalar( select(exists().where(TagAllowlist.tag_id == tag_id)) ) - if allowlisted: - return True - has_centroid = await self.session.scalar( - select( - exists().where(TagReferenceEmbedding.tag_id == tag_id) - ) - ) - return bool(has_centroid) + return bool(allowlisted) async def rename(self, tag_id: int, new_name: str) -> Tag: """Rename a tag. Raises TagMergeConflict if the new name collides @@ -573,7 +565,6 @@ class TagService: merged_count = await self._repoint_image_tags(source_id, target_id) await self._repoint_rejections(source_id, target_id) await self._repoint_allowlist(source_id, target_id) - await self._repoint_embedding(source_id) await self._repoint_aliases(source_id, target_id) await self._repoint_fandom_children( source_id, target_id, source_kind @@ -655,13 +646,6 @@ class TagService: .values(tag_id=tgt) ) - async def _repoint_embedding(self, src: int) -> None: - await self.session.execute( - text( - "DELETE FROM tag_reference_embedding WHERE tag_id = :src" - ), - {"src": src}, - ) async def _repoint_aliases(self, src: int, tgt: int) -> None: from ..models.tag_alias import TagAlias diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index f2aa6a7..b2d5e3c 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -1,9 +1,9 @@ -"""ML Celery tasks: per-image inference, backfill discovery, centroid -recompute, allowlist auto-apply, model self-heal. +"""ML Celery tasks: per-image inference, backfill discovery, head training, +allowlist auto-apply, model self-heal. -All run on the ml-worker (queue 'ml') except recompute_centroids and -apply_allowlist_tags sweeps which are 'maintenance' lane. Sync sessions -(Celery workers are sync processes), same pattern as FC-2a tasks. +All run on the ml-worker (queue 'ml') except apply_allowlist_tags sweeps which +are 'maintenance' lane. Sync sessions (Celery workers are sync processes), same +pattern as FC-2a tasks. """ import logging @@ -487,59 +487,6 @@ def _confidence_for_tag(session, tag, preds: dict) -> float | None: return best -@celery.task(name="backend.app.tasks.ml.recompute_centroid", bind=True) -def recompute_centroid(self, tag_id: int) -> bool: - import asyncio - - from ..services.ml.centroids import CentroidService - from ._async_session import async_session_factory - - async def _run() -> bool: - # Per-task NullPool engine bound to THIS asyncio.run loop — the shared - # process-wide engine reuses connections across loops and raises - # "Future attached to a different loop" on every call after the first. - async_factory, async_engine = async_session_factory() - try: - async with async_factory() as session: - svc = CentroidService(session) - result = await svc.recompute_for_tag(tag_id) - await session.commit() - return result - finally: - await async_engine.dispose() - - return asyncio.run(_run()) - - -@celery.task( - name="backend.app.tasks.ml.recompute_centroids", - bind=True, - # Audit 2026-06-02 — drifted-centroid rebuild over potentially - # hundreds of tags. - soft_time_limit=1800, time_limit=2100, -) -def recompute_centroids(self) -> int: - """Daily: find drifted centroids, enqueue recompute_centroid for each.""" - import asyncio - - from ..services.ml.centroids import CentroidService - from ._async_session import async_session_factory - - async def _list() -> list[int]: - # Per-task NullPool engine bound to this loop (see recompute_centroid). - async_factory, async_engine = async_session_factory() - try: - async with async_factory() as session: - return await CentroidService(session).list_drifted() - finally: - await async_engine.dispose() - - drifted = asyncio.run(_list()) - for tid in drifted: - recompute_centroid.delay(tid) - return len(drifted) - - @celery.task( name="backend.app.tasks.ml.tag_eval_run", bind=True, diff --git a/frontend/src/components/settings/CentroidRecomputeCard.vue b/frontend/src/components/settings/CentroidRecomputeCard.vue deleted file mode 100644 index f9c5df4..0000000 --- a/frontend/src/components/settings/CentroidRecomputeCard.vue +++ /dev/null @@ -1,36 +0,0 @@ - - - diff --git a/frontend/src/components/settings/MLThresholdSliders.vue b/frontend/src/components/settings/MLThresholdSliders.vue index eb1a8de..685009f 100644 --- a/frontend/src/components/settings/MLThresholdSliders.vue +++ b/frontend/src/components/settings/MLThresholdSliders.vue @@ -28,8 +28,7 @@
Tagger predictions below this confidence aren't stored — raising it keeps the image library lean. Suggestions can't be shown below the - floor; lower-confidence tags you actually want still surface through - the learned centroid path. + floor.
@@ -84,8 +83,7 @@ const store = useMLStore() // tagger store floor (nothing below the floor is stored to surface). const fields = [ { key: 'suggestion_threshold_character', label: 'Character', floorMin: true }, - { key: 'suggestion_threshold_general', label: 'General', floorMin: true }, - { key: 'centroid_similarity_threshold', label: 'Centroid similarity' } + { key: 'suggestion_threshold_general', label: 'General', floorMin: true } ] const local = reactive({}) watch(() => store.settings, (s) => { if (s) Object.assign(local, s) }, { immediate: true }) diff --git a/frontend/src/components/settings/MaintenancePanel.vue b/frontend/src/components/settings/MaintenancePanel.vue index 77a8035..aba2deb 100644 --- a/frontend/src/components/settings/MaintenancePanel.vue +++ b/frontend/src/components/settings/MaintenancePanel.vue @@ -1,9 +1,8 @@