Source code for langchain.vectorstores.cassandra
from __future__ import annotations
import typing
import uuid
from typing import (
    Any,
    Callable,
    Dict,
    Iterable,
    List,
    Optional,
    Tuple,
    Type,
    TypeVar,
    Union,
)
import numpy as np
if typing.TYPE_CHECKING:
    from cassandra.cluster import Session
from langchain.docstore.document import Document
from langchain.schema.embeddings import Embeddings
from langchain.schema.vectorstore import VectorStore
from langchain.vectorstores.utils import maximal_marginal_relevance
CVST = TypeVar("CVST", bound="Cassandra")
[docs]class Cassandra(VectorStore):
    """Wrapper around Apache Cassandra(R) for vector-store workloads.
    To use it, you need a recent installation of the `cassio` library
    and a Cassandra cluster / Astra DB instance supporting vector capabilities.
    Visit the cassio.org website for extensive quickstarts and code examples.
    Example:
        .. code-block:: python
                from langchain.vectorstores import Cassandra
                from langchain.embeddings.openai import OpenAIEmbeddings
                embeddings = OpenAIEmbeddings()
                session = ...             # create your Cassandra session object
                keyspace = 'my_keyspace'  # the keyspace should exist already
                table_name = 'my_vector_store'
                vectorstore = Cassandra(embeddings, session, keyspace, table_name)
    """
    _embedding_dimension: Union[int, None]
    @staticmethod
    def _filter_to_metadata(filter_dict: Optional[Dict[str, str]]) -> Dict[str, Any]:
        if filter_dict is None:
            return {}
        else:
            return filter_dict
    def _get_embedding_dimension(self) -> int: