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

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,        # in tokens (roughly 75% of character count)
    chunk_overlap=50,      # overlapping tokens at boundaries
    separators=["\n\n", "\n", ".", " ", ""]
    # Priority order: paragraph > line > sentence > word > character
)

chunks = splitter.split_text(document)
print(f"Generated {len(chunks)} chunks")
print(f"First chunk preview: {chunks[0][:100]}...")

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

splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # or "standard_deviation", "interquartile"
    breakpoint_threshold_amount=95           # split at top 5% distance changes
)

# Internal logic:
# 1. Embed each sentence
# 2. Calculate cosine distance between adjacent sentences
# 3. Split where distance exceeds the threshold

chunks = splitter.create_documents([long_document])

from langchain.text_splitter import MarkdownHeaderTextSplitter

# Split by Markdown headings
headers_to_split_on = [
    ("#", "h1"),
    ("##", "h2"),
    ("###", "h3"),
]

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False  # keep heading text in chunks
)

md_header_splits = md_splitter.split_text(markdown_document)

# Each chunk now has heading metadata:
# md_header_splits[0].metadata = {'h1': 'Product Overview', 'h2': 'Key Features'}
# This metadata enables filtered retrieval!

# If any heading-based chunk is still too large, split further
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.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Parent chunks: large context windows (sent to LLM)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

# Child chunks: small, focused units (used for embedding and search)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# Document store: holds the parent documents
docstore = InMemoryStore()  # use Redis or a DB in production

# Vector store: holds child chunk embeddings
vectorstore = FAISS.from_documents([], embedding_model)

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

# Adding documents automatically creates the parent-child hierarchy
retriever.add_documents(documents)

# At query time:
# 1. Find child chunks similar to the query (small, precise search)
# 2. Look up the parent ID for each matching child
# 3. Return the parent chunk (rich context for the LLM)
results = retriever.get_relevant_documents("What is the return policy?")

Leaf Nodes (original chunks):
[Product A Feature 1] [Product A Feature 2] [Product B Feature 1] [Product B Comparison]

Level 1 (cluster + summarize):
[Product A Summary: Feature 1, Feature 2 combined] [Product B Summary: Feature 1, comparison]

Level 2 (higher abstraction):
[Full Product Line Overview]

# RAPTOR implementation (conceptual code)
from sklearn.mixture import GaussianMixture
import numpy as np

def build_raptor_tree(chunks, embeddings, levels=3):
    """
    Recursively build a RAPTOR tree.
    At each level: cluster chunks, then summarize each cluster.
    """
    tree = {'level_0': chunks}

    current_chunks = chunks
    current_embeddings = embeddings

    for level in range(1, levels + 1):
        # 1. Cluster using GMM (select optimal k via BIC)
        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. Combine chunks in each cluster and summarize
        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 call
            summaries.append(summary)

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

        # Use current level summaries as input for the next level
        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

# Evaluation dataset: question + ground truth pairs
eval_questions = [
    "What is the return period?",
    "How much is shipping?",
    # ...
]
ground_truths = [
    "Returns accepted within 30 days",
    "Free shipping on orders over $50",
    # ...
]

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

# Compare strategies head-to-head
evaluate_chunking_strategy("Fixed-size 512", fixed_retriever)
evaluate_chunking_strategy("Semantic", semantic_retriever)
evaluate_chunking_strategy("Parent-Child", parent_child_retriever)