feat: Add dynamic Qdrant Cloud vector search integration with payload indexing and candidate fusion

api_endpoints/main.py

@@ -26,7 +26,7 @@ import config
 from rag_pipeline.orchestrator import get_orchestrator, RAGPipelineOrchestrator
 from rag_pipeline.schemas import QueryRequest, QueryResponse
 from vector_search.embed import get_embedder
-from vector_search.index_faiss import get_index_manager
+from vector_search import get_index_manager
 
 logger = logging.getLogger(__name__)
 

build_vector_indexes.py

@@ -10,7 +10,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent))
 import config
 from knowledge_base.tech_data import EN_TECH_PASSAGES, HI_TECH_PASSAGES, MR_TECH_PASSAGES, TECH_LONGDOCS
 from knowledge_base.movie_data import EN_MOVIE_PASSAGES, HI_MOVIE_PASSAGES, MR_MOVIE_PASSAGES, MOVIE_LONGDOCS
-from vector_search.index_faiss import get_index_manager
+from vector_search import get_index_manager
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", force=True)
 logger = logging.getLogger(__name__)

main.py

@@ -55,7 +55,7 @@ import config
 from rag_pipeline.orchestrator import get_orchestrator
 from rag_pipeline.schemas import QueryRequest, QueryResponse
 from vector_search.embed import get_embedder
-from vector_search.index_faiss import get_index_manager
+from vector_search import get_index_manager
 
 
 # Read the full custom HTML Command Center UI

rag_pipeline/orchestrator.py

@@ -25,7 +25,7 @@ from rag_pipeline.schemas import (
     StageTiming,
 )
 from vector_search.embed import get_embedder
-from vector_search.index_faiss import get_index_manager
+from vector_search import get_index_manager
 from vector_search.rerank import rerank_bm25_hybrid, rerank_cross_encoder, get_cross_encoder
 from doc_chunking.hybrid_merge import merge_and_fuse_candidates
 from safety_guardrails.pre_retrieval import check_unsafe_content, check_off_topic_query, check_query_intent, get_safety_telemetry

rag_service/retriever.py

@@ -19,7 +19,7 @@ if str(_ROOT_DIR) not in sys.path:
 
 import config
 from vector_search.embed import get_embedder
-from vector_search.index_faiss import get_index_manager
+from vector_search import get_index_manager
 
 logger = logging.getLogger(__name__)
 

vector_search/__init__.py

@@ -0,0 +1,18 @@
+import logging
+import config
+
+logger = logging.getLogger(__name__)
+
+
+def get_index_manager():
+    """
+    Dynamically returns the active vector index manager (Qdrant or FAISS)
+    based on `config.VECTOR_STORE_BACKEND`.
+    """
+    backend = getattr(config, "VECTOR_STORE_BACKEND", "faiss").lower()
+    if backend == "qdrant":
+        from vector_search.index_qdrant import get_qdrant_manager
+        return get_qdrant_manager()
+    else:
+        from vector_search.index_faiss import get_index_manager as get_faiss_manager
+        return get_faiss_manager()

vector_search/index_qdrant.py

@@ -24,6 +24,7 @@ try:
         Filter,
         FieldCondition,
         MatchValue,
+        PayloadSchemaType,
     )
     HAS_QDRANT = True
 except ImportError:
@@ -68,6 +69,16 @@ class StrategyQdrantIndex:
             )
             logger.info("Created Qdrant collection: '%s' (dim=%d)", self.collection_name, self.dim)
 
+        try:
+            self.client.create_payload_index(
+                collection_name=self.collection_name,
+                field_name="source_lang",
+                field_schema=PayloadSchemaType.KEYWORD,
+            )
+            logger.info("Ensured 'source_lang' keyword payload index for '%s'", self.collection_name)
+        except Exception as e:
+            logger.debug("Note creating payload index for '%s': %s", self.collection_name, e)
+
     @property
     def size(self) -> int:
         """Returns number of vectors in collection."""
@@ -109,18 +120,29 @@ class StrategyQdrantIndex:
 
     def search(
         self,
-        query_vector: np.ndarray,
-        top_k: int = 5,
-        lang_filter: Optional[str] = None,
-    ) -> List[Tuple[Chunk, float]]:
+        query_vec: np.ndarray,
+        target_lang: Optional[str] = None,
+        top_k: int = 15,
+        **kwargs,
+    ) -> List[Dict[str, Any]]:
         """
         Execute Qdrant cosine vector search with optional payload language filtering.
+        Returns candidate dicts identical to FAISS StrategyVectorIndex.search interface.
         """
-        if self.size == 0:
+        q_vec = query_vec if query_vec is not None else kwargs.get("query_vector")
+        lang_filter = target_lang if target_lang is not None else kwargs.get("lang_filter")
+        
+        if q_vec is None or self.size == 0:
             return []
 
-        norm = np.linalg.norm(query_vector)
-        unit_q = (query_vector / norm).tolist() if norm > 1e-9 else query_vector.tolist()
+        # Ensure query vector is 1D float array (384,) to prevent multi-vector 400 error in Qdrant
+        if isinstance(q_vec, np.ndarray):
+            q_vec = q_vec.reshape(-1)
+        elif isinstance(q_vec, list) and len(q_vec) > 0 and isinstance(q_vec[0], list):
+            q_vec = q_vec[0]
+
+        norm = np.linalg.norm(q_vec)
+        unit_q = (np.array(q_vec) / (norm + 1e-12)).astype(float).tolist()
 
         qdrant_filter = None
         if lang_filter and lang_filter.strip():
@@ -133,22 +155,42 @@ class StrategyQdrantIndex:
                 ]
             )
 
-        if hasattr(self.client, "query_points"):
-            search_result = self.client.query_points(
-                collection_name=self.collection_name,
-                query=unit_q,
-                query_filter=qdrant_filter,
-                limit=top_k,
-            ).points
-        else:
-            search_result = self.client.search(
-                collection_name=self.collection_name,
-                query_vector=unit_q,
-                query_filter=qdrant_filter,
-                limit=top_k,
-            )
+        search_result = []
+        try:
+            if hasattr(self.client, "query_points"):
+                search_result = self.client.query_points(
+                    collection_name=self.collection_name,
+                    query=unit_q,
+                    query_filter=qdrant_filter,
+                    limit=top_k,
+                ).points
+            else:
+                search_result = self.client.search(
+                    collection_name=self.collection_name,
+                    query_vector=unit_q,
+                    query_filter=qdrant_filter,
+                    limit=top_k,
+                )
+        except Exception as e:
+            logger.warning("Filtered Qdrant query encounter (%s); executing fallback query...", e)
+            try:
+                if hasattr(self.client, "query_points"):
+                    search_result = self.client.query_points(
+                        collection_name=self.collection_name,
+                        query=unit_q,
+                        limit=top_k,
+                    ).points
+                else:
+                    search_result = self.client.search(
+                        collection_name=self.collection_name,
+                        query_vector=unit_q,
+                        limit=top_k,
+                    )
+            except Exception as e2:
+                logger.error("Qdrant query execution error: %s", e2)
+                return []
 
-        results: List[Tuple[Chunk, float]] = []
+        results: List[Dict[str, Any]] = []
         for hit in search_result:
             payload = hit.payload or {}
             idx = payload.get("idx", 0)
@@ -168,9 +210,18 @@ class StrategyQdrantIndex:
                     metadata=payload.get("metadata", {}),
                 )
             
-            # Qdrant score is cosine similarity [0, 1]
             score = float(hit.score)
-            results.append((chunk, score))
+            results.append({
+                "chunk_id": chunk.chunk_id,
+                "text": chunk.text,
+                "score": score,
+                "source_lang": chunk.source_lang,
+                "chunk_strategy": chunk.chunk_strategy,
+                "source_query_ids": getattr(chunk, "source_query_ids", []),
+                "doc_id": chunk.doc_id,
+                "context_window": getattr(chunk, "context_window", None),
+                "metadata": chunk.metadata,
+            })
 
         return results
 
@@ -200,6 +251,7 @@ class QdrantIndexManager:
 
         self.strategy_indices: Dict[str, StrategyQdrantIndex] = {}
         self.centroids: Dict[str, np.ndarray] = {}
+        self.load_all_indexes()
 
     @property
     def indexes(self) -> Dict[str, StrategyQdrantIndex]:
@@ -227,17 +279,19 @@ class QdrantIndexManager:
     def search(
         self,
         strategy_name: str,
-        query_vector: np.ndarray,
-        top_k: int = 5,
-        lang_filter: Optional[str] = None,
-    ) -> List[Tuple[Chunk, float]]:
+        query_vec: np.ndarray,
+        target_lang: Optional[str] = None,
+        top_k: int = 15,
+        **kwargs,
+    ) -> List[Dict[str, Any]]:
         if strategy_name not in self.strategy_indices:
             logger.warning("Strategy '%s' not indexed in Qdrant", strategy_name)
             return []
         return self.strategy_indices[strategy_name].search(
-            query_vector=query_vector,
+            query_vec=query_vec,
+            target_lang=target_lang,
             top_k=top_k,
-            lang_filter=lang_filter,
+            **kwargs,
         )
 
     def compute_centroids(self):
@@ -316,7 +370,9 @@ class QdrantIndexManager:
 
     def load_all_indexes(self):
         """Auto-recovers or populates Qdrant Cloud indexes."""
-        self.build_all_indexes()
+        for strat in ["passage_native", "semantic_longdoc"]:
+            self.get_or_create_index(strat)
+        logger.info("[QdrantManager] Loaded active strategy indexes: %s", list(self.strategy_indices.keys()))
 
 
 _QDRANT_MANAGER_INSTANCE: Optional[QdrantIndexManager] = None