이 글은 누구를 위한 것인가
- 키워드 검색만으로는 의미 기반 질의를 처리 못하는 팀
- 벡터 검색만 도입했는데 정확도가 낮아 고민인 팀
- 검색 품질을 수치로 측정하고 싶은 검색 엔지니어
들어가며
"무릎이 아파요"라고 검색했을 때 "관절 질환 치료법"이 나오려면 키워드 매칭이 아닌 의미 이해가 필요하다. 하지만 "iPhone 15 Pro 256GB"처럼 정확한 스펙 검색은 키워드가 더 정확하다. 두 방식을 합치면 최선이 된다.
이 글은 bluefoxdev.kr의 검색 시스템 설계 를 참고하여 작성했습니다.
1. 하이브리드 검색 아키텍처
[검색 처리 흐름]
쿼리 입력
↓
쿼리 이해 (LLM 전처리)
- 오타 교정
- 동의어 확장
- 의도 분류 (탐색/구매/정보)
↓
병렬 검색 실행
├── BM25 키워드 검색 (Elasticsearch)
└── 벡터 유사도 검색 (pgvector / Pinecone)
↓
결과 병합 (RRF: Reciprocal Rank Fusion)
↓
재랭킹 (Cross-Encoder 모델)
↓
최종 결과 반환
[인덱싱 전략]
문서 청크:
제목: 별도 인덱싱 (가중치 높음)
본문: 512토큰 단위 청크
메타: 필터링용 (날짜, 카테고리)
임베딩 모델:
한국어: ko-sbert, multilingual-e5
영어: text-embedding-3-small (OpenAI)
도메인 특화: 자체 파인튜닝
[검색 품질 지표]
NDCG@10: 순위 품질
MRR: 첫 번째 정답 위치
Recall@K: K개 내 정답 포함률
2. 하이브리드 검색 구현
import anthropic
import numpy as np
from dataclasses import dataclass
client = anthropic.Anthropic()
@dataclass
class SearchResult:
doc_id: str
title: str
content: str
bm25_rank: int | None = None
vector_rank: int | None = None
rrf_score: float = 0.0
final_score: float = 0.0
def reciprocal_rank_fusion(
bm25_results: list[SearchResult],
vector_results: list[SearchResult],
k: int = 60,
) -> list[SearchResult]:
"""RRF로 두 검색 결과 병합"""
scores: dict[str, float] = {}
doc_map: dict[str, SearchResult] = {}
for rank, result in enumerate(bm25_results):
scores[result.doc_id] = scores.get(result.doc_id, 0) + 1 / (k + rank + 1)
result.bm25_rank = rank
doc_map[result.doc_id] = result
for rank, result in enumerate(vector_results):
scores[result.doc_id] = scores.get(result.doc_id, 0) + 1 / (k + rank + 1)
result.vector_rank = rank
if result.doc_id not in doc_map:
doc_map[result.doc_id] = result
merged = list(doc_map.values())
for result in merged:
result.rrf_score = scores[result.doc_id]
return sorted(merged, key=lambda x: x.rrf_score, reverse=True)
async def expand_query_with_llm(query: str) -> dict:
"""LLM으로 쿼리 전처리 및 확장"""
response = client.messages.create(
model="claude-haiku-4-5-20251001",
max_tokens=300,
messages=[{
"role": "user",
"content": f"""검색 쿼리를 분석하세요: "{query}"
JSON으로 반환:
{{
"corrected": "오타 교정된 쿼리",
"intent": "informational/navigational/transactional",
"expanded_terms": ["동의어1", "관련어2"],
"filters": {{"category": null, "date_range": null}}
}}"""
}]
)
import json
return json.loads(response.content[0].text)
async def rerank_with_llm(
query: str,
candidates: list[SearchResult],
top_k: int = 10,
) -> list[SearchResult]:
"""LLM으로 상위 후보 재랭킹"""
if len(candidates) <= top_k:
return candidates
docs_text = "\n".join(
f"{i+1}. [{r.doc_id}] {r.title}: {r.content[:150]}"
for i, r in enumerate(candidates[:20])
)
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=400,
messages=[{
"role": "user",
"content": f"""검색어: "{query}"
문서 목록:
{docs_text}
관련성 순으로 상위 {top_k}개 번호를 JSON 배열로 반환:
{{"ranked": [3, 1, 7, ...]}}"""
}]
)
import json
result = json.loads(response.content[0].text)
ranked_indices = [idx - 1 for idx in result["ranked"] if 1 <= idx <= len(candidates)]
reranked = [candidates[i] for i in ranked_indices[:top_k]]
for i, r in enumerate(reranked):
r.final_score = top_k - i
return reranked
async def hybrid_search(
query: str,
bm25_search_fn,
vector_search_fn,
top_k: int = 10,
) -> list[SearchResult]:
"""하이브리드 검색 통합 파이프라인"""
# 쿼리 확장
query_info = await expand_query_with_llm(query)
expanded_query = query_info["corrected"]
# 병렬 검색
import asyncio
bm25_results, vector_results = await asyncio.gather(
bm25_search_fn(expanded_query, top_k=50),
vector_search_fn(expanded_query, top_k=50),
)
# RRF 병합
merged = reciprocal_rank_fusion(bm25_results, vector_results)
# LLM 재랭킹
return await rerank_with_llm(query, merged[:20], top_k=top_k)
마무리
하이브리드 검색의 핵심은 BM25(정밀도)와 벡터 검색(재현율)의 상호 보완이다. RRF는 별도 하이퍼파라미터 튜닝 없이도 안정적인 성능을 낸다. 재랭킹 단계에 LLM을 쓰면 품질이 크게 오르지만 응답 시간이 늘어나므로 캐싱과 비동기 처리가 필수다.