RAG 파이프라인 고도화 전략: 청킹, 리랭킹, 하이브리드 검색 최적화

[인덱싱 단계]
문서 수집 → 전처리 → 청킹 → 임베딩 → 벡터 DB 저장 + 역인덱스 저장

[쿼리 단계]
질문 → 쿼리 변환 → 하이브리드 검색 (Dense + Sparse)
     → 리랭킹 → 상위 K개 선택 → 프롬프트 구성 → LLM 생성

from langchain_text_splitters import CharacterTextSplitter

# 고정 크기 청킹 - 가장 기본적인 방식
fixed_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=512,
    chunk_overlap=50,       # 10% 오버랩 권장
    length_function=len,
    is_separator_regex=False,
)

chunks = fixed_splitter.split_text(document_text)
print(f"총 {len(chunks)}개 청크 생성")

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 재귀적 청킹 - 프로덕션 기본 추천
recursive_splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", ". ", " ", ""],
    chunk_size=512,
    chunk_overlap=64,
    length_function=len,
    add_start_index=True,   # 원문 위치 추적용
)

chunks = recursive_splitter.split_documents(documents)

# 각 청크에 메타데이터 자동 포함
for chunk in chunks[:3]:
    print(f"크기: {len(chunk.page_content)}, "
          f"시작 위치: {chunk.metadata.get('start_index')}")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# 시맨틱 청킹 - 의미 단위 분할
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",  # percentile, standard_deviation, interquartile
    breakpoint_threshold_amount=75,          # 75 퍼센타일 이상 차이나면 분할
)

semantic_chunks = semantic_splitter.split_text(document_text)
print(f"시맨틱 청크 수: {len(semantic_chunks)}")
print(f"평균 청크 길이: {sum(len(c) for c in semantic_chunks) / len(semantic_chunks):.0f}")

from openai import OpenAI

client = OpenAI()

def agentic_chunk(text: str, max_chunks: int = 20) -> list[dict]:
    """LLM 기반 에이전틱 청킹"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "주어진 텍스트를 논리적 단위로 분할하세요. "
                    "각 청크는 하나의 완결된 주제를 다뤄야 합니다. "
                    "JSON 배열로 반환하세요: "
                    '[{"title": "청크 제목", "content": "청크 내용", "summary": "한줄 요약"}]'
                ),
            },
            {"role": "user", "content": text[:8000]},  # 토큰 제한 주의
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    import json
    result = json.loads(response.choices[0].message.content)
    return result.get("chunks", [])

# 사용 예시
chunks = agentic_chunk(long_document_text)
for chunk in chunks:
    print(f"[{chunk['title']}] {chunk['summary']}")

from langchain_openai import OpenAIEmbeddings

# OpenAI 임베딩 - 차원 축소 활용
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",
    dimensions=1024,  # 3072 -> 1024로 축소하여 비용/저장소 절약
)

# BGE-M3: Dense + Sparse 동시 생성 (하이브리드 검색에 최적)
from FlagEmbedding import BGEM3FlagModel

bge_model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)

# Dense와 Sparse 벡터를 동시에 생성
output = bge_model.encode(
    ["RAG 파이프라인 최적화 방법론"],
    return_dense=True,
    return_sparse=True,
)

dense_vector = output["dense_vecs"][0]    # (1024,) float 벡터
sparse_vector = output["lexical_weights"][0]  # 희소 벡터 (단어별 가중치)

print(f"Dense 차원: {len(dense_vector)}")
print(f"Sparse 활성 토큰 수: {len(sparse_vector)}")

질문: "Kubernetes HPA 설정에서 CPU 임계값은?"

Dense 검색 (벡터 유사도):
  → "컨테이너 오케스트레이션에서 자동 스케일링 설정 방법" (의미적 유사)

Sparse 검색 (BM25 키워드 매칭):
  → "HPA의 targetCPUUtilizationPercentage를 80으로 설정" (키워드 정확 매칭)

하이브리드 결합 (RRF 또는 가중합):
  → 두 결과를 병합하여 최적의 문서 반환

import weaviate, { WeaviateClient } from 'weaviate-client'

// Weaviate 클라이언트 연결
const client: WeaviateClient = await weaviate.connectToLocal({
  host: 'localhost',
  port: 8080,
})

// 하이브리드 검색 실행
const collection = client.collections.get('Documents')

const result = await collection.query.hybrid('Kubernetes HPA CPU 임계값', {
  alpha: 0.7, // 0 = 순수 BM25, 1 = 순수 벡터, 0.7 = 벡터 70%
  limit: 20, // 리랭킹 전 후보 수
  fusionType: 'RelativeScore', // RelativeScore 또는 Ranked
  returnMetadata: ['score', 'explainScore'],
  returnProperties: ['title', 'content', 'source'],
})

for (const item of result.objects) {
  console.log(`[${item.metadata?.score?.toFixed(3)}] ${item.properties.title}`)
}

from rank_bm25 import BM25Okapi
import numpy as np
from typing import List, Tuple

def hybrid_search(
    query: str,
    documents: list[dict],
    dense_scores: np.ndarray,
    k: int = 10,
    alpha: float = 0.7,
    rrf_k: int = 60,
) -> list[dict]:
    """
    RRF 기반 하이브리드 검색
    alpha: Dense 검색 가중치 (1-alpha가 Sparse 가중치)
    rrf_k: RRF 상수 (기본값 60, 논문 권장)
    """
    # Sparse 검색 (BM25)
    tokenized_docs = [doc["content"].split() for doc in documents]
    bm25 = BM25Okapi(tokenized_docs)
    sparse_scores = bm25.get_scores(query.split())

    # Dense 순위 계산
    dense_ranks = np.argsort(-dense_scores) + 1  # 1-indexed rank
    sparse_ranks = np.argsort(-sparse_scores) + 1

    # RRF 점수 계산
    rrf_scores = []
    for i in range(len(documents)):
        dense_rrf = alpha / (rrf_k + dense_ranks[i])
        sparse_rrf = (1 - alpha) / (rrf_k + sparse_ranks[i])
        rrf_scores.append(dense_rrf + sparse_rrf)

    # 상위 K개 반환
    top_indices = np.argsort(rrf_scores)[::-1][:k]
    return [
        {**documents[i], "hybrid_score": rrf_scores[i]}
        for i in top_indices
    ]

# 사용 예시
results = hybrid_search(
    query="Kubernetes HPA CPU threshold",
    documents=all_documents,
    dense_scores=embedding_similarity_scores,
    k=20,       # 리랭킹 전이므로 넉넉하게
    alpha=0.7,  # Dense 70%, Sparse 30%
)

import cohere

co = cohere.ClientV2(api_key="your-cohere-api-key")

def rerank_with_cohere(
    query: str,
    documents: list[str],
    top_n: int = 5,
) -> list[dict]:
    """Cohere Rerank 4로 문서 재순위"""
    response = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=documents,
        top_n=top_n,
        return_documents=True,
    )

    results = []
    for item in response.results:
        results.append({
            "index": item.index,
            "score": item.relevance_score,
            "text": item.document.text if item.document else documents[item.index],
        })
    return results

# 사용 예시: 하이브리드 검색 결과 20건을 리랭킹하여 상위 5건 선택
hybrid_results = hybrid_search(query, documents, dense_scores, k=20)
reranked = rerank_with_cohere(
    query="Kubernetes HPA CPU 임계값 설정 방법",
    documents=[r["content"] for r in hybrid_results],
    top_n=5,
)

for r in reranked:
    print(f"[{r['score']:.4f}] {r['text'][:80]}...")

from FlagEmbedding import FlagReranker

# BGE Reranker v2 M3 - 다국어 지원 경량 리랭커
reranker = FlagReranker(
    "BAAI/bge-reranker-v2-m3",
    use_fp16=True,  # GPU 메모리 절약
)

def rerank_with_bge(
    query: str,
    documents: list[str],
    top_n: int = 5,
) -> list[dict]:
    """BGE Reranker로 문서 재순위"""
    # 쿼리-문서 쌍 생성
    pairs = [[query, doc] for doc in documents]

    # 관련도 점수 계산
    scores = reranker.compute_score(pairs, normalize=True)

    # 점수 기준 정렬
    scored_docs = [
        {"index": i, "score": s, "text": d}
        for i, (s, d) in enumerate(zip(scores, documents))
    ]
    scored_docs.sort(key=lambda x: x["score"], reverse=True)

    return scored_docs[:top_n]

# 사용 예시
results = rerank_with_bge(
    query="RAG 파이프라인에서 청킹 크기를 결정하는 기준은?",
    documents=candidate_documents,
    top_n=5,
)

for r in results:
    print(f"[{r['score']:.4f}] {r['text'][:100]}...")

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class OptimizedRAGPipeline:
    """고도화된 RAG 파이프라인"""

    def __init__(self):
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512, chunk_overlap=64
        )
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-large", dimensions=1024
        )
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.reranker = FlagReranker("BAAI/bge-reranker-v2-m3", use_fp16=True)

    def query(self, question: str, top_k: int = 5) -> str:
        # 1. 하이브리드 검색: 후보 20건
        candidates = self._hybrid_search(question, k=20)

        # 2. 리랭킹: 상위 5건 선택
        reranked = self._rerank(question, candidates, top_n=top_k)

        # 3. 프롬프트 구성 및 LLM 생성
        context = "\n\n---\n\n".join([doc["text"] for doc in reranked])
        prompt = ChatPromptTemplate.from_messages([
            ("system", "다음 컨텍스트를 기반으로 질문에 답변하세요.\n\n{context}"),
            ("human", "{question}"),
        ])
        chain = prompt | self.llm
        response = chain.invoke({"context": context, "question": question})
        return response.content

    def _hybrid_search(self, query: str, k: int = 20) -> list[dict]:
        # Dense + Sparse 하이브리드 검색 (위 코드 참조)
        ...

    def _rerank(self, query: str, docs: list[dict], top_n: int) -> list[dict]:
        pairs = [[query, doc["text"]] for doc in docs]
        scores = self.reranker.compute_score(pairs, normalize=True)
        for doc, score in zip(docs, scores):
            doc["rerank_score"] = score
        docs.sort(key=lambda x: x["rerank_score"], reverse=True)
        return docs[:top_n]

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# 평가 데이터셋 구성
eval_data = {
    "question": [
        "Kubernetes HPA의 CPU 임계값 기본값은?",
        "RAG에서 청킹 크기를 결정하는 기준은?",
    ],
    "answer": [
        "HPA의 CPU 임계값 기본값은 80%입니다.",
        "쿼리 유형에 따라 256-1024 토큰 범위에서 결정합니다.",
    ],
    "contexts": [
        ["HPA는 targetCPUUtilizationPercentage 기본값 80을 사용합니다."],
        ["팩토이드 쿼리는 256-512, 분석형은 1024+ 토큰을 권장합니다."],
    ],
    "ground_truth": [
        "기본값은 80%이다.",
        "쿼리 유형과 문서 특성에 따라 결정한다.",
    ],
}

dataset = Dataset.from_dict(eval_data)

# 평가 실행
results = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# faithfulness: 0.92, answer_relevancy: 0.88,
# context_precision: 0.85, context_recall: 0.90

from deepeval import assert_test
from deepeval.test_case import LLMTestCase
from deepeval.metrics import (
    FaithfulnessMetric,
    ContextualRelevancyMetric,
    AnswerRelevancyMetric,
)

def test_rag_faithfulness():
    """RAG 응답이 컨텍스트에 충실한지 테스트"""
    test_case = LLMTestCase(
        input="Kubernetes HPA CPU 임계값은?",
        actual_output="HPA의 CPU 임계값 기본값은 80%입니다.",
        retrieval_context=[
            "HPA는 targetCPUUtilizationPercentage 기본값 80을 사용합니다."
        ],
    )

    faithfulness = FaithfulnessMetric(threshold=0.8)
    relevancy = ContextualRelevancyMetric(threshold=0.7)
    answer_rel = AnswerRelevancyMetric(threshold=0.7)

    assert_test(test_case, [faithfulness, relevancy, answer_rel])

# pytest로 실행: pytest test_rag.py -v

# Blue-Green 인덱스 전환 예시
import time

def reindex_with_blue_green(
    old_collection: str,
    new_collection: str,
    new_embedding_model: str,
):
    """무중단 재인덱싱"""
    # 1. 새 컬렉션에 인덱싱 (기존 서비스는 old_collection 사용 중)
    print(f"새 컬렉션 '{new_collection}' 인덱싱 시작...")
    create_and_populate_collection(new_collection, new_embedding_model)

    # 2. 검증: 새 컬렉션에 대해 테스트 쿼리 실행
    test_results = run_evaluation_suite(new_collection)
    if test_results["context_precision"] < 0.8:
        raise ValueError(
            f"새 인덱스 품질 미달: {test_results['context_precision']:.2f}"
        )

    # 3. 트래픽 전환: alias 또는 설정 변경
    update_active_collection(new_collection)
    print(f"트래픽 전환 완료: {old_collection} -> {new_collection}")

    # 4. 이전 컬렉션은 롤백 대비 일정 기간 유지
    schedule_cleanup(old_collection, delay_days=7)

import tiktoken

enc = tiktoken.encoding_for_model("text-embedding-3-small")

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    length_function=lambda text: len(enc.encode(text)),  # 토큰 수 기준
)

import asyncio

async def rerank_with_fallback(
    query: str,
    documents: list[str],
    top_n: int = 5,
    timeout: float = 2.0,
) -> list[dict]:
    """리랭킹 with 폴백"""
    try:
        # Primary: Cohere Rerank (타임아웃 2초)
        result = await asyncio.wait_for(
            cohere_rerank_async(query, documents, top_n),
            timeout=timeout,
        )
        return result
    except (asyncio.TimeoutError, Exception) as e:
        print(f"Cohere 리랭킹 실패, BGE 폴백: {e}")
        try:
            # Fallback: 셀프호스팅 BGE Reranker
            return rerank_with_bge(query, documents, top_n)
        except Exception as e2:
            print(f"BGE 리랭킹도 실패, 원본 순위 반환: {e2}")
            # 최종 폴백: 하이브리드 검색 결과 그대로 반환
            return [
                {"index": i, "score": 1.0 - i * 0.05, "text": d}
                for i, d in enumerate(documents[:top_n])
            ]