Cassandra Vector Database

Setup

Install Cassandra packages

uv pip install -U agno cassio cassandra-driver mistralai pypdf

Run Cassandra with Docker:

docker run -d \
  --name cassandra-db \
  -p 9042:9042 \
  cassandra:latest

Set your Mistral API key. The examples below use Mistral for embeddings and responses.

export MISTRAL_API_KEY=xxx

Example

agent_with_knowledge.py

from agno.agent import Agent
from agno.knowledge.knowledge import Knowledge
from agno.vectordb.cassandra import Cassandra
from agno.knowledge.embedder.mistral import MistralEmbedder
from agno.models.mistral import MistralChat
from cassandra.cluster import Cluster

# Set up your Cassandra DB

cluster = Cluster()

session = cluster.connect()
session.execute(
    """
    CREATE KEYSPACE IF NOT EXISTS testkeyspace
    WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }
    """
)

knowledge_base = Knowledge(
    vector_db=Cassandra(table_name="recipes", keyspace="testkeyspace", session=session, embedder=MistralEmbedder()),
)

knowledge_base.insert(
    url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"
)

agent = Agent(
    model=MistralChat(id="mistral-large-latest"),
    knowledge=knowledge_base,
)

agent.print_response(
    "What are the health benefits of Khao Niew Dam Piek Maphrao Awn?", markdown=True, show_full_reasoning=True
)

Cassandra creates the vector column using the configured embedder's dimensions. An existing table must match that dimension. The examples here use 1024-dimensional embeddings.

Async Support ⚡

Cassandra also supports asynchronous operations, enabling concurrency and leading to better performance.

async_cassandra.py

import asyncio

from agno.agent import Agent
from agno.knowledge.embedder.mistral import MistralEmbedder
from agno.knowledge.knowledge import Knowledge
from agno.models.mistral import MistralChat
from agno.vectordb.cassandra import Cassandra

try:
    from cassandra.cluster import Cluster  # type: ignore
except (ImportError, ModuleNotFoundError):
    raise ImportError(
        "Could not import cassandra-driver python package.Please install it with pip install cassandra-driver."
    )

cluster = Cluster()

session = cluster.connect()
session.execute(
    """
    CREATE KEYSPACE IF NOT EXISTS testkeyspace
    WITH REPLICATION = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 }
    """
)

knowledge_base = Knowledge(
    vector_db=Cassandra(
        table_name="recipes",
        keyspace="testkeyspace",
        session=session,
        embedder=MistralEmbedder(),
    ),
)

agent = Agent(
    model=MistralChat(),
    knowledge=knowledge_base,
)

if __name__ == "__main__":
    asyncio.run(knowledge_base.ainsert(
        url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
        )
    )

    # Create and use the agent
    asyncio.run(
        agent.aprint_response(
            "What are the health benefits of Khao Niew Dam Piek Maphrao Awn?",
            markdown=True,
        )
    )

Use ainsert() and aprint_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.

Cassandra Params

Parameter	Type	Default	Description
`table_name`	`str`	Required	Name of the table to store vectors and metadata in Cassandra
`keyspace`	`str`	Required	Keyspace name in Cassandra where the table will be created
`embedder`	`Optional[Embedder]`	`OpenAIEmbedder()`	Embedder instance to generate embeddings
`session`	`CassandraSession`	Required	Active Cassandra session object for database operations
`name`	`Optional[str]`	`None`	Name of the vector database
`description`	`Optional[str]`	`None`	Description of the vector database

Setup

Example

Async Support ⚡

Cassandra Params

On this page