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 now supports asynchronous operations for improved performance in production environments.
```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 now supports asynchronous operations for improved performance in production environments.
```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 now supports asynchronous operations for improved performance in production environments.
```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.
Qdrant now supports asynchronous operations for improved performance in production environments.
```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 now supports asynchronous operations for improved performance in production environments.
```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.