bvandeusen/FabledCurator
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 35 Wiki Activity

refactor(ml): retire the Camie tagger + allowlist bulk-apply (#1189)
CI / lint (push) Failing after 3s
Details
CI / frontend-build (push) Successful in 18s
Details
CI / backend-lint-and-test (push) Successful in 28s
Details
CI / integration (push) Successful in 3m27s
Details

Browse Source 
Heads + CCIP are the tag source and head auto-apply is the earned propagation.
The Camie tagger ran only to feed the allowlist bulk-apply (its ImagePrediction
rows had no other consumer), and the allowlist was a SECOND, un-earned auto-apply
path firing in parallel with heads on every accept — exactly the un-earned spray
the v2 pivot replaced. Retire both.

Behavior change: accepting a suggestion now applies the tag to THAT image only
(source='ml_accepted', a head-training positive) — it no longer allowlists +
fans the tag across the library via Camie. Propagation is heads' earned
auto-apply. (Loses instant cold-start propagation for booru-vocab tags; that was
un-earned and bypassed the precision gate.)

- tag_and_embed is now EMBED-ONLY (no Camie load/infer, no ImagePrediction
  writes); backfill enqueues it for images with no embedding.
- Removed: services/ml/tagger.py, apply_allowlist_tags + helpers + daily beat +
  every enqueue caller (accept/alias/merge/per-image), api/allowlist.py +
  blueprint, ImagePrediction + TagAllowlist models/tables (migration 0067),
  AllowlistTable.vue + allowlist store, the accept coverage-projection payload.
- AllowlistService gutted to accept/dismiss/undismiss/reject (the rejection store
  the rail still needs); accept returns nothing, API returns {accepted, tag_id}.
- tag merge no longer repoints/triggers the allowlist; _keep_as_alias now keys on
  ML-applied image_tag sources (incl. head_auto) instead of the allowlist.
- UI: MLBackfillCard relabelled to embedding-only; accept toast simplified;
  MaintenancePanel drops the allowlist tile.

Left for a follow-up hygiene pass (now-inert, harmless): the dead settings
columns (tagger_store_floor, tagger_model_version, suggestion_threshold_*,
video_min_tag_frames), image_record.tagger_model_version, MLThresholdSliders
trim, and the Camie model download in download_models.py.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ttrj5P7upUTueSfoJcxEqa
This commit is contained in:
Bryan Van Deusen
2026-06-30 13:04:31 -04:00
parent 3d77a38a25
commit 485387ff0b
31 changed files with 159 additions and 1710 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/tasks/ml.py
+24 -261
View File
@@ -1,20 +1,19 @@
"""ML Celery tasks: per-image inference, backfill discovery, head training,
allowlist auto-apply, model self-heal.
"""ML Celery tasks: per-image embedding, backfill discovery, head training,
model self-heal.
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.
All run on the ml-worker (queue 'ml'). Sync sessions (Celery workers are sync
processes), same pattern as FC-2a tasks.
"""
import logging
from pathlib import Path
from celery.exceptions import SoftTimeLimitExceeded
from sqlalchemy import delete, select
from sqlalchemy import select
from sqlalchemy.exc import DBAPIError, OperationalError
from ..celery_app import celery
from ..models import ImagePrediction, ImageRecord, MLSettings
from ..models import ImageRecord, MLSettings
from ._sync_engine import sync_session_factory as _sync_session_factory
log = logging.getLogger(__name__)
@@ -46,19 +45,16 @@ def _is_video(path: Path) -> bool:
time_limit=1200, # 20 min hard
)
def tag_and_embed(self, image_id: int) -> dict:
"""Run Camie + SigLIP on one image; store predictions + embedding;
then enqueue per-image allowlist application.
"""Compute + store one image's SigLIP embedding.
Video (#747): sample frames at a fixed cadence (ml_settings
video_frame_interval_seconds, capped at video_max_frames), keep a tag only if
it appears in >= video_min_tag_frames frames and average its confidence over
those frames (mean-pool, not max — kills one-frame noise); mean-pool the
SigLIP embeddings. On no-frames returns status='no_frames' (not an error).
video_frame_interval_seconds, capped at video_max_frames) and mean-pool the
per-frame SigLIP embeddings. On no-frames returns status='no_frames' (not an
error). (Camie tagging was retired #1189 — heads + CCIP are the tag source.)
"""
import time
from ..services.ml.embedder import get_embedder
from ..services.ml.tagger import get_tagger
# Phase + file context, so a timeout/crash names WHICH file and WHERE it
# died instead of a bare SoftTimeLimitExceeded() (operator-flagged 2026-06-08:
@@ -94,15 +90,13 @@ def tag_and_embed(self, image_id: int) -> dict:
return {"status": "file_missing", "image_id": image_id}
phase = "load_models"
tagger = get_tagger()
embedder = get_embedder(settings.embedder_model_name)
if is_vid:
# Layer-3 isolation: ffprobe (a separate process) validates
# the container before we burn ~20 GPU ops sampling frames
# from it. A corrupt video that would crash the frame
# decoder is rejected cleanly here instead of taking down
# the ml-worker. Operator-flagged 2026-05-28.
# the container before we burn GPU ops sampling frames from it.
# A corrupt video that would crash the frame decoder is rejected
# cleanly here instead of taking down the ml-worker.
phase = "video_probe"
from ..utils import safe_probe
vprobe = safe_probe.probe_video(src)
@@ -115,48 +109,23 @@ def tag_and_embed(self, image_id: int) -> dict:
"reason": vprobe.reason,
}
phase = "video_sample_frames"
t0 = time.monotonic()
frames = _sample_video_frames(
src,
interval=settings.video_frame_interval_seconds,
max_frames=settings.video_max_frames,
)
log.info(
"tag_and_embed sampled %d frame(s) in %.1fs: %s",
len(frames), time.monotonic() - t0, ctx,
)
if not frames:
return {"status": "no_frames", "image_id": image_id}
phase = "video_infer"
phase = "video_embed"
import numpy as np
preds = _aggregate_video_predictions(
[tagger.infer(f, store_floor=settings.tagger_store_floor)
for f in frames],
min_frames=settings.video_min_tag_frames,
)
# Mean-pool the per-frame SigLIP embeddings into one vector.
embedding = np.mean(
[embedder.infer(f) for f in frames], axis=0
).astype("float32")
log.info(
"tag_and_embed video aggregated %d tag(s) from %d frame(s) "
"(min_frames=%d): %s",
len(preds), len(frames), settings.video_min_tag_frames, ctx,
)
for f in frames:
f.unlink(missing_ok=True)
else:
phase = "tag"
t0 = time.monotonic()
raw = tagger.infer(src, store_floor=settings.tagger_store_floor)
log.info(
"tag_and_embed tagged in %.1fs (%d tags): %s",
time.monotonic() - t0, len(raw), ctx,
)
preds = {
name: {"category": p.category, "confidence": p.confidence}
for name, p in raw.items()
}
phase = "embed"
t0 = time.monotonic()
embedding = embedder.infer(src)
@@ -166,28 +135,9 @@ def tag_and_embed(self, image_id: int) -> dict:
)
phase = "persist"
record.tagger_model_version = settings.tagger_model_version
record.siglip_embedding = embedding.tolist()
record.siglip_model_version = settings.embedder_model_version
session.add(record)
# Write the normalized image_prediction rows (#768) — the sole home
# for predictions now (image_record.tagger_predictions was dropped in
# migration 0046). Delete-then-insert keeps a re-tag idempotent;
# tagger_store_floor was already applied in tagger.infer, so preds is
# the >=floor set.
session.execute(
delete(ImagePrediction).where(
ImagePrediction.image_record_id == image_id
)
)
session.add_all([
ImagePrediction(
image_record_id=image_id, raw_name=name,
category=p.get("category", "general"),
score=float(p.get("confidence", 0.0)),
)
for name, p in preds.items()
])
session.commit()
except SoftTimeLimitExceeded:
log.error(
@@ -210,11 +160,8 @@ def tag_and_embed(self, image_id: int) -> dict:
)
raise
log.info(
"tag_and_embed ok in %.1fs (%d tags): %s", _elapsed(), len(preds), ctx
)
apply_allowlist_tags.delay(image_id=image_id)
return {"status": "ok", "image_id": image_id, "tags": len(preds)}
log.info("tag_and_embed ok in %.1fs: %s", _elapsed(), ctx)
return {"status": "ok", "image_id": image_id}
def _sample_video_frames(
@@ -273,68 +220,24 @@ def _sample_video_frames(
return out
def _aggregate_video_predictions(per_frame: list[dict], *, min_frames: int) -> dict:
"""Aggregate per-frame {name: TagPrediction} into one prediction set (#747).
A tag is kept only if it appears (≥ the tagger store floor, already applied)
in at least `min_frames` of the sampled frames — because sampling is at a
fixed cadence, that means it was on screen for roughly min_frames×interval
seconds, so a single-frame flicker / scene-transition artifact is dropped
while a genuine scene-local tag in a long video survives. Confidence is the
MEAN over the frames where the tag appears (not max — max re-inflated the
one-frame noise this whole change exists to remove).
`min_frames` is clamped to the number of frames actually sampled so a very
short video (1–2 frames) still tags instead of dropping everything.
"""
n = len(per_frame)
if n == 0:
return {}
threshold = max(1, min(min_frames, n))
agg: dict[str, dict] = {}
for frame_preds in per_frame:
for name, p in frame_preds.items():
cur = agg.get(name)
if cur is None:
agg[name] = {"category": p.category, "sum": p.confidence, "count": 1}
else:
cur["sum"] += p.confidence
cur["count"] += 1
return {
name: {"category": v["category"], "confidence": v["sum"] / v["count"]}
for name, v in agg.items()
if v["count"] >= threshold
}
@celery.task(name="backend.app.tasks.ml.backfill", bind=True)
def backfill(self) -> int:
"""Enqueue tag_and_embed for images missing predictions/embeddings for
the current model versions. Keyset pagination by id ASC (restart-safe).
"""Enqueue tag_and_embed (embed-only) for images with no SigLIP embedding.
Keyset pagination by id ASC (restart-safe).
NB: a siglip MODEL-VERSION mismatch (an operator model swap, #1190) is NOT
re-embedded here — the CPU ml-worker can't churn the library at 384/512px;
the GPU agent owns version re-embeds via the 'embed' job.
"""
SessionLocal = _sync_session_factory()
enqueued = 0
last_id = 0
with SessionLocal() as session:
settings = session.execute(
select(MLSettings).where(MLSettings.id == 1)
).scalar_one()
while True:
rows = session.execute(
select(ImageRecord.id)
.where(ImageRecord.id > last_id)
.where(
(ImageRecord.tagger_model_version.is_(None))
| (
ImageRecord.tagger_model_version
!= settings.tagger_model_version
)
| (ImageRecord.siglip_embedding.is_(None))
# 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.
)
.where(ImageRecord.siglip_embedding.is_(None))
.order_by(ImageRecord.id.asc())
.limit(500)
).scalars().all()
@@ -347,146 +250,6 @@ def backfill(self) -> int:
return enqueued
@celery.task(
name="backend.app.tasks.ml.apply_allowlist_tags",
bind=True,
# Audit 2026-06-02 — the full-sweep mode (neither tag_id nor image_id)
# is O(images × allowlist) and legitimately runs >5 min on large
# libraries. Cap matches the maintenance queue's recovery threshold.
soft_time_limit=1800, time_limit=2100,
)
def apply_allowlist_tags(self, tag_id: int | None = None,
image_id: int | None = None) -> int:
"""Retroactively apply allowlisted tags.
Modes:
- tag_id only : scan all images for this tag.
- image_id only : scan all allowlisted tags for this image.
- both : just the (image, tag) pair.
- neither : full sweep (daily beat).
Skips: already-applied, rejected (tag_suggestion_rejection), or
confidence below the tag's allowlist min_confidence. Applied with
source='ml_auto'.
"""
from sqlalchemy import and_
from sqlalchemy import select as sa_select
from sqlalchemy.dialects.postgresql import insert as pg_insert
from ..models import TagAllowlist, TagSuggestionRejection
from ..models.tag import image_tag
SessionLocal = _sync_session_factory()
applied = 0
with SessionLocal() as session:
allow_rows = session.execute(
sa_select(TagAllowlist.tag_id, TagAllowlist.min_confidence)
if tag_id is None
else sa_select(
TagAllowlist.tag_id, TagAllowlist.min_confidence
).where(TagAllowlist.tag_id == tag_id)
).all()
allow = {r[0]: r[1] for r in allow_rows}
if not allow:
return 0
# Images that have any predictions (#768: from image_prediction, not
# the old JSON column), optionally narrowed to one image.
img_ids_query = sa_select(ImagePrediction.image_record_id).distinct()
if image_id is not None:
img_ids_query = img_ids_query.where(
ImagePrediction.image_record_id == image_id
)
for (img_id,) in session.execute(img_ids_query).all():
preds = _load_predictions_sync(session, img_id)
for a_tag_id, min_conf in allow.items():
exists = session.execute(
sa_select(image_tag.c.tag_id).where(
and_(
image_tag.c.image_record_id == img_id,
image_tag.c.tag_id == a_tag_id,
)
)
).scalar_one_or_none()
if exists is not None:
continue
rej = session.get(
TagSuggestionRejection, (img_id, a_tag_id)
)
if rej is not None:
continue
from ..models import Tag
tag = session.get(Tag, a_tag_id)
if tag is None:
continue
conf = _confidence_for_tag(session, tag, preds)
if conf is None or conf < min_conf:
continue
stmt = pg_insert(image_tag).values(
image_record_id=img_id,
tag_id=a_tag_id,
source="ml_auto",
)
stmt = stmt.on_conflict_do_nothing(
index_elements=["image_record_id", "tag_id"]
)
session.execute(stmt)
applied += 1
session.commit()
return applied
def _load_predictions_sync(session, image_id: int) -> dict:
"""Predictions for one image from image_prediction (#768), in the
{raw_name: {category, confidence}} shape _confidence_for_tag consumes —
keeps the allowlist resolution logic unchanged."""
from sqlalchemy import select as sa_select
rows = session.execute(
sa_select(
ImagePrediction.raw_name,
ImagePrediction.category,
ImagePrediction.score,
).where(ImagePrediction.image_record_id == image_id)
).all()
return {
r.raw_name: {"category": r.category, "confidence": r.score}
for r in rows
}
def _confidence_for_tag(session, tag, preds: dict) -> float | None:
"""Highest confidence among predictions that resolve to `tag` —
either the prediction name equals the tag name, or an alias maps
(prediction name, category) -> tag.id.
"""
from sqlalchemy import select as sa_select
from ..models import TagAlias
best: float | None = None
direct = preds.get(tag.name)
if direct is not None:
best = float(direct.get("confidence", 0.0))
alias_rows = session.execute(
sa_select(TagAlias.alias_string, TagAlias.alias_category).where(
TagAlias.canonical_tag_id == tag.id
)
).all()
for alias_string, alias_category in alias_rows:
p = preds.get(alias_string)
if p is None:
continue
if p.get("category") != alias_category:
continue
c = float(p.get("confidence", 0.0))
if best is None or c > best:
best = c
return best
@celery.task(
name="backend.app.tasks.ml.tag_eval_run",
bind=True,