Cassandra also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_cassandra.py 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, ) ) ```aload()
and aprint\_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.
ChromaDB also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_chroma_db.py # install chromadb - `pip install chromadb` import asyncio from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.vectordb.chroma import ChromaDb # Initialize ChromaDB vector_db = ChromaDb(collection="recipes", path="tmp/chromadb", persistent_client=True) # Create knowledge base knowledge_base = PDFUrlKnowledgeBase( urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], vector_db=vector_db, ) # Create and use the agent agent = Agent(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("How to make Tom Kha Gai", markdown=True)) ```aload()
and aprint\_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.
Clickhouse also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_clickhouse.py import asyncio from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.storage.agent.sqlite import SqliteAgentStorage from agno.vectordb.clickhouse import Clickhouse agent = Agent( storage=SqliteAgentStorage(table_name="recipe_agent"), knowledge=PDFUrlKnowledgeBase( urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], vector_db=Clickhouse( table_name="recipe_documents", host="localhost", port=8123, username="ai", password="ai", ), ), # Show tool calls in the response show_tool_calls=True, # Enable the agent to search the knowledge base search_knowledge=True, # Enable the agent to read the chat history read_chat_history=True, ) if __name__ == "__main__": # Comment out after first run asyncio.run(agent.knowledge.aload(recreate=False)) # Create and use the agent asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload()
and aprint\_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.
Several vector databases support asynchronous operations, offering improved performance through non-blocking operations, concurrent processing, reduced latency, and seamless integration with FastAPI and async agents.
aload
methods for async knowledge base loading in production environments.
LanceDB also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_lance_db.py # install lancedb - `pip install lancedb` import asyncio from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.vectordb.lancedb import LanceDb # Initialize LanceDB vector_db = LanceDb( table_name="recipes", uri="tmp/lancedb", # You can change this path to store data elsewhere ) # Create knowledge base knowledge_base = PDFUrlKnowledgeBase( urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], vector_db=vector_db, ) agent = Agent(knowledge=knowledge_base, show_tool_calls=True, debug_mode=True) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.aload(recreate=False)) # Comment out after first run # Create and use the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload()
and aprint\_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.
Milvus also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_milvus_db.py # install pymilvus - `pip install pymilvus` import asyncio from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.vectordb.milvus import Milvus # Initialize Milvus with local file vector_db = Milvus( collection="recipes", uri="tmp/milvus.db", # For local file-based storage ) # Create knowledge base knowledge_base = PDFUrlKnowledgeBase( urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], vector_db=vector_db, ) # Create agent with knowledge base agent = Agent(knowledge=knowledge_base) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.aload(recreate=False)) # Comment out after first run # Query the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload()
and aprint\_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.
MongoDB also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_mongodb.py import asyncio from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.vectordb.mongodb import MongoDb # MongoDB Atlas connection string """ Example connection strings: "mongodb+srv://aload()
and aprint\_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.
PgVector also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_pgvector.py import asyncio from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.vectordb.pgvector import PgVector db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" vector_db = PgVector(table_name="recipes", db_url=db_url) knowledge_base = PDFUrlKnowledgeBase( urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], vector_db=vector_db, ) agent = Agent(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("How to make Tom Kha Gai", markdown=True)) ```aload()
and aprint\_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.
Pinecone also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_pinecone.py import asyncio from os import getenv from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.vectordb.pineconedb import PineconeDb api_key = getenv("PINECONE_API_KEY") index_name = "thai-recipe-index" vector_db = PineconeDb( name=index_name, dimension=1536, metric="cosine", spec={"serverless": {"cloud": "aws", "region": "us-east-1"}}, api_key=api_key, ) knowledge_base = PDFUrlKnowledgeBase( urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], vector_db=vector_db, ) agent = Agent( knowledge=knowledge_base, # Show tool calls in the response show_tool_calls=True, # Enable the agent to search the knowledge base search_knowledge=True, # Enable the agent to read the chat history read_chat_history=True, ) if __name__ == "__main__": # Comment out after first run asyncio.run(knowledge_base.aload(recreate=False, upsert=True)) # Create and use the agent asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload()
and aprint\_response()
methods with asyncio.run()
for non-blocking operations in high-throughput applications.
Qdrant also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_qdrant_db.py import asyncio from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.vectordb.qdrant import Qdrant COLLECTION_NAME = "thai-recipes" # Initialize Qdrant with local instance vector_db = Qdrant( collection=COLLECTION_NAME, url="http://localhost:6333" ) # Create knowledge base knowledge_base = PDFUrlKnowledgeBase( urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], vector_db=vector_db, ) agent = Agent(knowledge=knowledge_base, show_tool_calls=True) if __name__ == "__main__": # Load knowledge base asynchronously asyncio.run(knowledge_base.aload(recreate=False)) # Comment out after first run # Create and use the agent asynchronously asyncio.run(agent.aprint_response("How to make Tom Kha Gai", markdown=True)) ```aload()
and aprint\_response()
with asyncio provides non-blocking operations, making your application more responsive under load.
Weaviate also supports asynchronous operations, enabling concurrency and leading to better performance.
```python async_weaviate_db.py import asyncio from agno.agent import Agent from agno.knowledge.pdf_url import PDFUrlKnowledgeBase from agno.vectordb.search import SearchType from agno.vectordb.weaviate import Distance, VectorIndex, Weaviate vector_db = Weaviate( collection="recipes_async", search_type=SearchType.hybrid, vector_index=VectorIndex.HNSW, distance=Distance.COSINE, local=True, # Set to False if using Weaviate Cloud and True if using local instance ) # Create knowledge base knowledge_base = PDFUrlKnowledgeBase( urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], vector_db=vector_db, ) agent = Agent( knowledge=knowledge_base, search_knowledge=True, 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("How to make Tom Kha Gai", markdown=True)) ```WeaviateAsyncClient
to provide non-blocking vector operations. This is particularly valuable for applications requiring high concurrency and throughput.