From 3d77a38a25c5bd94beb07a33042827849d40203f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 30 Jun 2026 11:48:09 -0400 Subject: [PATCH] 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 @@