fix

search_engine/indexer.py

@@ -12,6 +12,7 @@ import time
 import hashlib
 import os
 import json
+import threading
 
 from langchain_core.documents import Document
 from langchain_core.retrievers import BaseRetriever
@@ -24,12 +25,18 @@ from configuration.parameters import parameters
 
 logger = logging.getLogger(__name__)
 
+# Thread lock for manifest file access
+_manifest_lock = threading.Lock()
+
 
 def doc_id(doc) -> str:
+    """Generate a unique ID for a document based on source, page, chunk_id, and content hash."""
     src = doc.metadata.get("source", "")
     page = doc.metadata.get("page", "")
     chunk = doc.metadata.get("chunk_id", "")
-    base = f"{src}::{page}::{chunk}"
+    # Include content hash to ensure uniqueness even if chunk_id is missing
+    content = hashlib.sha256(doc.page_content.encode("utf-8")).hexdigest()[:16]
+    base = f"{src}::{page}::{chunk}::{content}"
     return hashlib.sha256(base.encode("utf-8")).hexdigest()
 
 
@@ -38,15 +45,28 @@ def content_hash(doc) -> str:
 
 
 def load_manifest(path):
+    """Thread-safe manifest loading."""
     if os.path.exists(path):
-        with open(path, "r") as f:
-            return json.load(f)
+        try:
+            with open(path, "r") as f:
+                return json.load(f)
+        except (json.JSONDecodeError, IOError) as e:
+            logger.warning(f"Failed to load manifest, starting fresh: {e}")
+            return {}
     return {}
 
 
 def save_manifest(path, manifest):
-    with open(path, "w") as f:
-        json.dump(manifest, f)
+    """Thread-safe manifest saving with atomic write."""
+    temp_path = path + ".tmp"
+    try:
+        with open(temp_path, "w") as f:
+            json.dump(manifest, f)
+        os.replace(temp_path, path)  # Atomic rename
+    except Exception as e:
+        logger.error(f"Failed to save manifest: {e}")
+        if os.path.exists(temp_path):
+            os.remove(temp_path)
 
 
 class EnsembleRetriever(BaseRetriever):
@@ -132,12 +152,13 @@ class RetrieverBuilder:
                 m.update(str(v).encode('utf-8'))
         return m.hexdigest()
 
-    def build_hybrid_retriever(self, docs) -> EnsembleRetriever:
+    def build_hybrid_retriever(self, docs, session_id: str = None) -> EnsembleRetriever:
         """
         Build hybrid retriever using BM25 and vector search.
         
         Args:
             docs: List of documents to index
+            session_id: Optional session ID for user isolation (recommended for multi-user)
             
         Returns:
             EnsembleRetriever combining BM25 and vector search
@@ -145,18 +166,31 @@ class RetrieverBuilder:
         logger.info(f"Building hybrid retriever with {len(docs)} documents...")
         if not docs:
             raise ValueError("No documents provided")
-        chroma_dir = parameters.CHROMA_DB_PATH
+        
+        # Use session-specific directory if provided (for multi-user isolation)
+        if session_id:
+            chroma_dir = os.path.join(parameters.CHROMA_DB_PATH, f"session_{session_id}")
+        else:
+            chroma_dir = parameters.CHROMA_DB_PATH
+            
         manifest_path = os.path.join(chroma_dir, "indexed_manifest.json")
         os.makedirs(chroma_dir, exist_ok=True)
-        manifest = load_manifest(manifest_path)
+        
+        # Thread-safe manifest access
+        with _manifest_lock:
+            manifest = load_manifest(manifest_path)
+        
+        t_vector_start = time.time()
         vector_store = Chroma(
             embedding_function=self.embeddings,
             persist_directory=chroma_dir,
         )
+        
         to_add = []
         ids_to_add = []
         to_delete_ids = []
         current_ids = set()
+        
         for d in docs:
             _id = doc_id(d)
             _hash = content_hash(d)
@@ -170,6 +204,7 @@ class RetrieverBuilder:
                 to_add.append(d)
                 ids_to_add.append(_id)
                 manifest[_id] = _hash             
+        
         if to_add:
             # Safety net: de-dupe before add_documents
             seen = set()
@@ -180,20 +215,35 @@ class RetrieverBuilder:
                 seen.add(_id)
                 uniq_docs.append(doc)
                 uniq_ids.append(_id)
-            # Debugging: show duplicate IDs and their sources
-            from collections import Counter
-            counts = Counter(ids_to_add)
-            dupes = [i for i, c in counts.items() if c > 1]
-            if dupes:
-                print("Duplicate IDs:", len(dupes))
-                for d in dupes[:10]:
-                    idxs = [k for k, x in enumerate(ids_to_add) if x == d]
-                    print("ID:", d, "examples:")
-                    for k in idxs[:3]:
-                        md = to_add[k].metadata
-                        print("  ", md.get("source"), md.get("page"), md.get("chunk_index"))
-            vector_store.add_documents(uniq_docs, ids=uniq_ids)
-        save_manifest(manifest_path, manifest)
+            
+            # Log duplicate count for debugging
+            dupe_count = len(to_add) - len(uniq_docs)
+            if dupe_count > 0:
+                logger.debug(f"Filtered {dupe_count} duplicate documents before indexing")
+            
+            # Batch add documents for better performance
+            logger.info(f"[PROFILE] Adding {len(uniq_docs)} new documents to vector store...")
+            t_add_start = time.time()
+            
+            # Add in batches for progress tracking and memory efficiency
+            batch_size = 100
+            for i in range(0, len(uniq_docs), batch_size):
+                batch_docs = uniq_docs[i:i+batch_size]
+                batch_ids = uniq_ids[i:i+batch_size]
+                vector_store.add_documents(batch_docs, ids=batch_ids)
+                if len(uniq_docs) > batch_size:
+                    logger.debug(f"[PROFILE] Indexed batch {i//batch_size + 1}/{(len(uniq_docs)-1)//batch_size + 1}")
+            
+            t_add_end = time.time()
+            logger.info(f"[PROFILE] Vector store add_documents: {t_add_end - t_add_start:.2f}s")
+        
+        t_vector_end = time.time()
+        logger.info(f"[PROFILE] Total vector store setup: {t_vector_end - t_vector_start:.2f}s")
+        
+        # Thread-safe manifest save
+        with _manifest_lock:
+            save_manifest(manifest_path, manifest)
+        
         # Create BM25 retriever
         t_bm25_start = time.time()
         texts = [doc.page_content for doc in docs]
@@ -203,6 +253,7 @@ class RetrieverBuilder:
         t_bm25_end = time.time()
         logger.info(f"[PROFILE] BM25 retriever creation: {t_bm25_end - t_bm25_start:.2f}s")
         logger.debug(f"BM25 indexed {len(texts)} texts, k={bm25_retriever.k}")
+        
         t_vec_retr_start = time.time()
         vector_retriever = vector_store.as_retriever(
             search_type="mmr",
@@ -215,6 +266,7 @@ class RetrieverBuilder:
         t_vec_retr_end = time.time()
         logger.info(f"[PROFILE] Vector retriever creation: {t_vec_retr_end - t_vec_retr_start:.2f}s")
         logger.debug("Vector retriever created")
+        
         t_ensemble_start = time.time()
         hybrid_retriever = EnsembleRetriever(
             retrievers=[bm25_retriever, vector_retriever],
@@ -224,5 +276,6 @@ class RetrieverBuilder:
         t_ensemble_end = time.time()
         logger.info(f"[PROFILE] Ensemble retriever creation: {t_ensemble_end - t_ensemble_start:.2f}s")
         logger.info(f"Hybrid retriever created (k={parameters.VECTOR_SEARCH_K})")
-        logger.info(f"[PROFILE] Total hybrid retriever build: {t_ensemble_end - t_bm25_start:.2f}s")
+        logger.info(f"[PROFILE] Total hybrid retriever build: {t_ensemble_end - t_vector_start:.2f}s")
+        
         return hybrid_retriever