diff --git a/alembic/versions/0080_tag_head_train_fingerprint.py b/alembic/versions/0080_tag_head_train_fingerprint.py new file mode 100644 index 0000000..b4bd224 --- /dev/null +++ b/alembic/versions/0080_tag_head_train_fingerprint.py @@ -0,0 +1,31 @@ +"""tag_head.train_fingerprint (#1317 phase 2) — incremental head retraining + +A per-head training-data fingerprint (positive + rejection count/latest-timestamp) +so a manual Retrain refits only the tags whose data changed; the nightly run +ignores it (full reconcile). Nullable — a NULL fingerprint (existing heads) forces +a refit on the first incremental run, then it's stamped. + +Revision ID: 0080 +Revises: 0079 +Create Date: 2026-07-06 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0080" +down_revision: Union[str, None] = "0079" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "tag_head", + sa.Column("train_fingerprint", sa.String(128), nullable=True), + ) + + +def downgrade() -> None: + op.drop_column("tag_head", "train_fingerprint") diff --git a/backend/app/models/tag_head.py b/backend/app/models/tag_head.py index 34a5bcb..0390810 100644 --- a/backend/app/models/tag_head.py +++ b/backend/app/models/tag_head.py @@ -73,5 +73,12 @@ class TagHead(Base): trained_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) + # Training-data fingerprint (positives + rejections) at last fit — the + # incremental-retrain change detector (#1317 p2). A manual Retrain refits only + # heads whose fingerprint moved; the nightly run ignores it (full reconcile). + # NULL forces a refit (pre-fingerprint heads). + train_fingerprint: Mapped[str | None] = mapped_column( + String(128), nullable=True + ) # Extra detail (auto-apply operating point, F1, etc.) — non-load-bearing. metrics: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py index bb18c25..41059cd 100644 --- a/backend/app/services/ml/heads.py +++ b/backend/app/services/ml/heads.py @@ -150,24 +150,103 @@ def _eligible_tag_ids(session: Session, min_pos: int) -> list[int]: return [r[0] for r in rows] +def _head_fingerprints(session: Session, tag_ids: list[int]) -> dict[int, str]: + """Per-tag training-data fingerprint: (positive count, latest positive + created_at) + (rejection count, latest rejected_at). It moves whenever a tag + gains/loses a positive or a rejection — the incremental-retrain change + detector (#1317 p2). A newly-added positive/rejection always has the latest + timestamp, so even a remove-one-add-one (unchanged count) is caught. The + sampled-unlabeled negative pool + the hygiene set drift GLOBALLY and are + reconciled by the nightly full run, not captured here.""" + if not tag_ids: + return {} + pos = session.execute( + select( + image_tag.c.tag_id, + func.count(image_tag.c.image_record_id), + func.max(image_tag.c.created_at), + ) + .where(image_tag.c.tag_id.in_(tag_ids)) + .group_by(image_tag.c.tag_id) + ).all() + pos_map = {t: (c, m) for t, c, m in pos} + rej = session.execute( + select( + TagSuggestionRejection.tag_id, + func.count(), + func.max(TagSuggestionRejection.rejected_at), + ) + .where(TagSuggestionRejection.tag_id.in_(tag_ids)) + .group_by(TagSuggestionRejection.tag_id) + ).all() + rej_map = {t: (c, m) for t, c, m in rej} + out = {} + for t in tag_ids: + pc, pm = pos_map.get(t, (0, None)) + rc, rm = rej_map.get(t, (0, None)) + out[t] = f"{pc}:{pm}:{rc}:{rm}" + return out + + +def _heads_needing_retrain( + session: Session, eligible: list[int], embedding_version: str, + fps: dict[int, str], full: bool, +) -> list[int]: + """The eligible tag_ids to (re)fit: no head yet, a head trained in a DIFFERENT + embedding space (a model swap), or a changed training-data fingerprint. + full=True forces every eligible tag. sklearn-free (train_head itself needs + scikit-learn) so the incremental decision is unit-testable on its own.""" + if full: + return list(eligible) + existing = { + tag_id: (fp, ev) + for tag_id, fp, ev in session.execute( + select( + TagHead.tag_id, TagHead.train_fingerprint, + TagHead.embedding_version, + ) + ).all() + } + out = [] + for tag_id in eligible: + prev = existing.get(tag_id) + if ( + prev is None + or prev[1] != embedding_version + or prev[0] != fps.get(tag_id) + ): + out.append(tag_id) + return out + + def train_all_heads( session: Session, params: dict[str, Any], run: HeadTrainingRun | None = None ) -> dict[str, int]: - """(Re)train a head for every eligible concept; prune heads whose tag is no - longer eligible. Commits per head so a SIGKILL leaves trained heads durable - (training is idempotent). Returns {n_trained, n_skipped}.""" + """(Re)train eligible concept heads, INCREMENTALLY by default (#1317 p2): + refit only the tags whose training data changed since last fit, so a manual + Retrain click is fast. `params["full"]=True` (the nightly run) refits every + head to reconcile sampled-negative + hygiene drift. Prunes heads whose tag is + no longer eligible. Commits per head so a SIGKILL leaves trained heads durable. + Returns {n_trained, n_skipped} (n_skipped = unchanged + too-few-examples).""" import numpy as np cfg = _normalize_params(session, params) embedding_version = _embedder_version(session) + full = bool((params or {}).get("full")) eligible = _eligible_tag_ids(session, cfg["min_positives"]) eligible_set = set(eligible) # Computed once per run, not per head — the hygiene set is identical for # every non-system concept. hygiene = _hygiene_excluded_ids(session) + fps = _head_fingerprints(session, eligible) + to_train = set( + _heads_needing_retrain(session, eligible, embedding_version, fps, full) + ) trained = 0 - skipped = 0 + failed = 0 for i, tag_id in enumerate(eligible): + if tag_id not in to_train: + continue try: ok = train_head( session, tag_id, embedding_version, cfg, np, hygiene=hygiene @@ -175,9 +254,15 @@ def train_all_heads( except Exception: log.exception("train_head failed for tag %d", tag_id) ok = False + if ok: + # Stamp the fingerprint we trained against so an unchanged tag is + # skipped on the next incremental run. + head = session.get(TagHead, tag_id) + if head is not None: + head.train_fingerprint = fps.get(tag_id) session.commit() trained += int(ok) - skipped += int(not ok) + failed += int(not ok) if run is not None and i % 10 == 0: run.last_progress_at = datetime.now(UTC) session.commit() @@ -188,7 +273,11 @@ def train_all_heads( else: session.execute(delete(TagHead)) session.commit() - return {"n_trained": trained, "n_skipped": skipped} + # n_skipped = unchanged (not attempted) + failed-to-fit (too few examples). + return { + "n_trained": trained, + "n_skipped": (len(eligible) - len(to_train)) + failed, + } def head_training_ids( diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index bfb8d0c..4959099 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -356,7 +356,10 @@ def scheduled_train_heads() -> str: if running is not None: return "already running" run = HeadTrainingRun( - params={"source": "scheduled"}, status="running", + # Nightly = FULL reconcile (refit every head) so sampled-negative + + # hygiene drift is folded in; the manual Retrain button stays + # incremental (#1317 p2). + params={"source": "scheduled", "full": True}, status="running", last_progress_at=datetime.now(UTC), ) session.add(run) diff --git a/tests/test_head_incremental.py b/tests/test_head_incremental.py new file mode 100644 index 0000000..b16b9dd --- /dev/null +++ b/tests/test_head_incremental.py @@ -0,0 +1,136 @@ +"""Incremental head retraining (#1317 phase 2). The refit-decision + fingerprint +are split out sklearn-free (train_head itself needs scikit-learn), so they're +tested directly via db_sync.""" +import pytest +from sqlalchemy import select + +from backend.app.models import ( + ImageRecord, + MLSettings, + Tag, + TagHead, + TagKind, + TagSuggestionRejection, +) +from backend.app.models.tag import image_tag +from backend.app.services.ml.heads import ( + _head_fingerprints, + _heads_needing_retrain, +) + +pytestmark = pytest.mark.integration + + +def _img(db, sha: str) -> ImageRecord: + img = ImageRecord( + path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg", + width=1, height=1, origin="imported_filesystem", + integrity_status="unknown", + ) + db.add(img) + db.flush() + return img + + +def _tag(db, name: str) -> Tag: + t = Tag(name=name, kind=TagKind.general) + db.add(t) + db.flush() + return t + + +def _apply(db, image_id: int, tag_id: int) -> None: + db.execute(image_tag.insert().values( + image_record_id=image_id, tag_id=tag_id, source="manual", + )) + + +def _reject(db, image_id: int, tag_id: int) -> None: + db.add(TagSuggestionRejection(image_record_id=image_id, tag_id=tag_id)) + db.flush() + + +def _version(db) -> str: + return db.execute( + select(MLSettings.embedder_model_version).where(MLSettings.id == 1) + ).scalar_one() + + +def _head(db, tag_id: int, fp: str | None, version: str) -> None: + db.add(TagHead( + tag_id=tag_id, embedding_version=version, + weights=[0.0] * 1152, bias=0.0, suggest_threshold=0.5, + auto_apply_threshold=None, n_pos=10, n_neg=30, + ap=0.8, precision_cv=0.9, recall=0.6, train_fingerprint=fp, + )) + db.flush() + + +def test_fingerprint_changes_on_new_positive(db_sync): + tag = _tag(db_sync, "glasses") + i1 = _img(db_sync, "a" * 64) + _apply(db_sync, i1.id, tag.id) + db_sync.commit() + fp1 = _head_fingerprints(db_sync, [tag.id])[tag.id] + + i2 = _img(db_sync, "b" * 64) + _apply(db_sync, i2.id, tag.id) + db_sync.commit() + assert _head_fingerprints(db_sync, [tag.id])[tag.id] != fp1 + + +def test_fingerprint_changes_on_new_rejection(db_sync): + tag = _tag(db_sync, "glasses") + i1 = _img(db_sync, "c" * 64) + _apply(db_sync, i1.id, tag.id) + db_sync.commit() + fp1 = _head_fingerprints(db_sync, [tag.id])[tag.id] + + _reject(db_sync, i1.id, tag.id) + db_sync.commit() + assert _head_fingerprints(db_sync, [tag.id])[tag.id] != fp1 + + +def test_needing_retrain_selects_only_changed(db_sync): + ver = _version(db_sync) + a = _tag(db_sync, "A") + _apply(db_sync, _img(db_sync, "d" * 64).id, a.id) + b = _tag(db_sync, "B") + _apply(db_sync, _img(db_sync, "e" * 64).id, b.id) + c = _tag(db_sync, "C") + _apply(db_sync, _img(db_sync, "f" * 64).id, c.id) + db_sync.commit() + + ids = [a.id, b.id, c.id] + fps = _head_fingerprints(db_sync, ids) + _head(db_sync, a.id, fps[a.id], ver) # A: head with CURRENT fp → skip + _head(db_sync, b.id, "stale", ver) # B: head with STALE fp → retrain + db_sync.commit() # C: no head → retrain + + need = set(_heads_needing_retrain(db_sync, ids, ver, fps, full=False)) + assert a.id not in need + assert {b.id, c.id} <= need + + +def test_stale_embedding_version_forces_retrain(db_sync): + ver = _version(db_sync) + a = _tag(db_sync, "A") + _apply(db_sync, _img(db_sync, "g" * 64).id, a.id) + db_sync.commit() + fps = _head_fingerprints(db_sync, [a.id]) + # Matching fingerprint but a DIFFERENT embedding space → must refit. + _head(db_sync, a.id, fps[a.id], "old-model-v0") + db_sync.commit() + assert a.id in set(_heads_needing_retrain(db_sync, [a.id], ver, fps, full=False)) + + +def test_full_forces_all(db_sync): + ver = _version(db_sync) + a = _tag(db_sync, "A") + _apply(db_sync, _img(db_sync, "h" * 64).id, a.id) + db_sync.commit() + fps = _head_fingerprints(db_sync, [a.id]) + _head(db_sync, a.id, fps[a.id], ver) # current fp → would be skipped + db_sync.commit() + # full=True ignores the fingerprint (nightly reconcile). + assert a.id in set(_heads_needing_retrain(db_sync, [a.id], ver, fps, full=True))