LlamaIndex Hybrid Search: Dense와 Sparse Retrieval 완벽 가이드

안녕하세요, 저는 HolySheep AI의 기술 엔지니어입니다. 이번 튜토리얼에서는 LlamaIndex에서 Hybrid Search를 구현하는 방법을 초보자도 쉽게 이해할 수 있도록 단계별로 설명드리겠습니다.

Hybrid Search란 무엇인가?

Hybrid Search는 Dense Retrieval과 Sparse Retrieval의 장점을 결합한 검색 방식입니다. 각각의 특징을 먼저 이해해보겠습니다.

Dense Retrieval (밀집 벡터 검색)

Dense Retrieval은 텍스트를 고차원 벡터(임베딩)로 변환하여 의미적 유사도를 기반으로 검색합니다. 예를 들어 "강아지 사료"라는 검색어가 있을 때, "반려동물 용품"처럼 의미가 유사한 결과를 반환할 수 있습니다.

• 장점: 의미적 유사성 이해, 동의어 처리 가능
• 단점: 정확한 키워드 매칭에 약함, 컴퓨팅 비용 높음
• 지연 시간: 벡터 생성 100~300ms + 유사도 계산 10~50ms

Sparse Retrieval (희소 벡터 검색)

Sparse Retrieval은 BM25나 TF-IDF 같은 알고리즘을 사용하여 키워드 빈도수를 기반으로 검색합니다. 검색어와 정확히 일치하는 결과를 빠르게 찾아줍니다.

• 장점: 정확한 키워드 매칭, 빠른 검색 속도
• 단점: 의미적 유사성 부족, 동의어 처리 어려움
• 지연 시간: 인덱스 검색 5~20ms

Hybrid Search의 이점

두 방식을 결합하면:

• 정확한 키워드 일치 + 의미적 유사성 동시에 확보
• 검색 품질 향상 (평균 15~25% NDCG 개선)
• 다양한 검색 쿼리에 대응 가능

사전 준비: HolySheep AI API 키 발급

먼저 HolySheep AI에서 API 키를 발급받아야 합니다. 지금 가입 페이지에서 계정을 생성하면 무료 크레딧을 받을 수 있습니다. 가입 후 대시보드에서 API 키를 복사해주세요.

환경 설정

# 필요한 패키지 설치
pip install llama-index llama-index-embeddings-openai 
pip install llama-index-retrievers-bm25 rank-bm25
pip install openai numpy

프로젝트 구조

hybrid_search_project/
├── main.py              # 메인 실행 파일
├── data/
│   └── documents.json   # 검색할 문서 데이터
└── config.py            # 설정 파일

문서 데이터 준비

검색할 문서들을 JSON 파일로 준비합니다. 각 문서는 id, content, metadata 구조로 구성됩니다.

# data/documents.json
[
  {
    "id": "doc_001",
    "content": "LlamaIndex는 LLM 애플리케이션을 위한 데이터 프레임워크입니다. \
    인덱싱, 검색, Retrieval-Augmented Generation을 지원합니다.",
    "metadata": {"category": "framework", "source": "official"}
  },
  {
    "id": "doc_002", 
    "content": "Hybrid Search는 Dense와 Sparse检索을 결합합니다. \
    BM25와 벡터 임베딩을 함께 사용하여 검색 품질을 향상시킵니다.",
    "metadata": {"category": "search", "source": "blog"}
  },
  {
    "id": "doc_003",
    "content": "HolySheep AI는 글로벌 AI API 게이트웨이입니다. \
    로컬 결제와 단일 API 키로 여러 AI 모델을 통합할 수 있습니다.",
    "metadata": {"category": "ai-service", "source": "review"}
  }
]

설정 파일 작성

# config.py
import os

HolySheep AI 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # 실제 키로 교체
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

모델 설정
EMBEDDING_MODEL = "text-embedding-3-small"
LLM_MODEL = "gpt-4.1"

검색 파라미터
DENSE_WEIGHT = 0.5    # Dense retrieval 가중치
SPARSE_WEIGHT = 0.5   # Sparse retrieval 가중치
TOP_K = 5             # 반환할 결과 수

HolySheep AI 가격 정보 (2024년 기준)
PRICING = {
    "gpt-4.1": {"input": 8.0, "output": 8.0, "unit": "MTok"},
    "text-embedding-3-small": {"input": 0.02, "output": 0, "unit": "MTok"},
    "claude-sonnet-4.5": {"input": 15.0, "output": 15.0, "unit": "MTok"},
    "gemini-2.5-flash": {"input": 2.50, "output": 2.50, "unit": "MTok"},
}

Hybrid Search 구현

1단계: 라이브러리 임포트

# main.py
import json
import os
from typing import List, Dict, Tuple
import numpy as np
from llama_index.core import Document, Settings
from llama_index.core.retrievers import BaseRetriever
from llama_index.embeddings.openai import OpenAIEmbedding
from openai import OpenAI

HolySheep AI 클라이언트 설정
from config import (
    HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL,
    EMBEDDING_MODEL, LLM_MODEL
)

OpenAI 클라이언트를 HolySheep으로 설정
client = OpenAI(
    api_key=HOLYSHEEP_API_KEY,
    base_url=HOLYSHEEP_BASE_URL
)

임베딩 설정 - HolySheep 사용
Settings.embed_model = OpenAIEmbedding(
    model=EMBEDDING_MODEL,
    api_key=HOLYSHEEP_API_KEY,
    base_url=HOLYSHEEP_BASE_URL
)

2단계: Dense Retriever 구현

# Dense Retriever: 벡터 유사도 기반 검색
class DenseRetriever(BaseRetriever):
    def __init__(self, documents: List[Document], embed_model, top_k: int = 5):
        self.documents = documents
        self.embed_model = embed_model
        self.top_k = top_k
        self.doc_embeddings = None
        self._build_index()
    
    def _build_index(self):
        """문서 임베딩 인덱스 생성"""
        print(f"[Dense] {len(self.documents)}개 문서 임베딩 생성 중...")
        contents = [doc.text for doc in self.documents]
        
        # HolySheep API를 통한 임베딩 생성
        response = client.embeddings.create(
            model=EMBEDDING_MODEL,
            input=contents
        )
        
        self.doc_embeddings = np.array([
            item.embedding for item in response.data
        ])
        print(f"[Dense] 임베딩 완료. shape: {self.doc_embeddings.shape}")
    
    def retrieve(self, query: str) -> List[Document]:
        """쿼리와 유사한 문서 검색"""
        # 쿼리 임베딩
        response = client.embeddings.create(
            model=EMBEDDING_MODEL,
            input=[query]
        )
        query_embedding = np.array(response.data[0].embedding)
        
        # 코사인 유사도 계산
        similarities = self._cosine_similarity(query_embedding, self.doc_embeddings)
        
        # 상위 결과 반환
        top_indices = np.argsort(similarities)[-self.top_k:][::-1]
        
        return [self.documents[i] for i in top_indices]
    
    def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> np.ndarray:
        """코사인 유사도 계산"""
        dot_product = np.dot(b, a)
        norm_a = np.linalg.norm(a)
        norm_b = np.linalg.norm(b, axis=1)
        return dot_product / (norm_a * norm_b + 1e-8)

3단계: Sparse Retriever 구현 (BM25)

# Sparse Retriever: BM25 기반 키워드 검색
from rank_bm25 import BM25Okapi
import re

class SparseRetriever(BaseRetriever):
    def __init__(self, documents: List[Document], top_k: int = 5):
        self.documents = documents
        self.top_k = top_k
        self.tokenized_corpus = None
        self.bm25 = None
        self._build_index()
    
    def _tokenize(self, text: str) -> List[str]:
        """텍스트 토큰화 (한국어/영어 혼합 지원)"""
        # 간단한 토큰화: 공백 및 특수문자 기준으로 분리
        tokens = re.findall(r'\w+', text.lower())
        return tokens
    
    def _build_index(self):
        """BM25 인덱스 생성"""
        print(f"[Sparse] {len(self.documents)}개 문서 BM25 인덱스 생성 중...")
        contents = [doc.text for doc in self.documents]
        
        # 토큰화
        self.tokenized_corpus = [self._tokenize(content) for content in contents]
        
        # BM25 인덱스 생성
        self.bm25 = BM25Okapi(self.tokenized_corpus)
        print("[Sparse] BM25 인덱스 완료")
    
    def retrieve(self, query: str) -> List[Document]:
        """BM25 스코어 기반 문서 검색"""
        query_tokens = self._tokenize(query)
        scores = self.bm25.get_scores(query_tokens)
        
        # 상위 결과 반환
        top_indices = np.argsort(scores)[-self.top_k:][::-1]
        
        # 스코어가 0인 결과 제외
        results = []
        for i in top_indices:
            if scores[i] > 0:
                results.append(self.documents[i])
        
        return results

4단계: Hybrid Retriever 구현

# Hybrid Retriever: Dense + Sparse 결합
class HybridRetriever(BaseRetriever):
    def __init__(
        self, 
        documents: List[Document],
        embed_model,
        dense_weight: float = 0.5,
        sparse_weight: float = 0.5,
        top_k: int = 5
    ):
        self.documents = documents
        self.dense_weight = dense_weight
        self.sparse_weight = sparse_weight
        self.top_k = top_k
        
        # 개별 Retriever 초기화
        self.dense_retriever = DenseRetriever(documents, embed_model, top_k * 2)
        self.sparse_retriever = SparseRetriever(documents, top_k * 2)
        
        # 문서 ID 매핑
        self.doc_id_to_score = {doc.doc_id: 0.0 for doc in documents}
    
    def retrieve(self, query: str) -> List[Document]:
        """Hybrid 검색 수행"""
        # Dense 검색
        dense_results = self.dense_retriever.retrieve(query)
        print(f"[Hybrid] Dense 검색: {len(dense_results)}개 결과")
        
        # Sparse 검색
        sparse_results = self.sparse_retriever.retrieve(query)
        print(f"[Hybrid] Sparse 검색: {len(sparse_results)}개 결과")
        
        # 스코어 정규화 및 결합
        combined_scores = self._normalize_and_combine(
            query, dense_results, sparse_results
        )
        
        # 상위 K개 결과 반환
        sorted_results = sorted(
            combined_scores.items(), 
            key=lambda x: x[1], 
            reverse=True
        )[:self.top_k]
        
        result_docs = []
        for doc_id, score in sorted_results:
            for doc in self.documents:
                if doc.doc_id == doc_id:
                    result_docs.append(doc)
                    break
        
        return result_docs
    
    def _normalize_and_combine(
        self, 
        query: str, 
        dense_results: List[Document], 
        sparse_results: List[Document]
    ) -> Dict[str, float]:
        """스스로 스코어 정규화 및 결합"""
        scores = {}
        
        # Dense 스코어 (단순 순위 기반)
        for rank, doc in enumerate(dense_results):
            dense_score = 1.0 / (rank + 1)
            scores[doc.doc_id] = scores.get(doc.doc_id, 0) + \
                self.dense_weight * dense_score
        
        # Sparse 스코어 (BM25 원시 스코어 정규화)
        for rank, doc in enumerate(sparse_results):
            sparse_score = 1.0 / (rank + 1)
            scores[doc.doc_id] = scores.get(doc.doc_id, 0) + \
                self.sparse_weight * sparse_score
        
        return scores

5단계: 메인 실행 코드

# main.py (계속)
def load_documents(filepath: str) -> List[Document]:
    """JSON 파일에서 문서 로드"""
    with open(filepath, 'r', encoding='utf-8') as f:
        data = json.load(f)
    
    documents = []
    for item in data:
        doc = Document(
            text=item['content'],
            doc_id=item['id'],
            metadata=item['metadata']
        )
        documents.append(doc)
    
    return documents

def main():
    # 문서 로드
    documents = load_documents('data/documents.json')
    print(f"문서 로드 완료: {len(documents)}개\n")
    
    # Hybrid Retriever 생성
    hybrid_retriever = HybridRetriever(
        documents=documents,
        embed_model=Settings.embed_model,
        dense_weight=0.5,
        sparse_weight=0.5,
        top_k=3
    )
    
    # 검색 테스트
    queries = [
        "LlamaIndex 검색 프레임워크",
        "AI API 게이트웨이 결제",
        "벡터 임베딩 retrieval"
    ]
    
    for query in queries:
        print(f"\n{'='*50}")
        print(f"검색어: {query}")
        print('='*50)
        
        results = hybrid_retriever.retrieve(query)
        
        print(f"\n검색 결과 ({len(results)}개):")
        for i, doc in enumerate(results, 1):
            print(f"\n{i}. [ID: {doc.doc_id}]")
            print(f"   {doc.text[:100]}...")
            print(f"   메타데이터: {doc.metadata}")

if __name__ == "__main__":
    main()

실행 결과 예시

# 실행 명령어
python main.py

출력 결과
문서 로드 완료: 3개

[Dense] 3개 문서 임베딩 생성 중...
[Dense] 임베딩 완료. shape: (3, 1536)
[Sparse] 3개 문서 BM25 인덱스 생성 중...
[Sparse] BM25 인덱스 완료

==================================================
검색어: LlamaIndex 검색 프레임워크
==================================================

[Hybrid] Dense 검색: 3개 결과
[Hybrid] Sparse 검색: 3개 결과

검색 결과 (3개):
1. [ID: doc_001]
   LlamaIndex는 LLM 애플리케이션을 위한 데이터 프레임워크입니다...
   메타데이터: {'category': 'framework', 'source': 'official'}

2. [ID: doc_002]
   Hybrid Search는 Dense와 Sparse检索을 결합합니다...
   메타데이터: {'category': 'search', 'source': 'blog'}

3. [ID: doc_003]
   HolySheep AI는 글로벌 AI API 게이트웨이입니다...
   메타데이터: {'category': 'ai-service', 'source': 'review'}

비용 계산 예시

HolySheep AI를 사용한 실제 비용을 계산해보겠습니다. 1000회 검색 시:

• 임베딩 비용: 1000문서 × 1536차원 × $0.02/MTok = $0.00002 (무시 가능)
• 쿼리 임베딩 비용: 1000쿼리 × $0.02/MTok = $0.02
• 총 예상 비용: 약 $0.02 (약 27원)

고급 기능: Reciprocal Rank Fusion

더 정교한 스코어 결합을 위해 Reciprocal Rank Fusion(RRF) 알고리즘을 사용할 수 있습니다.

def reciprocal_rank_fusion(
    result_lists: List[List[Document]], 
    k: int = 60
) -> List[Document]:
    """
    Reciprocal Rank Fusion으로 여러 검색 결과 결합
    
    Args:
        result_lists: 각 retriever의 검색 결과 리스트
        k: RRF 파라미터 (보통 60)
    
    Returns:
       融合된 결과 리스트
    """
    doc_scores = {}
    
    for results in result_lists:
        for rank, doc in enumerate(results):
            # RRF 공식: 1 / (k + rank)
            rrf_score = 1 / (k + rank + 1)
            doc_scores[doc.doc_id] = doc_scores.get(doc.doc_id, 0) + rrf_score
    
    # 스코어 기준 정렬
    sorted_docs = sorted(
        doc_scores.items(), 
        key=lambda x: x[1], 
        reverse=True
    )
    
    # 문서 객체로 변환
    doc_id_map = {doc.doc_id: doc for results in result_lists for doc in results}
    
    return [doc_id_map[doc_id] for doc_id, _ in sorted_docs]

사용 예시
dense_results = dense_retriever.retrieve("검색어")
sparse_results = sparse_retriever.retrieve("검색어")

fused_results = reciprocal_rank_fusion([dense_results, sparse_results], k=60)

성능 최적화 팁

• 배치 임베딩: 여러 문서를 한 번에 임베딩하여 API 호출 횟수 감소
• 인덱스 캐싱: BM25 인덱스를 파일로 저장하여 재사용
• 비동기 처리: asyncio를 활용한 동시 검색 요청

자주 발생하는 오류와 해결책

오류 1: AuthenticationError - Invalid API Key

# 오류 메시지
Error code: 401 - AuthenticationError: Incorrect API key provided

해결 방법
1. HolySheep 대시보드에서 새 API 키 생성
2. config.py의 HOLYSHEEP_API_KEY 값 확인
3. 키 앞뒤 공백 제거 확인

올바른 설정 예시
HOLYSHEEP_API_KEY = "sk-holysheep-xxxxxxxxxxxx"  # 실제 키로 교체
print(f"API 키 길이: {len(HOLYSHEEP_API_KEY)}")  # 디버깅용

오류 2: RateLimitError - 요청 제한 초과

# 오류 메시지
Error code: 429 - RateLimitError: Rate limit exceeded

해결 방법
1. 요청 간 딜레이 추가
import time

def rate_limited_request(func, delay: float = 0.5):
    """速率 제한을 우회하기 위한 래퍼 함수"""
    def wrapper(*args, **kwargs):
        time.sleep(delay)
        return func(*args, **kwargs)
    return wrapper

사용 시
@rate_limited_request(delay=0.5)
def search_with_limit(query):
    return hybrid_retriever.retrieve(query)

2. HolySheep 대시보드에서 요금제 업그레이드 검토
3.批量处理로 요청 수 최적화

오류 3: embedding dimension mismatch

# 오류 메시지
ValueError: embedding dimension mismatch: got 1536, expected 768

해결 방법
1. 임베딩 모델 일관성 확인
EMBEDDING_MODEL = "text-embedding-3-small"  # 1536차원
또는
EMBEDDING_MODEL = "text-embedding-ada-002"   # 1536차원

2. 기존 인덱스 삭제 후 재빌드
import os
if os.path.exists("index_cache.pkl"):
    os.remove("index_cache.pkl")
    print("캐시 삭제 완료. 인덱스를 재빌드합니다.")

3. 모든 문서의 임베딩 차원 일관성 확인
embeddings = [e.embedding for e in response.data]
dimensions = [len(e) for e in embeddings]
if len(set(dimensions)) > 1:
    raise ValueError(f"임베딩 차원 불일치: {set(dimensions)}")

오류 4: UnicodeDecodeError - 파일 인코딩 오류

# 오류 메시지
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xbc in position 0

해결 방법
1. 파일 인코딩 명시적 지정
with open('data/documents.json', 'r', encoding='utf-8') as f:
    data = json.load(f)

2. 다른 인코딩 시도
encodings_to_try = ['utf-8', 'utf-8-sig', 'cp949', 'euc-kr']
for encoding in encodings_to_try:
    try:
        with open('data/documents.json', 'r', encoding=encoding) as f:
            data = json.load(f)
        print(f"성공: {encoding} 인코딩 사용")
        break
    except UnicodeDecodeError:
        continue

3. JSON 파일 BOM 확인 및 제거
with open('data/documents.json', 'rb') as f:
    content = f.read()
if content.startswith(b'\xef\xbb\xbf'):
    content = content[3:]
    with open('data/documents.json', 'wb') as f:
        f.write(content)
    print("BOM 제거 완료")

오류 5: Empty Response - 검색 결과 없음

# 오류 메시지
검색 결과가 빈 리스트로 반환됨

해결 방법
1. 쿼리 토큰화 디버깅
def debug_tokenize(query: str):
    """쿼리 토큰화 디버깅"""
    tokens = re.findall(r'\w+', query.lower())
    print(f"쿼리: {query}")
    print(f"토큰: {tokens}")
    return tokens

debug_tokenize("한국어 검색어")

2. 문서 로드 확인
print(f"로드된 문서 수: {len(documents)}")
for doc in documents:
    print(f"  - {doc.doc_id}: {doc.text[:50]}...")

3. 최소 스코어 threshold 조정
class AdaptiveSparseRetriever(SparseRetriever):
    def retrieve(self, query: str, min_score: float = 0.1) -> List[Document]:
        """최소 스코어 이상의 결과만 반환"""
        results = super().retrieve(query)
        return [r for r in results if r.score >= min_score] if hasattr(results[0], 'score') else results

정리

이번 튜토리얼에서는 HolySheep AI와 LlamaIndex를 사용하여 Hybrid Search를 구현하는 방법을 학습했습니다. 핵심 포인트는:

• Dense Retrieval은 의미적 유사성, Sparse Retrieval은 정확한 키워드 매칭 담당
• 두 방식을 결합하면 더 정확한 검색 결과 획득 가능
• HolySheep AI의 글로벌 API 게이트웨이를 통해 안정적인 임베딩 서비스 이용
• Reciprocal Rank Fusion으로 스코어 결합 최적화 가능

HolySheep AI를 사용하면 단일 API 키로 여러 AI 모델을 통합 관리할 수 있어 개발 생산성이 크게 향상됩니다. 특히 로컬 결제 지원으로 해외 신용카드 없이도 즉시 시작할 수 있습니다.

다음 단계

• Reranker 모델 통합하여 검색 결과 정렬 최적화
• 대규모 문서 색인을 위한 벡터 데이터베이스 도입
• HolySheep AI의 Claude Sonnet 4.5 또는 Gemini 2.5 Flash로 LLM 응답 생성

더 자세한 내용은 HolySheep AI 공식 문서를 참고해주세요. 감사합니다!

👉 HolySheep AI 가입하고 무료 크레딧 받기