develop

.env.example

@@ -18,6 +18,8 @@ AI_PRIVACY_BATCH_WAIT_MS=4
 AI_PRIVACY_MAX_PENDING_FRAMES=128
 AI_PRIVACY_FACE_IMGSZ=640
 AI_PRIVACY_ENABLE_IDENTITY_EXCLUSION=true
+AI_PRIVACY_IDENTITY_RECHECK_INTERVAL=12
+AI_PRIVACY_IDENTITY_NEGATIVE_RECHECK_INTERVAL=90
 AI_PRIVACY_ME_IMAGE_PATH=me.jpeg
 AI_PRIVACY_FACE_MODEL_PATH=models/yolov8n-face.pt
 AI_PRIVACY_FACE_DETECTION_MODEL_PATH=models/face_detection_yunet_2023mar.onnx

README.md

@@ -32,13 +32,19 @@ Mac 로컬 실행은 위와 같이 CPU 모드를 사용한다. 운영 환경은
 처리 순서는 다음과 같다.
 
 1. YOLO FP16 배치로 얼굴을 탐지한다.
-2. 신규 또는 주기적 확인 대상 얼굴을 YuNet/SFace 배치로 식별한다.
+2. YOLO가 함께 출력한 5개 랜드마크로 얼굴을 정렬하고 SFace 배치로 식별한다.
+   랜드마크가 유효하지 않은 얼굴만 YuNet을 폴백으로 사용한다.
 3. 참조 얼굴(`AI_PRIVACY_ME_IMAGE_PATH` 또는 `/reference-face` API)을 제외한
    얼굴에 확장 타원 마스크와 Gaussian blur를 GPU에서 적용한다.
 4. 처리 결과를 원래 WebRTC 트랙으로 반환한다.
 
 `GET /health`에서 장치, ONNX provider, 큐 깊이, 배치 수, 평균 배치 크기와 평균
-처리 지연을 확인할 수 있다.
+처리 지연을 확인할 수 있다. 얼굴 대조 수와 지연, YOLO 랜드마크 사용 수 및 YuNet
+폴백 수도 별도 지표로 제공한다.
+
+한 프레임에서 새로 감지된 모든 얼굴은 ROI Align과 SFace의 단일 CUDA 배치로
+즉시 대조한다. 본인 트랙은 `AI_PRIVACY_IDENTITY_RECHECK_INTERVAL`, 비본인 트랙은 더 긴
+`AI_PRIVACY_IDENTITY_NEGATIVE_RECHECK_INTERVAL` 주기로 재확인한다.
 
 ### NVIDIA RTX 3090 배포
 

SERVER_STRUCTURE.md

@@ -183,12 +183,17 @@ Status: active → closing → closed
 | `ai_privacy_max_pending_frames` | 128 | 전체 스트림 공유 대기열 상한 |
 | `ai_privacy_face_imgsz` | 640 | YOLO 입력 크기 |
 | `ai_privacy_enable_identity_exclusion` | true | 참조 얼굴 블러 제외 여부 |
+| `ai_privacy_identity_recheck_interval` | 12 | 본인 트랙 재대조 프레임 주기 |
+| `ai_privacy_identity_negative_recheck_interval` | 90 | 비본인 트랙 재대조 프레임 주기 |
 | `ai_privacy_me_image_path` | null | 환경변수 기반 참조 얼굴 이미지 경로 |
 
 - 앱 프로세스마다 `SharedGpuRuntime` 하나가 YOLO/YuNet/SFace 모델과 배치 워커를 소유
 - WebRTC 스트림마다 `FacePrivacyFilter` 하나를 생성해 얼굴 추적 상태를 격리
 - 최대 4ms 동안 모은 서로 다른 스트림 프레임을 최대 8개까지 한 번에 추론
-- CUDA에서는 YOLO FP16, ONNX Runtime CUDA I/O binding, PyTorch 마스크/블러 사용
+- YOLO의 5개 얼굴 랜드마크를 SFace 정렬에 재사용하고, 유효하지 않을 때만 YuNet 실행
+- 한 프레임의 모든 신규 얼굴 crop·정렬·SFace 추론을 단일 CUDA 배치로 즉시 처리
+- 비본인 얼굴은 본인보다 긴 주기로 재대조해 군중 장면의 반복 추론량을 제한
+- CUDA에서는 YOLO FP16, ROI Align, ONNX Runtime I/O binding·TF32·비동기 실행 사용
 - `ProtectedVideoTrack` 종료 시 스트림 필터 상태를 해제하고, 처리 실패 시 검은 프레임 송신
 
 ## 6. WebSocket 시그널링 프로토콜

app/core/config.py

@@ -24,6 +24,8 @@ class Settings(BaseSettings):
     ai_privacy_max_pending_frames: int = 128
     ai_privacy_face_imgsz: int = 640
     ai_privacy_enable_identity_exclusion: bool = True
+    ai_privacy_identity_recheck_interval: int = 12
+    ai_privacy_identity_negative_recheck_interval: int = 90
     ai_privacy_me_image_path: str | None = None
     ai_privacy_face_model_path: str = "models/yolov8n-face.pt"
     ai_privacy_face_detection_model_path: str = (

app/services/ai/filter.py

@@ -61,6 +61,10 @@ def build_privacy_config(settings: Settings) -> PrivacyBlurConfig:
         batch_wait_ms=settings.ai_privacy_batch_wait_ms,
         max_pending_frames=settings.ai_privacy_max_pending_frames,
         enable_identity_exclusion=identity_enabled,
+        identity_recheck_interval=settings.ai_privacy_identity_recheck_interval,
+        identity_negative_recheck_interval=(
+            settings.ai_privacy_identity_negative_recheck_interval
+        ),
         me_image_path=reference_path or _PROJECT_ROOT / "me.jpeg",
         face_detection_model_path=_asset_path(
             settings.ai_privacy_face_detection_model_path

privacy_blur/config.py

@@ -34,6 +34,7 @@ class PrivacyBlurConfig:
     face_match_cosine_threshold: float = 0.40
     face_match_lost_cosine_threshold: float = 0.30
     identity_recheck_interval: int = 12
+    identity_negative_recheck_interval: int = 90
     identity_min_face_size: int = 36
 
     detect_every_n_frames: int = 1
@@ -71,6 +72,13 @@ def __post_init__(self) -> None:
             raise ValueError("detect_every_n_frames must be at least 1")
         if self.identity_detector_size < 32 or self.identity_detector_size % 32:
             raise ValueError("identity_detector_size must be a multiple of 32")
+        if self.identity_recheck_interval < 1:
+            raise ValueError("identity_recheck_interval must be at least 1")
+        if self.identity_negative_recheck_interval < self.identity_recheck_interval:
+            raise ValueError(
+                "identity_negative_recheck_interval must be at least "
+                "identity_recheck_interval"
+            )
 
     def with_overrides(self, **kwargs: Any) -> "PrivacyBlurConfig":
         return replace(self, **kwargs)

privacy_blur/core.py

@@ -48,6 +48,13 @@ def track_match_score(
     return max(iou, center_score * 0.75)
 
 
+@dataclass
+class FaceDetection:
+    xywh: np.ndarray
+    confidence: float
+    landmarks: np.ndarray | None = None
+
+
 @dataclass
 class FaceTrack:
     track_id: int
@@ -57,6 +64,7 @@ class FaceTrack:
     similarity: float
     last_seen_frame: int
     last_checked_frame: int
+    landmarks: np.ndarray | None = None
 
 
 class FacePrivacyEngine:
@@ -94,26 +102,93 @@ def needs_detection(self) -> bool:
                 or self.frame_index % self.config.detect_every_n_frames == 0
             )
 
-    def _find_matching_track(
+    def _match_detections_to_tracks(
         self,
-        face_xywh: np.ndarray,
+        detections: list[FaceDetection],
         tracks: list[FaceTrack],
-        matched_track_ids: set[int],
-    ) -> FaceTrack | None:
-        best_track: FaceTrack | None = None
-        best_score = 0.0
-        for track in tracks:
-            if track.track_id in matched_track_ids:
-                continue
-            score = track_match_score(face_xywh, track.xywh, self.config)
-            if score > best_score:
-                best_score = score
-                best_track = track
-        return best_track if best_score > 0.0 else None
+    ) -> list[FaceTrack | None]:
+        if not detections or not tracks:
+            return [None] * len(detections)
+
+        detection_boxes = np.stack([item.xywh[:4] for item in detections]).astype(
+            np.float32, copy=False
+        )
+        track_boxes = np.stack([item.xywh[:4] for item in tracks]).astype(
+            np.float32, copy=False
+        )
+        detection_x1 = detection_boxes[:, None, 0]
+        detection_y1 = detection_boxes[:, None, 1]
+        detection_x2 = detection_x1 + detection_boxes[:, None, 2]
+        detection_y2 = detection_y1 + detection_boxes[:, None, 3]
+        track_x1 = track_boxes[None, :, 0]
+        track_y1 = track_boxes[None, :, 1]
+        track_x2 = track_x1 + track_boxes[None, :, 2]
+        track_y2 = track_y1 + track_boxes[None, :, 3]
+
+        intersection_width = np.maximum(
+            0.0, np.minimum(detection_x2, track_x2) - np.maximum(detection_x1, track_x1)
+        )
+        intersection_height = np.maximum(
+            0.0, np.minimum(detection_y2, track_y2) - np.maximum(detection_y1, track_y1)
+        )
+        intersection = intersection_width * intersection_height
+        detection_area = detection_boxes[:, None, 2] * detection_boxes[:, None, 3]
+        track_area = track_boxes[None, :, 2] * track_boxes[None, :, 3]
+        union = detection_area + track_area - intersection
+        iou = np.divide(
+            intersection,
+            union,
+            out=np.zeros_like(intersection),
+            where=union > 1e-6,
+        )
+
+        detection_centers = detection_boxes[:, :2] + detection_boxes[:, 2:] * 0.5
+        track_centers = track_boxes[:, :2] + track_boxes[:, 2:] * 0.5
+        center_distance = np.linalg.norm(
+            detection_centers[:, None, :] - track_centers[None, :, :], axis=2
+        )
+        allowed_distance = np.maximum.reduce(
+            [
+                np.broadcast_to(detection_boxes[:, None, 2], iou.shape),
+                np.broadcast_to(detection_boxes[:, None, 3], iou.shape),
+                np.broadcast_to(track_boxes[None, :, 2], iou.shape),
+                np.broadcast_to(track_boxes[None, :, 3], iou.shape),
+            ]
+        ) * self.config.track_center_distance_scale
+        center_score = np.maximum(
+            0.0,
+            1.0
+            - np.divide(
+                center_distance,
+                allowed_distance,
+                out=np.full_like(center_distance, np.inf),
+                where=allowed_distance > 1e-6,
+            ),
+        )
+        scores = np.maximum(iou, center_score * 0.75)
+        scores = np.where(
+            (iou >= self.config.track_iou_threshold) | (center_score > 0.0),
+            scores,
+            0.0,
+        )
+
+        assignments: list[FaceTrack | None] = []
+        matched_indices: set[int] = set()
+        for detection_scores in scores:
+            if matched_indices:
+                detection_scores = detection_scores.copy()
+                detection_scores[list(matched_indices)] = -1.0
+            best_index = int(np.argmax(detection_scores))
+            if detection_scores[best_index] > 0.0:
+                assignments.append(tracks[best_index])
+                matched_indices.add(best_index)
+            else:
+                assignments.append(None)
+        return assignments
 
     def begin_frame(
         self,
-        detections: list[tuple[np.ndarray, float]] | None,
+        detections: list[FaceDetection] | None,
     ) -> list[FaceTrack]:
         """Update boxes and return tracks that need a batched identity check."""
         if detections is None:
@@ -122,14 +197,16 @@ def begin_frame(
         with self._lock:
             self.last_detection_count = len(detections)
             active_tracks = self._active_tracks()
+            matched_tracks = self._match_detections_to_tracks(
+                detections, active_tracks
+            )
             matched_track_ids: set[int] = set()
             updated_tracks: list[FaceTrack] = []
             identity_checks: list[FaceTrack] = []
 
-            for xywh, confidence in detections:
-                track = self._find_matching_track(
-                    xywh, active_tracks, matched_track_ids
-                )
+            for detection, track in zip(detections, matched_tracks):
+                xywh = detection.xywh
+                confidence = detection.confidence
                 needs_identity_check = False
                 if track is None:
                     track = FaceTrack(
@@ -139,21 +216,29 @@ def begin_frame(
                         is_me=False,
                         similarity=-1.0,
                         last_seen_frame=self.frame_index,
-                        last_checked_frame=self.frame_index,
+                        last_checked_frame=-1,
+                        landmarks=detection.landmarks,
                     )
                     self.next_track_id += 1
                     needs_identity_check = self.config.enable_identity_exclusion
                 else:
                     track.xywh = xywh
                     track.confidence = confidence
                     track.last_seen_frame = self.frame_index
+                    track.landmarks = detection.landmarks
+                    recheck_interval = (
+                        self.config.identity_recheck_interval
+                        if track.is_me
+                        else self.config.identity_negative_recheck_interval
+                    )
                     needs_identity_check = (
                         self.config.enable_identity_exclusion
-                        and self.frame_index - track.last_checked_frame
-                        >= self.config.identity_recheck_interval
+                        and (
+                            track.last_checked_frame < 0
+                            or self.frame_index - track.last_checked_frame
+                            >= recheck_interval
+                        )
                     )
-                    if needs_identity_check:
-                        track.last_checked_frame = self.frame_index
 
                 if needs_identity_check:
                     if (
@@ -174,13 +259,21 @@ def begin_frame(
                 if track.track_id not in matched_track_ids
             )
             self.tracks = updated_tracks
-            return identity_checks
+            return sorted(
+                identity_checks,
+                key=lambda track: (
+                    not track.is_me,
+                    track.last_checked_frame >= 0,
+                    -track.confidence,
+                ),
+            )
 
     def apply_identity_matches(self, matches: list[tuple[FaceTrack, float]]) -> None:
         with self._lock:
             for track, similarity in matches:
                 previous_is_me = track.is_me
                 track.similarity = similarity
+                track.last_checked_frame = self.frame_index
                 track.is_me = similarity >= self.config.face_match_cosine_threshold or (
                     previous_is_me
                     and similarity >= self.config.face_match_lost_cosine_threshold