Macbook
4b86b2a660
- server-dev start/stop/deploy 및 Gitea push 자동 배포 - local-dev 로컬 개발 환경 Co-authored-by: Cursor <cursoragent@cursor.com>
1132 lines
41 KiB
Python
1132 lines
41 KiB
Python
"""
|
|
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()}")
|
|
|