async changes

backend/Qdrant.py

@@ -1,8 +1,9 @@
 import os
-from qdrant_client import QdrantClient, models
-from qdrant_client.models import PayloadSchemaType
+from qdrant_client import QdrantClient, models, grpc
+from qdrant_client.http.models import PayloadSchemaType
 import logging
 from dotenv import load_dotenv
+import asyncio
 
 # Configure logging
 logger = logging.getLogger(__name__)
@@ -10,8 +11,6 @@ logger = logging.getLogger(__name__)
 load_dotenv()
 
 # Configuration
-# QDRANT_URL = "https://cc102304-2c06-4d51-9dee-d436f4413549.us-west-1-0.aws.cloud.qdrant.io"
-# QDRANT_API_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2Nlc3MiOiJtIn0.cHs27o6erIf1BQHCdTxE4L4qZg4vCdrp51oNNNghjWM"
 QDRANT_API_KEY = os.getenv("QDRANT_API_KEY")
 QDRANT_URL = os.getenv("QDRANT_URL")
 
@@ -20,10 +19,11 @@ class QdrantManager:
         self.qdrant_client = QdrantClient(
             url=QDRANT_URL,
             api_key=QDRANT_API_KEY,
+            prefer_grpc=True,
         )
         print("Connected to Qdrant")
 
-    def get_or_create_company_collection(self, collection_name: str) -> str:
+    async def get_or_create_company_collection(self, collection_name: str) -> str:
         """
         Get or create a collection for a company.
         
@@ -37,14 +37,13 @@ class QdrantManager:
             ValueError: If collection creation fails
         """
         try:
-
             print(f"Creating new collection: {collection_name}")
 
             # Vector size for text-embedding-3-small is 1536
             vector_size = 384
             
             # Create collection with vector configuration
-            self.qdrant_client.create_collection(
+            await self.qdrant_client.create_collection(
                 collection_name=collection_name,
                 vectors_config=models.VectorParams(
                     size=vector_size,
@@ -63,7 +62,7 @@ class QdrantManager:
             }
             
             for field_name, schema_type in payload_indices.items():
-                self.qdrant_client.create_payload_index(
+                await self.qdrant_client.create_payload_index(
                     collection_name=collection_name,
                     field_name=field_name,
                     field_schema=schema_type
@@ -78,11 +77,14 @@ class QdrantManager:
             raise ValueError(error_msg) from e
 
 # # Example usage
-# if __name__ == "__main__":
+# async def main():
 #     try:
 #         qdrant_manager = QdrantManager()
 #         collection_name = "ca-documents"
-#         result = qdrant_manager.get_or_create_company_collection(collection_name)
+#         result = await qdrant_manager.get_or_create_company_collection(collection_name)
 #         print(f"Collection name: {result}")
 #     except Exception as e:
-#         print(f"Error: {e}")
+#         print(f"Error: {e}")
+
+# if __name__ == "__main__":
+#     asyncio.run(main())

backend/backend_api.py

@@ -96,7 +96,7 @@ async def ask_question_stream(request: QuestionRequest):
         
         async def event_generator():
             try:
-                for chunk in rag_system.ask_question_stream(request.question):
+                async for chunk in rag_system.ask_question_stream(request.question):
                     if chunk:  # Only yield non-empty chunks
                         yield chunk
             except Exception as e:
@@ -138,7 +138,7 @@ async def upload_document(file: UploadFile = File(...)):
         try:
             # Process the uploaded file
             logger.info(f"Processing uploaded file: {file.filename}")
-            success = rag_system.upload_document(temp_file_path)
+            success = await rag_system.upload_document(temp_file_path)
             
             if success:
                 return {
@@ -170,7 +170,7 @@ async def search_documents(request: SearchRequest):
         if not rag_system:
             raise HTTPException(status_code=500, detail="RAG system not initialized")
         
-        results = rag_system.vector_store.search_similar(request.query, limit=request.limit)
+        results = await rag_system.vector_store.search_similar(request.query, limit=request.limit)
         
         return {
             "status": "success",
@@ -210,7 +210,7 @@ async def get_collection_info():
         if not rag_system:
             raise HTTPException(status_code=500, detail="RAG system not initialized")
         
-        info = rag_system.vector_store.get_collection_info()
+        info = await rag_system.vector_store.get_collection_info()
         return {
             "status": "success",
             "collection_info": info

backend/rag.py

@@ -5,6 +5,7 @@ from docx import Document
 from typing import List
 import os
 from langchain_text_splitters import RecursiveCharacterTextSplitter
+import asyncio
 
 class RAG:
     def __init__(self, google_api_key: str, collection_name: str = "ca-documents"):
@@ -17,10 +18,7 @@ class RAG:
         self.vector_store = VectorStore()
         
         # Verify vector store is properly initialized
-        if not self.vector_store.verify_collection_health():
-            print("Warning: Vector store collection health check failed")
-        else:
-            print("Vector store initialized successfully")
+        asyncio.run(self.vector_store.verify_collection_health())
         
         # Setup Text Splitter
         self.text_splitter = RecursiveCharacterTextSplitter(
@@ -53,7 +51,7 @@ class RAG:
         chunks = self.text_splitter.split_text(full_text)
         return [chunk.strip() for chunk in chunks if chunk.strip()]
     
-    def upload_document(self, file_path: str) -> bool:
+    async def upload_document(self, file_path: str) -> bool:
         """Upload and process document"""
         try:
             filename = os.path.basename(file_path)
@@ -73,7 +71,7 @@ class RAG:
             
             # Store chunks in Qdrant
             for i, chunk in enumerate(chunks):
-                self.vector_store.add_document(
+                await self.vector_store.add_document(
                     text=chunk,
                     metadata={"source": filename, "chunk_id": i}
                 )
@@ -136,7 +134,7 @@ class RAG:
         
         return False
 
-    def ask_question_stream(self, question: str):
+    async def ask_question_stream(self, question: str):
         """Ask a question and get a streaming answer"""
         try:
             # 1. Check if this is casual conversation
@@ -154,7 +152,7 @@ Respond naturally and warmly as a CA study assistant. Be helpful and mention tha
                 return
 
             # 2. For CA-specific questions, search for similar documents
-            similar_docs = self.vector_store.search_similar(question, limit=3)
+            similar_docs = await self.vector_store.search_similar(question, limit=3)
 
             if similar_docs and len(similar_docs) > 0:
                 # 3. Create context from similar documents

backend/vector_store.py

@@ -1,5 +1,5 @@
-from qdrant_client import QdrantClient, models
-from qdrant_client.models import PointStruct, PayloadSchemaType
+from qdrant_client import QdrantClient, models, grpc
+from qdrant_client.http.models import PointStruct, PayloadSchemaType
 from sentence_transformers import SentenceTransformer
 import uuid
 import os
@@ -7,6 +7,7 @@ import logging
 from typing import List, Dict, Any
 from dotenv import load_dotenv
 import time
+import asyncio
 
 # Load environment variables
 load_dotenv()
@@ -29,6 +30,7 @@ class VectorStore:
         self.client = QdrantClient(
             url=qdrant_url,
             api_key=qdrant_api_key,
+            prefer_grpc=True,
         )
         print("Connected to Qdrant")
         
@@ -36,7 +38,7 @@ class VectorStore:
         self.embedding_model = self._initialize_embedding_model()
         
         # Create collection with proper indices
-        self._ensure_collection_exists()
+        asyncio.run(self._ensure_collection_exists())
     
     def _initialize_embedding_model(self):
         """Initialize the embedding model from a local directory"""
@@ -52,7 +54,7 @@ class VectorStore:
             print(f"Failed to load local model: {e}")
             raise RuntimeError("Failed to initialize embedding model from local path")
 
-    def _collection_exists_and_accessible(self) -> bool:
+    async def _collection_exists_and_accessible(self) -> bool:
         """
         Check if collection exists and is accessible by trying to get its info.
         
@@ -61,14 +63,14 @@ class VectorStore:
         """
         try:
             # Try to get collection info - this is more reliable than just listing collections
-            collection_info = self.client.get_collection(self.collection_name)
+            collection_info = await self.client.get_collection(self.collection_name)
             print(f"Collection '{self.collection_name}' exists and is accessible")
             return True
         except Exception as e:
             print(f"Collection '{self.collection_name}' is not accessible: {e}")
             return False
 
-    def _create_collection(self) -> bool:
+    async def _create_collection(self) -> bool:
         """
         Create the collection with proper configuration.
         
@@ -82,7 +84,7 @@ class VectorStore:
             vector_size = 384
             
             # Create collection with vector configuration
-            self.client.create_collection(
+            await self.client.create_collection(
                 collection_name=self.collection_name,
                 vectors_config=models.VectorParams(
                     size=vector_size,
@@ -95,7 +97,7 @@ class VectorStore:
             )
             
             # Wait a moment for collection to be fully created
-            time.sleep(1)
+            await asyncio.sleep(1)
             
             # Create payload indices
             payload_indices = {
@@ -105,7 +107,7 @@ class VectorStore:
             
             for field_name, schema_type in payload_indices.items():
                 try:
-                    self.client.create_payload_index(
+                    await self.client.create_payload_index(
                         collection_name=self.collection_name,
                         field_name=field_name,
                         field_schema=schema_type
@@ -122,7 +124,7 @@ class VectorStore:
             print(error_msg)
             return False
 
-    def _ensure_collection_exists(self) -> bool:
+    async def _ensure_collection_exists(self) -> bool:
         """
         Ensure collection exists and is accessible, create if necessary.
         
@@ -131,12 +133,12 @@ class VectorStore:
         """
         try:
             # First, check if collection exists and is accessible
-            if self._collection_exists_and_accessible():
+            if await self._collection_exists_and_accessible():
                 return True
             
             # If not accessible, try to create it
             print(f"Collection '{self.collection_name}' not found or not accessible, creating...")
-            return self._create_collection()
+            return await self._create_collection()
             
         except Exception as e:
             error_msg = f"Failed to ensure collection exists: {str(e)}"
@@ -144,7 +146,7 @@ class VectorStore:
             print(error_msg)
             return False
     
-    def add_document(self, text: str, metadata: Dict = None) -> bool:
+    async def add_document(self, text: str, metadata: Dict = None) -> bool:
         """Add a document to the collection with retry logic"""
         max_retries = 3
         retry_delay = 1
@@ -152,9 +154,9 @@ class VectorStore:
         for attempt in range(max_retries):
             try:
                 # Ensure collection exists before adding document
-                if not self._collection_exists_and_accessible():
+                if not await self._collection_exists_and_accessible():
                     print("Collection not accessible, trying to recreate...")
-                    if not self._create_collection():
+                    if not await self._create_collection():
                         raise Exception("Failed to create collection")
                 
                 # Generate embedding
@@ -181,7 +183,7 @@ class VectorStore:
                 )
                 
                 # Store in Qdrant
-                result = self.client.upsert(
+                result = await self.client.upsert(
                     collection_name=self.collection_name,
                     points=[point]
                 )
@@ -200,11 +202,11 @@ class VectorStore:
                 if "Not found" in str(e) and "doesn't exist" in str(e):
                     # Collection doesn't exist, try to recreate
                     print("Collection not found, attempting to recreate...")
-                    self._create_collection()
+                    await self._create_collection()
                 
                 if attempt < max_retries - 1:
                     print(f"Retrying in {retry_delay} seconds...")
-                    time.sleep(retry_delay)
+                    await asyncio.sleep(retry_delay)
                     retry_delay *= 2  # Exponential backoff
                 else:
                     print(f"Failed to add document after {max_retries} attempts")
@@ -212,11 +214,11 @@ class VectorStore:
         
         return False
     
-    def search_similar(self, query: str, limit: int = 5) -> List[Dict]:
+    async def search_similar(self, query: str, limit: int = 5) -> List[Dict]:
         """Search for similar documents with error handling"""
         try:
             # Ensure collection exists before searching
-            if not self._collection_exists_and_accessible():
+            if not await self._collection_exists_and_accessible():
                 print("Collection not accessible for search")
                 return []
             
@@ -224,7 +226,7 @@ class VectorStore:
             query_embedding = self.embedding_model.encode([query])[0]
             
             # Search in Qdrant
-            results = self.client.search(
+            results = await self.client.search(
                 collection_name=self.collection_name,
                 query_vector=query_embedding.tolist(),
                 limit=limit
@@ -246,10 +248,10 @@ class VectorStore:
             print(f"Error searching: {e}")
             return []
     
-    def get_collection_info(self) -> Dict:
+    async def get_collection_info(self) -> Dict:
         """Get information about the collection"""
         try:
-            collection_info = self.client.get_collection(self.collection_name)
+            collection_info = await self.client.get_collection(self.collection_name)
             return {
                 "name": collection_info.config.name,
                 "vector_size": collection_info.config.params.vectors.size,
@@ -261,16 +263,16 @@ class VectorStore:
             print(f"Error getting collection info: {e}")
             return {}
     
-    def verify_collection_health(self) -> bool:
+    async def verify_collection_health(self) -> bool:
         """Verify that the collection is healthy and accessible"""
         try:
             # Try to get collection info
-            info = self.get_collection_info()
+            info = await self.get_collection_info()
             if not info:
                 return False
             
             # Try a simple search to verify functionality
-            test_results = self.search_similar("test query", limit=1)
+            test_results = await self.search_similar("test query", limit=1)
             # This should not fail even if no results are found
             
             print(f"Collection health check passed. Points count: {info.get('points_count', 0)}")