[LLMOps] Day 4: 검색 최적화와 하이브리드 서치 - 정확도를 높이는 고급 기법

서론: 벡터 검색의 한계

순수 벡터 검색(Semantic Search)은 강력하지만, 놓치는 것이 있다.

• 키워드 매칭 실패: “GPT-4”를 검색할 때, “GPT-4”가 정확히 포함된 문서를 찾아야 하는데 의미적으로 유사한 “LLM” 문서가 상위에 올라올 수 있다.
• 희귀 용어 처리 부족: “PostgreSQL 15의 pg_stat_statements” 같은 구체적 기술 용어는 임베딩 공간에서 제대로 표현되지 않는다.
• 순서 민감도: “Python에서 Rust 호출”과 “Rust에서 Python 호출”은 의미가 다른데, 벡터는 순서를 제대로 구분 못할 수 있다.

프로덕션 RAG 시스템은 하이브리드 검색(Hybrid Search)을 사용한다. 벡터 검색과 키워드 검색을 결합하여 두 가지 장점을 모두 가져간다.

1. 하이브리드 검색: BM25 + 벡터 검색

1.1 BM25 (Best Matching 25)

정보 검색의 고전적 알고리즘이지만, 여전히 강력하다.

핵심 아이디어:

• TF (Term Frequency): 문서에서 단어가 많이 등장할수록 관련성이 높다.
• IDF (Inverse Document Frequency): 드물게 등장하는 단어일수록 중요하다 (예: “the”보다 “PyO3”가 중요).
• 문서 길이 정규화: 긴 문서가 유리하지 않도록 보정.

수식:

Score(D, Q) = Σ IDF(qi) · (f(qi, D) · (k1 + 1)) / (f(qi, D) + k1 · (1 - b + b · |D| / avgdl))

실전 구현:

from rank_bm25 import BM25Okapi
import numpy as np

# 문서 토큰화
documents = [
    "Rust는 메모리 안전성을 보장하는 시스템 프로그래밍 언어이다",
    "PyO3를 사용하면 Rust로 Python 확장을 작성할 수 있다",
    "Python은 생산성이 높지만 성능은 C보다 느리다"
]

tokenized_docs = [doc.split() for doc in documents]

# BM25 인덱스 생성
bm25 = BM25Okapi(tokenized_docs)

# 검색
query = "Rust Python 확장"
tokenized_query = query.split()

# 각 문서의 BM25 점수
scores = bm25.get_scores(tokenized_query)
print(scores)
# [0.43, 1.28, 0.21]  ← 두 번째 문서가 가장 관련성 높음

# Top-K 검색
top_n = np.argsort(scores)[::-1][:2]
print([documents[i] for i in top_n])

1.2 하이브리드 검색 구현

from typing import List, Tuple
import numpy as np

class HybridSearcher:
    def __init__(self, documents: List[str], embeddings: List[List[float]]):
        # BM25 인덱스
        tokenized = [doc.split() for doc in documents]
        self.bm25 = BM25Okapi(tokenized)

        # 벡터 인덱스
        self.embeddings = np.array(embeddings)
        self.documents = documents

    def vector_search(self, query_embedding: List[float], top_k: int = 10) -> List[Tuple[int, float]]:
        """코사인 유사도 기반 벡터 검색"""
        query_vec = np.array(query_embedding)

        # 코사인 유사도 계산
        similarities = np.dot(self.embeddings, query_vec) / (
            np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_vec)
        )

        # Top-K 인덱스
        top_indices = np.argsort(similarities)[::-1][:top_k]

        return [(idx, similarities[idx]) for idx in top_indices]

    def keyword_search(self, query: str, top_k: int = 10) -> List[Tuple[int, float]]:
        """BM25 기반 키워드 검색"""
        tokenized_query = query.split()
        scores = self.bm25.get_scores(tokenized_query)

        top_indices = np.argsort(scores)[::-1][:top_k]

        return [(idx, scores[idx]) for idx in top_indices]

    def hybrid_search(
        self,
        query: str,
        query_embedding: List[float],
        alpha: float = 0.5,
        top_k: int = 5
    ) -> List[Tuple[int, float]]:
        """
        하이브리드 검색

        alpha: 벡터 검색 가중치 (0~1)
               0 = 순수 키워드, 1 = 순수 벡터, 0.5 = 균형
        """
        # 두 검색 수행
        vector_results = self.vector_search(query_embedding, top_k=20)
        keyword_results = self.keyword_search(query, top_k=20)

        # 점수 정규화 (0~1 범위)
        def normalize_scores(results):
            scores = [score for _, score in results]
            min_score, max_score = min(scores), max(scores)
            if max_score == min_score:
                return [(idx, 0.5) for idx, _ in results]
            return [
                (idx, (score - min_score) / (max_score - min_score))
                for idx, score in results
            ]

        vector_results = normalize_scores(vector_results)
        keyword_results = normalize_scores(keyword_results)

        # 점수 병합
        combined_scores = {}

        for idx, score in vector_results:
            combined_scores[idx] = alpha * score

        for idx, score in keyword_results:
            if idx in combined_scores:
                combined_scores[idx] += (1 - alpha) * score
            else:
                combined_scores[idx] = (1 - alpha) * score

        # 정렬 후 Top-K
        sorted_results = sorted(
            combined_scores.items(),
            key=lambda x: x[1],
            reverse=True
        )[:top_k]

        return sorted_results

# 사용 예시
results = searcher.hybrid_search(
    query="Rust로 Python 확장 만들기",
    query_embedding=query_embedding,
    alpha=0.7  # 벡터 검색에 70% 가중치
)

for idx, score in results:
    print(f"{score:.2f}: {documents[idx][:50]}...")

1.3 Reciprocal Rank Fusion (RRF)

점수를 직접 합치는 대신, 순위를 기반으로 병합하는 방법이다.

def reciprocal_rank_fusion(
    vector_results: List[Tuple[int, float]],
    keyword_results: List[Tuple[int, float]],
    k: int = 60
) -> List[Tuple[int, float]]:
    """
    RRF 공식: Score = Σ 1 / (k + rank)

    k: 상수 (보통 60), 상위 순위와 하위 순위 간 점수 차이 조절
    """
    rrf_scores = {}

    # 벡터 검색 결과
    for rank, (doc_id, _) in enumerate(vector_results):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + 1 / (k + rank + 1)

    # 키워드 검색 결과
    for rank, (doc_id, _) in enumerate(keyword_results):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + 1 / (k + rank + 1)

    # 정렬
    return sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)

# 사용 예시
rrf_results = reciprocal_rank_fusion(vector_results, keyword_results)

장점: 점수 스케일이 다른 두 검색 결과를 합칠 때, 정규화 없이 사용 가능.

2. 리랭킹 (Reranking)

검색은 빠르게, 정확도는 리랭커에서.

2.1 Cross-Encoder vs Bi-Encoder

특성	Bi-Encoder (임베딩)	Cross-Encoder (리랭커)
속도	매우 빠름 (사전 계산 가능)	느림 (실시간 계산 필요)
정확도	중간	매우 높음
사용 시점	1차 검색 (Top-100)	2차 정렬 (Top-10)
구조	Query와 Doc 독립적으로 인코딩	Query와 Doc을 함께 입력

핵심 전략: Bi-Encoder로 100개 후보를 빠르게 찾고, Cross-Encoder로 10개로 압축.

2.2 Cohere Rerank API

import cohere

co = cohere.Client("YOUR_API_KEY")

# 1차 검색 결과
documents = [
    "Rust는 메모리 안전성을 보장한다",
    "PyO3를 사용하면 Rust로 Python 확장을 만들 수 있다",
    "Python은 생산성이 높다"
]

query = "Rust로 Python 확장 만들기"

# 리랭킹
rerank_response = co.rerank(
    model="rerank-multilingual-v3.0",
    query=query,
    documents=documents,
    top_n=3
)

for result in rerank_response.results:
    print(f"Score: {result.relevance_score:.2f}, Doc: {documents[result.index]}")

# 출력:
# Score: 0.95, Doc: PyO3를 사용하면 Rust로 Python 확장을 만들 수 있다
# Score: 0.42, Doc: Rust는 메모리 안전성을 보장한다
# Score: 0.12, Doc: Python은 생산성이 높다

비용: $2.00 per 1,000 searches (문서 10개 기준)

2.3 오픈소스 리랭커

from sentence_transformers import CrossEncoder

# 모델 로드 (한 번만)
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# 리랭킹
query = "Rust로 Python 확장 만들기"
documents = [...]  # 검색 결과

# Query-Document 쌍 생성
pairs = [[query, doc] for doc in documents]

# 점수 계산
scores = reranker.predict(pairs)

# 정렬
reranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)

for doc, score in reranked[:3]:
    print(f"{score:.2f}: {doc[:50]}...")

성능:

• 레이턴시: ~50ms per document (CPU), ~5ms (GPU)
• 정확도: Cohere 대비 약 85% 수준
• 비용: 무료 (호스팅 비용만)

3. 쿼리 확장 (Query Expansion)

사용자 질문을 다양한 형태로 변형하여 검색 범위를 넓힌다.

3.1 HyDE (Hypothetical Document Embeddings)

아이디어: 질문을 직접 임베딩하지 말고, “이런 답변이 있을 것이다”라는 가상의 문서를 생성한 후 임베딩한다.

from openai import OpenAI

client = OpenAI()

def hyde_search(query: str, vector_db):
    # 1. LLM으로 가상 답변 생성
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "user",
            "content": f"다음 질문에 대한 답변을 200자 이내로 작성하세요:\n{query}"
        }],
        max_tokens=200
    )

    hypothetical_answer = response.choices[0].message.content

    # 2. 가상 답변을 임베딩
    embedding = get_embedding(hypothetical_answer)

    # 3. 벡터 검색
    results = vector_db.search(embedding, top_k=5)

    return results

# 사용 예시
results = hyde_search("Rust로 Python 확장 만드는 방법")

효과: 기술 문서 검색에서 10~15% 정확도 향상 (특히 “how-to” 질문)

3.2 Multi-Query

쿼리를 여러 개로 변형하여 검색 후 결과를 병합한다.

def multi_query_search(query: str, vector_db):
    # 1. 쿼리 변형 생성
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "user",
            "content": f"""다음 질문을 다른 표현으로 3가지 변형하세요:

질문: {query}

변형1:
변형2:
변형3:"""
        }]
    )

    # 변형된 쿼리 추출 (파싱 로직 생략)
    queries = [query] + parse_variations(response.choices[0].message.content)

    # 2. 각 쿼리로 검색
    all_results = []
    for q in queries:
        embedding = get_embedding(q)
        results = vector_db.search(embedding, top_k=10)
        all_results.append(results)

    # 3. RRF로 병합
    return reciprocal_rank_fusion(*all_results)

4. 실전 파이프라인: 모든 기법 결합

class AdvancedRAG:
    def __init__(self, vector_db, bm25_index, reranker):
        self.vector_db = vector_db
        self.bm25 = bm25_index
        self.reranker = reranker

    def search(self, query: str, top_k: int = 5):
        # 1. 쿼리 확장 (Multi-Query)
        expanded_queries = self.expand_query(query)

        # 2. 하이브리드 검색
        all_candidates = []
        for q in expanded_queries:
            # 벡터 검색
            vector_results = self.vector_db.search(q, top_k=20)
            # 키워드 검색
            keyword_results = self.bm25.search(q, top_k=20)
            # RRF 병합
            candidates = reciprocal_rank_fusion(vector_results, keyword_results)
            all_candidates.extend(candidates)

        # 중복 제거
        unique_docs = list(set([doc_id for doc_id, _ in all_candidates]))

        # 3. 리랭킹
        reranked = self.reranker.rerank(query, unique_docs, top_k=top_k)

        return reranked

성능:

• Top-5 정확도: 순수 벡터 검색 대비 25% 향상
• 레이턴시: ~400ms (캐싱 없을 때)
• 비용: 쿼리당 $0.003 (임베딩 + 리랭킹)

5. A/B 테스트 필수

각 기법의 효과는 데이터셋마다 다르다.

# 평가 세트 준비
test_queries = [
    ("Rust로 Python 확장 만들기", ["doc_123", "doc_456"]),  # (쿼리, 정답 문서 ID)
    # ...
]

# 정확도 측정
def evaluate(search_fn, test_set):
    hits = 0
    for query, ground_truth in test_set:
        results = search_fn(query, top_k=5)
        result_ids = [r[0] for r in results]

        # Top-5에 정답이 있는지 확인
        if any(gt in result_ids for gt in ground_truth):
            hits += 1

    return hits / len(test_set)

# 비교
print(f"순수 벡터: {evaluate(vector_search, test_queries):.2%}")
print(f"하이브리드: {evaluate(hybrid_search, test_queries):.2%}")
print(f"하이브리드 + 리랭킹: {evaluate(advanced_search, test_queries):.2%}")

다음 단계

검색 품질을 높였으니, 이제 프로덕션 배포를 준비할 차례다. Day 5에서는 모니터링, 비용 최적화, 장애 대응을 다룰 것이다.

---