rishik1111 commited on
Commit
4bb5758
·
1 Parent(s): 13d81de

feat: Add Qdrant vector database engine adapter with dual FAISS/Qdrant backend support

Browse files
config.py CHANGED
@@ -105,6 +105,10 @@ if _MAX_PASSAGES_ENV is None or _MAX_PASSAGES_ENV.strip() == "":
105
  MAX_INDEX_PASSAGES_PER_LANG = None # explicit: no cap, not a silent fallback
106
  else:
107
  MAX_INDEX_PASSAGES_PER_LANG = int(_MAX_PASSAGES_ENV)
 
 
 
 
108
  HNSW_M = 32
109
  HNSW_EF_CONSTRUCTION = 200
110
  HNSW_EF_SEARCH = 64
 
105
  MAX_INDEX_PASSAGES_PER_LANG = None # explicit: no cap, not a silent fallback
106
  else:
107
  MAX_INDEX_PASSAGES_PER_LANG = int(_MAX_PASSAGES_ENV)
108
+ # Vector Database Engine Selection ("faiss" or "qdrant")
109
+ VECTOR_STORE_BACKEND = os.getenv("VECTOR_STORE_BACKEND", "faiss").lower()
110
+ QDRANT_STORAGE_PATH = Path(os.getenv("QDRANT_STORAGE_PATH", DATA_DIR / "qdrant_db"))
111
+
112
  HNSW_M = 32
113
  HNSW_EF_CONSTRUCTION = 200
114
  HNSW_EF_SEARCH = 64
requirements.txt CHANGED
@@ -7,6 +7,7 @@ transformers>=4.36.0
7
  sentence-transformers>=2.2.0
8
  onnxruntime>=1.16.0
9
  faiss-cpu>=1.7.4
 
10
  rank-bm25>=0.2.2
11
  sarvamai>=0.1.30
12
  datasets>=2.16.0
 
7
  sentence-transformers>=2.2.0
8
  onnxruntime>=1.16.0
9
  faiss-cpu>=1.7.4
10
+ qdrant-client>=1.12.0
11
  rank-bm25>=0.2.2
12
  sarvamai>=0.1.30
13
  datasets>=2.16.0
vector_search/index_faiss.py CHANGED
@@ -454,9 +454,12 @@ class IndexManager:
454
  _INDEX_MANAGER: Optional[IndexManager] = None
455
 
456
 
457
- def get_index_manager() -> IndexManager:
458
- """Singleton getter for IndexManager."""
459
  global _INDEX_MANAGER
 
 
 
460
  if _INDEX_MANAGER is None:
461
  _INDEX_MANAGER = IndexManager()
462
  _INDEX_MANAGER.load_all_indexes()
 
454
  _INDEX_MANAGER: Optional[IndexManager] = None
455
 
456
 
457
+ def get_index_manager() -> Any:
458
+ """Singleton getter for IndexManager (supports FAISS and Qdrant vector backends)."""
459
  global _INDEX_MANAGER
460
+ if getattr(config, "VECTOR_STORE_BACKEND", "faiss").lower() == "qdrant":
461
+ from vector_search.index_qdrant import get_qdrant_manager
462
+ return get_qdrant_manager()
463
  if _INDEX_MANAGER is None:
464
  _INDEX_MANAGER = IndexManager()
465
  _INDEX_MANAGER.load_all_indexes()
vector_search/index_qdrant.py ADDED
@@ -0,0 +1,254 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """
2
+ Qdrant Vector Database Integration Engine.
3
+
4
+ Provides native Qdrant vector indexing and HNSW cosine similarity search:
5
+ - In-memory or local persistent Qdrant engine via `qdrant-client`.
6
+ - Native payload metadata language filtering.
7
+ - Dual compatibility with FAISSIndexManager interface.
8
+ """
9
+
10
+ import json
11
+ import logging
12
+ import os
13
+ import sys
14
+ from pathlib import Path
15
+ from typing import Any, Dict, List, Optional, Tuple
16
+ import numpy as np
17
+
18
+ try:
19
+ from qdrant_client import QdrantClient
20
+ from qdrant_client.models import (
21
+ Distance,
22
+ VectorParams,
23
+ PointStruct,
24
+ Filter,
25
+ FieldCondition,
26
+ MatchValue,
27
+ )
28
+ HAS_QDRANT = True
29
+ except ImportError:
30
+ HAS_QDRANT = False
31
+
32
+ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
33
+ import config
34
+ from doc_chunking.metadata import Chunk
35
+ from vector_search.embed import get_embedder
36
+
37
+ logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", force=True)
38
+ logger = logging.getLogger(__name__)
39
+
40
+
41
+ class StrategyQdrantIndex:
42
+ """
43
+ Manages a Qdrant collection for a specific chunking strategy.
44
+ """
45
+ def __init__(
46
+ self,
47
+ strategy_name: str,
48
+ client: "QdrantClient",
49
+ dim: int = config.EMBEDDING_DIM,
50
+ ):
51
+ if not HAS_QDRANT:
52
+ raise ImportError("qdrant-client is not installed. Run `pip install qdrant-client`")
53
+
54
+ self.strategy_name = strategy_name
55
+ self.client = client
56
+ self.dim = dim
57
+ self.collection_name = f"vector_{strategy_name}"
58
+ self.chunks: List[Chunk] = []
59
+
60
+ # Create or recreate collection in Qdrant
61
+ if not self.client.collection_exists(self.collection_name):
62
+ self.client.create_collection(
63
+ collection_name=self.collection_name,
64
+ vectors_config=VectorParams(size=self.dim, distance=Distance.COSINE),
65
+ )
66
+ logger.info("Created Qdrant collection: '%s' (dim=%d)", self.collection_name, self.dim)
67
+
68
+ @property
69
+ def size(self) -> int:
70
+ """Returns number of vectors in collection."""
71
+ try:
72
+ info = self.client.get_collection(self.collection_name)
73
+ return info.points_count or len(self.chunks)
74
+ except Exception:
75
+ return len(self.chunks)
76
+
77
+ def add_chunks(self, chunks: List[Chunk], vectors: np.ndarray):
78
+ """Add chunks and vector embeddings into Qdrant index."""
79
+ if not chunks or len(chunks) == 0:
80
+ return
81
+ if len(chunks) != len(vectors):
82
+ raise ValueError(f"Chunk count ({len(chunks)}) != vector count ({len(vectors)})")
83
+
84
+ start_idx = len(self.chunks)
85
+ points: List[PointStruct] = []
86
+
87
+ for idx_offset, (chunk, vec) in enumerate(zip(chunks, vectors)):
88
+ global_id = start_idx + idx_offset
89
+ # Unit normalize for cosine distance
90
+ norm = np.linalg.norm(vec)
91
+ unit_vec = (vec / norm).tolist() if norm > 1e-9 else vec.tolist()
92
+
93
+ payload = {
94
+ "chunk_id": chunk.chunk_id,
95
+ "doc_id": chunk.doc_id,
96
+ "text": chunk.text,
97
+ "source_lang": chunk.source_lang.lower(),
98
+ "metadata": chunk.metadata,
99
+ "idx": global_id,
100
+ }
101
+ points.append(PointStruct(id=global_id, vector=unit_vec, payload=payload))
102
+ self.chunks.append(chunk)
103
+
104
+ self.client.upsert(collection_name=self.collection_name, points=points)
105
+ logger.info("Indexed %d vectors into Qdrant collection '%s'", len(points), self.collection_name)
106
+
107
+ def search(
108
+ self,
109
+ query_vector: np.ndarray,
110
+ top_k: int = 5,
111
+ lang_filter: Optional[str] = None,
112
+ ) -> List[Tuple[Chunk, float]]:
113
+ """
114
+ Execute Qdrant cosine vector search with optional payload language filtering.
115
+ """
116
+ if self.size == 0:
117
+ return []
118
+
119
+ norm = np.linalg.norm(query_vector)
120
+ unit_q = (query_vector / norm).tolist() if norm > 1e-9 else query_vector.tolist()
121
+
122
+ qdrant_filter = None
123
+ if lang_filter and lang_filter.strip():
124
+ qdrant_filter = Filter(
125
+ must=[
126
+ FieldCondition(
127
+ key="source_lang",
128
+ match=MatchValue(value=lang_filter.lower().strip()),
129
+ )
130
+ ]
131
+ )
132
+
133
+ if hasattr(self.client, "query_points"):
134
+ search_result = self.client.query_points(
135
+ collection_name=self.collection_name,
136
+ query=unit_q,
137
+ query_filter=qdrant_filter,
138
+ limit=top_k,
139
+ ).points
140
+ else:
141
+ search_result = self.client.search(
142
+ collection_name=self.collection_name,
143
+ query_vector=unit_q,
144
+ query_filter=qdrant_filter,
145
+ limit=top_k,
146
+ )
147
+
148
+ results: List[Tuple[Chunk, float]] = []
149
+ for hit in search_result:
150
+ payload = hit.payload or {}
151
+ idx = payload.get("idx", 0)
152
+
153
+ if 0 <= idx < len(self.chunks):
154
+ chunk = self.chunks[idx]
155
+ else:
156
+ chunk_text = payload.get("text", "")
157
+ chunk = Chunk(
158
+ chunk_id=payload.get("chunk_id", ""),
159
+ doc_id=payload.get("doc_id", ""),
160
+ text=chunk_text,
161
+ embed_text=chunk_text,
162
+ source_lang=payload.get("source_lang", "en"),
163
+ chunk_strategy=self.strategy_name,
164
+ token_count=len(chunk_text.split()),
165
+ metadata=payload.get("metadata", {}),
166
+ )
167
+
168
+ # Qdrant score is cosine similarity [0, 1]
169
+ score = float(hit.score)
170
+ results.append((chunk, score))
171
+
172
+ return results
173
+
174
+
175
+ class QdrantIndexManager:
176
+ """
177
+ Manager for Qdrant Vector Engine across all chunking strategies.
178
+ """
179
+ def __init__(self, in_memory: bool = True, storage_path: Optional[str] = None):
180
+ if not HAS_QDRANT:
181
+ raise ImportError("qdrant-client package is required for Qdrant vector backend.")
182
+
183
+ if in_memory:
184
+ self.client = QdrantClient(":memory:")
185
+ logger.info("Initialized Qdrant In-Memory Vector Storage Engine.")
186
+ else:
187
+ path = storage_path or str(config.DATA_DIR / "qdrant_db")
188
+ os.makedirs(path, exist_ok=True)
189
+ self.client = QdrantClient(path=path)
190
+ logger.info("Initialized Qdrant Persistent Storage Engine at '%s'.", path)
191
+
192
+ self.strategy_indices: Dict[str, StrategyQdrantIndex] = {}
193
+ self.centroids: Dict[str, np.ndarray] = {}
194
+
195
+ def get_or_create_index(self, strategy_name: str) -> StrategyQdrantIndex:
196
+ if strategy_name not in self.strategy_indices:
197
+ self.strategy_indices[strategy_name] = StrategyQdrantIndex(
198
+ strategy_name=strategy_name,
199
+ client=self.client,
200
+ )
201
+ return self.strategy_indices[strategy_name]
202
+
203
+ def search(
204
+ self,
205
+ strategy_name: str,
206
+ query_vector: np.ndarray,
207
+ top_k: int = 5,
208
+ lang_filter: Optional[str] = None,
209
+ ) -> List[Tuple[Chunk, float]]:
210
+ if strategy_name not in self.strategy_indices:
211
+ logger.warning("Strategy '%s' not indexed in Qdrant", strategy_name)
212
+ return []
213
+ return self.strategy_indices[strategy_name].search(
214
+ query_vector=query_vector,
215
+ top_k=top_k,
216
+ lang_filter=lang_filter,
217
+ )
218
+
219
+ def compute_centroids(self):
220
+ """Compute corpus language centroids for guardrails."""
221
+ lang_vectors: Dict[str, List[np.ndarray]] = {}
222
+ embedder = get_embedder()
223
+
224
+ for idx_obj in self.strategy_indices.values():
225
+ for chunk in idx_obj.chunks:
226
+ lang = chunk.source_lang.lower()
227
+ if lang not in lang_vectors:
228
+ lang_vectors[lang] = []
229
+ vec = embedder.embed_text(chunk.text)
230
+ lang_vectors[lang].append(vec)
231
+
232
+ for lang, vecs in lang_vectors.items():
233
+ arr = np.array(vecs)
234
+ mean_v = np.mean(arr, axis=0)
235
+ norm = np.linalg.norm(mean_v)
236
+ self.centroids[lang] = (mean_v / norm) if norm > 1e-9 else mean_v
237
+
238
+ logger.info("Computed Qdrant corpus centroids for languages: %s", list(self.centroids.keys()))
239
+
240
+ def get_centroid(self, lang: str) -> Optional[np.ndarray]:
241
+ return self.centroids.get(lang.lower().strip())
242
+
243
+ def get_all_centroids(self) -> Dict[str, np.ndarray]:
244
+ return self.centroids
245
+
246
+
247
+ _QDRANT_MANAGER_INSTANCE: Optional[QdrantIndexManager] = None
248
+
249
+
250
+ def get_qdrant_manager(in_memory: bool = True) -> QdrantIndexManager:
251
+ global _QDRANT_MANAGER_INSTANCE
252
+ if _QDRANT_MANAGER_INSTANCE is None:
253
+ _QDRANT_MANAGER_INSTANCE = QdrantIndexManager(in_memory=in_memory)
254
+ return _QDRANT_MANAGER_INSTANCE