하이브리드 검색 (Hybrid Search)

                     ┌──────────────────────┐
              ┌────> │ Dense (벡터 검색)     │ ──> 순위 A ──┐
              │      │ "의미가 비슷한가?"     │              │
 질문 ────────┤      └──────────────────────┘              ▼
              │      ┌──────────────────────┐         ┌──────────┐
              └────> │ Sparse (BM25)        │ ──> 순위 B ─>  │  결합     │ ──> 최종 순위
                     │ "같은 단어가 있는가?"  │         │ RRF 등    │
                     └──────────────────────┘         └──────────┘

질문: "ERR-4031은 왜 나요?"

Dense top-3                              BM25 top-3
1. "에러가 발생하면 로그를..."  0.81      1. "ERR-4031: 결제 토큰 만료..."  14.2  ← 정답
2. "ERR-4032: 재고 부족..."     0.80      2. "ERR-4031 대응 런북..."        11.8
3. "결제 실패 시 재시도..."     0.79      3. "에러 코드 목록 ERR-4001~..."   6.3

Dense 점수: 코사인 유사도      0.70 ~ 0.85 정도에 몰려 있다
BM25 점수:  상한이 없는 값     0 ~ 20 이상, 질문마다 범위가 다르다

→ 그냥 더하면 BM25가 결과를 지배한다

RRF(d) = Σ  1 / (k + rank_r(d))
        r∈R

  R         : 결합할 검색 결과 목록들 (예: Dense, BM25)
  rank_r(d) : 목록 r에서 문서 d의 순위 (1부터 시작)
  k         : 상수. 원 논문에서 60을 사용했고, 지금도 흔한 기본값이다
  목록에 없는 문서는 그 목록에서 0점

Dense 순위:  1. A   2. B   3. C   4. D
BM25 순위:   1. C   2. A   3. E   4. B

import numpy as np
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer

chunks = [
    "ERR-4031: 결제 토큰 만료. 토큰 TTL은 30분이며 refresh_token으로 갱신한다.",
    "ERR-4032: 재고 부족. 주문 수량이 재고보다 많을 때 발생한다.",
    "에러가 발생하면 먼저 애플리케이션 로그를 확인하세요.",
    "유급휴가는 입사 1년 후 연 15일이 부여된다.",
]

# --- Sparse ---
def tokenize(text: str) -> list[str]:
    # 단순 공백 분할. 한국어는 형태소 분석기(예: kiwipiepy)로 바꿔야 조사가 분리된다
    return text.lower().split()

bm25 = BM25Okapi([tokenize(c) for c in chunks])

# --- Dense ---
model = SentenceTransformer("BAAI/bge-m3")
vecs = model.encode(chunks, normalize_embeddings=True)

def dense_rank(q: str, n: int) -> list[int]:
    qv = model.encode([q], normalize_embeddings=True)[0]
    return list(np.argsort(-(vecs @ qv))[:n])

def sparse_rank(q: str, n: int) -> list[int]:
    return list(np.argsort(-bm25.get_scores(tokenize(q)))[:n])

# --- RRF ---
def rrf(rankings: list[list[int]], k: int = 60) -> list[tuple[int, float]]:
    scores: dict[int, float] = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank)
    return sorted(scores.items(), key=lambda x: -x[1])

def hybrid_search(q: str, top_k: int = 3, n: int = 20):
    fused = rrf([dense_rank(q, n), sparse_rank(q, n)])
    return [(chunks[i], round(s, 4)) for i, s in fused[:top_k]]

print(hybrid_search("ERR-4031 원인"))

A, B, C, D, E = range(5)
result = [doc for doc, _ in rrf([[A, B, C, D], [C, A, E, B]])]
assert result == [A, C, B, E, D]

WITH dense AS (
  SELECT id, ROW_NUMBER() OVER (ORDER BY embedding <=> :query_vec) AS rnk
  FROM chunks
  ORDER BY embedding <=> :query_vec
  LIMIT 50
),
sparse AS (
  SELECT id, ROW_NUMBER() OVER (ORDER BY ts_rank_cd(tsv, q) DESC) AS rnk
  FROM chunks, plainto_tsquery('simple', :query_text) q
  WHERE tsv @@ q
  ORDER BY ts_rank_cd(tsv, q) DESC
  LIMIT 50
)
SELECT c.id, c.content,
       COALESCE(1.0 / (60 + d.rnk), 0) + COALESCE(1.0 / (60 + s.rnk), 0) AS rrf_score
FROM dense d
FULL OUTER JOIN sparse s ON d.id = s.id
JOIN chunks c ON c.id = COALESCE(d.id, s.id)
ORDER BY rrf_score DESC
LIMIT 10;

문서: "유급휴가는 연 15일이 부여된다"

BM25 색인:   {유급휴가: w1, 연: w2, 15일: w3, 부여: w4}

SPLADE 벡터: {유급휴가: 2.1, 휴가: 1.8, 연차: 1.2, 15일: 1.5, 부여: 0.9, ...}
                                    ↑ 문서에 없는 단어도 확장되어 들어간다

질문 "연차 며칠?" → BM25는 0점, SPLADE는 "연차" 차원에서 매칭