AI Application Patterns with Cassandra

  Ingest path:
  Documents → Chunker → Embeddings API → Cassandra (chunks + vectors + metadata)

  Query path:
  User Query → Embeddings API → ANN Search → Retrieved Chunks → LLM → Answer

CREATE TABLE documents (
    doc_id      uuid,
    chunk_id    int,
    content     text,
    embedding   vector<float, 1536>,
    metadata    map<text, text>,
    created_at  timestamp,
    PRIMARY KEY (doc_id, chunk_id)
);

-- ANN index for semantic retrieval
CREATE CUSTOM INDEX ON documents (embedding)
    USING 'StorageAttachedIndex'
    WITH OPTIONS = {'similarity_function': 'cosine'};

-- SAI index on metadata for hybrid filtering
CREATE CUSTOM INDEX ON documents (metadata)
    USING 'StorageAttachedIndex';

-- Retrieve the 5 most relevant chunks within a specific category
SELECT content, metadata
FROM documents
WHERE metadata CONTAINS KEY 'category'
ORDER BY embedding ANN OF ?
LIMIT 5;

-- Filter by a specific metadata value
SELECT content, doc_id, chunk_id
FROM documents
WHERE metadata['category'] = 'legal'
ORDER BY embedding ANN OF ?
LIMIT 10;

-- Single table: structured data and embedding co-located
CREATE TABLE products (
    product_id      uuid PRIMARY KEY,
    name            text,
    description     text,
    category        text,
    price           decimal,
    description_emb vector<float, 1536>
);

CREATE TABLE articles (
    article_id    uuid PRIMARY KEY,
    title         text,
    body          text,
    summary       text,
    title_emb     vector<float, 1536>,    -- for title-based retrieval
    body_emb      vector<float, 1536>,    -- for full-text semantic search
    summary_emb   vector<float, 768>      -- smaller model for summary
);

CREATE CUSTOM INDEX ON articles (title_emb)   USING 'StorageAttachedIndex';
CREATE CUSTOM INDEX ON articles (body_emb)    USING 'StorageAttachedIndex';
CREATE CUSTOM INDEX ON articles (summary_emb) USING 'StorageAttachedIndex';

-- Prerequisite indexes
CREATE CUSTOM INDEX ON products (category)  USING 'StorageAttachedIndex';
CREATE CUSTOM INDEX ON products (price)     USING 'StorageAttachedIndex';
CREATE CUSTOM INDEX ON products (description_emb) USING 'StorageAttachedIndex';

-- Semantic search scoped to a category and price range
SELECT name, description, price
FROM products
WHERE category = 'electronics'
  AND price < 500
ORDER BY description_emb ANN OF ?
LIMIT 10;

SELECT name, price
FROM products
WHERE (category = 'electronics' OR category = 'computers')
  AND price < 1000
ORDER BY description_emb ANN OF ?
LIMIT 20;

CREATE TABLE conversations (
    session_id  uuid,
    message_id  timeuuid,
    role        text,           -- 'user', 'assistant', or 'system'
    content     text,
    embedding   vector<float, 1536>,
    PRIMARY KEY (session_id, message_id)
) WITH CLUSTERING ORDER BY (message_id DESC);

CREATE CUSTOM INDEX ON conversations (embedding)
    USING 'StorageAttachedIndex'
    WITH OPTIONS = {'similarity_function': 'cosine'};

-- Fetch the last 20 messages for a session
SELECT role, content
FROM conversations
WHERE session_id = ?
LIMIT 20;

-- Find the 5 most semantically relevant past messages in this session
-- (not necessarily the most recent)
SELECT role, content, message_id
FROM conversations
WHERE session_id = ?
ORDER BY embedding ANN OF ?
LIMIT 5;

-- Insert a message with a 30-day TTL
INSERT INTO conversations (session_id, message_id, role, content, embedding)
VALUES (?, ?, ?, ?, ?)
USING TTL 2592000;   -- 30 days in seconds

pip install cassio langchain-community langchain-openai

from langchain_community.vectorstores import Cassandra
from langchain_openai import OpenAIEmbeddings
import cassio

# Initialize cassio with your cluster connection
cassio.init(contact_points=["127.0.0.1"], keyspace="my_keyspace")

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Create or connect to a Cassandra vector store
vectorstore = Cassandra(
    embedding=embeddings,
    table_name="documents",
    session=None,      # uses cassio session
    keyspace=None,     # uses cassio keyspace
)

# Add documents
vectorstore.add_texts(["chunk one", "chunk two"], metadatas=[{}, {}])

# Similarity search
results = vectorstore.similarity_search("my query", k=5)

from llama_index.vector_stores.cassandra import CassandraVectorStore
from llama_index.core import VectorStoreIndex, StorageContext
import cassio

# pip install cassio llama-index llama-index-vector-stores-cassandra

cassio.init(contact_points=["127.0.0.1"], keyspace="my_keyspace")

vector_store = CassandraVectorStore(
    table="llamaindex_docs",
    embedding_dimension=1536,
)

storage_context = StorageContext.from_defaults(vector_store=vector_store)

# Build an index from documents
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context,
)

# Query
query_engine = index.as_query_engine(similarity_top_k=5)
response = query_engine.query("What is Cassandra's consistency model?")