Python Search

vector_search_study.python_search

Scalar pure-Python exact search implementations.

PythonSortSearcher

Bases: BaseExactSearcher

Exhaustively score into a Python list and fully sort it.

Source code in src/vector_search_study/python_search.py

class PythonSortSearcher(BaseExactSearcher):
    """Exhaustively score into a Python list and fully sort it."""

    def __init__(
        self,
        corpus: FloatMatrix,
        *,
        objective: SearchObjective | str = SearchObjective.NORMALIZED_COSINE,
    ) -> None:
        """Build the scalar corpus representation outside search timing."""
        super().__init__(corpus, objective=objective)
        self._rows = tuple(tuple(float(value) for value in row) for row in self._corpus)

    def _search_prepared(self, queries: PreparedQueries, k: int) -> SearchResult:
        """Score every vector, then sort all candidates."""
        all_indices: list[list[int]] = []
        all_scores: list[list[float]] = []
        for query_array in queries.values:
            query = tuple(float(value) for value in query_array)
            ranked: list[tuple[float, int]] = []
            for index, vector in enumerate(self._rows):
                score = _scalar_score(vector, query, self.objective)
                ranked.append((score, index))
            ranked.sort(key=lambda item: (-item[0], item[1]))
            selected = ranked[:k]
            all_indices.append([index for _, index in selected])
            all_scores.append([score for score, _ in selected])
        return SearchResult(
            indices=np.asarray(all_indices, dtype=np.int64, order="C"),
            scores=np.asarray(all_scores, dtype=np.float64, order="C"),
        )

__init__

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

Build the scalar corpus representation outside search timing.

Source code in src/vector_search_study/python_search.py

def __init__(
    self,
    corpus: FloatMatrix,
    *,
    objective: SearchObjective | str = SearchObjective.NORMALIZED_COSINE,
) -> None:
    """Build the scalar corpus representation outside search timing."""
    super().__init__(corpus, objective=objective)
    self._rows = tuple(tuple(float(value) for value in row) for row in self._corpus)

PythonHeapSearcher

Bases: BaseExactSearcher

Stream exhaustive scalar scores through a bounded size-k heap.

Source code in src/vector_search_study/python_search.py

class PythonHeapSearcher(BaseExactSearcher):
    """Stream exhaustive scalar scores through a bounded size-k heap."""

    def __init__(
        self,
        corpus: FloatMatrix,
        *,
        objective: SearchObjective | str = SearchObjective.NORMALIZED_COSINE,
    ) -> None:
        """Build the scalar corpus representation outside search timing."""
        super().__init__(corpus, objective=objective)
        self._rows = tuple(tuple(float(value) for value in row) for row in self._corpus)

    def _search_prepared(self, queries: PreparedQueries, k: int) -> SearchResult:
        """Retain only the best k candidates while streaming scores."""
        all_indices: list[list[int]] = []
        all_scores: list[list[float]] = []
        for query_array in queries.values:
            query = tuple(float(value) for value in query_array)
            heap: list[tuple[float, int, int]] = []
            for index, vector in enumerate(self._rows):
                score = _scalar_score(vector, query, self.objective)
                item = (score, -index, index)
                if len(heap) < k:
                    heapq.heappush(heap, item)
                elif item[:2] > heap[0][:2]:
                    _ = heapq.heapreplace(heap, item)
            selected = sorted(heap, key=lambda item: (-item[0], item[2]))
            all_indices.append([index for _, _, index in selected])
            all_scores.append([score for score, _, _ in selected])
        return SearchResult(
            indices=np.asarray(all_indices, dtype=np.int64, order="C"),
            scores=np.asarray(all_scores, dtype=np.float64, order="C"),
        )

__init__

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

Build the scalar corpus representation outside search timing.

Source code in src/vector_search_study/python_search.py

def __init__(
    self,
    corpus: FloatMatrix,
    *,
    objective: SearchObjective | str = SearchObjective.NORMALIZED_COSINE,
) -> None:
    """Build the scalar corpus representation outside search timing."""
    super().__init__(corpus, objective=objective)
    self._rows = tuple(tuple(float(value) for value in row) for row in self._corpus)