File size: 3,965 Bytes
f5eeb1c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """
FaceIndexProvider — reverse face search via the indexed FaceIndex.
This provider integrates the FaceIndex (sqlite-vec + ArcFace embeddings)
with the orchestrator's provider pipeline. When the user submits a
search request, the orchestrator calls this provider's execute() with
a PipelineOutput; the provider extracts the query embedding from the
largest detected face and searches the FaceIndex for matches.
Note: For direct API usage, prefer the /faces/enroll and /faces/search
endpoints which bypass the orchestrator and call FaceIndexService
directly. This provider exists so the orchestrator can include
reverse-face-search in full-pipeline /jobs runs.
"""
from __future__ import annotations
from typing import Any
from config.settings import Settings, settings as _default_settings
from models.providers import ProviderCapability
from pipeline.feature_extraction import PipelineOutput
from providers.base import BaseProvider, ProviderResult
class FaceIndexProvider(BaseProvider):
"""Reverse face search via the indexed FaceIndex."""
name = "face_index"
capability = ProviderCapability.REVERSE_FACE_SEARCH
def __init__(self, settings: Settings | None = None) -> None:
super().__init__(settings=settings or _default_settings)
# The FaceIndex instance is injected at runtime by the orchestrator
# via the pipeline_output's metadata, OR looked up lazily.
# For now, we mark availability based on the enable flag.
self._available = bool(getattr(self._settings, "enable_face_index", True))
def is_available(self) -> bool:
return self._available
def _run(self, pipeline_output: PipelineOutput) -> tuple[dict, dict]:
"""
Search the FaceIndex for matches to the largest face in the image.
The FaceIndex instance is NOT available here directly (providers are
isolated). Instead, this provider reads the embedding from the
pipeline_output (set by InsightFaceProvider if it ran first) and
returns the embedding for the orchestrator to use.
For direct indexed search, use the /faces/search API endpoint instead.
"""
if not pipeline_output.face_crops:
raw = {"num_query_faces": 0, "matches": [], "note": "No faces detected in query image"}
normalized = {"num_query_faces": 0, "matches": []}
return raw, normalized
# Use the largest face
largest = max(
pipeline_output.face_crops,
key=lambda c: getattr(c, "width", 0) * getattr(c, "height", 1),
)
embedding = getattr(largest, "embedding", None)
if embedding is None:
raw = {
"num_query_faces": len(pipeline_output.face_crops),
"matches": [],
"note": "No embedding available — InsightFace provider did not run or failed. "
"Use the /faces/search API endpoint for direct indexed search.",
}
normalized = {
"num_query_faces": len(pipeline_output.face_crops),
"matches": [],
}
return raw, normalized
# The actual index search is performed by the FaceIndexService
# (called from the API layer). This provider just signals that
# the query embedding is ready.
import numpy as np
if not isinstance(embedding, np.ndarray):
embedding = np.array(embedding, dtype=np.float32)
raw = {
"num_query_faces": len(pipeline_output.face_crops),
"query_embedding_dim": int(embedding.shape[0]),
"matches": [],
"note": "Query embedding computed. Use /faces/search to search the index.",
}
normalized = {
"num_query_faces": len(pipeline_output.face_crops),
"query_embedding": embedding.tolist(),
"matches": [],
}
return raw, normalized
|