Skip to content

Faiss Search

Exact Faiss flat-index adapters.

FaissFlatL2Searcher

Bases: BaseExactSearcher

Exact Faiss IndexFlatL2 search with negative squared-distance scores.

Source code in src/vector_search_study/faiss_search.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class FaissFlatL2Searcher(BaseExactSearcher):
    """Exact Faiss IndexFlatL2 search with negative squared-distance scores."""

    def __init__(self, corpus: FloatMatrix) -> None:
        """Build a float32 flat L2 index outside search timing."""
        _require_float32(corpus)
        super().__init__(corpus, objective=SearchObjective.SQUARED_L2)
        faiss = import_optional("faiss", extra="benchmark-backends")
        faiss.omp_set_num_threads(1)
        self._index: Any = faiss.IndexFlatL2(self.dimension)
        self._index.add(self._corpus)

    def _search_prepared(self, queries: PreparedQueries, k: int) -> SearchResult:
        """Search the flat L2 index and invert its distances."""
        distances, indices = self._index.search(queries.values, k)
        return canonical_result(indices, -distances)

__init__

__init__(corpus: FloatMatrix) -> None

Build a float32 flat L2 index outside search timing.

Source code in src/vector_search_study/faiss_search.py
19
20
21
22
23
24
25
26
def __init__(self, corpus: FloatMatrix) -> None:
    """Build a float32 flat L2 index outside search timing."""
    _require_float32(corpus)
    super().__init__(corpus, objective=SearchObjective.SQUARED_L2)
    faiss = import_optional("faiss", extra="benchmark-backends")
    faiss.omp_set_num_threads(1)
    self._index: Any = faiss.IndexFlatL2(self.dimension)
    self._index.add(self._corpus)

FaissFlatIPSearcher

Bases: BaseExactSearcher

Exact Faiss IndexFlatIP search for inner product or cosine.

Source code in src/vector_search_study/faiss_search.py
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class FaissFlatIPSearcher(BaseExactSearcher):
    """Exact Faiss IndexFlatIP search for inner product or cosine."""

    def __init__(
        self,
        corpus: FloatMatrix,
        *,
        objective: SearchObjective | str = SearchObjective.INNER_PRODUCT,
    ) -> None:
        """Build a float32 flat inner-product index outside search timing."""
        resolved = resolve_search_objective(objective)
        if resolved not in {SearchObjective.INNER_PRODUCT, SearchObjective.NORMALIZED_COSINE}:
            raise UnsupportedObjectiveError("Faiss IndexFlatIP supports only inner product and normalized cosine")
        _require_float32(corpus)
        super().__init__(corpus, objective=resolved)
        faiss = import_optional("faiss", extra="benchmark-backends")
        faiss.omp_set_num_threads(1)
        self._index: Any = faiss.IndexFlatIP(self.dimension)
        self._index.add(self._corpus)

    def _search_prepared(self, queries: PreparedQueries, k: int) -> SearchResult:
        """Search the flat inner-product index."""
        scores, indices = self._index.search(queries.values, k)
        return canonical_result(indices, scores)

__init__

__init__(
    corpus: FloatMatrix,
    *,
    objective: SearchObjective
    | str = SearchObjective.INNER_PRODUCT,
) -> None

Build a float32 flat inner-product index outside search timing.

Source code in src/vector_search_study/faiss_search.py
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    corpus: FloatMatrix,
    *,
    objective: SearchObjective | str = SearchObjective.INNER_PRODUCT,
) -> None:
    """Build a float32 flat inner-product index outside search timing."""
    resolved = resolve_search_objective(objective)
    if resolved not in {SearchObjective.INNER_PRODUCT, SearchObjective.NORMALIZED_COSINE}:
        raise UnsupportedObjectiveError("Faiss IndexFlatIP supports only inner product and normalized cosine")
    _require_float32(corpus)
    super().__init__(corpus, objective=resolved)
    faiss = import_optional("faiss", extra="benchmark-backends")
    faiss.omp_set_num_threads(1)
    self._index: Any = faiss.IndexFlatIP(self.dimension)
    self._index.add(self._corpus)