diff --git a/alembic/versions/0073_drop_tag_eval_run.py b/alembic/versions/0073_drop_tag_eval_run.py new file mode 100644 index 0000000..4aedb38 --- /dev/null +++ b/alembic/versions/0073_drop_tag_eval_run.py @@ -0,0 +1,46 @@ +"""drop tag_eval_run — the head-vs-centroid eval harness is retired + +The eval (#1130) existed to prove the heads tagging spine on the operator's own +data. It did; the operator accepted the system and retired the harness +(2026-07-02) — card, API, task, model and this table all go. The eval's data +loaders + metric helpers live on in services/ml/training_data.py, where the +production heads trainer uses them nightly. + +Revision ID: 0073 +Revises: 0072 +Create Date: 2026-07-02 +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0073" +down_revision: Union[str, None] = "0072" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.drop_index("ix_tag_eval_run_status", table_name="tag_eval_run") + op.drop_table("tag_eval_run") + + +def downgrade() -> None: + # Recreates the shape from 0056 (data is not restorable). + op.create_table( + "tag_eval_run", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("params", postgresql.JSONB(), nullable=False), + sa.Column("status", sa.String(length=16), nullable=False, + server_default="running"), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.func.now()), + sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("report", postgresql.JSONB(), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("last_progress_at", sa.DateTime(timezone=True), + nullable=True), + ) + op.create_index("ix_tag_eval_run_status", "tag_eval_run", ["status"]) diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index 8345f2d..f500acd 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -38,7 +38,6 @@ def all_blueprints() -> list[Blueprint]: from .suggestions import suggestions_bp from .system_activity import system_activity_bp from .system_backup import system_backup_bp - from .tag_eval import tag_eval_bp from .tags import tags_bp from .thumbnails import thumbnails_bp return [ @@ -58,7 +57,6 @@ def all_blueprints() -> list[Blueprint]: import_admin_bp, suggestions_bp, aliases_bp, - tag_eval_bp, heads_bp, gpu_bp, ccip_bp, diff --git a/backend/app/api/tag_eval.py b/backend/app/api/tag_eval.py deleted file mode 100644 index 31fe10b..0000000 --- a/backend/app/api/tag_eval.py +++ /dev/null @@ -1,70 +0,0 @@ -"""Tag-eval API (#1130): trigger + revisit the head-vs-centroid eval. - -The run + full report live in the tag_eval_run row, so the admin card rehydrates -from GET (history / detail) on mount — the report survives navigation rather than -living in transient frontend state. -""" - -from quart import Blueprint, jsonify, request -from sqlalchemy import select - -from ..extensions import get_session -from ..models import TagEvalRun -from ..services.ml.tag_eval import EvalAlreadyRunning, start_tag_eval_run - -tag_eval_bp = Blueprint("tag_eval", __name__, url_prefix="/api/tag-eval") - - -def _serialize(run: TagEvalRun, *, include_report: bool) -> dict: - out = { - "id": run.id, - "params": run.params, - "status": run.status, - "started_at": run.started_at.isoformat() if run.started_at else None, - "finished_at": run.finished_at.isoformat() if run.finished_at else None, - "error": run.error, - } - if include_report: - out["report"] = run.report - return out - - -@tag_eval_bp.route("", methods=["POST"]) -async def create(): - body = await request.get_json(silent=True) or {} - params = body.get("params") or body or {} - async with get_session() as session: - try: - run_id = await session.run_sync( - lambda s: start_tag_eval_run(s, params) - ) - except EvalAlreadyRunning as running: - return jsonify({ - "error": "eval_already_running", - "running_id": int(running.args[0]), - }), 409 - await session.commit() - return jsonify({"run_id": run_id, "status": "running"}), 202 - - -@tag_eval_bp.route("", methods=["GET"]) -async def history(): - try: - limit = min(int(request.args.get("limit", "20")), 100) - except ValueError: - return jsonify({"error": "invalid_limit"}), 400 - async with get_session() as session: - rows = (await session.execute( - select(TagEvalRun).order_by(TagEvalRun.id.desc()).limit(limit) - )).scalars().all() - # List is light — no full report (the detail endpoint carries it). - return jsonify({"runs": [_serialize(r, include_report=False) for r in rows]}) - - -@tag_eval_bp.route("/", methods=["GET"]) -async def detail(run_id: int): - async with get_session() as session: - run = await session.get(TagEvalRun, run_id) - if run is None: - return jsonify({"error": "not_found"}), 404 - return jsonify(_serialize(run, include_report=True)) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 5cb1ab1..0600988 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -183,10 +183,6 @@ def make_celery() -> Celery: "task": "backend.app.tasks.maintenance.recover_stalled_library_audit_runs", "schedule": 300.0, }, - "recover-stalled-tag-eval-runs": { - "task": "backend.app.tasks.maintenance.recover_stalled_tag_eval_runs", - "schedule": 300.0, - }, "recover-stalled-head-training-runs": { "task": "backend.app.tasks.maintenance.recover_stalled_head_training_runs", "schedule": 300.0, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 8707467..5080b50 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -33,7 +33,6 @@ from .subscribestar_failed_media import SubscribeStarFailedMedia from .subscribestar_seen_media import SubscribeStarSeenMedia from .tag import Tag, TagKind, image_tag from .tag_alias import TagAlias -from .tag_eval_run import TagEvalRun from .tag_head import TagHead from .tag_positive_confirmation import TagPositiveConfirmation from .tag_suggestion_rejection import TagSuggestionRejection @@ -75,7 +74,6 @@ __all__ = [ "HeadMetricsSnapshot", "HeadTrainingRun", "TagAlias", - "TagEvalRun", "TagHead", "TagPositiveConfirmation", "TagSuggestionRejection", diff --git a/backend/app/models/head_training_run.py b/backend/app/models/head_training_run.py index 150357c..fd5858e 100644 --- a/backend/app/models/head_training_run.py +++ b/backend/app/models/head_training_run.py @@ -1,7 +1,7 @@ """HeadTrainingRun — persisted lifecycle of a head-training batch (#114). -Mirrors TagEvalRun so the run SURVIVES navigation and the admin card can show -live + historical status instead of holding it in transient frontend state. +A persisted run row (not transient frontend state) so the run SURVIVES +navigation and the admin card can show live + historical status. Training is idempotent (it upserts tag_head rows), so a SIGKILL'd run is harmless — a maintenance recovery sweep flips a stalled `running` row to `error`, and the next run re-trains. State machine: running → ready / error. @@ -37,8 +37,8 @@ class HeadTrainingRun(Base): n_trained: Mapped[int | None] = mapped_column(Integer, nullable=True) n_skipped: Mapped[int | None] = mapped_column(Integer, nullable=True) error: Mapped[str | None] = mapped_column(Text, nullable=True) - # Last time the task made progress — the recovery sweep tells a live run from - # a SIGKILL'd one by this (mirrors TagEvalRun). + # Last time the task made progress — the recovery sweep tells a live run + # from a SIGKILL'd one by this. last_progress_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True ) diff --git a/backend/app/models/tag_eval_run.py b/backend/app/models/tag_eval_run.py deleted file mode 100644 index d0775ed..0000000 --- a/backend/app/models/tag_eval_run.py +++ /dev/null @@ -1,45 +0,0 @@ -"""TagEvalRun — persisted lifecycle of a head-vs-centroid tagging eval (#1130). - -Mirrors LibraryAuditRun so the result SURVIVES navigation: the run + its full -report live in this row, and the admin card rehydrates from it on mount instead -of holding the report in transient frontend state. State machine: -running → ready / error. The async ml-queue task writes `report` (JSONB) when -done; a maintenance recovery sweep flips a stalled `running` row to `error`. -""" - -from datetime import datetime -from typing import Any - -from sqlalchemy import DateTime, Integer, String, Text, func -from sqlalchemy.dialects.postgresql import JSONB -from sqlalchemy.orm import Mapped, mapped_column - -from .base import Base - - -class TagEvalRun(Base): - __tablename__ = "tag_eval_run" - - id: Mapped[int] = mapped_column(Integer, primary_key=True) - # The eval parameters: {concepts: [...], curve_points: [...], neg_ratio, - # cv_folds, ...} — echoed back so the report is self-describing. - params: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False) - status: Mapped[str] = mapped_column( - String(16), nullable=False, default="running", index=True, - ) - # running | ready | error - started_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), nullable=False, server_default=func.now(), - ) - finished_at: Mapped[datetime | None] = mapped_column( - DateTime(timezone=True), nullable=True, - ) - # The full result: per-concept metrics (head vs centroid), learning-curve - # points, and example image ids. Null until the task finishes. - report: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True) - error: Mapped[str | None] = mapped_column(Text, nullable=True) - # Last time the task made progress — the recovery sweep tells a live run - # from a SIGKILL'd one by this (mirrors LibraryAuditRun). - last_progress_at: Mapped[datetime | None] = mapped_column( - DateTime(timezone=True), nullable=True, - ) diff --git a/backend/app/services/ml/heads.py b/backend/app/services/ml/heads.py index b234897..2f47401 100644 --- a/backend/app/services/ml/heads.py +++ b/backend/app/services/ml/heads.py @@ -1,12 +1,13 @@ """Production heads: train + score the per-concept classifiers (#114). -The eval (#1130, tag_eval.py) proved the spine; this is its production form. +The eval harness (#1130) proved the spine, then retired 2026-07-02 once the +tagging system was accepted; this is the production form. - TRAIN (sync, ml worker — needs scikit-learn): for every general/character tag with enough labelled positives, fit a logistic-regression head on the FROZEN SigLIP embeddings (positives + negatives = rejections + sampled unlabeled), derive an honest suggest threshold + earned-auto-apply point from CROSS- - VALIDATED scores, and upsert a TagHead row. Reuses tag_eval's proven data - loaders + metric helpers so production heads match the eval's measured numbers. + VALIDATED scores, and upsert a TagHead row. Uses the eval-proven data loaders + + metric helpers (training_data.py) so heads match the measured numbers. - SCORE (async, API worker — numpy via pgvector, NO scikit-learn): score one image's embedding against all current heads → the suggestions the rail shows, REPLACING Camie predictions + per-tag centroids. @@ -37,7 +38,7 @@ from ...models import ( TagSuggestionRejection, ) from ...models.tag import image_tag -from .tag_eval import ( +from .training_data import ( _auto_apply_point, _ids_with_tag, _l2norm, diff --git a/backend/app/services/ml/tag_eval.py b/backend/app/services/ml/tag_eval.py deleted file mode 100644 index abff374..0000000 --- a/backend/app/services/ml/tag_eval.py +++ /dev/null @@ -1,430 +0,0 @@ -"""Head-vs-centroid tagging eval (#1130, milestone #114 slice 1). - -Proves the "frozen embedding + small trained head (with negatives)" spine on the -operator's OWN data, reusing the SigLIP embeddings already stored on -image_record. For each concept tag it compares: - - CENTROID baseline (the old approach): cosine to the mean of positive vectors. - - HEAD (the new approach): logistic regression trained on positives + negatives. -and reports cross-validated precision/recall/AP for both, a LEARNING CURVE -(accuracy as the number of tagged positives grows), and example image ids to -eyeball. - -numpy + scikit-learn are imported LAZILY inside run_eval so the API worker (base -image, no ML stack) can still import start_tag_eval_run to enqueue the ml-queue -task — the heavy compute only runs on the ml worker. -""" - -from __future__ import annotations - -import logging -from datetime import UTC, datetime -from typing import Any - -from sqlalchemy import func, select -from sqlalchemy.orm import Session - -from ...models import ( - ImageRecord, - Tag, - TagEvalRun, - TagKind, - TagPositiveConfirmation, - TagSuggestionRejection, -) -from ...models.tag import image_tag - -log = logging.getLogger(__name__) - -# The operator's real concept list (mix of whole-ish + small/local cues). The -# admin trigger can override; this is the default eval set. -DEFAULT_CONCEPTS = [ - "glasses", "cat", "dog", "horse", "goblin", - "cum", "lactation", "fellatio", "xray", "stomach bulge", -] -DEFAULT_CURVE_POINTS = [10, 30, 100, 300] -DEFAULT_NEG_RATIO = 3 # negatives per positive (rejections + sampled unlabeled) -DEFAULT_CV_FOLDS = 5 -MIN_POSITIVES = 8 # below this, a concept can't be evaluated meaningfully -_UNLABELED_POOL = 4000 # cap on sampled unlabeled rows pulled per concept -_EXAMPLES_K = 12 - - -def start_tag_eval_run(session: Session, params: dict[str, Any]) -> int: - """Create a TagEvalRun (status='running') and dispatch the ml-queue task. - Returns the new run id. Light guard: one running eval at a time.""" - existing = session.execute( - select(TagEvalRun.id).where(TagEvalRun.status == "running") - ).scalar_one_or_none() - if existing is not None: - raise EvalAlreadyRunning(existing) - norm = _normalize_params(params) - run = TagEvalRun(params=norm, status="running", last_progress_at=datetime.now(UTC)) - session.add(run) - session.flush() - run_id = run.id - # Same enqueue-by-import pattern api/suggestions.py uses for ml tasks; the - # commit happens in the API handler so row + dispatch are visible together. - from ...tasks.ml import tag_eval_run as _task - _task.delay(run_id) - return run_id - - -class EvalAlreadyRunning(Exception): - """Raised by start_tag_eval_run when an eval is already in flight.""" - - -def _normalize_params(params: dict[str, Any] | None) -> dict[str, Any]: - params = params or {} - concepts = [str(c).strip() for c in (params.get("concepts") or []) if str(c).strip()] - try: - neg_ratio = max(1, int(params.get("neg_ratio", DEFAULT_NEG_RATIO))) - except (TypeError, ValueError): - neg_ratio = DEFAULT_NEG_RATIO - try: - cv_folds = max(2, int(params.get("cv_folds", DEFAULT_CV_FOLDS))) - except (TypeError, ValueError): - cv_folds = DEFAULT_CV_FOLDS - try: - auto_top_n = min(max(int(params.get("auto_top_n", 0) or 0), 0), 200) - except (TypeError, ValueError): - auto_top_n = 0 - try: - precision_target = min(max(float(params.get("precision_target", 0.97)), 0.5), 0.999) - except (TypeError, ValueError): - precision_target = 0.97 - # No explicit concepts and auto-discovery off → fall back to the hand list. - if not concepts and not auto_top_n: - concepts = list(DEFAULT_CONCEPTS) - curve = params.get("curve_points") or DEFAULT_CURVE_POINTS - curve = sorted({int(n) for n in curve if int(n) > 0}) - return { - "concepts": concepts, - "neg_ratio": neg_ratio, - "cv_folds": cv_folds, - "auto_top_n": auto_top_n, - "precision_target": round(precision_target, 4), - "curve_points": curve, - } - - -def _top_general_concepts(session: Session, n: int, min_count: int) -> list[str]: - """The n most-tagged general (concept) tags with >= min_count images — a fast - server-side way to broaden the eval beyond the hand-picked list (counts all - sources; source-aware filtering is a separate concern).""" - rows = session.execute( - select(Tag.name) - .join(image_tag, image_tag.c.tag_id == Tag.id) - .where(Tag.kind == TagKind.general) - .group_by(Tag.id) - .having(func.count(image_tag.c.image_record_id) >= min_count) - .order_by(func.count(image_tag.c.image_record_id).desc()) - .limit(n) - ).all() - return [r[0] for r in rows] - - -def _resolve_tag_id(session: Session, name: str) -> int | None: - """Case-insensitive tag-name match; if several share a name, take the one - applied to the most images (the one the operator actually uses).""" - rows = session.execute( - select(Tag.id, func.count(image_tag.c.image_record_id)) - .outerjoin(image_tag, image_tag.c.tag_id == Tag.id) - .where(func.lower(Tag.name) == name.lower()) - .group_by(Tag.id) - .order_by(func.count(image_tag.c.image_record_id).desc()) - ).all() - return rows[0][0] if rows else None - - -def _ids_with_tag(session: Session, tag_id: int) -> list[int]: - return [ - r[0] for r in session.execute( - select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tag_id) - ).all() - ] - - -def _rejected_ids(session: Session, tag_id: int) -> list[int]: - return [ - r[0] for r in session.execute( - select(TagSuggestionRejection.image_record_id) - .where(TagSuggestionRejection.tag_id == tag_id) - ).all() - ] - - -def _confirmed_ids(session: Session, tag_id: int) -> set[int]: - """Positives the operator explicitly affirmed ('keep') — excluded from the - doubts list so confirmed-correct images don't resurface every run.""" - return { - r[0] for r in session.execute( - select(TagPositiveConfirmation.image_record_id) - .where(TagPositiveConfirmation.tag_id == tag_id) - ).all() - } - - -def _sample_unlabeled(session: Session, exclude: set[int], limit: int) -> list[int]: - """Random image ids (with an embedding) NOT carrying the tag. Concepts are - sparse, so an untagged image is almost always a true negative.""" - stmt = ( - select(ImageRecord.id) - .where(ImageRecord.siglip_embedding.is_not(None)) - .order_by(func.random()) - .limit(limit) - ) - if exclude: - stmt = stmt.where(ImageRecord.id.not_in(exclude)) - return [r[0] for r in session.execute(stmt).all()] - - -def _load_embeddings(session: Session, ids: list[int]) -> dict[int, Any]: - import numpy as np - - out: dict[int, Any] = {} - if not ids: - return out - # Chunk the IN list to stay well under psycopg's parameter ceiling. - for i in range(0, len(ids), 2000): - chunk = ids[i:i + 2000] - for rid, emb in session.execute( - select(ImageRecord.id, ImageRecord.siglip_embedding) - .where(ImageRecord.id.in_(chunk)) - .where(ImageRecord.siglip_embedding.is_not(None)) - ).all(): - out[rid] = np.asarray(emb, dtype=np.float32) - return out - - -def run_eval(session: Session, params: dict[str, Any]) -> dict[str, Any]: - """Compute the full report. Per-concept failures are captured, not fatal.""" - import numpy as np - - cfg = _normalize_params(params) - # Auto-discovery: union the explicit concepts with the top-N most-tagged - # general tags (server-side, fast) so the eval can broaden itself. - concepts = list(cfg["concepts"]) - if cfg["auto_top_n"]: - seen = {c.lower() for c in concepts} - for name in _top_general_concepts(session, cfg["auto_top_n"], MIN_POSITIVES): - if name.lower() not in seen: - concepts.append(name) - seen.add(name.lower()) - cfg["concepts"] = concepts - concepts_out = [] - for name in cfg["concepts"]: - try: - concepts_out.append(_eval_concept(session, name, cfg, np)) - except Exception as exc: # one bad concept shouldn't kill the run - log.exception("tag-eval concept %r failed", name) - concepts_out.append({"name": name, "skipped": f"error: {exc}"}) - return { - "generated_at": datetime.now(UTC).isoformat(), - "params": cfg, - "concepts": concepts_out, - } - - -def _eval_concept(session: Session, name: str, cfg: dict, np) -> dict[str, Any]: - tag_id = _resolve_tag_id(session, name) - if tag_id is None: - return {"name": name, "skipped": "no such tag"} - pos_ids = _ids_with_tag(session, tag_id) - if len(pos_ids) < MIN_POSITIVES: - return {"name": name, "tag_id": tag_id, "n_pos": len(pos_ids), - "skipped": f"too few positives (<{MIN_POSITIVES})"} - - neg_ratio = cfg["neg_ratio"] - pos_set = set(pos_ids) - rejected = [i for i in _rejected_ids(session, tag_id) if i not in pos_set] - want_neg = max(len(pos_ids) * neg_ratio, _EXAMPLES_K * 4) - sampled = _sample_unlabeled(session, pos_set | set(rejected), - min(_UNLABELED_POOL, want_neg)) - neg_ids = rejected + [i for i in sampled if i not in pos_set] - - emb = _load_embeddings(session, pos_ids + neg_ids) - pos = [(i, emb[i]) for i in pos_ids if i in emb] - neg = [(i, emb[i]) for i in neg_ids if i in emb] - if len(pos) < MIN_POSITIVES or len(neg) < MIN_POSITIVES: - return {"name": name, "tag_id": tag_id, "n_pos": len(pos), - "n_neg": len(neg), "skipped": "too few embedded examples"} - - ids = np.array([i for i, _ in pos] + [i for i, _ in neg]) - X = np.vstack([v for _, v in pos] + [v for _, v in neg]).astype(np.float32) - y = np.array([1] * len(pos) + [0] * len(neg)) - Xn = _l2norm(X, np) - - head = _eval_head(Xn, y, cfg["cv_folds"], cfg["precision_target"], np) - centroid = _eval_centroid(Xn, y, cfg["cv_folds"], np) - curve = _learning_curve(Xn, y, cfg["curve_points"], neg_ratio, np) - confirmed = _confirmed_ids(session, tag_id) - examples = _examples(session, Xn, y, ids, np, set(rejected), confirmed) - - return { - "name": name, "tag_id": tag_id, - "n_pos": len(pos), "n_neg": len(neg), - "n_rejected": len(rejected), - "head": head, "centroid": centroid, - "curve": curve, "examples": examples, - } - - -def _l2norm(X, np): - n = np.linalg.norm(X, axis=1, keepdims=True) - n[n == 0] = 1.0 - return X / n - - -def _metrics_from_scores(y, scores, np) -> dict[str, float]: - from sklearn.metrics import average_precision_score, precision_recall_curve - - ap = float(average_precision_score(y, scores)) - prec, rec, thr = precision_recall_curve(y, scores) - f1 = (2 * prec * rec) / np.clip(prec + rec, 1e-9, None) - best = int(np.argmax(f1)) - # thr has len = len(prec)-1; map best index safely. - t = float(thr[min(best, len(thr) - 1)]) if len(thr) else 0.5 - return { - "ap": round(ap, 4), - "precision": round(float(prec[best]), 4), - "recall": round(float(rec[best]), 4), - "f1": round(float(f1[best]), 4), - "threshold": round(t, 4), - } - - -def _safe_folds(y, folds, np) -> int: - minority = int(min(np.bincount(y))) - return max(2, min(folds, minority)) - - -def _eval_head(Xn, y, folds, target, np) -> dict[str, float]: - from sklearn.linear_model import LogisticRegression - from sklearn.model_selection import StratifiedKFold, cross_val_predict - - clf = LogisticRegression(max_iter=1000, class_weight="balanced") - cv = StratifiedKFold(n_splits=_safe_folds(y, folds, np), shuffle=True, - random_state=0) - probs = cross_val_predict(clf, Xn, y, cv=cv, method="predict_proba")[:, 1] - m = _metrics_from_scores(y, probs, np) - m["auto_apply"] = _auto_apply_point(y, probs, target, np) - return m - - -def _auto_apply_point(y, scores, target, np) -> dict | None: - """The auto-apply operating point: the threshold that yields the MOST recall - while holding precision >= target. This answers 'could this concept fire - without a human, and how much would it catch?' Returns None if no threshold - reaches the precision target (concept not auto-apply-ready).""" - from sklearn.metrics import precision_recall_curve - - prec, rec, thr = precision_recall_curve(y, scores) - best = None # (threshold, precision, recall) maximizing recall s.t. prec>=target - for i in range(len(thr)): # thr[i] corresponds to prec[i], rec[i] - if prec[i] >= target and (best is None or rec[i] > best[2]): - best = (float(thr[i]), float(prec[i]), float(rec[i])) - if best is None: - return None - return { - "target": round(float(target), 4), - "threshold": round(best[0], 4), - "precision": round(best[1], 4), - "recall": round(best[2], 4), - } - - -def _eval_centroid(Xn, y, folds, np) -> dict[str, float]: - """Cross-validated cosine-to-positive-mean — the OLD method's quality.""" - from sklearn.model_selection import StratifiedKFold - - cv = StratifiedKFold(n_splits=_safe_folds(y, folds, np), shuffle=True, - random_state=0) - scores = np.zeros(len(y), dtype=np.float32) - for train, test in cv.split(Xn, y): - c = Xn[train][y[train] == 1].mean(axis=0) - cn = c / (np.linalg.norm(c) or 1.0) - scores[test] = Xn[test] @ cn - return _metrics_from_scores(y, scores, np) - - -def _learning_curve(Xn, y, points, neg_ratio, np) -> list[dict[str, float]]: - """Hold out a fixed test split; train the head on a growing number of - positives and watch AP/F1 climb — answers 'does tagging more sharpen it?'""" - from sklearn.linear_model import LogisticRegression - from sklearn.model_selection import train_test_split - - rng = np.random.default_rng(0) - idx = np.arange(len(y)) - try: - tr, te = train_test_split(idx, test_size=0.3, stratify=y, random_state=0) - except ValueError: - return [] - tr_pos = tr[y[tr] == 1] - tr_neg = tr[y[tr] == 0] - out = [] - for n in points: - if n > len(tr_pos): - break - sp = rng.choice(tr_pos, size=n, replace=False) - nn = min(len(tr_neg), n * neg_ratio) - sn = rng.choice(tr_neg, size=nn, replace=False) - sub = np.concatenate([sp, sn]) - clf = LogisticRegression(max_iter=1000, class_weight="balanced") - clf.fit(Xn[sub], y[sub]) - prob = clf.predict_proba(Xn[te])[:, 1] - m = _metrics_from_scores(y[te], prob, np) - out.append({"n_pos": int(n), "ap": m["ap"], "f1": m["f1"]}) - return out - - -def _examples(session, Xn, y, ids, np, rejected_set, confirmed_set) -> dict[str, list[dict]]: - """Train on all data, then surface: top-scoring negatives the operator has - NOT already rejected (= fresh suggestions) and lowest-scoring POSITIVES the - operator has NOT already confirmed (= unreviewed doubts). Excluding rejected - ids stops an adjudicated near-miss from resurfacing in 'would suggest'; - excluding confirmed ids stops a 'kept' correct positive from resurfacing in - 'head doubts' every run. Resolves thumbnail urls for a self-contained report.""" - from sklearn.linear_model import LogisticRegression - - clf = LogisticRegression(max_iter=1000, class_weight="balanced") - clf.fit(Xn, y) - s = clf.predict_proba(Xn)[:, 1] - neg_idx = np.where(y == 0)[0] - pos_idx = np.where(y == 1)[0] - top_neg = [] - for i in neg_idx[np.argsort(s[neg_idx])[::-1]]: # high score → low - rid = int(ids[i]) - if rid in rejected_set: - continue # already told the head 'no' — don't re-suggest it - top_neg.append(rid) - if len(top_neg) >= _EXAMPLES_K: - break - low_pos = [] - for i in pos_idx[np.argsort(s[pos_idx])]: # low score → high - rid = int(ids[i]) - if rid in confirmed_set: - continue # already kept/confirmed — don't re-doubt it - low_pos.append(rid) - if len(low_pos) >= _EXAMPLES_K: - break - thumbs = _resolve_thumbs(session, top_neg + low_pos) - return { - "head_would_suggest": [thumbs[i] for i in top_neg if i in thumbs], - "head_doubts_positive": [thumbs[i] for i in low_pos if i in thumbs], - } - - -def _resolve_thumbs(session, ids: list[int]) -> dict[int, dict]: - from ..gallery_service import thumbnail_url - - out: dict[int, dict] = {} - if not ids: - return out - for rid, tp, sha, mime in session.execute( - select( - ImageRecord.id, ImageRecord.thumbnail_path, - ImageRecord.sha256, ImageRecord.mime, - ).where(ImageRecord.id.in_(ids)) - ).all(): - out[rid] = {"id": rid, "thumbnail_url": thumbnail_url(tp, sha, mime)} - return out diff --git a/backend/app/services/ml/training_data.py b/backend/app/services/ml/training_data.py new file mode 100644 index 0000000..d01acf7 --- /dev/null +++ b/backend/app/services/ml/training_data.py @@ -0,0 +1,121 @@ +"""Shared data-selection + validated-metric helpers for the heads trainer. + +Born in the head-vs-centroid eval harness (#1130, tag_eval.py) that proved the +"frozen embedding + small trained head (with negatives)" spine; the harness was +retired 2026-07-02 (operator: the tagging system is proven, the eval isn't +needed) and these survivors moved here — they ARE the heads' production data +pipeline (heads.py trains and scores with them nightly). + +numpy/scikit-learn are imported lazily inside the functions that need them so +the API worker (base image, no ML stack) can import this module. +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from ...models import ImageRecord, TagSuggestionRejection +from ...models.tag import image_tag + + +def _ids_with_tag(session: Session, tag_id: int) -> list[int]: + return [ + r[0] for r in session.execute( + select(image_tag.c.image_record_id).where(image_tag.c.tag_id == tag_id) + ).all() + ] + + +def _rejected_ids(session: Session, tag_id: int) -> list[int]: + return [ + r[0] for r in session.execute( + select(TagSuggestionRejection.image_record_id) + .where(TagSuggestionRejection.tag_id == tag_id) + ).all() + ] + + +def _sample_unlabeled(session: Session, exclude: set[int], limit: int) -> list[int]: + """Random image ids (with an embedding) NOT carrying the tag. Concepts are + sparse, so an untagged image is almost always a true negative.""" + stmt = ( + select(ImageRecord.id) + .where(ImageRecord.siglip_embedding.is_not(None)) + .order_by(func.random()) + .limit(limit) + ) + if exclude: + stmt = stmt.where(ImageRecord.id.not_in(exclude)) + return [r[0] for r in session.execute(stmt).all()] + + +def _load_embeddings(session: Session, ids: list[int]) -> dict[int, Any]: + import numpy as np + + out: dict[int, Any] = {} + if not ids: + return out + # Chunk the IN list to stay well under psycopg's parameter ceiling. + for i in range(0, len(ids), 2000): + chunk = ids[i:i + 2000] + for rid, emb in session.execute( + select(ImageRecord.id, ImageRecord.siglip_embedding) + .where(ImageRecord.id.in_(chunk)) + .where(ImageRecord.siglip_embedding.is_not(None)) + ).all(): + out[rid] = np.asarray(emb, dtype=np.float32) + return out + + +def _l2norm(X, np): + n = np.linalg.norm(X, axis=1, keepdims=True) + n[n == 0] = 1.0 + return X / n + + +def _metrics_from_scores(y, scores, np) -> dict[str, float]: + from sklearn.metrics import average_precision_score, precision_recall_curve + + ap = float(average_precision_score(y, scores)) + prec, rec, thr = precision_recall_curve(y, scores) + f1 = (2 * prec * rec) / np.clip(prec + rec, 1e-9, None) + best = int(np.argmax(f1)) + # thr has len = len(prec)-1; map best index safely. + t = float(thr[min(best, len(thr) - 1)]) if len(thr) else 0.5 + return { + "ap": round(ap, 4), + "precision": round(float(prec[best]), 4), + "recall": round(float(rec[best]), 4), + "f1": round(float(f1[best]), 4), + "threshold": round(t, 4), + } + + +def _safe_folds(y, folds, np) -> int: + minority = int(min(np.bincount(y))) + return max(2, min(folds, minority)) + + +def _auto_apply_point(y, scores, target, np) -> dict | None: + """The auto-apply operating point: the threshold that yields the MOST recall + while holding precision >= target. This answers 'could this concept fire + without a human, and how much would it catch?' Returns None if no threshold + reaches the precision target (concept not auto-apply-ready).""" + from sklearn.metrics import precision_recall_curve + + prec, rec, thr = precision_recall_curve(y, scores) + best = None # (threshold, precision, recall) maximizing recall s.t. prec>=target + for i in range(len(thr)): # thr[i] corresponds to prec[i], rec[i] + if prec[i] >= target and (best is None or rec[i] > best[2]): + best = (float(thr[i]), float(prec[i]), float(rec[i])) + if best is None: + return None + return { + "target": round(float(target), 4), + "threshold": round(best[0], 4), + "precision": round(best[1], 4), + "recall": round(best[2], 4), + } diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 842df0f..e403cf5 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -21,7 +21,6 @@ from ..models import ( ImportTask, LibraryAuditRun, Source, - TagEvalRun, TaskRun, ) from ..utils.phash import compute_phash @@ -96,9 +95,6 @@ BACKUP_DB_STALL_THRESHOLD_MINUTES = 40 # Library audit: scan_library_for_rule has time_limit=7500s (2h5m). # 2h15m gives a 10-min buffer. LIBRARY_AUDIT_STALL_THRESHOLD_MINUTES = 135 -# tag-eval (#1130) has a 30-min soft limit; flag a run with no progress past 40. -TAG_EVAL_STALL_THRESHOLD_MINUTES = 40 -TAG_EVAL_KEEP_RUNS = 20 # head training (#114) has a 60-min soft limit; flag no-progress past 75. HEAD_TRAINING_STALL_THRESHOLD_MINUTES = 75 HEAD_TRAINING_KEEP_RUNS = 20 @@ -743,46 +739,6 @@ def recover_stalled_library_audit_runs() -> int: return recovered -@celery.task(name="backend.app.tasks.maintenance.recover_stalled_tag_eval_runs") -def recover_stalled_tag_eval_runs() -> int: - """Flip TagEvalRun rows stuck in 'running' past the stall threshold to - 'error', and prune old runs to the last TAG_EVAL_KEEP_RUNS (retention, - rule 89). Runs every 5 min on the maintenance lane; no-op when idle.""" - SessionLocal = _sync_session_factory() - now = datetime.now(UTC) - cutoff = now - timedelta(minutes=TAG_EVAL_STALL_THRESHOLD_MINUTES) - with SessionLocal() as session: - result = session.execute( - update(TagEvalRun) - .where(TagEvalRun.status == "running") - .where( - func.coalesce(TagEvalRun.last_progress_at, TagEvalRun.started_at) - < cutoff - ) - .values( - status="error", finished_at=now, - error=( - f"stranded by recovery sweep (no progress for " - f"{TAG_EVAL_STALL_THRESHOLD_MINUTES} min)" - ), - ) - ) - # Retention: keep only the most recent N runs. - keep = session.execute( - select(TagEvalRun.id).order_by(TagEvalRun.id.desc()) - .limit(TAG_EVAL_KEEP_RUNS) - ).scalars().all() - if keep: - session.execute( - delete(TagEvalRun).where(TagEvalRun.id.not_in(keep)) - ) - session.commit() - recovered = result.rowcount or 0 - if recovered: - log.info("recover_stalled_tag_eval_runs: recovered %d rows", recovered) - return recovered - - @celery.task(name="backend.app.tasks.maintenance.recover_stalled_head_training_runs") def recover_stalled_head_training_runs() -> int: """Flip HeadTrainingRun rows stuck in 'running' past the stall threshold to diff --git a/backend/app/tasks/ml.py b/backend/app/tasks/ml.py index 895f7e2..737c36c 100644 --- a/backend/app/tasks/ml.py +++ b/backend/app/tasks/ml.py @@ -250,51 +250,6 @@ def backfill(self) -> int: return enqueued -@celery.task( - name="backend.app.tasks.ml.tag_eval_run", - bind=True, - # The head-vs-centroid eval (#1130) loads embeddings + fits sklearn heads - # for several concepts — minutes, not seconds. Runs on the ml queue because - # only that worker has numpy/scikit-learn. - soft_time_limit=1800, time_limit=2100, -) -def tag_eval_run(self, run_id: int) -> str: - """Compute the eval report into the persisted TagEvalRun row so it survives - navigation (the admin card rehydrates from the row, not transient state).""" - from datetime import UTC, datetime - - from ..models import TagEvalRun - from ..services.ml.tag_eval import run_eval - - SessionLocal = _sync_session_factory() - with SessionLocal() as session: - run = session.get(TagEvalRun, run_id) - if run is None: - return "missing" - run.last_progress_at = datetime.now(UTC) - session.commit() - try: - report = run_eval(session, run.params) - except SoftTimeLimitExceeded: - run.status = "error" - run.error = "timed out" - run.finished_at = datetime.now(UTC) - session.commit() - raise - except Exception as exc: - log.exception("tag_eval_run %d failed", run_id) - run.status = "error" - run.error = str(exc) - run.finished_at = datetime.now(UTC) - session.commit() - return "error" - run.report = report - run.status = "ready" - run.finished_at = datetime.now(UTC) - session.commit() - return "ready" - - @celery.task( name="backend.app.tasks.ml.train_heads", bind=True, diff --git a/frontend/src/components/modal/SuggestionItem.vue b/frontend/src/components/modal/SuggestionItem.vue index b9df764..e0e196d 100644 --- a/frontend/src/components/modal/SuggestionItem.vue +++ b/frontend/src/components/modal/SuggestionItem.vue @@ -132,8 +132,8 @@ const hasMenu = computed(() => color: rgb(var(--v-theme-on-surface-variant, var(--v-theme-on-surface))); font-family: 'JetBrains Mono', monospace; } -/* Green ✓ / red ✗ verdict pair — same circular language as the eval card - (TagEvalCard .fc-act) so accept/reject read identically across surfaces. */ +/* Green ✓ / red ✗ verdict pair — circular buttons so accept/reject read + identically across surfaces. */ .fc-suggestion__acts { flex: 0 0 auto; display: flex; gap: 4px; } diff --git a/frontend/src/components/settings/MaintenancePanel.vue b/frontend/src/components/settings/MaintenancePanel.vue index 8957496..69acbcc 100644 --- a/frontend/src/components/settings/MaintenancePanel.vue +++ b/frontend/src/components/settings/MaintenancePanel.vue @@ -28,7 +28,6 @@ - @@ -55,7 +54,6 @@ import MLThresholdSliders from './MLThresholdSliders.vue' import HeadsCard from './HeadsCard.vue' import GpuAgentCard from './GpuAgentCard.vue' import AliasTable from './AliasTable.vue' -import TagEvalCard from './TagEvalCard.vue' import BackupCard from './BackupCard.vue' import { useSystemActivityStore } from '../../stores/systemActivity.js' diff --git a/frontend/src/components/settings/TagEvalCard.vue b/frontend/src/components/settings/TagEvalCard.vue deleted file mode 100644 index 1e1af56..0000000 --- a/frontend/src/components/settings/TagEvalCard.vue +++ /dev/null @@ -1,303 +0,0 @@ - - - - - diff --git a/frontend/src/stores/tagEval.js b/frontend/src/stores/tagEval.js deleted file mode 100644 index 4abbe1b..0000000 --- a/frontend/src/stores/tagEval.js +++ /dev/null @@ -1,57 +0,0 @@ -import { defineStore } from 'pinia' - -import { useApi } from '../composables/useApi.js' - -// Tag-eval (#1130): trigger + revisit the head-vs-centroid learning-curve eval. -// The run + full report live server-side (tag_eval_run), so the card rehydrates -// from getRun() on mount — the report survives navigation. -export const useTagEvalStore = defineStore('tagEval', () => { - const api = useApi() - - async function start(params) { - return await api.post('/api/tag-eval', { body: { params } }) - } - - async function getRun(id) { - return await api.get(`/api/tag-eval/${id}`) // includes the full report - } - - // The most recent run (light row, no report) — the card calls getRun() with - // its id to pull the persisted report on mount. - async function latest() { - const body = await api.get('/api/tag-eval', { params: { limit: 1 } }) - return (body.runs && body.runs[0]) || null - } - - // --- Acting on the head's example lists (closes the learn-from-tags loop). - // These write the SAME tables the head trains on: image_tag (positives) and - // tag_suggestion_rejection (negatives, via the dismiss endpoint). - - // "Yes, it is this" — apply the tag (new positive). - async function applyTag(imageId, tagId) { - return await api.post(`/api/images/${imageId}/tags`, - { body: { tag_id: tagId, source: 'manual' } }) - } - - // "No, it's not" on an UNtagged suggestion — record a rejection (hard negative). - async function rejectTag(imageId, tagId) { - return await api.post(`/api/images/${imageId}/suggestions/dismiss`, - { body: { tag_id: tagId } }) - } - - // "Not it" on one of YOUR positives the head doubts — remove the tag AND - // record the rejection (kills the bad positive, leaves a hard negative). - async function removeTag(imageId, tagId) { - await api.delete(`/api/images/${imageId}/tags/${tagId}`) - return await api.post(`/api/images/${imageId}/suggestions/dismiss`, - { body: { tag_id: tagId } }) - } - - // "Keep" — affirm a doubted positive is correct. Records a confirmation so it - // stops resurfacing in the doubts list (it stays a positive either way). - async function confirmTag(imageId, tagId) { - return await api.post(`/api/images/${imageId}/tags/${tagId}/confirm`) - } - - return { start, getRun, latest, applyTag, rejectTag, removeTag, confirmTag } -}) diff --git a/tests/test_api_tag_eval.py b/tests/test_api_tag_eval.py deleted file mode 100644 index 2addd1d..0000000 --- a/tests/test_api_tag_eval.py +++ /dev/null @@ -1,77 +0,0 @@ -import pytest - -from backend.app.models import TagEvalRun -from backend.app.services.ml.tag_eval import ( - DEFAULT_CONCEPTS, - _normalize_params, -) - -pytestmark = pytest.mark.integration - - -def test_normalize_params_defaults_and_overrides(): - d = _normalize_params(None) - assert d["concepts"] == DEFAULT_CONCEPTS - assert d["neg_ratio"] >= 1 and d["cv_folds"] >= 2 - over = _normalize_params( - {"concepts": ["glasses", " ", "cat"], "neg_ratio": "4", - "cv_folds": "1", "curve_points": [30, 10, 10]} - ) - assert over["concepts"] == ["glasses", "cat"] # blanks dropped - assert over["neg_ratio"] == 4 - assert over["cv_folds"] == 2 # clamped to >=2 - assert over["curve_points"] == [10, 30] # deduped + sorted - - -@pytest.mark.asyncio -async def test_history_and_detail_rehydrate(client, db): - # A finished run with a report — the persisted row IS the survives-navigation - # source: history is light (no report), detail carries it. - run = TagEvalRun( - params={"concepts": ["glasses"]}, - status="ready", - report={"concepts": [{"name": "glasses", "head": {"ap": 0.9}}]}, - ) - db.add(run) - await db.flush() - await db.commit() - rid = run.id - - h = await client.get("/api/tag-eval?limit=10") - assert h.status_code == 200 - hbody = await h.get_json() - row = next(r for r in hbody["runs"] if r["id"] == rid) - assert row["status"] == "ready" - assert "report" not in row # list stays light - - d = await client.get(f"/api/tag-eval/{rid}") - assert d.status_code == 200 - dbody = await d.get_json() - assert dbody["report"]["concepts"][0]["name"] == "glasses" - - -@pytest.mark.asyncio -async def test_create_enqueues_running(client, db, monkeypatch): - monkeypatch.setattr( - "backend.app.tasks.ml.tag_eval_run.delay", lambda *a, **k: None - ) - resp = await client.post("/api/tag-eval", json={"params": {"concepts": ["cat"]}}) - assert resp.status_code == 202 - body = await resp.get_json() - assert body["status"] == "running" - got = await db.get(TagEvalRun, body["run_id"]) - assert got is not None and got.status == "running" - - -@pytest.mark.asyncio -async def test_create_conflicts_when_one_running(client, db, monkeypatch): - monkeypatch.setattr( - "backend.app.tasks.ml.tag_eval_run.delay", lambda *a, **k: None - ) - db.add(TagEvalRun(params={}, status="running")) - await db.flush() - await db.commit() - resp = await client.post("/api/tag-eval", json={"params": {}}) - assert resp.status_code == 409 - body = await resp.get_json() - assert body["error"] == "eval_already_running"