Setup
Install cassandra packagesuv pip install cassandra-driver
docker run -d \
--name cassandra-db \
-p 9042:9042 \
cassandra:latest
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
# (Optional) 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(provider="mistral-large-latest", api_key=os.getenv("MISTRAL_API_KEY")),
knowledge=knowledge_base,
)
agent.print_response(
"What are the health benefits of Khao Niew Dam Piek Maphrao Awn?", markdown=True, show_full_reasoning=True
)
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
aload() and aprint_response() methods with asyncio.run() for non-blocking operations in high-throughput applications.Cassandra Params
| Parameter | Type | Default | Description |
|---|---|---|---|
table_name | str | None | Name of the table to store vectors and metadata in Cassandra |
keyspace | str | None | Keyspace name in Cassandra where the table will be created |
embedder | Optional[Embedder] | OpenAIEmbedder() | Embedder instance to generate embeddings |
session | CassandraSession | None | Active Cassandra session object for database operations |