9장: 하이브리드 검색 구현

import math
 
def bm25_score(
    query_terms: list[str],
    document: list[str],
    doc_freq: dict[str, int],
    total_docs: int,
    avg_doc_length: float,
    k1: float = 1.2,
    b: float = 0.75
) -> float:
    """BM25 점수 계산 (개념 설명용)"""
    score = 0.0
    doc_length = len(document)
 
    for term in query_terms:
        if term not in document:
            continue
 
        # TF: 문서 내 용어 빈도
        tf = document.count(term)
 
        # IDF: 역문서 빈도
        df = doc_freq.get(term, 0)
        idf = math.log((total_docs - df + 0.5) / (df + 0.5) + 1)
 
        # BM25 공식
        numerator = tf * (k1 + 1)
        denominator = tf + k1 * (1 - b + b * doc_length / avg_doc_length)
        score += idf * numerator / denominator
 
    return score

def reciprocal_rank_fusion(
    rankings: list[list[str]],
    k: int = 60
) -> list[tuple[str, float]]:
    """RRF 구현"""
    scores: dict[str, float] = {}
 
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            if doc_id not in scores:
                scores[doc_id] = 0.0
            scores[doc_id] += 1.0 / (k + rank)
 
    # 점수 내림차순 정렬
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)
 
# 사용 예시
vector_results = ["doc_3", "doc_1", "doc_7", "doc_5", "doc_2"]
bm25_results = ["doc_1", "doc_4", "doc_3", "doc_8", "doc_5"]
 
fused = reciprocal_rank_fusion([vector_results, bm25_results])
# doc_1과 doc_3이 양쪽에서 높은 순위 -> 상위로 올라감

def relative_score_fusion(
    vector_results: list[tuple[str, float]],
    bm25_results: list[tuple[str, float]],
    alpha: float = 0.75  # 벡터 비중
) -> list[tuple[str, float]]:
    """Relative Score Fusion"""
    def normalize(results):
        if not results:
            return {}
        scores = [s for _, s in results]
        min_s, max_s = min(scores), max(scores)
        if max_s == min_s:
            return {doc_id: 1.0 for doc_id, _ in results}
        return {
            doc_id: (s - min_s) / (max_s - min_s)
            for doc_id, s in results
        }
 
    norm_vector = normalize(vector_results)
    norm_bm25 = normalize(bm25_results)
 
    all_docs = set(norm_vector.keys()) | set(norm_bm25.keys())
    fused = {}
    for doc_id in all_docs:
        vec_score = norm_vector.get(doc_id, 0.0)
        bm25_score = norm_bm25.get(doc_id, 0.0)
        fused[doc_id] = alpha * vec_score + (1 - alpha) * bm25_score
 
    return sorted(fused.items(), key=lambda x: x[1], reverse=True)

from sentence_transformers import CrossEncoder
 
# Cross-Encoder 리랭커
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")
 
def search_with_reranking(
    query: str,
    initial_results: list[dict],
    top_k: int = 10
) -> list[dict]:
    """하이브리드 검색 + 리랭킹 파이프라인"""
 
    # 리랭킹: 쿼리-문서 쌍의 관련도 평가
    pairs = [(query, doc["content"]) for doc in initial_results]
    rerank_scores = reranker.predict(pairs)
 
    # 리랭킹 점수로 재정렬
    for doc, score in zip(initial_results, rerank_scores):
        doc["rerank_score"] = float(score)
 
    reranked = sorted(
        initial_results,
        key=lambda x: x["rerank_score"],
        reverse=True
    )
 
    return reranked[:top_k]

import cohere
 
co = cohere.Client("YOUR_COHERE_KEY")
 
results = co.rerank(
    model="rerank-multilingual-v3.0",
    query="벡터 데이터베이스의 HNSW 파라미터",
    documents=[doc["content"] for doc in initial_results],
    top_n=10
)
 
for result in results.results:
    print(f"Index: {result.index}, Score: {result.relevance_score:.4f}")

articles = client.collections.get("Article")
 
response = articles.query.hybrid(
    query="HNSW 인덱스 최적화",
    alpha=0.75,
    fusion_type="relative_score",
    limit=20
)

from qdrant_client.models import Prefetch, FusionQuery, Fusion
 
# RRF 기반 하이브리드 검색
results = client.query_points(
    collection_name="articles",
    prefetch=[
        Prefetch(
            query=[0.1, 0.2, ...],     # 밀집 벡터 검색
            using="dense",
            limit=20
        ),
        Prefetch(
            query=sparse_vector,        # 희소 벡터 (BM25/SPLADE)
            using="sparse",
            limit=20
        ),
    ],
    query=FusionQuery(fusion=Fusion.RRF),  # RRF로 결합
    limit=10
)

# 밀집 + 희소 벡터 하이브리드 검색
results = index.query(
    vector=dense_embedding,          # 밀집 벡터
    sparse_vector={
        "indices": sparse_indices,   # SPLADE 등으로 생성
        "values": sparse_values
    },
    top_k=10
)

-- 벡터 검색 결과
WITH vector_results AS (
    SELECT id, title,
        ROW_NUMBER() OVER (ORDER BY embedding <=> query_vec) AS vec_rank
    FROM articles
    ORDER BY embedding <=> query_vec
    LIMIT 50
),
-- BM25 (tsvector) 검색 결과
text_results AS (
    SELECT id, title,
        ROW_NUMBER() OVER (ORDER BY ts_rank(tsv, query) DESC) AS text_rank
    FROM articles
    WHERE tsv @@ plainto_tsquery('korean', 'HNSW 인덱스')
    LIMIT 50
)
-- RRF 결합
SELECT
    COALESCE(v.id, t.id) AS id,
    COALESCE(v.title, t.title) AS title,
    COALESCE(1.0 / (60 + v.vec_rank), 0) +
    COALESCE(1.0 / (60 + t.text_rank), 0) AS rrf_score
FROM vector_results v
FULL OUTER JOIN text_results t ON v.id = t.id
ORDER BY rrf_score DESC
LIMIT 10;