근사 검색 알고리즘, HNSW (ANN & FAISS)

import faiss
import numpy as np
import torch

def understand_faiss_hnsw(query_tensor, db_tensors):
    """
    차원(d)이 512인 PyTorch 벡터들을 FAISS HNSW 인덱스에 넣고 검색합니다.
    """
    # 1. PyTorch Tensor -> Numpy Array 변환 (FAISS 필수 조건)
    # CPU 메모리로 옮기고 연속적인 float32 배열로 캐스팅합니다.
    db_vectors = db_tensors.cpu().numpy().astype('float32')
    query_vector = query_tensor.cpu().numpy().astype('float32')
    
    d = db_vectors.shape[1] # 벡터 차원 수 (예: 512)
    
    # 2. HNSW 인덱스 생성
    # 32는 노드당 연결할 최대 이웃의 수(M)를 의미합니다. (파라미터 튜닝 요소)
    index = faiss.IndexHNSWFlat(d, 32)
    
    # 거리 측정 방식을 코사인 유사도와 동일한 효과를 내는 L2 거리(내적)로 설정
    index.hnsw.efConstruction = 40 # 색인 시 탐색 깊이 (정확도 vs 빌드 속도)
    
    # 3. 데이터 색인 (메모리에 그래프 빌드)
    index.add(db_vectors)
    print(f"HNSW 그래프에 총 {index.ntotal}개의 벡터 색인 완료")
    
    # 4. 검색 파라미터 세팅 및 검색
    index.hnsw.efSearch = 64 # 검색 시 탐색 깊이 (정확도 vs 검색 속도)
    k = 5 # 상위 5개
    
    # distances: 쿼리와 가장 가까운 노드들의 거리 값 반환
    # indices: 해당 노드들의 DB 내 배열 인덱스(ID) 반환
    distances, indices = index.search(query_vector, k)
    
    print(f"찾아낸 Top-{k} ID   : {indices[0]}")
    print(f"해당 ID의 거리 값 : {distances[0]}")

import faiss
import numpy as np

class HNSWSearchEngine:
    """실무 환경에서 사용하는 FAISS 기반 벡터 저장/검색 엔진"""
    
    def __init__(self, dimension: int = 512, m: int = 32, ef_search: int = 128):
        self.dimension = dimension
        # L2 거리를 기반으로 하는 HNSW 인덱스 생성 (코사인 유사도용)
        self.index = faiss.IndexHNSWFlat(dimension, m)
        self.index.hnsw.efConstruction = 100
        self.index.hnsw.efSearch = ef_search
        
    def _normalize(self, vectors: np.ndarray) -> np.ndarray:
        """
        [매우 중요] FAISS는 내장 코사인 유사도 검색이 까다롭습니다.
        하지만 벡터를 L2 정규화(길이를 1로 만듦) 한 뒤 L2 거리(Euclidean)를 구하면, 
        수학적으로 코사인 유사도 순위와 완벽하게 동일해집니다.
        """
        norms = np.linalg.norm(vectors, axis=1, keepdims=True)
        # 0으로 나누는 것 방지
        norms = np.where(norms == 0, 1e-10, norms)
        return vectors / norms

    def build_index(self, product_ids: list, vectors: np.ndarray, save_path: str = "hnsw.index"):
        """전체 상품 벡터를 정규화하여 그래프에 올리고 파일로 저장합니다."""
        assert len(product_ids) == vectors.shape[0]
        
        normalized_vecs = self._normalize(vectors).astype('float32')
        self.index.add(normalized_vecs)
        
        # 실무에서는 인덱스(0, 1, 2...)와 실제 상품 ID(item_99)를 매핑하는 딕셔너리를 따로 저장해야 합니다.
        self.id_map = {i: pid for i, pid in enumerate(product_ids)}
        
        # 재시작 시 1초 만에 복구하기 위해 인덱스를 디스크에 덤프
        faiss.write_index(self.index, save_path)
        print(f"✅ HNSW 인덱스 저장 완료: {save_path} (총 {self.index.ntotal}건)")

    def load_index(self, load_path: str = "hnsw.index"):
        """서버 부팅 시 디스크에서 인덱스를 메모리로 로드합니다."""
        self.index = faiss.read_index(load_path)
        print(f"✅ HNSW 인덱스 로드 완료 (총 {self.index.ntotal}건)")

    def search(self, query_vector: np.ndarray, top_k: int = 10):
        """실시간 검색 쿼리를 처리합니다."""
        # 쿼리 벡터 역시 정규화 필수
        query_normalized = self._normalize(query_vector).astype('float32')
        
        # 검색 수행 (수 밀리초 소요)
        distances, indices = self.index.search(query_normalized, top_k)
        
        results = []
        for dist, idx in zip(distances[0], indices[0]):
            if idx != -1: # -1은 찾지 못했음을 의미
                results.append({
                    "product_id": self.id_map.get(idx, "Unknown"),
                    # L2 거리를 직관적인 점수로 변환 (거리가 짧을수록 점수가 높게)
                    "score": round(float(1 / (1 + dist)), 4) 
                })
        return results

# --- 실무 API 서버 활용 예시 ---
# hnsw_db = HNSWSearchEngine(dimension=512)
# hnsw_db.load_index()
#
# @app.post("/search/vector")
# def vector_search(query: str):
#     query_vec = clip_encoder.get_vector(query) # Step 2에서 만든 CLIP 객체
#     results = hnsw_db.search(np.array([query_vec]), top_k=20)
#     return {"items": results}