Розробка системи ідентифікації обличчя (Face Identification, 1:N)

Розробка систем ідентифікації облич (1:N)

Ідентифікація облич — пошук людини в базі без попередньоготаким указання кандидата. Система отримує фото обличчя та повертає: хто це зі зареєстрованих людей або "невідомо". Технічно складніше за верифікацію: потрібно масштабоване сховище, швидкий ANN-пошук, правильне керування порогами в міру зростання бази.

Архітектурне рішення для масштабування

При базі > 100k облич грубий силовий пошук стає занадто повільним. Ієрархія підходів:

import faiss
import numpy as np
from dataclasses import dataclass

@dataclass
class IdentificationResult:
    person_id: str | None
    person_name: str | None
    similarity: float
    identified: bool

class FaceIdentificationSystem:
    def __init__(self, embedding_dim: int = 512,
                 n_lists: int = 100,  # параметр IVF
                 threshold: float = 0.45):
        self.dim = embedding_dim
        self.threshold = threshold

        # FAISS IVFFlat індекс з Inner Product (косинус через нормалізацію)
        quantizer = faiss.IndexFlatIP(embedding_dim)
        self.index = faiss.IndexIDMap(
            faiss.IndexIVFFlat(quantizer, embedding_dim, n_lists,
                               faiss.METRIC_INNER_PRODUCT)
        )
        self.index.nprobe = 20  # більше = точніше але повільніше

        self.id_map = {}  # faiss_int_id -> {'person_id': str, 'name': str}
        self._next_id = 0

    def register(self, person_id: str, name: str,
                 embeddings: np.ndarray) -> int:
        """Реєстрація декількох фото однієї людини"""
        faiss.normalize_L2(embeddings)
        ids = np.arange(self._next_id, self._next_id + len(embeddings))

        if not self.index.is_trained:
            # Потрібен мінімальний набір даних для навчання IVF
            self.index.train(embeddings)

        self.index.add_with_ids(embeddings, ids)

        for fid in ids:
            self.id_map[int(fid)] = {'person_id': person_id, 'name': name}

        self._next_id += len(embeddings)
        return len(embeddings)

    def identify(self, query_embedding: np.ndarray,
                 k: int = 5) -> IdentificationResult:
        query = query_embedding.reshape(1, -1).copy()
        faiss.normalize_L2(query)

        similarities, faiss_ids = self.index.search(query, k)

        best_sim = float(similarities[0][0])
        best_id = int(faiss_ids[0][0])

        if best_id == -1 or best_sim < self.threshold:
            return IdentificationResult(None, None, best_sim, False)

        person_info = self.id_map[best_id]
        return IdentificationResult(
            person_info['person_id'],
            person_info['name'],
            best_sim,
            True
        )

Декілька зображень на людину

Реєстрація декількох фото під різними кутами та умовами освітлення підвищує recall. При ідентифікації з агрегацією за всіма фото людини:

def identify_with_aggregation(self, query_emb: np.ndarray,
                               k: int = 10) -> IdentificationResult:
    """k кандидатів → агрегація голосів за person_id"""
    query = query_emb.reshape(1, -1).copy()
    faiss.normalize_L2(query)
    similarities, faiss_ids = self.index.search(query, k)

    votes = {}
    for sim, fid in zip(similarities[0], faiss_ids[0]):
        if fid == -1:
            continue
        pid = self.id_map[int(fid)]['person_id']
        votes[pid] = votes.get(pid, 0) + float(sim)

    if not votes:
        return IdentificationResult(None, None, 0.0, False)

    best_pid = max(votes, key=votes.get)
    best_score = votes[best_pid] / k  # normalize

    if best_score < self.threshold:
        return IdentificationResult(None, None, best_score, False)

    name = self.id_map[next(
        fid for fid, info in self.id_map.items()
        if info['person_id'] == best_pid
    )]['name']
    return IdentificationResult(best_pid, name, best_score, True)

Закрита множина vs відкрита множина ідентифікація

Закрита множина: всі запити належать одній із зареєстрованих людей. Завдання зводиться до ранжування.

Відкрита множина: система повинна відхилити "невідомих" — людей не з бази. Вимагає налаштування порога відхилення. В міру зростання бази поріг може потребувати коригування: від 1000 до 100k людей ймовірність випадкового збігу зростає.

Оновлення бази в реальному часі

Додавання нових користувачів без переіндексації: index.add_with_ids() працює інкрементально. Видалення: IndexIDMap.remove_ids(). Персистентність через faiss.write_index().

Метрики production-системи

• CMC (Cumulative Match Characteristic): Rank-1, Rank-5, Rank-10 точність
• DIR@FAR (Detection and Identification Rate): для відкритої множини
• Latency p95/p99 під піковим навантаженням
• QPS (queries per second) — для планування інфраструктури