Module vexpresso.retrievers.base#
View Source
import abc
from dataclasses import dataclass
from typing import Any, Iterable, List, Optional, Union
import numpy as np
@dataclass
class RetrievalOutput:
embeddings: Any
indices: Union[np.ndarray, Iterable[int]]
scores: Union[np.ndarray, Iterable[float]]
query_embeddings: Optional[Any] = None
class BaseRetriever(metaclass=abc.ABCMeta):
SUPPORTED_TYPES = [np.dtype]
@abc.abstractmethod
def retrieve(
self, query_embeddings: np.ndarray, embeddings: List[Any], *args, **kwargs
) -> Union[List[RetrievalOutput], RetrievalOutput]:
"""
Queries embeddings with query embedding vector and returns nearest embeddings and their corresponding ids
Args:
query_embeddings (np.ndarray): query, used to find nearest embeddings in set.
embeddings (List[Any]): embeddings set, query is compared to this.
Returns:
Union[List[QueryOutput], QueryOutput]: dataclasses containing returned embeddings and corresponding indices.
When this has more than one entry, that means that the call was batched
"""
Classes#
BaseRetriever#
class BaseRetriever(
/,
*args,
**kwargs
)
View Source
class BaseRetriever(metaclass=abc.ABCMeta):
SUPPORTED_TYPES = [np.dtype]
@abc.abstractmethod
def retrieve(
self, query_embeddings: np.ndarray, embeddings: List[Any], *args, **kwargs
) -> Union[List[RetrievalOutput], RetrievalOutput]:
"""
Queries embeddings with query embedding vector and returns nearest embeddings and their corresponding ids
Args:
query_embeddings (np.ndarray): query, used to find nearest embeddings in set.
embeddings (List[Any]): embeddings set, query is compared to this.
Returns:
Union[List[QueryOutput], QueryOutput]: dataclasses containing returned embeddings and corresponding indices.
When this has more than one entry, that means that the call was batched
"""
Descendants#
- vexpresso.retrievers.faiss.FaissRetriever
- vexpresso.retrievers.np.Retriever
Class variables#
SUPPORTED_TYPES
Methods#
retrieve#
def retrieve(
self,
query_embeddings: numpy.ndarray,
embeddings: List[Any],
*args,
**kwargs
) -> Union[List[vexpresso.retrievers.base.RetrievalOutput], vexpresso.retrievers.base.RetrievalOutput]
Queries embeddings with query embedding vector and returns nearest embeddings and their corresponding ids
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
| query_embeddings | np.ndarray | query, used to find nearest embeddings in set. | None |
| embeddings | List[Any] | embeddings set, query is compared to this. | None |
Returns:
| Type | Description |
|---|---|
| Union[List[QueryOutput], QueryOutput] | dataclasses containing returned embeddings and corresponding indices. When this has more than one entry, that means that the call was batched |
View Source
@abc.abstractmethod
def retrieve(
self, query_embeddings: np.ndarray, embeddings: List[Any], *args, **kwargs
) -> Union[List[RetrievalOutput], RetrievalOutput]:
"""
Queries embeddings with query embedding vector and returns nearest embeddings and their corresponding ids
Args:
query_embeddings (np.ndarray): query, used to find nearest embeddings in set.
embeddings (List[Any]): embeddings set, query is compared to this.
Returns:
Union[List[QueryOutput], QueryOutput]: dataclasses containing returned embeddings and corresponding indices.
When this has more than one entry, that means that the call was batched
"""
RetrievalOutput#
class RetrievalOutput(
embeddings: Any,
indices: Union[numpy.ndarray, Iterable[int]],
scores: Union[numpy.ndarray, Iterable[float]],
query_embeddings: Optional[Any] = None
)
RetrievalOutput(embeddings: Any, indices: Union[numpy.ndarray, Iterable[int]], scores: Union[numpy.ndarray, Iterable[float]], query_embeddings: Optional[Any] = None)
View Source
@dataclass
class RetrievalOutput:
embeddings: Any
indices: Union[np.ndarray, Iterable[int]]
scores: Union[np.ndarray, Iterable[float]]
query_embeddings: Optional[Any] = None
Class variables#
query_embeddings