diff --git a/app/celery_app.py b/app/celery_app.py index ab8f51b..c15c4dc 100644 --- a/app/celery_app.py +++ b/app/celery_app.py @@ -120,6 +120,24 @@ def make_celery(app=None): 'task': 'app.tasks.scan.update_system_stats', 'schedule': 21600, # Every 6 hours }, + + # ML-pipeline self-maintenance. Each is internally self-gating: + # backfill returns immediately when nothing's missing, + # recompute_all_centroids only enqueues for tags whose member + # count changed, and apply_auto_accept_predictions short-circuits + # when no above-threshold predictions are unattached. + 'ml-backfill-sweep': { + 'task': 'app.tasks.ml.backfill', + 'schedule': int(os.environ.get('ML_BACKFILL_EVERY_SECONDS', '86400')), # daily + }, + 'apply-auto-accept-sweep': { + 'task': 'app.tasks.maintenance.apply_auto_accept_predictions', + 'schedule': int(os.environ.get('AUTO_ACCEPT_EVERY_SECONDS', '86400')), # daily + }, + 'recompute-centroids-sweep': { + 'task': 'app.tasks.ml.recompute_all_centroids', + 'schedule': int(os.environ.get('CENTROIDS_EVERY_SECONDS', '604800')), # weekly + }, }, ) diff --git a/app/tasks/maintenance.py b/app/tasks/maintenance.py index f45cb02..efc4ff1 100644 --- a/app/tasks/maintenance.py +++ b/app/tasks/maintenance.py @@ -363,12 +363,58 @@ def apply_auto_accept_predictions(batch_size: int = 100) -> dict: # Lazy-import the suggestion service so the maintenance worker doesn't # pay its overhead on unrelated tasks. from app.services.tag_suggestions import ( - _config, _existing_tag_names, get_suggestions, + _DEFAULTS, _config, _existing_tag_names, get_suggestions, ) + from app.ml.wd14 import MODEL_VERSION as WD14_VER cfg = _config() existing_all = _existing_tag_names() + # Fast-path: if no WD14 prediction at/above the threshold exists for + # an image-tag combo that isn't already attached and isn't user-rejected, + # there's nothing to do. Daily-scheduled runs will hit this branch on + # most days once the library has settled, so the full walk only runs + # after fresh predictions land. + try: + threshold = float(cfg.get('auto_accept_general_threshold', + _DEFAULTS['auto_accept_general_threshold'])) + except (TypeError, ValueError): + threshold = float(_DEFAULTS['auto_accept_general_threshold']) + + pending_exists = db.session.execute(text(""" + SELECT 1 + FROM image_tag_prediction p + WHERE p.confidence >= :thr + AND p.tag_category = 'general' + AND p.model_version = :wd14_ver + AND NOT EXISTS ( + SELECT 1 FROM image_tags it + JOIN tag t ON t.id = it.tag_id + WHERE it.image_id = p.image_id + AND t.name = p.tag_name + AND t.kind = 'user' + ) + AND NOT EXISTS ( + SELECT 1 FROM suggestion_feedback sf + WHERE sf.image_id = p.image_id + AND sf.tag_name = p.tag_name + AND sf.decision = 'rejected' + ) + LIMIT 1 + """), {'thr': threshold, 'wd14_ver': WD14_VER}).first() + + if pending_exists is None: + log.info( + "apply_auto_accept_predictions: no candidates above threshold=%.3f; skipping walk", + threshold, + ) + return { + 'scanned': 0, + 'images_with_applies': 0, + 'tags_applied': 0, + 'skipped_no_candidates': True, + } + scanned = 0 images_with_applies = 0 tags_applied = 0 diff --git a/app/tasks/ml.py b/app/tasks/ml.py index 39a8c4f..e6613ce 100644 --- a/app/tasks/ml.py +++ b/app/tasks/ml.py @@ -319,17 +319,25 @@ def recompute_centroid(self, tag_name: str): @celery.task(bind=True, name='app.tasks.ml.recompute_all_centroids', soft_time_limit=None, time_limit=None) def recompute_all_centroids(self): - """Enqueue recompute_centroid for every eligible tag with enough reference images. + """Enqueue recompute_centroid for every eligible tag whose member count + has changed since its centroid was last computed. - Uses a single aggregate query to find tags with >= min_reference_images applied - images, then enqueues one recompute_centroid task per tag on the ml queue. + Without the count-delta gate, the daily-scheduled call would re-enqueue + a recompute for every eligible tag every run, even if no images had + been attached or detached — wasteful both in work and in ml-queue + contention with `tag_and_embed`. + + Joins `tag_reference_embedding` on `(tag_name, model_version)` so the + aggregate `HAVING` can compare current count against stored + `reference_count`. NULL stored count (no centroid yet) always + qualifies. Equal counts skip. """ + from app.ml.siglip import MODEL_VERSION as SIGLIP_VER from app.services.tag_suggestions import ELIGIBLE_CENTROID_KINDS min_refs_row = TagSuggestionConfig.query.filter_by(key='min_reference_images').first() min_refs = int(min_refs_row.value) if min_refs_row else 5 - # Build an IS NULL / IN filter that covers ELIGIBLE_CENTROID_KINDS including None. kinds_not_null = [k for k in ELIGIBLE_CENTROID_KINDS if k is not None] allow_null = None in ELIGIBLE_CENTROID_KINDS @@ -337,18 +345,43 @@ def recompute_all_centroids(self): if allow_null: kind_filter = kind_filter | Tag.kind.is_(None) + # LEFT JOIN to TagReferenceEmbedding for the SigLIP version so absent + # rows surface as NULL stored_count (always !=, always recompute). rows = ( - db.session.query(Tag.name, func.count(image_tags.c.image_id).label('n')) + db.session.query( + Tag.name, + func.count(image_tags.c.image_id).label('current_count'), + TagReferenceEmbedding.reference_count.label('stored_count'), + ) .join(image_tags, image_tags.c.tag_id == Tag.id) + .outerjoin( + TagReferenceEmbedding, + and_( + TagReferenceEmbedding.tag_name == Tag.name, + TagReferenceEmbedding.model_version == SIGLIP_VER, + ), + ) .filter(kind_filter) - .group_by(Tag.name) + .group_by(Tag.name, TagReferenceEmbedding.reference_count) .having(func.count(image_tags.c.image_id) >= min_refs) .all() ) enqueued = 0 - for tag_name, n in rows: + skipped_unchanged = 0 + for tag_name, current_count, stored_count in rows: + if stored_count is not None and stored_count == current_count: + skipped_unchanged += 1 + continue recompute_centroid.apply_async(args=[tag_name], queue='ml') enqueued += 1 - log.info(f"recompute_all_centroids: enqueued {enqueued} tags (min_refs={min_refs})") - return {'status': 'ok', 'enqueued': enqueued, 'min_refs': min_refs} + log.info( + "recompute_all_centroids: enqueued=%d skipped_unchanged=%d min_refs=%d", + enqueued, skipped_unchanged, min_refs, + ) + return { + 'status': 'ok', + 'enqueued': enqueued, + 'skipped_unchanged': skipped_unchanged, + 'min_refs': min_refs, + }