Hybrid Search 완전 가이드: BM25와 벡터 검색을 합치면 RAG가 달라진다

score(D, Q) = Σ IDF(qi) × [f(qi, D) × (k1 + 1)] / [f(qi, D) + k1 × (1 - b + b × |D| / avgdl)]

여기서:
  qi     = 쿼리의 각 단어
  f(qi, D)  = 문서 D에서 단어 qi의 출현 빈도 (Term Frequency)
  |D|    = 문서 D의 길이
  avgdl  = 전체 문서의 평균 길이
  k1, b  = 조율 파라미터 (일반적으로 k1=1.5, b=0.75)
  IDF(qi) = Inverse Document Frequency = log((N - df + 0.5) / (df + 0.5))

벡터 검색의 약점:
- 정확한 키워드 매칭에 약함 (모델 번호, 제품 코드)
- "semantic drift": 유사하지만 다른 개념을 가져옴
- 희귀어, 신조어, 약어에 취약

BM25의 약점:
- 동의어를 이해 못 함 ("automobile" vs "car")
- 문맥을 이해 못 함 (단어 순서 무시)
- 오타에 취약
- 다국어 처리에 약함

def reciprocal_rank_fusion(results_list: list, k: int = 60) -> list:
    """
    여러 순위 결과 목록을 RRF로 결합.

    RRF 점수 = Σ 1/(k + rank_i)
    k=60은 낮은 순위 아이템의 영향을 완화하는 상수

    Args:
        results_list: 각 검색 결과 목록의 리스트 (doc_id 순서)
        k: 순위 안정화 상수 (기본값 60)

    Returns:
        RRF 점수 기준 정렬된 doc_id 목록
    """
    scores: dict = {}

    for results in results_list:
        for rank, doc_id in enumerate(results, start=1):
            scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank)

    return sorted(scores.keys(), key=lambda x: scores[x], reverse=True)


# 구체적인 예시로 이해하기
vector_results = ["doc_A", "doc_C", "doc_B"]  # 벡터 검색 결과
bm25_results   = ["doc_B", "doc_A", "doc_D"]  # BM25 키워드 검색 결과

# 점수 계산:
# doc_A: 1/(60+1) + 1/(60+2) = 0.01639 + 0.01613 = 0.03252
# doc_B: 1/(60+3) + 1/(60+1) = 0.01587 + 0.01639 = 0.03226
# doc_C: 1/(60+2) = 0.01613
# doc_D: 1/(60+2) = 0.01613

fused = reciprocal_rank_fusion([vector_results, bm25_results])
print(fused)
# ['doc_A', 'doc_B', 'doc_C', 'doc_D']
# doc_A가 두 검색 모두에서 상위권이라 1위
# doc_B는 BM25에서 1위지만 벡터에서 3위 → 2위

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 가정: docs는 langchain Document 객체 리스트
embedding_model = OpenAIEmbeddings()

# BM25 retriever 생성
bm25_retriever = BM25Retriever.from_documents(
    docs,
    k=5  # 상위 5개 반환
)

# 벡터 스토어 생성 및 retriever 설정
vectorstore = FAISS.from_documents(docs, embedding_model)
vector_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# 앙상블 리트리버: 두 검색을 가중치로 결합
hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25 40%, 벡터 60%
)

# 사용
query = "iPhone 15 Pro Max 256GB 재고"
results = hybrid_retriever.get_relevant_documents(query)

from elasticsearch import Elasticsearch
from langchain_community.vectorstores import ElasticsearchStore

es_client = Elasticsearch(["http://localhost:9200"])

# Elasticsearch에서 hybrid search 쿼리
def hybrid_search_es(query: str, query_embedding: list, k: int = 5):
    response = es_client.search(
        index="documents",
        body={
            "query": {
                "bool": {
                    "should": [
                        # BM25 키워드 검색
                        {
                            "match": {
                                "content": {
                                    "query": query,
                                    "boost": 0.4
                                }
                            }
                        }
                    ]
                }
            },
            # KNN 벡터 검색 (ES 8.x+)
            "knn": {
                "field": "content_vector",
                "query_vector": query_embedding,
                "k": k,
                "num_candidates": 100,
                "boost": 0.6
            },
            "size": k
        }
    )
    return response["hits"]["hits"]

# 한국어 형태소 분석기와 BM25 결합
from konlpy.tag import Okt

okt = Okt()

def korean_tokenizer(text: str) -> list:
    # 형태소 분석으로 토크나이징
    tokens = okt.morphs(text, stem=True)  # 어간 추출
    return tokens

# BM25 생성 시 커스텀 토크나이저 적용
bm25_retriever = BM25Retriever.from_documents(
    docs,
    preprocess_func=korean_tokenizer,
    k=5
)