Vector Stores

ragbits.core.vector_stores.base.VectorStoreEntry

Bases: BaseModel

An object representing a vector database entry.

id instance-attribute

id: str

key instance-attribute

key: str

vector instance-attribute

vector: list[float]

metadata instance-attribute

metadata: dict

ragbits.core.vector_stores.base.VectorStoreOptions

Bases: BaseModel, ABC

An object representing the options for the vector store.

k class-attribute instance-attribute

k: int = 5

max_distance class-attribute instance-attribute

max_distance: float | None = None

ragbits.core.vector_stores.base.VectorStore

VectorStore(default_options: VectorStoreOptions | None = None, metadata_store: MetadataStore | None = None)

Bases: ABC

A class with an implementation of Vector Store, allowing to store and retrieve vectors by similarity function.

Constructs a new VectorStore instance.

PARAMETER	DESCRIPTION
default_options	The default options for querying the vector store. TYPE: VectorStoreOptions | None DEFAULT: None
metadata_store	The metadata store to use. TYPE: MetadataStore | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/base.py

def __init__(
    self,
    default_options: VectorStoreOptions | None = None,
    metadata_store: MetadataStore | None = None,
) -> None:
    """
    Constructs a new VectorStore instance.

    Args:
        default_options: The default options for querying the vector store.
        metadata_store: The metadata store to use.
    """
    super().__init__()
    self._default_options = default_options or VectorStoreOptions()
    self._metadata_store = metadata_store

from_config classmethod

from_config(config: dict) -> VectorStore

Creates and returns an instance of the Reranker class from the given configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing the configuration for initializing the Reranker instance. TYPE: dict

RETURNS	DESCRIPTION
VectorStore	An initialized instance of the Reranker class.

RAISES	DESCRIPTION
NotImplementedError	If the class cannot be created from the provided configuration.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/base.py

@classmethod
def from_config(cls, config: dict) -> "VectorStore":
    """
    Creates and returns an instance of the Reranker class from the given configuration.

    Args:
        config: A dictionary containing the configuration for initializing the Reranker instance.

    Returns:
        An initialized instance of the Reranker class.

    Raises:
        NotImplementedError: If the class cannot be created from the provided configuration.
    """
    raise NotImplementedError(f"Cannot create class {cls.__name__} from config.")

store abstractmethod async

store(entries: list[VectorStoreEntry]) -> None

Store entries in the vector store.

PARAMETER	DESCRIPTION
entries	The entries to store. TYPE: list[VectorStoreEntry]

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/base.py

@abstractmethod
async def store(self, entries: list[VectorStoreEntry]) -> None:
    """
    Store entries in the vector store.

    Args:
        entries: The entries to store.
    """

retrieve abstractmethod async

retrieve(vector: list[float], options: VectorStoreOptions | None = None) -> list[VectorStoreEntry]

Retrieve entries from the vector store.

PARAMETER	DESCRIPTION
vector	The vector to search for. TYPE: list[float]
options	The options for querying the vector store. TYPE: VectorStoreOptions | None DEFAULT: None

RETURNS	DESCRIPTION
list[VectorStoreEntry]	The entries.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/base.py

@abstractmethod
async def retrieve(self, vector: list[float], options: VectorStoreOptions | None = None) -> list[VectorStoreEntry]:
    """
    Retrieve entries from the vector store.

    Args:
        vector: The vector to search for.
        options: The options for querying the vector store.

    Returns:
        The entries.
    """

list abstractmethod async

list(where: WhereQuery | None = None, limit: int | None = None, offset: int = 0) -> list[VectorStoreEntry]

List entries from the vector store. The entries can be filtered, limited and offset.

PARAMETER	DESCRIPTION
where	The filter dictionary - the keys are the field names and the values are the values to filter by. Not specifying the key means no filtering. TYPE: WhereQuery | None DEFAULT: None
limit	The maximum number of entries to return. TYPE: int | None DEFAULT: None
offset	The number of entries to skip. TYPE: int DEFAULT: 0

RETURNS	DESCRIPTION
list[VectorStoreEntry]	The entries.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/base.py

@abstractmethod
async def list(
    self, where: WhereQuery | None = None, limit: int | None = None, offset: int = 0
) -> list[VectorStoreEntry]:
    """
    List entries from the vector store. The entries can be filtered, limited and offset.

    Args:
        where: The filter dictionary - the keys are the field names and the values are the values to filter by.
            Not specifying the key means no filtering.
        limit: The maximum number of entries to return.
        offset: The number of entries to skip.

    Returns:
        The entries.
    """

ragbits.core.vector_stores.in_memory.InMemoryVectorStore

InMemoryVectorStore(default_options: VectorStoreOptions | None = None, metadata_store: MetadataStore | None = None)

Bases: VectorStore

A simple in-memory implementation of Vector Store, storing vectors in memory.

Constructs a new InMemoryVectorStore instance.

PARAMETER	DESCRIPTION
default_options	The default options for querying the vector store. TYPE: VectorStoreOptions | None DEFAULT: None
metadata_store	The metadata store to use. TYPE: MetadataStore | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/in_memory.py

def __init__(
    self,
    default_options: VectorStoreOptions | None = None,
    metadata_store: MetadataStore | None = None,
) -> None:
    """
    Constructs a new InMemoryVectorStore instance.

    Args:
        default_options: The default options for querying the vector store.
        metadata_store: The metadata store to use.
    """
    super().__init__(default_options=default_options, metadata_store=metadata_store)
    self._storage: dict[str, VectorStoreEntry] = {}

from_config classmethod

from_config(config: dict) -> InMemoryVectorStore

Creates and returns an instance of the InMemoryVectorStore class from the given configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing the configuration for initializing the InMemoryVectorStore instance. TYPE: dict

RETURNS	DESCRIPTION
InMemoryVectorStore	An initialized instance of the InMemoryVectorStore class.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/in_memory.py

@classmethod
def from_config(cls, config: dict) -> "InMemoryVectorStore":
    """
    Creates and returns an instance of the InMemoryVectorStore class from the given configuration.

    Args:
        config: A dictionary containing the configuration for initializing the InMemoryVectorStore instance.

    Returns:
        An initialized instance of the InMemoryVectorStore class.
    """
    return cls(
        default_options=VectorStoreOptions(**config.get("default_options", {})),
        metadata_store=get_metadata_store(config.get("metadata_store")),
    )

store async

store(entries: list[VectorStoreEntry]) -> None

Store entries in the vector store.

PARAMETER	DESCRIPTION
entries	The entries to store. TYPE: list[VectorStoreEntry]

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/in_memory.py

@traceable
async def store(self, entries: list[VectorStoreEntry]) -> None:
    """
    Store entries in the vector store.

    Args:
        entries: The entries to store.
    """
    for entry in entries:
        self._storage[entry.id] = entry

retrieve async

retrieve(vector: list[float], options: VectorStoreOptions | None = None) -> list[VectorStoreEntry]

Retrieve entries from the vector store.

PARAMETER	DESCRIPTION
vector	The vector to search for. TYPE: list[float]
options	The options for querying the vector store. TYPE: VectorStoreOptions | None DEFAULT: None

RETURNS	DESCRIPTION
list[VectorStoreEntry]	The entries.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/in_memory.py

@traceable
async def retrieve(self, vector: list[float], options: VectorStoreOptions | None = None) -> list[VectorStoreEntry]:
    """
    Retrieve entries from the vector store.

    Args:
        vector: The vector to search for.
        options: The options for querying the vector store.

    Returns:
        The entries.
    """
    options = self._default_options if options is None else options
    entries = sorted(
        (
            (entry, float(np.linalg.norm(np.array(entry.vector) - np.array(vector))))
            for entry in self._storage.values()
        ),
        key=lambda x: x[1],
    )
    return [
        entry
        for entry, distance in entries[: options.k]
        if options.max_distance is None or distance <= options.max_distance
    ]

list async

list(where: WhereQuery | None = None, limit: int | None = None, offset: int = 0) -> list[VectorStoreEntry]

List entries from the vector store. The entries can be filtered, limited and offset.

PARAMETER	DESCRIPTION
where	The filter dictionary - the keys are the field names and the values are the values to filter by. Not specifying the key means no filtering. TYPE: WhereQuery | None DEFAULT: None
limit	The maximum number of entries to return. TYPE: int | None DEFAULT: None
offset	The number of entries to skip. TYPE: int DEFAULT: 0

RETURNS	DESCRIPTION
list[VectorStoreEntry]	The entries.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/in_memory.py

@traceable
async def list(
    self, where: WhereQuery | None = None, limit: int | None = None, offset: int = 0
) -> list[VectorStoreEntry]:
    """
    List entries from the vector store. The entries can be filtered, limited and offset.

    Args:
        where: The filter dictionary - the keys are the field names and the values are the values to filter by.
            Not specifying the key means no filtering.
        limit: The maximum number of entries to return.
        offset: The number of entries to skip.

    Returns:
        The entries.
    """
    entries = iter(self._storage.values())

    if where:
        entries = (
            entry for entry in entries if all(entry.metadata.get(key) == value for key, value in where.items())
        )

    if offset:
        entries = islice(entries, offset, None)

    if limit:
        entries = islice(entries, limit)

    return list(entries)

ragbits.core.vector_stores.chroma.ChromaVectorStore

ChromaVectorStore(client: ClientAPI, index_name: str, distance_method: Literal['l2', 'ip', 'cosine'] = 'cosine', default_options: VectorStoreOptions | None = None, metadata_store: MetadataStore | None = None)

Bases: VectorStore

Vector store implementation using Chroma.

Constructs a new ChromaVectorStore instance.

PARAMETER	DESCRIPTION
client	The ChromaDB client. TYPE: ClientAPI
index_name	The name of the index. TYPE: str
distance_method	The distance method to use. TYPE: Literal['l2', 'ip', 'cosine'] DEFAULT: 'cosine'
default_options	The default options for querying the vector store. TYPE: VectorStoreOptions | None DEFAULT: None
metadata_store	The metadata store to use. If None, the metadata will be stored in ChromaDB. TYPE: MetadataStore | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/chroma.py

def __init__(
    self,
    client: ClientAPI,
    index_name: str,
    distance_method: Literal["l2", "ip", "cosine"] = "cosine",
    default_options: VectorStoreOptions | None = None,
    metadata_store: MetadataStore | None = None,
) -> None:
    """
    Constructs a new ChromaVectorStore instance.

    Args:
        client: The ChromaDB client.
        index_name: The name of the index.
        distance_method: The distance method to use.
        default_options: The default options for querying the vector store.
        metadata_store: The metadata store to use. If None, the metadata will be stored in ChromaDB.
    """
    super().__init__(default_options=default_options, metadata_store=metadata_store)
    self._client = client
    self._index_name = index_name
    self._distance_method = distance_method
    self._collection = self._client.get_or_create_collection(
        name=self._index_name,
        metadata={"hnsw:space": self._distance_method},
    )

from_config classmethod

from_config(config: dict) -> ChromaVectorStore

Creates and returns an instance of the ChromaVectorStore class from the given configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing the configuration for initializing the ChromaVectorStore instance. TYPE: dict

RETURNS	DESCRIPTION
ChromaVectorStore	An initialized instance of the ChromaVectorStore class.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/chroma.py

@classmethod
def from_config(cls, config: dict) -> "ChromaVectorStore":
    """
    Creates and returns an instance of the ChromaVectorStore class from the given configuration.

    Args:
        config: A dictionary containing the configuration for initializing the ChromaVectorStore instance.

    Returns:
        An initialized instance of the ChromaVectorStore class.
    """
    client_cls = get_cls_from_config(config["client"]["type"], chromadb)
    return cls(
        client=client_cls(**config["client"].get("config", {})),
        index_name=config["index_name"],
        distance_method=config.get("distance_method", "cosine"),
        default_options=VectorStoreOptions(**config.get("default_options", {})),
        metadata_store=get_metadata_store(config.get("metadata_store")),
    )

store async

store(entries: list[VectorStoreEntry]) -> None

Stores entries in the ChromaDB collection.

PARAMETER	DESCRIPTION
entries	The entries to store. TYPE: list[VectorStoreEntry]

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/chroma.py

@traceable
async def store(self, entries: list[VectorStoreEntry]) -> None:
    """
    Stores entries in the ChromaDB collection.

    Args:
        entries: The entries to store.
    """
    if not entries:
        return

    ids = [entry.id for entry in entries]
    documents = [entry.key for entry in entries]
    embeddings = [entry.vector for entry in entries]
    metadatas = [entry.metadata for entry in entries]

    # Flatten metadata
    flattened_metadatas = [flatten_dict(metadata) for metadata in metadatas]

    metadatas = (
        flattened_metadatas
        if self._metadata_store is None
        else await self._metadata_store.store(ids, flattened_metadatas)  # type: ignore
    )

    self._collection.add(ids=ids, embeddings=embeddings, metadatas=metadatas, documents=documents)  # type: ignore

retrieve async

retrieve(vector: list[float], options: VectorStoreOptions | None = None) -> list[VectorStoreEntry]

Retrieves entries from the ChromaDB collection.

PARAMETER	DESCRIPTION
vector	The vector to query. TYPE: list[float]
options	The options for querying the vector store. TYPE: VectorStoreOptions | None DEFAULT: None

RETURNS	DESCRIPTION
list[VectorStoreEntry]	The retrieved entries.

RAISES	DESCRIPTION
MetadataNotFoundError	If the metadata is not found.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/chroma.py

@traceable
async def retrieve(self, vector: list[float], options: VectorStoreOptions | None = None) -> list[VectorStoreEntry]:
    """
    Retrieves entries from the ChromaDB collection.

    Args:
        vector: The vector to query.
        options: The options for querying the vector store.

    Returns:
        The retrieved entries.

    Raises:
        MetadataNotFoundError: If the metadata is not found.
    """
    options = self._default_options if options is None else options

    results = self._collection.query(
        query_embeddings=vector,
        n_results=options.k,
        include=["metadatas", "embeddings", "distances", "documents"],
    )

    ids = results.get("ids") or []
    embeddings = results.get("embeddings") or []
    distances = results.get("distances") or []
    documents = results.get("documents") or []
    metadatas = [
        [metadata for batch in results.get("metadatas", []) for metadata in batch]  # type: ignore
        if self._metadata_store is None
        else await self._metadata_store.get(*ids)
    ]

    return [
        VectorStoreEntry(
            id=id,
            key=document,
            vector=list(embeddings),
            metadata=unflatten_dict(metadata) if metadata else {},  # type: ignore
        )
        for batch in zip(ids, metadatas, embeddings, distances, documents, strict=True)
        for id, metadata, embeddings, distance, document in zip(*batch, strict=True)
        if options.max_distance is None or distance <= options.max_distance
    ]

list async

list(where: WhereQuery | None = None, limit: int | None = None, offset: int = 0) -> list[VectorStoreEntry]

List entries from the vector store. The entries can be filtered, limited and offset.

PARAMETER	DESCRIPTION
where	The filter dictionary - the keys are the field names and the values are the values to filter by. Not specifying the key means no filtering. TYPE: WhereQuery | None DEFAULT: None
limit	The maximum number of entries to return. TYPE: int | None DEFAULT: None
offset	The number of entries to skip. TYPE: int DEFAULT: 0

RETURNS	DESCRIPTION
list[VectorStoreEntry]	The entries.

RAISES	DESCRIPTION
MetadataNotFoundError	If the metadata is not found.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/chroma.py

@traceable
async def list(
    self, where: WhereQuery | None = None, limit: int | None = None, offset: int = 0
) -> list[VectorStoreEntry]:
    """
    List entries from the vector store. The entries can be filtered, limited and offset.

    Args:
        where: The filter dictionary - the keys are the field names and the values are the values to filter by.
            Not specifying the key means no filtering.
        limit: The maximum number of entries to return.
        offset: The number of entries to skip.

    Returns:
        The entries.

    Raises:
        MetadataNotFoundError: If the metadata is not found.
    """
    # Cast `where` to chromadb's Where type
    where_chroma: chromadb.Where | None = dict(where) if where else None

    results = self._collection.get(
        where=where_chroma,
        limit=limit,
        offset=offset,
        include=["metadatas", "embeddings", "documents"],
    )

    ids = results.get("ids") or []
    embeddings = results.get("embeddings") or []
    documents = results.get("documents") or []
    metadatas = (
        results.get("metadatas") or [] if self._metadata_store is None else await self._metadata_store.get(ids)
    )

    return [
        VectorStoreEntry(
            id=id,
            key=document,
            vector=list(embedding),
            metadata=unflatten_dict(metadata) if metadata else {},  # type: ignore
        )
        for id, metadata, embedding, document in zip(ids, metadatas, embeddings, documents, strict=True)
    ]

ragbits.core.vector_stores.qdrant.QdrantVectorStore

QdrantVectorStore(client: AsyncQdrantClient, index_name: str, distance_method: Distance = Distance.COSINE, default_options: VectorStoreOptions | None = None, metadata_store: MetadataStore | None = None)

Bases: VectorStore

Vector store implementation using Qdrant.

Constructs a new QdrantVectorStore instance.

PARAMETER	DESCRIPTION
client	An instance of the Qdrant client. TYPE: AsyncQdrantClient
index_name	The name of the index. TYPE: str
distance_method	The distance metric to use when creating the collection. TYPE: Distance DEFAULT: COSINE
default_options	The default options for querying the vector store. TYPE: VectorStoreOptions | None DEFAULT: None
metadata_store	The metadata store to use. If None, the metadata will be stored in Qdrant. TYPE: MetadataStore | None DEFAULT: None

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/qdrant.py

def __init__(
    self,
    client: AsyncQdrantClient,
    index_name: str,
    distance_method: Distance = Distance.COSINE,
    default_options: VectorStoreOptions | None = None,
    metadata_store: MetadataStore | None = None,
) -> None:
    """
    Constructs a new QdrantVectorStore instance.

    Args:
        client: An instance of the Qdrant client.
        index_name: The name of the index.
        distance_method: The distance metric to use when creating the collection.
        default_options: The default options for querying the vector store.
        metadata_store: The metadata store to use. If None, the metadata will be stored in Qdrant.
    """
    super().__init__(default_options=default_options, metadata_store=metadata_store)
    self._client = client
    self._index_name = index_name
    self._distance_method = distance_method

from_config classmethod

from_config(config: dict) -> QdrantVectorStore

Creates and returns an instance of the QdrantVectorStore class from the given configuration.

PARAMETER	DESCRIPTION
config	A dictionary containing the configuration for initializing the QdrantVectorStore instance. TYPE: dict

RETURNS	DESCRIPTION
QdrantVectorStore	An initialized instance of the QdrantVectorStore class.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/qdrant.py

@classmethod
def from_config(cls, config: dict) -> "QdrantVectorStore":
    """
    Creates and returns an instance of the QdrantVectorStore class from the given configuration.

    Args:
        config: A dictionary containing the configuration for initializing the QdrantVectorStore instance.

    Returns:
        An initialized instance of the QdrantVectorStore class.
    """
    client_cls = get_cls_from_config(config["client"]["type"], qdrant_client)
    return cls(
        client=client_cls(**config["client"].get("config", {})),
        index_name=config["index_name"],
        distance_method=config.get("distance_method", Distance.COSINE),
        default_options=VectorStoreOptions(**config.get("default_options", {})),
        metadata_store=get_metadata_store(config.get("metadata_store")),
    )

store async

store(entries: list[VectorStoreEntry]) -> None

Stores vector entries in the Qdrant collection.

PARAMETER	DESCRIPTION
entries	List of VectorStoreEntry objects to store TYPE: list[VectorStoreEntry]

RAISES	DESCRIPTION
QdrantException	If upload to collection fails.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/qdrant.py

@traceable
async def store(self, entries: list[VectorStoreEntry]) -> None:
    """
    Stores vector entries in the Qdrant collection.

    Args:
        entries: List of VectorStoreEntry objects to store

    Raises:
        QdrantException: If upload to collection fails.
    """
    if not entries:
        return

    if not await self._client.collection_exists(self._index_name):
        await self._client.create_collection(
            collection_name=self._index_name,
            vectors_config=VectorParams(size=len(entries[0].vector), distance=self._distance_method),
        )

    ids = [entry.id for entry in entries]
    embeddings = [entry.vector for entry in entries]
    payloads = [{"document": entry.key} for entry in entries]
    metadatas = [entry.metadata for entry in entries]

    metadatas = (
        [{"metadata": json.dumps(metadata, default=str)} for metadata in metadatas]
        if self._metadata_store is None
        else await self._metadata_store.store(ids, metadatas)  # type: ignore
    )
    if metadatas is not None:
        payloads = [{**payload, **metadata} for (payload, metadata) in zip(payloads, metadatas, strict=True)]

    self._client.upload_collection(
        collection_name=self._index_name,
        vectors=embeddings,
        payload=payloads,
        ids=ids,
        wait=True,
    )

retrieve async

retrieve(vector: list[float], options: VectorStoreOptions | None = None) -> list[VectorStoreEntry]

Retrieves entries from the Qdrant collection based on vector similarity.

PARAMETER	DESCRIPTION
vector	The vector to query. TYPE: list[float]
options	The options for querying the vector store. TYPE: VectorStoreOptions | None DEFAULT: None

RETURNS	DESCRIPTION
list[VectorStoreEntry]	The retrieved entries.

RAISES	DESCRIPTION
MetadataNotFoundError	If metadata cannot be retrieved

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/qdrant.py

@traceable
async def retrieve(self, vector: list[float], options: VectorStoreOptions | None = None) -> list[VectorStoreEntry]:
    """
    Retrieves entries from the Qdrant collection based on vector similarity.

    Args:
        vector: The vector to query.
        options: The options for querying the vector store.

    Returns:
        The retrieved entries.

    Raises:
        MetadataNotFoundError: If metadata cannot be retrieved
    """
    options = options or self._default_options
    score_threshold = 1 - options.max_distance if options.max_distance else None

    results = await self._client.query_points(
        collection_name=self._index_name,
        query=vector,
        limit=options.k,
        score_threshold=score_threshold,
        with_payload=True,
        with_vectors=True,
    )

    ids = [point.id for point in results.points]
    vectors = [point.vector for point in results.points]
    documents = [point.payload["document"] for point in results.points]  # type: ignore
    metadatas = (
        [json.loads(point.payload["metadata"]) for point in results.points]  # type: ignore
        if self._metadata_store is None
        else await self._metadata_store.get(ids)  # type: ignore
    )

    return [
        VectorStoreEntry(
            id=str(id),
            key=document,
            vector=vector,  # type: ignore
            metadata=metadata,
        )
        for id, document, vector, metadata in zip(ids, documents, vectors, metadatas, strict=True)
    ]

list async

list(where: Filter | None = None, limit: int | None = None, offset: int = 0) -> list[VectorStoreEntry]

List entries from the vector store. The entries can be filtered, limited and offset.

PARAMETER	DESCRIPTION
where	Conditions for filtering results. TYPE: Filter | None DEFAULT: None
Reference	https://qdrant.tech/documentation/concepts/filtering
limit	The maximum number of entries to return. TYPE: int | None DEFAULT: None
offset	The number of entries to skip. TYPE: int DEFAULT: 0

RETURNS	DESCRIPTION
list[VectorStoreEntry]	The entries.

RAISES	DESCRIPTION
MetadataNotFoundError	If the metadata is not found.

Source code in packages/ragbits-core/src/ragbits/core/vector_stores/qdrant.py

@traceable
async def list(  # type: ignore
    self,
    where: Filter | None = None,  # type: ignore
    limit: int | None = None,
    offset: int = 0,
) -> list[VectorStoreEntry]:
    """
    List entries from the vector store. The entries can be filtered, limited and offset.

    Args:
        where: Conditions for filtering results.
        Reference: https://qdrant.tech/documentation/concepts/filtering
        limit: The maximum number of entries to return.
        offset: The number of entries to skip.

    Returns:
        The entries.

    Raises:
        MetadataNotFoundError: If the metadata is not found.
    """
    results = await self._client.query_points(
        collection_name=self._index_name,
        query_filter=where,
        limit=limit or 10,
        offset=offset,
        with_payload=True,
        with_vectors=True,
    )

    ids = [point.id for point in results.points]
    vectors = [point.vector for point in results.points]
    documents = [point.payload["document"] for point in results.points]  # type: ignore
    metadatas = (
        [json.loads(point.payload["metadata"]) for point in results.points]  # type: ignore
        if self._metadata_store is None
        else await self._metadata_store.get(ids)  # type: ignore
    )

    return [
        VectorStoreEntry(
            id=str(id),
            key=document,
            vector=vector,  # type: ignore
            metadata=metadata,
        )
        for id, document, vector, metadata in zip(ids, documents, vectors, metadatas, strict=True)
    ]