하이브리드 검색 RAG: 벡터检索+키워드 BM25融合 구현 완전 가이드

안녕하세요, 저는 HolySheep AI의 기술 작가입니다. 이번 튜토리얼에서는 하이브리드 검색 RAG를 구현하는 방법을 초보자도 이해할 수 있도록 단계별로 알려드리겠습니다.

하이브리드 검색 RAG란?

RAG(Retrieval-Augmented Generation)는 AI가 외부 지식을 검색해서 답변의 정확성을 높이는 기술입니다. 흔히 사용하는 벡터 검색만으로는 정확한 키워드 매칭이 어렵고, 키워드 검색만으로는 의미적 유사성을 파악하지 못합니다.

하이브리드 검색은 이 두 가지 방법의 장점을 결합합니다:

• 벡터 검색: 의미적 유사성 파악에优秀 (예: "큰犬" 검색 시 "개" 관련 문서도 찾음)
• BM25 키워드 검색: 정확한 단어 매칭에优秀 (예: 제품명, 고유명사 검색)

필수 개념 정리

벡터检索(임베딩)

문서를 숫자 나열(벡터)로 변환하여 의미적 유사도를 계산합니다. HolySheep AI의 임베딩 모델을 사용하면 한글 문서에 최적화된 벡터를 생성할 수 있습니다.

BM25

기존 검색 엔진에서 사용하는 알고리즘으로, 키워드의 중요도와 문서 빈도를 고려하여 순위를 매깁니다. "Okapi BM25"라고도 불립니다.

개발 환경 설정

# 필수 패키지 설치
pip install numpy scikit-learn rank-bm25 openai python-dotenv

패키지 설명:
- numpy: 벡터 연산
- scikit-learn: 코사인 유사도 계산
- rank-bm25: BM25 알고리즘 구현
- openai: HolySheep AI API 호출
- python-dotenv: 환경 변수 관리

1단계: HolySheep AI API 키 설정

# .env 파일 생성
HolySheep AI에서 받은 API 키를 입력하세요
https://www.holysheep.ai/register 에서 무료 가입 후 API 키 발급

echo "HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY" > .env

Python에서 환경 변수 로드
from dotenv import load_dotenv
import os

load_dotenv()

API_KEY = os.getenv("HOLYSHEEP_API_KEY")
BASE_URL = "https://api.holysheep.ai/v1"

⚠️ 중요: base_url에 절대 api.openai.com 사용 금지
HolySheep AI가 모든 모델을 통합 게이트웨이로 제공합니다

2단계: 문서 임베딩 생성

from openai import OpenAI
import numpy as np

HolySheep AI 클라이언트 초기화
client = OpenAI(
    api_key=API_KEY,
    base_url=BASE_URL  # 반드시 HolySheep 게이트웨이 사용
)

def create_embeddings(texts, model="embedding-3"):
    """문서 목록에서 HolySheep AI 임베딩 생성"""
    response = client.embeddings.create(
        model=model,
        input=texts
    )
    # 반환된 임베딩을 numpy 배열로 변환
    embeddings = [np.array(item.embedding) for item in response.data]
    return np.array(embeddings)

샘플 문서 준비
documents = [
    "Python은 Interpreted 언어이다",
    "Java는 객체지향 프로그래밍 언어이다",
    "JavaScript는 웹 개발에 사용되는 스크립트 언어이다",
    "TypeScript는 JavaScript에 타입을 추가한 언어이다",
    "Go는 구글이 개발한 프로그래밍 언어이다",
    "Rust는 안전한 메모리 관리 시스템을 가진 언어이다",
    "Swift는 Apple이 개발한 프로그래밍 언어이다",
    "Kotlin은 Android 개발에 사용되는 언어이다"
]

문서 임베딩 생성
print("문서 임베딩 생성 중...")
doc_embeddings = create_embeddings(documents)
print(f"✅ 생성 완료: {len(documents)}개 문서, 각 {doc_embeddings.shape[1]}차원 벡터")

3단계: BM25 키워드 검색 구현

import re
from rank_bm25 import BM25Okapi

def tokenize_korean(text):
    """한국어 토큰화 (공백 기반 단순 분할)"""
    # 실제 프로젝트에서는 KoNLPy 등 전문 한국어 토크나이저 사용 권장
    tokens = re.sub(r'[^\w\s]', ' ', text.lower())
    return tokens.split()

문서 토큰화
tokenized_docs = [tokenize_korean(doc) for doc in documents]

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

def search_bm25(query, top_k=3):
    """BM25 키워드 검색 수행"""
    query_tokens = tokenize_korean(query)
    scores = bm25.get_scores(query_tokens)
    
    # 점수 기준 상위 문서 선택
    top_indices = np.argsort(scores)[::-1][:top_k]
    
    results = []
    for idx in top_indices:
        results.append({
            'document': documents[idx],
            'score': scores[idx]
        })
    return results

테스트 검색
query = "Apple 개발 언어"
results = search_bm25(query)
print(f"BM25 검색 결과 ('{query}'):")
for i, r in enumerate(results, 1):
    print(f"  {i}. {r['document']} (점수: {r['score']:.4f})")

4단계: 벡터 검색 구현

from sklearn.metrics.pairwise import cosine_similarity

def search_vector(query, doc_embeddings, documents, top_k=3):
    """벡터 유사도 검색 수행"""
    # 쿼리 임베딩 생성
    query_embedding = create_embeddings([query])
    
    # 코사인 유사도 계산
    similarities = cosine_similarity(query_embedding, doc_embeddings)[0]
    
    # 상위 결과 선택
    top_indices = np.argsort(similarities)[::-1][:top_k]
    
    results = []
    for idx in top_indices:
        results.append({
            'document': documents[idx],
            'score': float(similarities[idx])
        })
    return results

테스트 검색
query = "Apple 개발 언어"
results = search_vector(query, doc_embeddings, documents)
print(f"벡터 검색 결과 ('{query}'):")
for i, r in enumerate(results, 1):
    print(f"  {i}. {r['document']} (유사도: {r['score']:.4f})")

5단계: 하이브리드 검색 Fusion 구현

def normalize_scores(scores, method='min-max'):
    """점수 정규화 (0~1 범위로 변환)"""
    scores = np.array(scores)
    if method == 'min-max':
        if scores.max() == scores.min():
            return np.ones_like(scores) * 0.5
        return (scores - scores.min()) / (scores.max() - scores.min())
    elif method == 'rank':
        # 순위 기반 정규화
        return np.argsort(np.argsort(scores)) / len(scores)
    return scores

def hybrid_search(query, documents, doc_embeddings, bm25, top_k=3, alpha=0.5):
    """
    하이브리드 검색: 벡터 검색 + BM25 결합
    
    Args:
        alpha: 벡터 검색 가중치 (1-alpha는 BM25 가중치)
        alpha=0.5: 두 검색 방법 동등하게对待
        alpha=0.7: 벡터 검색 중심
        alpha=0.3: 키워드 검색 중심
    """
    # 각 검색 방법 수행
    vector_results = search_vector(query, doc_embeddings, documents, top_k=len(documents))
    bm25_results = search_bm25(query, top_k=len(documents))
    
    # 점수 추출
    vector_scores = [r['score'] for r in vector_results]
    bm25_scores = [r['score'] for r in bm25_results]
    
    # 정규화
    vector_scores_norm = normalize_scores(vector_scores)
    bm25_scores_norm = normalize_scores(bm25_scores)
    
    # 결합 점수 계산
    combined_scores = []
    for i, doc in enumerate(documents):
        v_score = vector_scores_norm[i]
        b_score = bm25_scores_norm[i]
        combined = alpha * v_score + (1 - alpha) * b_score
        combined_scores.append({
            'document': doc,
            'vector_score': v_score,
            'bm25_score': b_score,
            'combined_score': combined
        })
    
    # 결합 점수 기준 정렬
    combined_scores.sort(key=lambda x: x['combined_score'], reverse=True)
    
    return combined_scores[:top_k]

하이브리드 검색 테스트
print("=" * 60)
query = "Apple 개발 언어"
print(f"하이브리드 검색 쿼리: '{query}'")
print("=" * 60)

results = hybrid_search(query, documents, doc_embeddings, bm25, alpha=0.5)
print("\n🔍 검색 결과 (벡터 50% + BM25 50%):")
for i, r in enumerate(results, 1):
    print(f"  {i}. {r['document']}")
    print(f"     벡터 점수: {r['vector_score']:.4f} | BM25 점수: {r['bm25_score']:.4f} | 결합: {r['combined_score']:.4f}")
    print()

6단계: HolySheep AI로 RAG 답변 생성

def generate_rag_answer(query, documents, doc_embeddings, bm25, alpha=0.5):
    """
    하이브리드 검색 기반 RAG 답변 생성
    HolySheep AI의 저렴한 모델로 비용 최적화
    """
    # 1단계: 하이브리드 검색으로 관련 문서 검색
    search_results = hybrid_search(query, documents, doc_embeddings, bm25, top_k=3, alpha=alpha)
    context = "\n".join([f"- {r['document']}" for r in search_results])
    
    # 2단계: HolySheep AI로 답변 생성
    # 💰 비용 최적화 팁: DeepSeek V3.2는 $0.42/MTok으로 매우 저렴
    response = client.chat.completions.create(
        model="deepseek-chat",  # HolySheep AI에서 제공하는 모델
        messages=[
            {"role": "system", "content": "당신은 질문에 도움을 주는 AI 어시스턴트입니다. 검색된 정보를 바탕으로 답변하세요."},
            {"role": "user", "content": f"질문: {query}\n\n검색된 관련 문서:\n{context}\n\n위 문서를 참고하여 답변해 주세요."}
        ],
        temperature=0.3,
        max_tokens=500
    )
    
    answer = response.choices[0].message.content
    
    return {
        'answer': answer,
        'sources': [r['document'] for r in search_results],
        'model': 'deepseek-chat',
        'usage': {
            'prompt_tokens': response.usage.prompt_tokens,
            'completion_tokens': response.usage.completion_tokens,
            'total_tokens': response.usage.total_tokens
        }
    }

RAG 테스트
print("=" * 60)
query = "Apple이 개발한 언어와 관련된 프로그래밍 언어들을 알려주세요"
print(f"RAG 쿼리: '{query}'")
print("=" * 60)

result = generate_rag_answer(query, documents, doc_embeddings, bm25, alpha=0.5)

print("\n📝 AI 답변:")
print(result['answer'])
print(f"\n📚 참고 문서: {result['sources']}")
print(f"💰 사용된 토큰: {result['usage']['total_tokens']} (DeepSeek V3.2 기준 약 ${result['usage']['total_tokens']/1000000*0.42:.6f})")

검색 방법 비교 테스트

def compare_search_methods(query, documents, doc_embeddings, bm25):
    """세 가지 검색 방법 비교"""
    print(f"\n{'='*60}")
    print(f"검색 쿼리: '{query}'")
    print('='*60)
    
    # BM25만
    print("\n📌 BM25 키워드 검색 결과:")
    bm25_results = search_bm25(query, top_k=3)
    for i, r in enumerate(bm25_results, 1):
        print(f"  {i}. {r['document']} (점수: {r['score']:.4f})")
    
    # 벡터만
    print("\n📌 벡터 검색 결과:")
    vector_results = search_vector(query, doc_embeddings, documents, top_k=3)
    for i, r in enumerate(vector_results, 1):
        print(f"  {i}. {r['document']} (유사도: {r['score']:.4f})")
    
    # 하이브리드
    print("\n📌 하이브리드 검색 결과 (alpha=0.5):")
    hybrid_results = hybrid_search(query, documents, doc_embeddings, bm25, top_k=3, alpha=0.5)
    for i, r in enumerate(hybrid_results, 1):
        print(f"  {i}. {r['document']}")
        print(f"     (결합 점수: {r['combined_score']:.4f})")

다양한 쿼리로 테스트
test_queries = [
    "Apple 개발 언어",
    "안전한 메모리 관리",
    "웹 개발에 사용되는"
]

for query in test_queries:
    compare_search_methods(query, documents, doc_embeddings, bm25)

자주 발생하는 오류와 해결

오류 1: API 키 미설정 오류

# ❌ 잘못된 코드
client = OpenAI(api_key=YOUR_HOLYSHEEP_API_KEY)  # 환경 변수에서 로드 안 함

✅ 올바른 코드
from dotenv import load_dotenv
import os

load_dotenv()  # .env 파일 로드 필수

client = OpenAI(
    api_key=os.getenv("HOLYSHEEP_API_KEY"),  # 환경 변수에서 안전하게 로드
    base_url="https://api.holysheep.ai/v1"
)

⚠️ HolySheep AI 가입: https://www.holysheep.ai/register

오류 2: base_url 설정 누락

# ❌ 잘못된 코드 - API 호출 시 오류 발생
client = OpenAI(api_key=API_KEY)
API가 openai.com으로 직접 연결되어 HolySheep 요금제에 접근 불가

✅ 올바른 코드 - HolySheep 게이트웨이 명시적 지정
client = OpenAI(
    api_key=API_KEY,
    base_url="https://api.holysheep.ai/v1"  # 반드시 설정
)

오류 3: 임베딩 API 응답 형식 오류

# ❌ 잘못된 코드
embeddings = response.data  # 리스트 형태 그대로 사용

✅ 올바른 코드 - 각 임베딩 접근 방식
response = client.embeddings.create(
    model="embedding-3",
    input=texts
)

올바른 접근 방식 1: 리스트 컴프리헨션
embeddings = [np.array(item.embedding) for item in response.data]

올바른 접근 방식 2: 인덱스로 접근
if len(response.data) > 0:
    first_embedding = np.array(response.data[0].embedding)
    print(f"임베딩 차원: {len(first_embedding)}")

오류 4: BM25 빈 결과 반환

# ❌ 문제 상황: 한국어 토큰화 미흡으로 인한 빈 결과
query = "Apple 개발 언어"
tokenize_korean(query)  # ['apple', '개발', '언어'] - 정확도 낮음

✅ 해결 방법 1: 한국어 전용 토크나이저 사용
from konlpy.tag import Okt

okt = Okt()
def tokenize_korean_morpheme(text):
    """형태소 분석 기반 토큰화"""
    morphemes = okt.morphs(text.lower())
    return morphemes

✅ 해결 방법 2: 영어+한국어 혼합 토큰화
def tokenize_mixed(text):
    """영어는 그대로, 한국어는 형태소 분석"""
    text_lower = text.lower()
    # 영어 알파벳 단어 추출
    english_words = re.findall(r'[a-z]+', text_lower)
    # 한국어 형태소 분석
    korean_morphemes = okt.morphs(text)
    return english_words + korean_morphemes

테스트
print(tokenize_korean_morpheme("Apple 개발 언어"))  # ['apple', '개발', '언어']
print(tokenize_mixed("Apple 개발 언어"))  # ['apple', '개발', '언어']

오류 5: 토큰 제한 초과

# ❌ 문제 상황: 긴 문서 여러 개 임베딩 시 토큰 초과
long_documents = ["очень длинный текст..." * 1000 for _ in range(100)]

✅ 해결 방법: 문서 청킹
def chunk_text(text, chunk_size=500, overlap=50):
    """긴 텍스트를 작은 청크로 분할"""
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunk = ' '.join(words[start:end])
        chunks.append(chunk)
        start = end - overlap  # 오버랩으로 문맥 유지
    return chunks

HolySheep AI 임베딩 모델별 최대 입력
embedding-3: 최대 8191 토큰
embedding-3-small: 최대 8191 토큰
MAX_TOKENS = 8000  # 안전 마진 포함

def create_embeddings_safe(client, texts, model="embedding-3"):
    """토큰 제한을 고려한 안전한 임베딩 생성"""
    all_embeddings = []
    for text in texts:
        # 토큰 수 추정 (대략적으로 글자수/2)
        estimated_tokens = len(text) // 2
        if estimated_tokens > MAX_TOKENS:
            # 청킹 후 개별 임베딩
            chunks = chunk_text(text, chunk_size=4000, overlap=100)
            chunk_embeddings = []
            for chunk in chunks:
                chunk_emb = create_embeddings([chunk])
                chunk_embeddings.append(chunk_emb[0])
            # 청크 임베딩 평균
            avg_embedding = np.mean(chunk_embeddings, axis=0)
            all_embeddings.append(avg_embedding)
        else:
            all_embeddings.append(create_embeddings([text])[0])
    return np.array(all_embeddings)

성능 최적화 팁

• 가중치 조정: alpha=0.5에서 시작하여 도메인에 따라 조절 (기술 문서는 벡터 검색 비중 높게, 제품 카탈로그는 BM25 비중 높게)
• 인덱싱 최적화: 문서 수 많으면 FAISS, Annoy 등 벡터 데이터베이스 사용
• 캐싱: 자주 검색되는 쿼리의 임베딩 결과를 Redis에 캐싱
• 모델 선택: HolySheep AI에서 DeepSeek V3.2($0.42/MTok)로 RAG 구성 시 비용大幅 절감

결론

이번 튜토리얼에서는 하이브리드 검색 RAG를 구현하는 방법을 다루었습니다. 벡터 검색의 의미적 이해력과 BM25의 정확한 키워드 매칭을 결합하면 더 정확한 검색 결과를 얻을 수 있습니다.

HolySheep AI를 사용하면:

• 단일 API 키로 임베딩 + 채팅 모델 통합 접근 가능
• DeepSeek V3.2($0.42/MTok)로 RAG 파이프라인 비용 최적화
• 해외 신용카드 없이 로컬 결제 지원

궁금한 점이 있으면 지금 가입하여 HolySheep AI의 기술 문서를 참고하세요.

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