RAG 청킹 전략 완전 가이드: 나이브 분할부터 RAPTOR까지

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,        # 토큰 수 기준 (대략 문자 수의 75%)
    chunk_overlap=50,      # 겹치는 토큰 수
    separators=["\n\n", "\n", ".", " ", ""]
    # 위에서부터 우선순위: 단락 > 줄 > 문장 > 단어 > 문자
)

chunks = splitter.split_text(document)
print(f"총 {len(chunks)}개 청크 생성됨")
print(f"첫 번째 청크: {chunks[0][:100]}...")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # 또는 "standard_deviation", "interquartile"
    breakpoint_threshold_amount=95           # 상위 5% 거리 변화를 경계로 사용
)

# 내부 동작:
# 1. 각 문장을 임베딩
# 2. 인접 문장 간 코사인 거리 계산
# 3. 거리가 임계값 이상인 지점에서 분할

chunks = splitter.create_documents([long_document])

from langchain.text_splitter import MarkdownHeaderTextSplitter

# Markdown 헤딩 기반 분할
headers_to_split_on = [
    ("#", "h1"),
    ("##", "h2"),
    ("###", "h3"),
]

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False  # 헤딩을 청크 내에 유지
)

md_header_splits = md_splitter.split_text(markdown_document)

# 각 청크에 메타데이터로 헤딩 정보가 포함됨
# md_header_splits[0].metadata = {'h1': '제품 소개', 'h2': '주요 기능'}
# 이 메타데이터를 필터링에 활용 가능

# 헤딩으로 나눈 후 크기가 너무 크면 추가로 분할
from langchain.text_splitter import RecursiveCharacterTextSplitter

secondary_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512, chunk_overlap=50
)

final_chunks = secondary_splitter.split_documents(md_header_splits)

from langchain.document_loaders import BSHTMLLoader
from bs4 import BeautifulSoup

# HTML 태그 구조를 보존하면서 파싱
# article, section, div.content 등의 의미 있는 태그 단위로 분할

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 부모 청크: 큰 컨텍스트 (LLM에게 전달)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

# 자식 청크: 작고 정확한 단위 (임베딩 및 검색에 사용)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 저장소: 부모 문서 보관
docstore = InMemoryStore()  # 프로덕션에서는 Redis나 DB 사용

# 벡터 스토어: 자식 청크의 임베딩 저장
vectorstore = FAISS.from_documents([], embedding_model)

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=docstore,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# 문서 추가 시 자동으로 부모-자식 계층 생성
retriever.add_documents(documents)

# 검색 시:
# 1. 쿼리와 유사한 자식 청크 찾기 (작고 정확한 검색)
# 2. 해당 자식의 부모 ID 조회
# 3. 부모 청크 반환 (풍부한 컨텍스트)
results = retriever.get_relevant_documents("검색 쿼리")

원본 청크들 (Leaf Nodes):
[제품A 기능1] [제품A 기능2] [제품B 기능1] [제품B 비교]

클러스터링 + 요약 (Level 1):
[제품A 개요: 기능1, 기능2 요약] [제품B 개요: 기능1, 비교 요약]

더 높은 수준 요약 (Level 2):
[전체 제품군 비교 요약]

# RAPTOR 구현 (개념 코드)
from sklearn.mixture import GaussianMixture
import numpy as np

def build_raptor_tree(chunks, embeddings, levels=3):
    """
    RAPTOR 트리를 재귀적으로 구성.
    각 레벨에서 클러스터링 후 요약 생성.
    """
    tree = {'level_0': chunks}

    current_chunks = chunks
    current_embeddings = embeddings

    for level in range(1, levels + 1):
        # 1. GMM으로 클러스터링 (BIC로 최적 k 선택)
        n_clusters = max(1, len(current_chunks) // 5)
        gm = GaussianMixture(n_components=n_clusters, random_state=42)
        gm.fit(current_embeddings)
        labels = gm.predict(current_embeddings)

        # 2. 각 클러스터 내 청크들을 합쳐서 요약
        summaries = []
        for cluster_id in range(n_clusters):
            cluster_chunks = [
                current_chunks[i]
                for i in range(len(current_chunks))
                if labels[i] == cluster_id
            ]
            combined_text = "\n\n".join(cluster_chunks)
            summary = llm.summarize(combined_text)  # LLM으로 요약
            summaries.append(summary)

        tree[f'level_{level}'] = summaries

        # 다음 레벨 입력으로 현재 레벨 요약 사용
        current_chunks = summaries
        current_embeddings = embed(summaries)

    return tree

from ragas.metrics import context_precision, context_recall
from ragas import evaluate
from datasets import Dataset

# 평가 데이터셋: 질문 + 정답 쌍
eval_questions = [
    "제품의 반품 기간은?",
    "배송비는 얼마인가?",
    # ...
]
ground_truths = [
    "30일 이내 반품 가능",
    "5만원 이상 무료 배송",
    # ...
]

# 각 청킹 전략으로 RAG 실행 후 결과 수집
def evaluate_chunking_strategy(strategy_name, retriever):
    results = []
    for q, gt in zip(eval_questions, ground_truths):
        retrieved_docs = retriever.get_relevant_documents(q)
        contexts = [doc.page_content for doc in retrieved_docs]
        results.append({
            "question": q,
            "contexts": contexts,
            "ground_truth": gt
        })

    dataset = Dataset.from_list(results)
    scores = evaluate(dataset, metrics=[context_precision, context_recall])

    print(f"\n=== {strategy_name} ===")
    print(f"Context Precision: {scores['context_precision']:.3f}")
    print(f"Context Recall:    {scores['context_recall']:.3f}")
    return scores

# 각 전략 비교
evaluate_chunking_strategy("Fixed-size 512", fixed_retriever)
evaluate_chunking_strategy("Semantic", semantic_retriever)
evaluate_chunking_strategy("Parent-Child", parent_child_retriever)