Source code for langchain.vectorstores.cassandra
"""Wrapper around Cassandra vector-store capabilities, based on cassIO."""
from __future__ import annotations
import typing
import uuid
from typing import Any, Callable, Iterable, List, Optional, Tuple, Type, TypeVar
import numpy as np
if typing.TYPE_CHECKING:
    from cassandra.cluster import Session
from langchain.docstore.document import Document
from langchain.embeddings.base import Embeddings
from langchain.vectorstores.base import VectorStore
from langchain.vectorstores.utils import maximal_marginal_relevance
CVST = TypeVar("CVST", bound="Cassandra")
[docs]class Cassandra(VectorStore):
    """Wrapper around Cassandra embeddings platform.
    There is no notion of a default table name, since each embedding
    function implies its own vector dimension, which is part of the schema.
    Example:
        .. code-block:: python
                from langchain.vectorstores import Cassandra
                from langchain.embeddings.openai import OpenAIEmbeddings
                embeddings = OpenAIEmbeddings()
                session = ...
                keyspace = 'my_keyspace'
                vectorstore = Cassandra(embeddings, session, keyspace, 'my_doc_archive')
    """
    _embedding_dimension: int | None
    def _get_embedding_dimension(self) -> int:
        if self._embedding_dimension is None:
            self._embedding_dimension = len(
                self.embedding.embed_query("This is a sample sentence.")
            )
        return self._embedding_dimension
[docs]    def __init__(
        self,
        embedding: Embeddings,
        session: Session,
        keyspace: str,
        table_name: str,
        ttl_seconds: Optional[int] = None,
    ) -> None:
        try:
            from cassio.vector import VectorTable