""" vector_store.py ──────────────────────────────────────────── 벡터 DB 추상화 레이어: FAISS / Qdrant 통합 인터페이스 """ import os import pickle import re import uuid from abc import ABC, abstractmethod from datetime import datetime, timezone from pathlib import Path from typing import List, Dict, Any, Optional, Tuple import numpy as np import faiss # ──────────────────────────────────────────── # 환경 변수 설정 # ──────────────────────────────────────────── VECTOR_STORE = os.getenv("VECTOR_STORE", "faiss").lower() QDRANT_HOST = os.getenv("QDRANT_HOST", "localhost") QDRANT_PORT = int(os.getenv("QDRANT_PORT", "6333")) QDRANT_COLLECTION = os.getenv("QDRANT_COLLECTION", "qa_vectors") HYBRID_SEARCH_ENABLED = os.getenv("HYBRID_SEARCH_ENABLED", "false").lower() == "true" SPARSE_SEARCH_ENABLED = os.getenv("SPARSE_SEARCH_ENABLED", "true").lower() == "true" SPARSE_TOP_K = int(os.getenv("SPARSE_TOP_K", "30")) HYBRID_MERGE_TOP_K = int(os.getenv("HYBRID_MERGE_TOP_K", "40")) # 결정적 point ID 생성을 위한 고정 네임스페이스 (변경 금지: 바꾸면 기존 ID와 불일치) POINT_ID_NAMESPACE = uuid.UUID("b1e7c0de-0000-4a00-8000-71636861746b") def make_point_id(meta: Dict[str, Any]) -> str: """ FAQ 메타데이터로부터 결정적(deterministic) point ID(UUID) 생성. - 질문(q) 텍스트(trim) 기준 → 동일 질문은 항상 같은 ID - reingest·업로드·어드민 추가 시 upsert / 중복 방지 """ q = str(meta.get("q") or meta.get("question") or "").strip() if not q: key = f"rand:{uuid.uuid4()}" else: key = f"q:{q}" return str(uuid.uuid5(POINT_ID_NAMESPACE, key)) def source_created_at_key(value: Any) -> Tuple[int, int, int]: """source_created_at 정렬 키 — 큰 값일수록 최신. 빈/파싱불가 → (0,0,0).""" s = str(value or "").strip() if not s: return (0, 0, 0) m = re.match(r"^(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})", s) if m: return (int(m.group(1)), int(m.group(2)), int(m.group(3))) m = re.match(r"^(\d{4})(\d{2})(\d{2})$", s) if m: return (int(m.group(1)), int(m.group(2)), int(m.group(3))) if re.match(r"^\d{4}$", s): return (int(s), 0, 0) m = re.search(r"\b(20\d{2})\b", s) if m: return (int(m.group(1)), 0, 0) return (0, 0, 0) def source_created_at_cmp(incoming: Any, existing: Any) -> int: """incoming vs existing — 양수면 incoming이 더 최신.""" a, b = source_created_at_key(incoming), source_created_at_key(existing) if a < b: return -1 if a > b: return 1 return 0 def _now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") # ──────────────────────────────────────────── # 추상 인터페이스 # ──────────────────────────────────────────── class VectorStore(ABC): """벡터 DB 추상 인터페이스""" @abstractmethod def add_vectors( self, vectors: np.ndarray, metadatas: List[Dict[str, Any]], skip_if_older: bool = True, ) -> None: """벡터 추가""" pass @abstractmethod def search( self, query_vector: np.ndarray, top_k: int = 30, threshold: Optional[float] = None, filter_dict: Optional[Dict[str, Any]] = None ) -> List[Dict[str, Any]]: """ 벡터 검색 Returns: [{"meta": {...}, "score": float}, ...] """ pass @abstractmethod def save(self, directory: str) -> None: """인덱스 저장""" pass @abstractmethod def load(self, directory: str) -> None: """인덱스 로드""" pass @abstractmethod def count(self) -> int: """벡터 개수""" pass # ── 어드민(조회/삭제)용 선택 기능 ── # 기본은 미지원. 증분 운영이 가능한 스토어(Qdrant)에서만 오버라이드한다. def get_by_id(self, point_id: str) -> Optional[Dict[str, Any]]: raise NotImplementedError("get_by_id는 Qdrant 스토어에서만 지원합니다") def list_points( self, limit: int = 50, offset: Optional[str] = None, filter_dict: Optional[Dict[str, Any]] = None, ) -> Tuple[List[Dict[str, Any]], Optional[str]]: raise NotImplementedError("list_points는 Qdrant 스토어에서만 지원합니다") def delete_by_id(self, point_id: str) -> None: raise NotImplementedError("delete_by_id는 Qdrant 스토어에서만 지원합니다") def delete_by_ids(self, point_ids: List[str]) -> None: raise NotImplementedError("delete_by_ids는 Qdrant 스토어에서만 지원합니다") def keyword_search( self, keyword: str, fields: Tuple[str, ...] = ("q", "a"), filter_dict: Optional[Dict[str, Any]] = None, limit: int = 50, skip: int = 0, batch: int = 256, max_scan: int = 500000, count_total: bool = True, ) -> Tuple[List[Dict[str, Any]], int, bool]: raise NotImplementedError("keyword_search는 Qdrant 스토어에서만 지원합니다") def distinct_payload_values(self, field: str) -> Dict[str, int]: raise NotImplementedError("distinct_payload_values는 Qdrant 스토어에서만 지원합니다") # ──────────────────────────────────────────── # FAISS 구현 # ──────────────────────────────────────────── class FAISSStore(VectorStore): """FAISS 기반 벡터 스토어 (단순/고속)""" def __init__(self): self.index: Optional[faiss.IndexFlatIP] = None self.metadatas: List[Dict[str, Any]] = [] self.dimension: Optional[int] = None print("[FAISSStore] 초기화 완료") def add_vectors( self, vectors: np.ndarray, metadatas: List[Dict[str, Any]], skip_if_older: bool = True, ) -> None: """벡터 추가 (정규화된 벡터 전제)""" if vectors.shape[0] != len(metadatas): raise ValueError(f"벡터 수({vectors.shape[0]})와 메타데이터 수({len(metadatas)}) 불일치") vectors = vectors.astype("float32") # 인덱스 초기화 if self.index is None: self.dimension = vectors.shape[1] self.index = faiss.IndexFlatIP(self.dimension) print(f"[FAISSStore] 새 인덱스 생성: dim={self.dimension}") # 정규화 확인 (IndexFlatIP는 정규화된 벡터 필요) norms = np.linalg.norm(vectors, axis=1) if not np.allclose(norms, 1.0, atol=1e-5): print(f"[FAISSStore] ⚠️ 정규화되지 않은 벡터 감지 (norm 범위: [{norms.min():.4f}, {norms.max():.4f}])") faiss.normalize_L2(vectors) print("[FAISSStore] 자동 정규화 완료") # 추가 self.index.add(vectors) self.metadatas.extend(metadatas) print(f"[FAISSStore] {len(metadatas)}개 벡터 추가 (총 {self.count()}개)") def search( self, query_vector: np.ndarray, top_k: int = 30, threshold: Optional[float] = None, filter_dict: Optional[Dict[str, Any]] = None ) -> List[Dict[str, Any]]: """벡터 검색""" if self.index is None or self.count() == 0: return [] # 쿼리 벡터 정규화 query_vector = query_vector.astype("float32") if query_vector.ndim == 1: query_vector = query_vector[np.newaxis, :] norm = np.linalg.norm(query_vector) if not np.isclose(norm, 1.0, atol=1e-5): faiss.normalize_L2(query_vector) # 검색 D, I = self.index.search(query_vector, k=min(top_k, self.count())) # 결과 조합 results = [] for score, idx in zip(D[0], I[0]): if idx == -1: # padding continue if threshold is not None and score < threshold: continue results.append({ "meta": self.metadatas[idx], "score": float(score) }) # 메타데이터 필터링 (FAISS는 사후 필터링) if filter_dict: results = [ r for r in results if all(r["meta"].get(k) == v for k, v in filter_dict.items()) ] return results def save(self, directory: str) -> None: """인덱스 및 메타데이터 저장""" directory = Path(directory) directory.mkdir(parents=True, exist_ok=True) index_path = directory / "qa.index" meta_path = directory / "qa_meta.pkl" if self.index is None: raise ValueError("저장할 인덱스가 없습니다") faiss.write_index(self.index, str(index_path)) with open(meta_path, "wb") as f: pickle.dump(self.metadatas, f) print(f"[FAISSStore] 저장 완료: {index_path} ({self.count()}개 벡터)") def load(self, directory: str) -> None: """인덱스 및 메타데이터 로드""" directory = Path(directory) index_path = directory / "qa.index" meta_path = directory / "qa_meta.pkl" if not index_path.exists(): raise FileNotFoundError(f"인덱스 파일 없음: {index_path}") if not meta_path.exists(): raise FileNotFoundError(f"메타데이터 파일 없음: {meta_path}") self.index = faiss.read_index(str(index_path)) with open(meta_path, "rb") as f: self.metadatas = pickle.load(f) self.dimension = self.index.d print(f"[FAISSStore] 로드 완료: {index_path} ({self.count()}개 벡터, dim={self.dimension})") def count(self) -> int: """벡터 개수""" return self.index.ntotal if self.index else 0 # ──────────────────────────────────────────── # Qdrant 구현 (확장용) # ──────────────────────────────────────────── class QdrantStore(VectorStore): """Qdrant 기반 벡터 스토어 (확장 가능)""" def __init__( self, host: str = QDRANT_HOST, port: int = QDRANT_PORT, collection_name: str = QDRANT_COLLECTION ): try: from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct from qdrant_client.models import ( SparseIndexParams, SparseVector, SparseVectorParams, ) except ImportError: raise ImportError( "Qdrant를 사용하려면 qdrant-client 설치 필요: pip install qdrant-client" ) self.client = QdrantClient(host=host, port=port) self.collection_name = collection_name self.dimension: Optional[int] = None self._PointStruct = PointStruct self._Distance = Distance self._VectorParams = VectorParams self._SparseVector = SparseVector self._SparseVectorParams = SparseVectorParams self._SparseIndexParams = SparseIndexParams self.hybrid_enabled = HYBRID_SEARCH_ENABLED or collection_name.endswith("_v2") self.sparse_search_enabled = SPARSE_SEARCH_ENABLED print( f"[QdrantStore] 초기화: {host}:{port}, collection={collection_name}, " f"hybrid={self.hybrid_enabled}" ) def _ensure_collection(self, dimension: int) -> None: """컬렉션 생성 (존재하지 않을 경우)""" collections = self.client.get_collections().collections exists = any(c.name == self.collection_name for c in collections) if not exists: if self.hybrid_enabled: self.client.create_collection( collection_name=self.collection_name, vectors_config={ "dense": self._VectorParams( size=dimension, distance=self._Distance.COSINE, ) }, sparse_vectors_config={ "sparse": self._SparseVectorParams( index=self._SparseIndexParams(on_disk=False) ) }, ) print( f"[QdrantStore] hybrid 컬렉션 생성: " f"{self.collection_name} (dense_dim={dimension}, sparse=sparse)" ) else: self.client.create_collection( collection_name=self.collection_name, vectors_config=self._VectorParams( size=dimension, distance=self._Distance.COSINE # 정규화된 벡터 + 내적 = 코사인 ) ) print(f"[QdrantStore] 컬렉션 생성: {self.collection_name} (dim={dimension})") else: print(f"[QdrantStore] 기존 컬렉션 사용: {self.collection_name}") def find_by_exact_q(self, question: str) -> Optional[Dict[str, Any]]: """payload.q 정확일치로 단건 조회 (레거시 source_id 기반 ID 마이그레이션용).""" q = str(question or "").strip() if not q: return None try: from qdrant_client.models import Filter, FieldCondition, MatchValue flt = Filter(must=[FieldCondition(key="q", match=MatchValue(value=q))]) points, _ = self.client.scroll( collection_name=self.collection_name, limit=1, with_payload=True, with_vectors=False, scroll_filter=flt, ) except Exception as e: print(f"[QdrantStore] find_by_exact_q 실패: {e}") return None if not points: return None p = points[0] return {"id": p.id, "meta": p.payload} def upsert_vectors( self, vectors: np.ndarray, metadatas: List[Dict[str, Any]], skip_if_older: bool = True, ) -> Dict[str, Any]: """ 벡터 upsert. 동일 q는 q 기반 ID로 덮어쓰기. skip_if_older=True 이면 기존 source_created_at이 더 최신이면 스킵. Returns: inserted, updated, skipped, last_id, last_action, last_skip_reason """ if vectors.shape[0] != len(metadatas): raise ValueError(f"벡터 수({vectors.shape[0]})와 메타데이터 수({len(metadatas)}) 불일치") vectors = vectors.astype("float32") self.dimension = vectors.shape[1] self._ensure_collection(self.dimension) point_ids = [make_point_id(m) for m in metadatas] existing_by_id: Dict[Any, Any] = {} try: records = self.client.retrieve( collection_name=self.collection_name, ids=point_ids, with_payload=True, with_vectors=False, ) existing_by_id = {r.id: r for r in records} except Exception as e: print(f"[QdrantStore] batch retrieve 실패: {e}") points = [] legacy_delete_ids: List[str] = [] inserted = updated = skipped = 0 last_id: Optional[str] = None last_action: Optional[str] = None last_skip_reason: Optional[str] = None for vec, meta, point_id in zip(vectors, metadatas, point_ids): incoming = dict(meta) q = str(incoming.get("q") or incoming.get("question") or "").strip() if q: incoming["q"] = q incoming.pop("question", None) record = existing_by_id.get(point_id) existing_item: Optional[Dict[str, Any]] = None if record is not None: existing_item = {"id": record.id, "meta": record.payload or {}} elif q: existing_item = self.find_by_exact_q(q) existing_meta = (existing_item or {}).get("meta") or {} legacy_id = existing_item.get("id") if existing_item else None if skip_if_older and existing_meta: incoming_date = incoming.get("source_created_at") existing_date = existing_meta.get("source_created_at") if source_created_at_cmp(incoming_date, existing_date) < 0: skipped += 1 last_id = str(legacy_id or point_id) last_action = "skipped" last_skip_reason = "older_source_created_at" continue now = _now_iso() if existing_meta: merged = dict(existing_meta) merged.update(incoming) merged["indexed_at"] = existing_meta.get("indexed_at") or now merged["updated_at"] = now payload = merged updated += 1 last_action = "updated" else: payload = dict(incoming) payload.setdefault("indexed_at", now) payload.setdefault("updated_at", now) inserted += 1 last_action = "inserted" payload["point_id"] = point_id if self.hybrid_enabled: from sparse_encoder import encode_document sparse = encode_document(payload) vector = { "dense": vec.tolist(), "sparse": self._SparseVector( indices=sparse["indices"], values=sparse["values"], ), } else: vector = vec.tolist() points.append( self._PointStruct( id=point_id, vector=vector, payload=payload, ) ) last_id = point_id if legacy_id is not None and str(legacy_id) != str(point_id): legacy_delete_ids.append(str(legacy_id)) if points: self.client.upsert(collection_name=self.collection_name, points=points) if legacy_delete_ids: self.delete_by_ids(list(dict.fromkeys(legacy_delete_ids))) if inserted or updated: print( f"[QdrantStore] upsert 완료: +{inserted} 갱신 {updated} 스킵 {skipped} " f"(총 {self.count()}개)" ) elif skipped: print(f"[QdrantStore] {skipped}건 스킵 (기존 source_created_at이 더 최신)") return { "inserted": inserted, "updated": updated, "skipped": skipped, "last_id": last_id, "last_action": last_action, "last_skip_reason": last_skip_reason, } def add_vectors( self, vectors: np.ndarray, metadatas: List[Dict[str, Any]], skip_if_older: bool = True, ) -> None: """벡터 upsert (Qdrant).""" self.upsert_vectors(vectors, metadatas, skip_if_older=skip_if_older) def search( self, query_vector: np.ndarray, top_k: int = 30, threshold: Optional[float] = None, filter_dict: Optional[Dict[str, Any]] = None ) -> List[Dict[str, Any]]: """벡터 검색""" query_vector = query_vector.astype("float32") if query_vector.ndim == 2: query_vector = query_vector[0] # Qdrant 필터 구성 query_filter = None if filter_dict: from qdrant_client.models import Filter, FieldCondition, MatchValue conditions = [ FieldCondition(key=k, match=MatchValue(value=v)) for k, v in filter_dict.items() ] query_filter = Filter(must=conditions) query_kwargs = {"using": "dense"} if self.hybrid_enabled else {} # 검색 (qdrant-client 1.10+ 권장 API: query_points) response = self.client.query_points( collection_name=self.collection_name, query=query_vector.tolist(), limit=top_k, query_filter=query_filter, score_threshold=threshold, with_payload=True, **query_kwargs, ) # 결과 변환 (point id 포함 → 어드민에서 삭제 등에 사용) return [ { "id": r.id, "meta": r.payload, "score": r.score, "source": "dense", "dense_score": r.score, } for r in response.points ] def sparse_search( self, query_text: str, top_k: int = SPARSE_TOP_K, filter_dict: Optional[Dict[str, Any]] = None, ) -> List[Dict[str, Any]]: """Qdrant sparse vector 검색 (hybrid collection 전용).""" if not self.hybrid_enabled or not self.sparse_search_enabled: return [] from sparse_encoder import encode_query sparse = encode_query(query_text) if not sparse["indices"]: return [] query_filter = None if filter_dict: from qdrant_client.models import Filter, FieldCondition, MatchValue conditions = [ FieldCondition(key=k, match=MatchValue(value=v)) for k, v in filter_dict.items() ] query_filter = Filter(must=conditions) response = self.client.query_points( collection_name=self.collection_name, query=self._SparseVector( indices=sparse["indices"], values=sparse["values"], ), using="sparse", limit=top_k, query_filter=query_filter, with_payload=True, ) return [ { "id": r.id, "meta": r.payload, "score": r.score, "source": "sparse", "sparse_score": r.score, } for r in response.points ] def hybrid_search( self, query_vector: np.ndarray, query_text: str, top_k: int = 30, threshold: Optional[float] = None, filter_dict: Optional[Dict[str, Any]] = None, sparse_top_k: int = SPARSE_TOP_K, merge_top_k: int = HYBRID_MERGE_TOP_K, ) -> List[Dict[str, Any]]: """Dense + sparse 후보를 병합해 reranker 후보로 반환.""" dense_results = self.search( query_vector, top_k=top_k, threshold=threshold, filter_dict=filter_dict, ) sparse_results = self.sparse_search( query_text, top_k=sparse_top_k, filter_dict=filter_dict, ) merged: Dict[str, Dict[str, Any]] = {} for result in dense_results + sparse_results: key = str(result.get("id") or make_point_id(result.get("meta") or {})) existing = merged.get(key) if not existing: item = dict(result) meta = dict(item.get("meta") or {}) item["meta"] = meta item["sources"] = {item.get("source")} merged[key] = item continue existing["sources"].add(result.get("source")) if result.get("dense_score") is not None: existing["dense_score"] = result.get("dense_score") if result.get("sparse_score") is not None: existing["sparse_score"] = result.get("sparse_score") existing["score"] = max( float(existing.get("score") or 0.0), float(result.get("score") or 0.0), ) items = [] for item in merged.values(): sources = {s for s in item.pop("sources", set()) if s} source = "both" if len(sources) > 1 else next(iter(sources), "unknown") item["source"] = source item["meta"]["_search_source"] = source item["meta"]["_dense_score"] = item.get("dense_score") item["meta"]["_sparse_score"] = item.get("sparse_score") items.append(item) def sort_key(item: Dict[str, Any]) -> Tuple[int, float]: source_bonus = {"both": 2, "sparse": 1, "dense": 0}.get(item.get("source"), 0) return (source_bonus, float(item.get("score") or 0.0)) items.sort(key=sort_key, reverse=True) return items[:merge_top_k] def save(self, directory: str) -> None: """Qdrant는 자체 영속성 지원 (별도 저장 불필요)""" print(f"[QdrantStore] Qdrant는 자동으로 영속화됩니다 (스냅샷 경로: {directory} 무시)") def load(self, directory: str) -> None: """Qdrant는 자체 영속성 지원 (별도 로드 불필요)""" # 컬렉션 존재 확인만 수행 try: info = self.client.get_collection(self.collection_name) vectors = info.config.params.vectors if isinstance(vectors, dict): dense = vectors.get("dense") or next(iter(vectors.values())) self.dimension = getattr(dense, "size", None) else: self.dimension = getattr(vectors, "size", None) print(f"[QdrantStore] 컬렉션 연결: {self.collection_name} ({self.count()}개 벡터)") except Exception as e: print(f"[QdrantStore] ⚠️ 컬렉션 없음: {e}") def count(self) -> int: """벡터 개수 (정확 카운트)""" try: return self.client.count( collection_name=self.collection_name, exact=True ).count except Exception: return 0 # ── 어드민(조회/삭제)용 기능 ── def get_by_id(self, point_id: str) -> Optional[Dict[str, Any]]: """point ID로 단건 조회""" try: records = self.client.retrieve( collection_name=self.collection_name, ids=[point_id], with_payload=True, with_vectors=False, ) except Exception as e: print(f"[QdrantStore] get_by_id 실패: {e}") return None if not records: return None r = records[0] return {"id": r.id, "meta": r.payload} def _build_filter(self, filter_dict: Optional[Dict[str, Any]]): """payload 정확일치 필터 생성 (None/빈값은 무시). category=__EMPTY__ → 미분류""" if not filter_dict: return None from qdrant_client.models import Filter, FieldCondition, MatchValue category = filter_dict.get("category") if isinstance(category, str): category = category.strip() if category in ("__EMPTY__", "미분류"): should = [FieldCondition(key="category", match=MatchValue(value=""))] try: from qdrant_client.models import IsEmptyCondition, IsNullCondition, PayloadField field = PayloadField(key="category") should.append(IsEmptyCondition(is_empty=field)) should.append(IsNullCondition(is_null=field)) except ImportError: try: from qdrant_client.models import IsEmpty should.append(IsEmpty(key="category")) except ImportError: pass return Filter(should=should) conditions = [ FieldCondition(key=k, match=MatchValue(value=v)) for k, v in filter_dict.items() if k != "category" and v is not None and str(v) != "" ] if category is not None and str(category).strip() != "": conditions.append( FieldCondition(key="category", match=MatchValue(value=str(category))) ) return Filter(must=conditions) if conditions else None def ensure_admin_payload_indexes(self) -> None: """어드민 목록 필터/order_by용 payload index (없으면 생성, 실패는 무시).""" if not hasattr(self.client, "create_payload_index"): return try: from qdrant_client.models import PayloadSchemaType for field, schema in ( ("category", PayloadSchemaType.KEYWORD), ("source", PayloadSchemaType.KEYWORD), ("updated_at", PayloadSchemaType.DATETIME), ): try: self.client.create_payload_index( collection_name=self.collection_name, field_name=field, field_schema=schema, ) print(f"[QdrantStore] payload index ensured: {field}") except Exception as e: msg = str(e).lower() if "already exists" in msg or "already exist" in msg: continue print(f"[QdrantStore] payload index {field} skip: {e}") except Exception as e: print(f"[QdrantStore] ensure_admin_payload_indexes 실패(무시): {e}") def _scroll_collect( self, size: int, scroll_offset, filter_dict: Optional[Dict[str, Any]], ) -> Tuple[List[Dict[str, Any]], Optional[Any]]: """ scroll로 size개 point 수집. payload가 크면 Qdrant가 limit보다 적게 반환할 수 있어 배치를 반복한다. """ flt = self._build_filter(filter_dict) collected: List[Dict[str, Any]] = [] offset = scroll_offset while len(collected) < size: batch = min(size - len(collected), 64) try: points, offset = self.client.scroll( collection_name=self.collection_name, limit=batch, offset=offset, with_payload=True, with_vectors=False, scroll_filter=flt, ) except Exception as e: print(f"[QdrantStore] scroll collect 실패: {e}") break if not points: break for p in points: collected.append({"id": p.id, "meta": p.payload}) if len(collected) >= size: break if offset is None: break return collected, offset def _scroll_skip( self, skip: int, filter_dict: Optional[Dict[str, Any]], scroll_offset=None, ) -> Tuple[int, Optional[Any]]: """payload 없이 scroll하며 skip 건수만큼 건너뛴다.""" flt = self._build_filter(filter_dict) skipped = 0 offset = scroll_offset while skipped < skip: batch = min(256, skip - skipped) try: points, offset = self.client.scroll( collection_name=self.collection_name, limit=batch, offset=offset, with_payload=False, with_vectors=False, scroll_filter=flt, ) except Exception as e: print(f"[QdrantStore] scroll skip 실패: {e}") return skipped, offset if not points: break skipped += len(points) if offset is None: break return skipped, offset def list_points( self, limit: int = 50, offset: Optional[str] = None, filter_dict: Optional[Dict[str, Any]] = None, ) -> Tuple[List[Dict[str, Any]], Optional[str]]: """ point 페이지네이션 조회 (scroll). filter_dict로 category/source 등 정확일치 필터 가능. Returns: (항목 리스트, 다음 페이지 offset). offset이 None이면 마지막 페이지. """ limit = max(1, min(limit, 200)) items, next_offset = self._scroll_collect(limit, offset, filter_dict) return items, next_offset def count_filtered(self, filter_dict: Optional[Dict[str, Any]] = None) -> int: """필터 조건에 맞는 벡터 정확 건수""" try: flt = self._build_filter(filter_dict) return self.client.count( collection_name=self.collection_name, count_filter=flt, exact=True, ).count except Exception as e: print(f"[QdrantStore] count_filtered 실패: {e}") return 0 def _scroll_batch( self, limit: int, scroll_offset, filter_dict: Optional[Dict[str, Any]], with_payload: bool, order_by=None, ): flt = self._build_filter(filter_dict) kwargs = dict( collection_name=self.collection_name, limit=limit, offset=scroll_offset, with_payload=with_payload, with_vectors=False, scroll_filter=flt, ) if order_by is not None: kwargs["order_by"] = order_by return self.client.scroll(**kwargs) def _scroll_skip_and_collect( self, skip: int, size: int, filter_dict: Optional[Dict[str, Any]], order_by, ) -> List[Any]: """scroll로 skip 건수만큼 건너뛴 뒤 size 만큼 point 반환.""" scroll_offset = None skipped = 0 while skipped < skip: batch = min(256, skip - skipped) points, scroll_offset = self._scroll_batch( batch, scroll_offset, filter_dict, False, order_by ) if not points: return [] skipped += len(points) if scroll_offset is None: return [] points, _ = self._scroll_batch( size, scroll_offset, filter_dict, True, order_by ) return points or [] def list_points_page( self, page: int, size: int, filter_dict: Optional[Dict[str, Any]] = None, ) -> Tuple[List[Dict[str, Any]], int]: """ 페이지 번호 기반 목록. list_points와 동일 scroll 경로 사용 + 페이지 내 updated_at 정렬. """ page = max(0, page) size = max(1, min(size, 200)) total = self.count_filtered(filter_dict) if total == 0 or page * size >= total: return [], total skip = page * size skipped, offset = self._scroll_skip(skip, filter_dict) if skip > 0 and skipped < skip: return [], total items, _ = self._scroll_collect(size, offset, filter_dict) if items: items.sort( key=lambda x: (x.get("meta") or {}).get("updated_at") or "", reverse=True, ) return items, total def delete_by_id(self, point_id: str) -> None: """point ID로 단건 삭제""" self.delete_by_ids([point_id]) def delete_by_ids(self, point_ids: List[str]) -> None: """point ID 목록으로 일괄 삭제""" if not point_ids: return from qdrant_client.models import PointIdsList self.client.delete( collection_name=self.collection_name, points_selector=PointIdsList(points=list(point_ids)), ) print(f"[QdrantStore] {len(point_ids)}건 삭제") def keyword_search( self, keyword: str, fields: Tuple[str, ...] = ("q", "a"), filter_dict: Optional[Dict[str, Any]] = None, limit: int = 50, skip: int = 0, batch: int = 256, max_scan: int = 500000, count_total: bool = True, ) -> Tuple[List[Dict[str, Any]], int, bool]: """ DB 전체 대상 문자열 포함(부분일치, 대소문자 무시) 검색. Returns: (matches, total_matches, exhausted) - matches: skip 이후 limit 개 - total_matches: 전체 매칭 건수. count_total=False이면 조기 중단 시점까지의 매칭 수 - exhausted: 전체 스캔 완료 여부 """ kw = (keyword or "").lower().strip() if not kw: return [], 0, True skip = max(0, skip) flt = self._build_filter(filter_dict) page_items: List[Dict[str, Any]] = [] offset = None scanned = 0 total_matches = 0 exhausted = True while True: try: points, offset = self.client.scroll( collection_name=self.collection_name, limit=batch, offset=offset, with_payload=True, with_vectors=False, scroll_filter=flt, ) except Exception as e: print(f"[QdrantStore] keyword_search 실패: {e}") break if not points: break for p in points: scanned += 1 payload = p.payload or {} parts = [] for field in fields: if field == "question": parts.append(str(payload.get("question") or payload.get("q") or "")) elif field == "answer": parts.append(str(payload.get("answer") or payload.get("a") or "")) else: parts.append(str(payload.get(field) or "")) if kw in " ".join(parts).lower(): if total_matches >= skip and len(page_items) < limit: page_items.append({"id": p.id, "meta": payload}) total_matches += 1 if not count_total and len(page_items) >= limit: exhausted = False return page_items, total_matches, exhausted if offset is None: break if scanned >= max_scan: exhausted = False break return page_items, total_matches, exhausted def distinct_payload_values( self, field: str, batch: int = 512, max_scan: int = 100000 ) -> Dict[str, int]: """payload 특정 필드의 distinct 값과 개수 (예: category 목록). 전체 스캔.""" counts: Dict[str, int] = {} offset = None scanned = 0 while True: try: points, offset = self.client.scroll( collection_name=self.collection_name, limit=batch, offset=offset, with_payload=True, with_vectors=False, ) except Exception as e: print(f"[QdrantStore] distinct_payload_values 실패: {e}") break if not points: break for p in points: scanned += 1 value = (p.payload or {}).get(field) if value is None or str(value).strip() == "": if field == "category": counts["미분류"] = counts.get("미분류", 0) + 1 else: key = str(value) counts[key] = counts.get(key, 0) + 1 if offset is None or scanned >= max_scan: break return counts # ──────────────────────────────────────────── # 팩토리 함수 # ──────────────────────────────────────────── def get_vector_store(store_type: Optional[str] = None) -> VectorStore: """ 환경 변수 기반 벡터 스토어 생성 Args: store_type: "faiss" or "qdrant" (None이면 환경변수 VECTOR_STORE 사용) Returns: VectorStore 인스턴스 """ store_type = (store_type or VECTOR_STORE).lower() if store_type == "faiss": return FAISSStore() elif store_type == "qdrant": return QdrantStore() else: raise ValueError(f"지원하지 않는 벡터 스토어: {store_type} (faiss 또는 qdrant 사용)") if __name__ == "__main__": # 간단한 테스트 print("=== 벡터 스토어 테스트 ===") print(f"선택된 벡터 스토어: {VECTOR_STORE}") store = get_vector_store() print(f"현재 벡터 수: {store.count()}")