Elasticsearch 8.x Native Vector Search와 하이브리드 아키텍처

// 1. 인덱스 생성 및 Mapping 정의 (REST API)
PUT /home_shopping_products
{
  "mappings": {
    "properties": {
      "product_id": { "type": "keyword" },
      "brand_name": { "type": "keyword" },
      "image_vector": {
        "type": "dense_vector",
        "dims": 512,
        "index": true,             // Lucene HNSW 인덱싱 활성화
        "similarity": "cosine",    // 코사인 유사도 기반 거리 계산
        "index_options": {         // HNSW 세부 파라미터 튜닝
          "type": "hnsw",
          "m": 16,
          "ef_construction": 100
        }
      }
    }
  }
}

// 2. 단일 벡터 k-NN 검색 쿼리 (Pre-filter 포함)
POST /home_shopping_products/_search
{
  "knn": {
    "field": "image_vector",
    "query_vector": [0.012, 0.453, -0.123, ...], // 512차원 벡터 배열
    "k": 10,                                     // 최종 반환할 문서 수
    "num_candidates": 100,                       // HNSW 탐색 노드 수 (정확도 vs 속도)
    "filter": {                                  // 그래프 탐색 전 적용할 역색인 필터
      "term": { "brand_name": "Nike" }
    }
  },
  "_source": ["product_id", "brand_name"]
}

from elasticsearch import Elasticsearch
from typing import List, Dict, Any

class ElasticVectorSearchEngine:
    """
    Elasticsearch 8.x 기반의 하이브리드 검색 처리 엔진 (RRF 지원).
    """
    def __init__(self, es_client: Elasticsearch, index_name: str = "home_shopping_products"):
        self.es = es_client
        self.index_name = index_name

    def hybrid_search_with_filter(
        self, 
        vector_query: List[float], 
        keyword: str, 
        brand_filter: str, 
        top_k: int = 10
    ) -> List[Dict[str, Any]]:
        """
        벡터 유사도와 키워드 매칭을 결합하고 특정 브랜드로 필터링하는 하이브리드 쿼리를 실행합니다.
        """
        
        # 1. k-NN 쿼리 블록 (벡터 유사도 기반)
        # Pre-filter를 적용하여 해당 브랜드의 BitSet이 켜진 노드만 HNSW 그래프에서 탐색
        knn_query = {
            "field": "image_vector",
            "query_vector": vector_query,
            "k": top_k,
            "num_candidates": 50,
            "filter": {
                "term": { "brand_name": brand_filter }
            }
        }
        
        # 2. 전문 검색(Full-text Search) 쿼리 블록 (BM25 형태소 분석 기반)
        match_query = {
            "bool": {
                "must": [
                    { "match": { "product_name": keyword } }
                ],
                "filter": [
                    { "term": { "brand_name": brand_filter } }
                ]
            }
        }

        # 3. ES 8.x 하이브리드 쿼리 바디 구성
        # knn 파라미터와 query 파라미터를 동위에 선언.
        # rrf(Reciprocal Rank Fusion) 블록을 통해 두 검색 엔진의 결과를 순위 기반 정규화하여 병합
        request_body = {
            "knn": knn_query,
            "query": match_query,
            "rank": {
                "rrf": { 
                    "window_size": 50,
                    "rank_constant": 60
                }
            },
            "_source": ["product_id", "brand_name", "product_name"]
        }

        # 4. 쿼리 실행 및 파싱
        response = self.es.search(index=self.index_name, body=request_body)
        
        results = []
        # ES 8.x RRF 결과는 hit['_score']가 절대값이 아닌 병합된 역수 랭크 스코어입니다.
        for hit in response['hits']['hits']:
            results.append({
                "product_id": hit['_source']['product_id'],
                "brand": hit['_source']['brand_name'],
                "name": hit['_source']['product_name'],
                "hybrid_score": hit['_score'] 
            })
            
        return results

# --- 실무 API 라우터 사용 예시 ---
# es_client = Elasticsearch("https://localhost:9200", basic_auth=("elastic", "password"))
# search_engine = ElasticVectorSearchEngine(es_client)
# 
# @app.get("/search/hybrid")
# def search(keyword: str, brand: str):
#     # CLIP Text Encoder를 통해 검색어 벡터화 (약 15ms)
#     query_vector = text_encoder.encode(keyword) 
#     
#     # 하이브리드 검색 실행 (네트워크 I/O 포함 약 30ms 이내)
#     results = search_engine.hybrid_search_with_filter(
#         vector_query=query_vector, 
#         keyword=keyword, 
#         brand_filter=brand
#     )
#     return {"data": results}