Source code for langchain.vectorstores.bageldb
from __future__ import annotations
import uuid
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Dict,
    Iterable,
    List,
    Optional,
    Tuple,
    Type,
)
if TYPE_CHECKING:
    import bagel
    import bagel.config
    from bagel.api.types import ID, OneOrMany, Where, WhereDocument
from langchain.docstore.document import Document
from langchain.schema.embeddings import Embeddings
from langchain.schema.vectorstore import VectorStore
from langchain.utils import xor_args
DEFAULT_K = 5
def _results_to_docs(results: Any) -> List[Document]:
    return [doc for doc, _ in _results_to_docs_and_scores(results)]
def _results_to_docs_and_scores(results: Any) -> List[Tuple[Document, float]]:
    return [
        (Document(page_content=result[0], metadata=result[1] or {}), result[2])
        for result in zip(
            results["documents"][0],
            results["metadatas"][0],
            results["distances"][0],
        )
    ]
[docs]class Bagel(VectorStore):
    """``BagelDB.ai`` vector store.
    To use, you should have the ``betabageldb`` python package installed.
    Example:
        .. code-block:: python
                from langchain.vectorstores import Bagel
                vectorstore = Bagel(cluster_name="langchain_store")
    """
    _LANGCHAIN_DEFAULT_CLUSTER_NAME = "langchain"
[docs]    def __init__(
        self,
        cluster_name: str = _LANGCHAIN_DEFAULT_CLUSTER_NAME,
        client_settings: Optional[bagel.config.Settings] = None,