feat: replace face pipeline with InsightFace, add content classifier

Face detection/recognition:
- Replace YuNet + SFace with InsightFace buffalo_l (RetinaFace + ArcFace)
- 512-d ArcFace embeddings (was 128-d SFace), migration 0006 resizes column
- Remove YOLO person-bbox workaround — RetinaFace is accurate enough
- Detection threshold 0.65 cleanly separates real faces (0.72+) from
  false positives on dogs/paintings (0.56-0.61)

Content-type classification:
- CLIP zero-shot classifier using native PyTorch text encoder + ONNX
  image encoder for high-quality text-image similarity
- Categories: photograph, screenshot, document, receipt, meme, artwork
- Writes Tag(kind=content_type) per photo via photo_tags
- Margin-based confidence: top-1 vs top-2 score difference
- New ClassifierSettings in config (enabled, min_confidence)
- Wired into vision_fanout pipeline

Tested: 6 real faces from 4 photos (zero false positives), 11/13 photos
classified (8 photograph, 2 artwork, 1 meme).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-10 13:49:02 +02:00
parent f48e099bd2
commit fa9b21856f
14 changed files with 331 additions and 73 deletions

View File

@@ -93,6 +93,8 @@ def vision_fanout(photo_id: str):
detect_objects.delay(photo_id)
if settings.vision.faces.enabled:
extract_faces.delay(photo_id)
if settings.vision.classifier.enabled:
classify_content.delay(photo_id)
return {'status': 'dispatched', 'photo_id': photo_id}
@@ -206,6 +208,66 @@ def detect_objects(photo_id: str):
return {'status': 'success', 'photo_id': photo_id, 'objects': len(detections)}
@shared_task(name='classify_content', queue='vision')
def classify_content(photo_id: str):
"""Classify image content type (screenshot, document, artwork, etc.)
using CLIP zero-shot classification. Writes Tag(kind=content_type)."""
if not settings.vision.enabled or not settings.vision.classifier.enabled:
return {'status': 'skipped', 'reason': 'classifier disabled'}
image = _load_thumb(photo_id, "medium")
if image is None:
return {'status': 'error', 'message': 'thumbnail not found'}
from app.services.vision.registry import registry
classifier = registry.get_classifier()
results = classifier.classify(image)
if not results:
logger.info("No confident classification for photo %s", photo_id)
return {'status': 'success', 'photo_id': photo_id, 'content_type': None}
from app.models.tags import Tag, photo_tags
source_name = "vision:clip_classifier"
best = results[0]
session = _get_sync_session()
try:
# Wipe previous classification for this photo
session.execute(
delete(photo_tags).where(
photo_tags.c.photo_id == photo_id,
photo_tags.c.source == source_name,
)
)
# Find or create content_type tag
tag = session.execute(
select(Tag).where(Tag.name == best.label, Tag.kind == 'content_type')
).scalar_one_or_none()
if not tag:
tag = Tag(name=best.label, kind='content_type', source=source_name)
session.add(tag)
session.flush()
session.execute(
photo_tags.insert().values(
photo_id=photo_id,
tag_id=tag.id,
confidence=best.confidence,
source=source_name,
)
)
session.commit()
finally:
session.close()
logger.info("Classified photo %s as '%s' (%.2f)", photo_id, best.label, best.confidence)
return {'status': 'success', 'photo_id': photo_id, 'content_type': best.label}
def _load_original(photo_id: str) -> np.ndarray | None:
"""Load the original photo file as an RGB numpy array, resized to
max 1280px on the longest edge for face detection."""
@@ -242,72 +304,26 @@ def _load_original(photo_id: str) -> np.ndarray | None:
return None
def _iou(a: list[float], b: list[float]) -> float:
"""Intersection-over-area of box a within box b (how much of a is inside b).
Boxes are [x1, y1, x2, y2] normalized 0-1."""
x1 = max(a[0], b[0])
y1 = max(a[1], b[1])
x2 = min(a[2], b[2])
y2 = min(a[3], b[3])
inter = max(0, x2 - x1) * max(0, y2 - y1)
area_a = max(0, a[2] - a[0]) * max(0, a[3] - a[1])
return inter / area_a if area_a > 0 else 0
def _face_inside_person(face_bbox: list[float], person_bboxes: list[list[float]]) -> bool:
"""Return True if the face bbox overlaps at least 50% with any YOLO
'person' detection. Filters out faces on dogs, paintings, etc."""
for pb in person_bboxes:
if _iou(face_bbox, pb) >= 0.5:
return True
return False
@shared_task(name='extract_faces', queue='vision')
def extract_faces(photo_id: str):
"""Detect faces and store recognition embeddings. Only keeps faces
that overlap with a YOLO 'person' detection to filter out animal
and cartoon false positives."""
"""Detect faces and store recognition embeddings using InsightFace
(RetinaFace + ArcFace). No YOLO workaround needed — RetinaFace has
strong human-vs-non-human precision on its own."""
if not settings.vision.enabled or not settings.vision.faces.enabled:
return {'status': 'skipped', 'reason': 'faces disabled'}
# Use original file for face detection — thumbnails are often too
# small (240px) for reliable face detection.
image = _load_original(photo_id)
if image is None:
image = _load_thumb(photo_id, "large")
if image is None:
return {'status': 'error', 'message': 'no image available'}
# Step 1: run YOLO to find "person" bounding boxes
from app.services.vision.registry import registry
detector = registry.get_detector()
thumb = _load_thumb(photo_id, "medium")
person_bboxes = []
if thumb is not None:
detections = detector.detect(thumb)
person_bboxes = [d.bbox for d in detections if d.label == 'person']
# Step 2: run face detection
face_proc = registry.get_face_processor()
faces = face_proc.process(image)
if not faces:
logger.info("No faces detected for photo %s", photo_id)
return _save_faces(photo_id, [])
# Step 3: filter — keep only faces inside a person bbox
if person_bboxes:
verified = [f for f in faces if _face_inside_person(f.bbox, person_bboxes)]
dropped = len(faces) - len(verified)
if dropped:
logger.info("Dropped %d non-person face(s) for photo %s", dropped, photo_id)
faces = verified
else:
# No person detected by YOLO → drop all face detections
# (no human body visible = likely false positives)
logger.info("No YOLO person detected, dropping %d face(s) for photo %s", len(faces), photo_id)
faces = []
return _save_faces(photo_id, faces)