Deploying RAG

app.py

@@ -11,6 +11,9 @@ from langchain_core.prompts import PromptTemplate
 from langchain.schema.output_parser import StrOutputParser
 from langchain.schema.runnable import RunnablePassthrough
 from langchain.schema.runnable.config import RunnableConfig
+from tqdm.asyncio import tqdm_asyncio
+import asyncio
+from tqdm.asyncio import tqdm
 
 # GLOBAL SCOPE - ENTIRE APPLICATION HAS ACCESS TO VALUES SET IN THIS SCOPE #
 # ---- ENV VARIABLES ---- # 
@@ -49,25 +52,57 @@ hf_embeddings = HuggingFaceEndpointEmbeddings(
     huggingfacehub_api_token=HF_TOKEN,
 )
 
-if os.path.exists("./data/vectorstore"):
-    vectorstore = FAISS.load_local(
-        "./data/vectorstore", 
-        hf_embeddings, 
-        allow_dangerous_deserialization=True # this is necessary to load the vectorstore from disk as it's stored as a `.pkl` file.
-    )
-    hf_retriever = vectorstore.as_retriever()
-    print("Loaded Vectorstore")
-else:
+async def add_documents_async(vectorstore, documents):
+    await vectorstore.aadd_documents(documents)
+
+async def process_batch(vectorstore, batch, is_first_batch, pbar):
+    if is_first_batch:
+        result = await FAISS.afrom_documents(batch, hf_embeddings)
+    else:
+        await add_documents_async(vectorstore, batch)
+        result = vectorstore
+    pbar.update(len(batch))
+    return result
+
+async def main():
     print("Indexing Files")
-    os.makedirs("./data/vectorstore", exist_ok=True)
-    for i in range(0, len(split_documents), 32):
-        if i == 0:
-            vectorstore = FAISS.from_documents(split_documents[i:i+32], hf_embeddings)
-            continue
-        vectorstore.add_documents(split_documents[i:i+32])
-    vectorstore.save_local("./data/vectorstore")
-
-hf_retriever = vectorstore.as_retriever()
+    
+    vectorstore = None
+    batch_size = 32
+    
+    batches = [split_documents[i:i+batch_size] for i in range(0, len(split_documents), batch_size)]
+    
+    async def process_all_batches():
+        nonlocal vectorstore
+        tasks = []
+        pbars = []
+        
+        for i, batch in enumerate(batches):
+            pbar = tqdm(total=len(batch), desc=f"Batch {i+1}/{len(batches)}", position=i)
+            pbars.append(pbar)
+            
+            if i == 0:
+                vectorstore = await process_batch(None, batch, True, pbar)
+            else:
+                tasks.append(process_batch(vectorstore, batch, False, pbar))
+        
+        if tasks:
+            await asyncio.gather(*tasks)
+        
+        for pbar in pbars:
+            pbar.close()
+    
+    await process_all_batches()
+    
+    hf_retriever = vectorstore.as_retriever()
+    print("\nIndexing complete. Vectorstore is ready for use.")
+    return hf_retriever
+
+async def run():
+    retriever = await main()
+    return retriever
+
+hf_retriever = asyncio.run(run())
 
 # -- AUGMENTED -- #
 """