중급 과정으로 돌아가기
Chapter 2: 하이브리드 검색 전략
키워드와 벡터 검색의 시너지 활용하기
2.1 하이브리드 검색이 필요한 이유
각 검색 방식의 장단점과 보완 관계
🔤 키워드 검색 (BM25)
장점:
- • 정확한 단어 매칭
- • 희귀 용어, 고유명사에 강함
- • 검색 결과 설명 가능
- • 빠른 속도
단점:
- • 동의어 처리 어려움
- • 문맥 이해 부족
- • 철자 오류에 취약
🧠 벡터 검색 (Semantic)
장점:
- • 의미적 유사성 파악
- • 동의어, 유사어 처리
- • 문맥 기반 이해
- • 다국어 지원
단점:
- • 고유명사, ID에 약함
- • 계산 비용 높음
- • 블랙박스 성격
실제 사례로 보는 차이점
쿼리: "SKU-12345의 재고 현황"
키워드 검색 ✅
정확히 SKU-12345를 포함한 문서 검색
벡터 검색 ❌
유사한 제품 코드들을 반환할 수 있음
쿼리: "차가운 음료"
키워드 검색 ❌
"차가운"과 "음료"를 정확히 포함한 문서만
벡터 검색 ✅
"아이스커피", "냉음료", "시원한 음료" 등도 검색
2.2 BM25 알고리즘 구현
Best Matching 25의 원리와 최적화
BM25 수식 이해하기
BM25는 TF-IDF의 확률론적 해석으로, 문서 길이를 정규화하여 더 정확한 점수를 계산합니다.
score(D,Q) = Σ IDF(qi) · (f(qi,D) · (k1+1)) / (f(qi,D) + k1·(1-b+b·|D|/avgdl))
- • k1: 용어 빈도의 포화점 조절 (일반적으로 1.2)
- • b: 문서 길이 정규화 강도 (일반적으로 0.75)
- • avgdl: 평균 문서 길이
Python 구현 예제
from typing import List, Dict
import math
from collections import Counter
class BM25:
def __init__(self, corpus: List[List[str]], k1=1.2, b=0.75):
self.k1 = k1
self.b = b
self.corpus = corpus
self.avgdl = sum(len(doc) for doc in corpus) / len(corpus)
self.doc_freqs = self._calc_doc_freqs()
self.idf = self._calc_idf()
def _calc_doc_freqs(self) -> Dict[str, int]:
"""각 용어가 나타나는 문서 수 계산"""
doc_freqs = {}
for doc in self.corpus:
for word in set(doc):
doc_freqs[word] = doc_freqs.get(word, 0) + 1
return doc_freqs
def _calc_idf(self) -> Dict[str, float]:
"""역문서빈도(IDF) 계산"""
idf = {}
N = len(self.corpus)
for word, freq in self.doc_freqs.items():
idf[word] = math.log(((N - freq + 0.5) / (freq + 0.5)) + 1)
return idf
def score(self, query: List[str], doc_idx: int) -> float:
"""쿼리와 문서의 BM25 점수 계산"""
doc = self.corpus[doc_idx]
doc_len = len(doc)
scores = 0.0
doc_freqs = Counter(doc)
for term in query:
if term not in self.idf:
continue
term_freq = doc_freqs[term]
idf = self.idf[term]
numerator = idf * term_freq * (self.k1 + 1)
denominator = term_freq + self.k1 * (1 - self.b + self.b * doc_len / self.avgdl)
scores += numerator / denominator
return scores
# 사용 예제
corpus = [
["신용카드", "결제", "시스템", "오류", "발생"],
["카드", "결제", "실패", "환불", "처리"],
["시스템", "점검", "안내", "공지사항"],
]
bm25 = BM25(corpus)
query = ["카드", "결제", "오류"]
for i, doc in enumerate(corpus):
score = bm25.score(query, i)
print(f"문서 {i}: {' '.join(doc)} - 점수: {score:.3f}")2.3 검색 결과 결합 및 재순위화
효과적인 하이브리드 검색 구현 방법
점수 정규화 기법
1. Min-Max 정규화
def min_max_normalize(scores):
min_score = min(scores)
max_score = max(scores)
if max_score == min_score:
return [0.5] * len(scores)
return [(s - min_score) / (max_score - min_score) for s in scores]2. Z-Score 정규화
def z_score_normalize(scores):
mean = sum(scores) / len(scores)
std = (sum((s - mean) ** 2 for s in scores) / len(scores)) ** 0.5
if std == 0:
return [0.0] * len(scores)
return [(s - mean) / std for s in scores]3. Reciprocal Rank Fusion (RRF)
def reciprocal_rank_fusion(rankings, k=60):
"""여러 순위를 결합하는 RRF 알고리즘"""
scores = {}
for ranking in rankings:
for rank, doc_id in enumerate(ranking):
if doc_id not in scores:
scores[doc_id] = 0
scores[doc_id] += 1 / (k + rank + 1)
return sorted(scores.items(), key=lambda x: x[1], reverse=True)완전한 하이브리드 검색 파이프라인
from typing import List, Tuple
import numpy as np
class HybridSearchEngine:
def __init__(self, vector_db, bm25_engine, alpha=0.5):
"""
alpha: 벡터 검색 가중치 (0~1)
1-alpha: 키워드 검색 가중치
"""
self.vector_db = vector_db
self.bm25 = bm25_engine
self.alpha = alpha
def search(self, query: str, top_k: int = 10) -> List[Tuple[str, float]]:
# 1. 벡터 검색 수행
query_embedding = self.embed_query(query)
vector_results = self.vector_db.similarity_search(
query_embedding, k=top_k * 2
)
# 2. BM25 키워드 검색 수행
query_tokens = self.tokenize(query)
bm25_scores = []
for i, doc in enumerate(self.bm25.corpus):
score = self.bm25.score(query_tokens, i)
bm25_scores.append((i, score))
bm25_results = sorted(bm25_scores, key=lambda x: x[1], reverse=True)[:top_k * 2]
# 3. 점수 정규화
vector_scores = [r[1] for r in vector_results]
vector_norm = self.min_max_normalize(vector_scores)
bm25_scores = [r[1] for r in bm25_results]
bm25_norm = self.min_max_normalize(bm25_scores)
# 4. 결과 병합 및 재순위화
combined_scores = {}
for (doc_id, _), norm_score in zip(vector_results, vector_norm):
combined_scores[doc_id] = self.alpha * norm_score
for (doc_id, _), norm_score in zip(bm25_results, bm25_norm):
if doc_id in combined_scores:
combined_scores[doc_id] += (1 - self.alpha) * norm_score
else:
combined_scores[doc_id] = (1 - self.alpha) * norm_score
# 5. 최종 순위 결정
final_results = sorted(
combined_scores.items(),
key=lambda x: x[1],
reverse=True
)[:top_k]
return final_results
def adaptive_alpha(self, query: str) -> float:
"""쿼리 특성에 따라 alpha 값 동적 조정"""
# 숫자, 코드, ID가 포함된 경우 키워드 검색 가중치 증가
if any(c.isdigit() for c in query) or '-' in query:
return 0.3
# 일반적인 질문인 경우 벡터 검색 가중치 증가
if any(word in query for word in ['무엇', '어떻게', '왜', '설명']):
return 0.7
return 0.5 # 기본값
# 사용 예제
hybrid_search = HybridSearchEngine(vector_db, bm25_engine, alpha=0.5)
results = hybrid_search.search("SKU-12345 제품의 재고 현황", top_k=5)
for doc_id, score in results:
print(f"문서 {doc_id}: {score:.3f}")2.4 실제 적용 사례와 성능 향상
기업들의 하이브리드 검색 도입 결과
🏢 이커머스 플랫폼 사례
문제 상황
- • 상품명/SKU 검색 정확도 낮음
- • "빨간 운동화" → "레드 스니커즈" 매칭 안됨
- • 브랜드명 오타 처리 불가
개선 결과
- • 검색 정확도 35% 향상
- • 클릭률(CTR) 28% 증가
- • 검색 포기율 40% 감소
핵심 전략: SKU, 브랜드명은 BM25로, 상품 설명은 벡터 검색으로 처리. 쿼리 타입에 따라 가중치 동적 조정.
📚 기술 문서 검색 시스템
구현 상세
# 문서 타입별 가중치 설정
WEIGHT_CONFIG = {
"api_reference": {"bm25": 0.7, "vector": 0.3}, # 함수명, 파라미터 중요
"tutorial": {"bm25": 0.3, "vector": 0.7}, # 개념 설명 중요
"error_guide": {"bm25": 0.6, "vector": 0.4}, # 에러 코드 중요
"conceptual": {"bm25": 0.2, "vector": 0.8} # 의미 이해 중요
}
# 메타데이터 부스팅
if "error" in query and doc.type == "error_guide":
score *= 1.5 # 에러 관련 쿼리는 에러 가이드 문서 우선92%
정답 포함률 (Top 5)
1.2초
평균 응답 시간
4.7/5
사용자 만족도
실습 과제
하이브리드 검색 시스템 구축
📋 요구사항
- 1. Elasticsearch와 벡터 DB를 사용한 하이브리드 검색 구현
- 2. 한국어 형태소 분석기 적용 (Nori, Komoran 등)
- 3. 쿼리 타입 자동 분류 (키워드형, 자연어형, 혼합형)
- 4. A/B 테스트를 통한 최적 가중치 찾기
- 5. 검색 품질 메트릭 측정 (MRR, NDCG, Precision@K)
🎯 평가 데이터셋
- • 1000개의 문서 (뉴스, 제품 설명, FAQ 혼합)
- • 100개의 테스트 쿼리와 정답 셋
- • 키워드형 30%, 자연어형 50%, 혼합형 20%
💡 도전 과제
검색 로그를 분석하여 사용자의 검색 패턴을 학습하고, 개인화된 가중치를 적용하는 시스템으로 확장해보세요.