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

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

where:
  qi        = each query term
  f(qi, D)  = term frequency of qi in document D
  |D|       = document length
  avgdl     = average document length across corpus
  k1, b     = tuning parameters (typically k1=1.5, b=0.75)
  IDF(qi)   = log((N - df + 0.5) / (df + 0.5))

Vector search weaknesses:
- Poor at exact keyword matching (model numbers, product codes)
- "Semantic drift": retrieves semantically similar but incorrect content
- Vulnerable to rare terms, neologisms, abbreviations

BM25 weaknesses:
- No synonym understanding ("automobile" vs "car")
- No contextual understanding (word order ignored)
- Vulnerable to typos
- Weak multilingual support

def reciprocal_rank_fusion(results_list: list, k: int = 60) -> list:
    """
    Combine multiple ranked result lists using RRF.

    RRF score = Σ 1/(k + rank_i)
    k=60 is a constant that dampens the impact of low-ranked items.

    Args:
        results_list: list of result lists (each is a list of doc IDs in rank order)
        k: rank stabilization constant (default 60)

    Returns:
        List of doc IDs sorted by descending RRF score
    """
    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)


# Concrete example
vector_results = ["doc_A", "doc_C", "doc_B"]  # from embedding search
bm25_results   = ["doc_B", "doc_A", "doc_D"]  # from BM25 keyword search

# Score calculation:
# 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 ranks first because it's highly ranked in both lists
# doc_B ranks second: #1 in BM25 but #3 in vector

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

# Assume: docs is a list of LangChain Document objects
embedding_model = OpenAIEmbeddings()

# Create BM25 retriever
bm25_retriever = BM25Retriever.from_documents(
    docs,
    k=5  # return top 5
)

# Create vector store and retriever
vectorstore = FAISS.from_documents(docs, embedding_model)
vector_retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5}
)

# EnsembleRetriever: combines both searches with weights
hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25 40%, Vector 60%
)

# Usage
query = "iPhone 15 Pro Max 256GB in stock"
results = hybrid_retriever.get_relevant_documents(query)

from elasticsearch import Elasticsearch

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

def hybrid_search_es(query: str, query_embedding: list, k: int = 5):
    response = es_client.search(
        index="documents",
        body={
            "query": {
                "bool": {
                    "should": [
                        # BM25 keyword search
                        {
                            "match": {
                                "content": {
                                    "query": query,
                                    "boost": 0.4
                                }
                            }
                        }
                    ]
                }
            },
            # KNN vector search (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"]

import nltk
from nltk.tokenize import word_tokenize
from nltk.stem import PorterStemmer
from nltk.corpus import stopwords

nltk.download('punkt')
nltk.download('stopwords')

stemmer = PorterStemmer()
stop_words = set(stopwords.words('english'))

def technical_tokenizer(text: str) -> list:
    # Tokenize
    tokens = word_tokenize(text.lower())
    # Remove stopwords and apply stemming
    tokens = [
        stemmer.stem(token)
        for token in tokens
        if token.isalnum() and token not in stop_words
    ]
    return tokens

# Apply custom tokenizer when creating BM25 retriever
bm25_retriever = BM25Retriever.from_documents(
    docs,
    preprocess_func=technical_tokenizer,
    k=5
)

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA

# Initialize all components
bm25_retriever = BM25Retriever.from_documents(docs, k=5)
vectorstore = FAISS.from_documents(docs, embedding_model)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]
)

# Build the RAG chain
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=hybrid_retriever,
    return_source_documents=True
)

# Run a query
response = qa_chain.invoke({"query": "iPhone 15 Pro Max 256GB price"})
print(response["result"])
print("\nSources:")
for doc in response["source_documents"]:
    print(f"  - {doc.metadata.get('source', 'unknown')}")