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.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. 親チャンクを返す（LLMへの豊富なコンテキスト）
results = retriever.get_relevant_documents("検索クエリ")

リーフノード（元のチャンク）:
[製品A機能1] [製品A機能2] [製品B機能1] [製品B比較]

クラスタリング+要約（レベル1）:
[製品A概要：機能1、2の要約] [製品B概要：機能1、比較の要約]

より高い抽象レベル（レベル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,000円以上で送料無料",
    # ...
]

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)