Install cassandra packages
pip install cassandra-driver
Run cassandra
docker run -d \
--name cassandra-db\
-p 9042:9042 \
cassandra:latest
from agno.agent import Agent
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.vectordb.cassandra import Cassandra
from agno.embedder.mistral import MistralEmbedder
from agno.models.mistral import MistralChat
# (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 = PDFUrlKnowledgeBase(
urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=Cassandra(table_name="recipes", keyspace="testkeyspace", session=session, embedder=MistralEmbedder()),
)
# knowledge_base.load(recreate=False) # Comment out after first run
agent = Agent(
model=MistralChat(provider="mistral-large-latest", api_key=os.getenv("MISTRAL_API_KEY")),
knowledge=knowledge_base,
show_tool_calls=True,
)
agent.print_response(
"What are the health benefits of Khao Niew Dam Piek Maphrao Awn?", markdown=True, show_full_reasoning=True
)
Cassandra also supports asynchronous operations, enabling concurrency and leading to better performance.
import asyncio
from agno.agent import Agent
from agno.embedder.mistral import MistralEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
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 = PDFUrlKnowledgeBase(
urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=Cassandra(
table_name="recipes",
keyspace="testkeyspace",
session=session,
embedder=MistralEmbedder(),
),
)
agent = Agent(
model=MistralChat(),
knowledge=knowledge_base,
show_tool_calls=True,
)
if __name__ == "__main__":
# Comment out after first run
asyncio.run(knowledge_base.aload(recreate=False))
# 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.
Was this page helpful?
Install cassandra packages
pip install cassandra-driver
Run cassandra
docker run -d \
--name cassandra-db\
-p 9042:9042 \
cassandra:latest
from agno.agent import Agent
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
from agno.vectordb.cassandra import Cassandra
from agno.embedder.mistral import MistralEmbedder
from agno.models.mistral import MistralChat
# (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 = PDFUrlKnowledgeBase(
urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=Cassandra(table_name="recipes", keyspace="testkeyspace", session=session, embedder=MistralEmbedder()),
)
# knowledge_base.load(recreate=False) # Comment out after first run
agent = Agent(
model=MistralChat(provider="mistral-large-latest", api_key=os.getenv("MISTRAL_API_KEY")),
knowledge=knowledge_base,
show_tool_calls=True,
)
agent.print_response(
"What are the health benefits of Khao Niew Dam Piek Maphrao Awn?", markdown=True, show_full_reasoning=True
)
Cassandra also supports asynchronous operations, enabling concurrency and leading to better performance.
import asyncio
from agno.agent import Agent
from agno.embedder.mistral import MistralEmbedder
from agno.knowledge.pdf_url import PDFUrlKnowledgeBase
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 = PDFUrlKnowledgeBase(
urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
vector_db=Cassandra(
table_name="recipes",
keyspace="testkeyspace",
session=session,
embedder=MistralEmbedder(),
),
)
agent = Agent(
model=MistralChat(),
knowledge=knowledge_base,
show_tool_calls=True,
)
if __name__ == "__main__":
# Comment out after first run
asyncio.run(knowledge_base.aload(recreate=False))
# 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.
Was this page helpful?