feat(ml): cadence-based video frame sampling + min-frame tag aggregation (#747)

Video tag noise root cause: frames were a FIXED count (6) max-pooled — a tag
firing on one frame survived at peak confidence, and a fixed count under-samples
long multi-scene videos so real scene-local tags looked like noise.

Redesign (operator-steered):
- Sample at a fixed CADENCE — one frame every `video_frame_interval_seconds`
  (default 4) across the 5–95% window — so a tag's frame-presence reflects real
  screen time independent of video length. Capped at `video_max_frames` (default
  64): a long video stretches the spacing instead of exploding into hundreds of
  inferences, bounding per-video cost on the single ml-worker (per-frame ffmpeg
  timeout also cut 60s→30s).
- Aggregate with `_aggregate_video_predictions`: keep a tag only if it appears in
  >= `video_min_tag_frames` sampled frames (≈ that many × interval seconds on
  screen — duration-independent noise rejection), with confidence = MEAN over the
  frames it appears in (not max). Clamps the threshold to the sample count so a
  1–2-frame short video still tags.
- All three knobs are DB-backed ml_settings (migration 0053), patchable via
  /api/ml/settings + sliders in the ML settings card — replaces the
  VIDEO_ML_FRAMES env var (product-not-project).

Tests: aggregation drops one-frame noise + means corroborated tags + clamps on
short videos; settings round-trip + min>max validation. Replaced the
_maxpool_predictions unit test.

NOTE: this is the QUALITY half of #747. The perf half — the ml-worker runs
CPU-only — is GPU enablement, tracked separately in #872.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/api/ml_admin.py

@@ -14,6 +14,9 @@ _EDITABLE = (
     "centroid_similarity_threshold",
     "min_reference_images",
     "tagger_store_floor",
+    "video_frame_interval_seconds",
+    "video_max_frames",
+    "video_min_tag_frames",
 )
 
 
@@ -32,6 +35,9 @@ async def get_settings():
             "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,
+            "video_min_tag_frames": s.video_min_tag_frames,
             "tagger_model_version": s.tagger_model_version,
             "embedder_model_version": s.embedder_model_version,
         }
@@ -85,6 +91,15 @@ def _validate(p: dict) -> str | None:
                 f"suggestion_threshold_{cat} cannot be below tagger_store_floor "
                 f"({floor}) — predictions below the floor are not stored"
             )
+    # Video tagging (#747).
+    if p["video_frame_interval_seconds"] <= 0:
+        return "video_frame_interval_seconds must be > 0"
+    if p["video_max_frames"] < 1:
+        return "video_max_frames must be >= 1"
+    if p["video_min_tag_frames"] < 1:
+        return "video_min_tag_frames must be >= 1"
+    if p["video_min_tag_frames"] > p["video_max_frames"]:
+        return "video_min_tag_frames cannot exceed video_max_frames"
     return None
 
 

backend/app/models/ml_settings.py

@@ -40,6 +40,21 @@ class MLSettings(Base):
     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
+    # of inferences (the cadence stretches past the cap). A tag is kept only if it
+    # appears in >= video_min_tag_frames sampled frames (≈ that many × interval
+    # seconds on screen) — duration-independent noise rejection. Operator-tunable.
+    video_frame_interval_seconds: Mapped[float] = mapped_column(
+        Float, nullable=False, default=4.0
+    )
+    video_max_frames: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=64
+    )
+    video_min_tag_frames: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=3
+    )
     tagger_model_version: Mapped[str] = mapped_column(
         String(128), nullable=False, default="camie-tagger-v2"
     )

backend/app/tasks/ml.py

@@ -49,11 +49,12 @@ def tag_and_embed(self, image_id: int) -> dict:
     """Run Camie + SigLIP on one image; store predictions + embedding;
     then enqueue per-image allowlist application.
 
-    Video: sample frames between 10% and 90% of duration (VIDEO_ML_FRAMES,
-    default 6). Max-pool tagger confidences across frames, mean-pool the
+    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).
     """
-    import os
     import time
 
     from ..services.ml.embedder import get_embedder
@@ -116,7 +117,9 @@ def tag_and_embed(self, image_id: int) -> dict:
                 phase = "video_sample_frames"
                 t0 = time.monotonic()
                 frames = _sample_video_frames(
-                    src, int(os.environ.get("VIDEO_ML_FRAMES", "6"))
+                    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",
@@ -127,13 +130,19 @@ def tag_and_embed(self, image_id: int) -> dict:
                 phase = "video_infer"
                 import numpy as np
 
-                preds = _maxpool_predictions(
+                preds = _aggregate_video_predictions(
                     [tagger.infer(f, store_floor=settings.tagger_store_floor)
-                     for f in frames]
+                     for f in frames],
+                    min_frames=settings.video_min_tag_frames,
                 )
                 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:
@@ -208,9 +217,17 @@ def tag_and_embed(self, image_id: int) -> dict:
     return {"status": "ok", "image_id": image_id, "tags": len(preds)}
 
 
-def _sample_video_frames(src: Path, n: int) -> list[Path]:
-    """Extract n frames evenly between 10% and 90% of duration via ffmpeg.
-    Returns temp file paths (caller deletes). Empty list on failure."""
+def _sample_video_frames(
+    src: Path, *, interval: float, max_frames: int,
+) -> list[Path]:
+    """Sample frames at a fixed CADENCE — one every `interval` seconds — so a
+    tag's frame-presence reflects real screen time regardless of video length
+    (#747). The count is capped at `max_frames`: a video longer than
+    interval×max_frames stretches the spacing instead of exploding the frame
+    count (keeps cost bounded so a long video can't hog the single ml-worker).
+    Frames are taken across the 5%–95% window (skip intro/outro black/cards) via
+    per-frame fast-seek. Returns temp file paths (caller deletes); [] on failure.
+    """
     import json
     import subprocess
     import tempfile
@@ -229,20 +246,25 @@ def _sample_video_frames(src: Path, n: int) -> list[Path]:
     if duration <= 0:
         return []
 
-    start, end = duration * 0.10, duration * 0.90
-    step = (end - start) / max(n - 1, 1)
+    start, end = duration * 0.05, duration * 0.95
+    span = max(end - start, 0.0)
+    # Cadence count, clamped to [1, max_frames]. int(span/interval)+1 ≈ one frame
+    # per `interval` seconds across the window; the cap stretches spacing on very
+    # long videos.
+    n = max(1, min(int(span / interval) + 1, max(1, max_frames)))
+    step = span / max(n - 1, 1)
     out: list[Path] = []
     tmpdir = Path(tempfile.mkdtemp(prefix="fc_vid_"))
     for i in range(n):
         ts = start + i * step
-        dest = tmpdir / f"frame_{i:02d}.jpg"
+        dest = tmpdir / f"frame_{i:04d}.jpg"
         try:
             subprocess.run(
                 [
                     "ffmpeg", "-ss", f"{ts:.2f}", "-i", str(src),
                     "-frames:v", "1", "-q:v", "3", "-y", str(dest),
                 ],
-                check=True, capture_output=True, timeout=60,
+                check=True, capture_output=True, timeout=30,
             )
             if dest.is_file():
                 out.append(dest)
@@ -251,18 +273,38 @@ def _sample_video_frames(src: Path, n: int) -> list[Path]:
     return out
 
 
-def _maxpool_predictions(per_frame: list[dict]) -> dict:
-    """Aggregate per-frame {name: TagPrediction} dicts by max confidence."""
-    merged: dict[str, dict] = {}
+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 = merged.get(name)
-            if cur is None or p.confidence > cur["confidence"]:
-                merged[name] = {
-                    "category": p.category,
-                    "confidence": p.confidence,
-                }
-    return merged
+            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)