Cassandra

Langchain Vector store for Apache Cassandra.

Install

pip install langchain-cassandra

or

uv add langchain-cassandra

Requires a running Cassandra cluster (default: localhost:9042).

API Reference

__init__

def __init__(
    self,
    embedding_function: Embeddings | None = None,
    contact_points: list[str] | None = None,
    port: int = 9042,
    keyspace: str = "langchain",
    table_name: str = "documents",
    **kwargs,
)

from_texts

def from_texts(
    cls,
    texts: list[str],
    embedding: Embeddings | None = None,
    metadatas: list[dict] | None = None,
    ids: list[str] | None = None,
    contact_points: list[str] | None = None,
    port: int | None = None,
    keyspace: str | None = None,
    table_name: str | None = None,
    **kwargs,
) -> Cassandra

Returns: Cassandra — a new vector store with the texts indexed.

add_texts

def add_texts(
    self,
    texts: list[str],
    metadatas: list[dict] | None = None,
    ids: list[str] | None = None,
    **kwargs,
) -> list[str]

Returns: list[str] — the IDs of the added texts.

delete

def delete(
    self,
    ids: list[str] | None = None,
    **kwargs,
) -> bool | None

Returns: bool — True if deletion succeeded. Raises ValueError if ids is None.

similarity_search

def similarity_search(
    self,
    query: str,
    k: int = 4,
    **kwargs,
) -> list[Document]

Returns: list[Document] — documents most similar to the query. Uses cosine similarity (fetches all embeddings, computes in-memory).

similarity_search_with_score

def similarity_search_with_score(
    self,
    query: str,
    k: int = 4,
    **kwargs,
) -> list[tuple[Document, float]]

Returns: list[tuple[Document, float]] — tuples of (Document, cosine_similarity). Score range [-1, 1]. Higher = more similar.