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

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

Идентификация лица — поиск человека в базе без предварительного указания кандидата. Система получает фото лица и возвращает: кто это из зарегистрированных в базе или «незнакомец». Это технически сложнее верификации: нужно масштабируемое хранилище, быстрый ANN-поиск, корректное управление порогами при растущей базе.

Архитектурное решение для scale

При базе > 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 (cosine через нормализацию)
        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)

Closed-set vs Open-set идентификация

Closed-set: все запросы принадлежат одному из зарегистрированных людей. Задача сводится к ранжированию.

Open-set: система должна отвергать «незнакомцев» — людей не из базы. Требует настройки порога отвержения. При растущей базе порог может нуждаться в корректировке: с 1000 до 100k людей вероятность случайного совпадения растёт.

Обновление базы в реальном времени

Добавление новых пользователей без переиндексации: index.add_with_ids() работает инкрементально. Удаление: IndexIDMap.remove_ids(). Персистентность через faiss.write_index().

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

• CMC (Cumulative Match Characteristic): Rank-1, Rank-5, Rank-10 accuracy
• DIR@FAR (Detection and Identification Rate): для open-set
• Latency p95/p99 при пиковой нагрузке
• QPS (queries per second) — для планирования инфраструктуры