機械学習と情報技術

【RAG】テキストチャンキング戦略の比較と実装

NLP

RAG（検索拡張生成）システムの性能を大きく左右するのが、文書をどのように分割（チャンキング）するかです。チャンクが小さすぎると文脈が失われ、大きすぎると検索精度が低下します。

本記事では、テキストチャンキングの各種戦略を数学的な観点から分析し、Pythonでの実装方法を解説します。

本記事の内容

  • チャンキングの基本概念と重要性
  • 固定長チャンキング、オーバーラップチャンキング
  • セマンティックチャンキング
  • 各手法のPython実装と比較

前提知識

この記事を読む前に、以下の記事を読んでおくと理解が深まります。

チャンキングとは

チャンキングとは、長い文書を小さな単位（チャンク）に分割する処理です。RAGにおいては、各チャンクが検索の単位となります。

チャンキングの目的

  1. 検索精度の向上: 関連性の高い部分のみを取得
  2. コンテキスト長の制約: LLMの入力長制限に対応
  3. 計算効率: 必要な情報のみを処理

トレードオフ

チャンクサイズには以下のトレードオフがあります。

チャンクサイズ メリット デメリット
小さい 検索精度が高い 文脈が失われやすい
大きい 文脈が保持される ノイズが増える

固定長チャンキング

アルゴリズム

最もシンプルな方法は、固定の文字数またはトークン数で分割することです。

テキスト $T$ を長さ $L$ のチャンクに分割:

$$ C_i = T[i \cdot L : (i+1) \cdot L], \quad i = 0, 1, 2, \ldots, \lfloor |T|/L \rfloor $$

ここで $|T|$ はテキストの長さです。

Pythonでの実装

def fixed_length_chunking(text, chunk_size=500):
    """固定長チャンキング"""
    chunks = []
    for i in range(0, len(text), chunk_size):
        chunk = text[i:i + chunk_size]
        chunks.append(chunk)
    return chunks

# 使用例
text = """人工知能（AI）は、機械が人間のような知能を持つ技術です。
機械学習は、AIの一分野で、データから学習するアルゴリズムを研究します。
深層学習は、多層のニューラルネットワークを使った機械学習の手法です。
大規模言語モデル（LLM）は、深層学習を用いて自然言語を理解・生成します。
RAGは、検索と生成を組み合わせた手法で、LLMの性能を向上させます。"""

chunks = fixed_length_chunking(text, chunk_size=100)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i}: {chunk[:50]}...")

問題点

固定長チャンキングには以下の問題があります。

  1. 文の途中で分割: 意味のある単位で分割されない
  2. コンテキストの分断: 関連する情報が別のチャンクに分かれる

オーバーラップチャンキング

アルゴリズム

チャンク間に重複部分（オーバーラップ）を設けることで、境界での情報損失を軽減します。

オーバーラップ $O$ を持つチャンク:

$$ C_i = T[i \cdot (L – O) : i \cdot (L – O) + L] $$

ストライド（移動量）は $S = L – O$ となります。

数学的性質

オーバーラップ率 $r = O / L$ とすると、チャンク数は:

$$ N = \left\lceil \frac{|T| – L}{L(1-r)} \right\rceil + 1 $$

オーバーラップなし（$r=0$）と比較した冗長度:

$$ \text{Redundancy} = \frac{1}{1-r} $$

例えば、$r=0.2$（20%オーバーラップ）の場合、チャンク数は約1.25倍になります。

Pythonでの実装

def overlap_chunking(text, chunk_size=500, overlap=100):
    """オーバーラップチャンキング"""
    chunks = []
    stride = chunk_size - overlap

    for i in range(0, len(text), stride):
        chunk = text[i:i + chunk_size]
        if len(chunk) > overlap:  # 最後の短いチャンクを除外
            chunks.append(chunk)
        if i + chunk_size >= len(text):
            break

    return chunks

# 使用例
text = "A" * 1000  # 1000文字のテキスト

chunks_no_overlap = fixed_length_chunking(text, chunk_size=200)
chunks_with_overlap = overlap_chunking(text, chunk_size=200, overlap=50)

print(f"No overlap: {len(chunks_no_overlap)} chunks")
print(f"With overlap (50): {len(chunks_with_overlap)} chunks")

文・段落ベースのチャンキング

文単位の分割

自然な境界（文の終わり）で分割することで、意味の分断を防ぎます。

import re

def sentence_chunking(text, max_chunk_size=500):
    """文単位でチャンキング（最大サイズ制約付き）"""
    # 文に分割（日本語対応）
    sentences = re.split(r'(?<=[。！？\.\!\?])', text)
    sentences = [s.strip() for s in sentences if s.strip()]

    chunks = []
    current_chunk = ""

    for sentence in sentences:
        if len(current_chunk) + len(sentence) <= max_chunk_size:
            current_chunk += sentence
        else:
            if current_chunk:
                chunks.append(current_chunk)
            current_chunk = sentence

    if current_chunk:
        chunks.append(current_chunk)

    return chunks

# 使用例
text = """人工知能は急速に発展しています。特に深層学習の進歩は目覚ましいです。
大規模言語モデルは自然言語処理を革新しました。GPTやClaude等が代表例です。
RAGは検索と生成を組み合わせた技術です。企業での活用が進んでいます。"""

chunks = sentence_chunking(text, max_chunk_size=100)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i}: {chunk}")

段落単位の分割

段落は通常、一つのトピックを扱うため、意味的なまとまりとして適しています。

def paragraph_chunking(text, max_chunk_size=1000):
    """段落単位でチャンキング"""
    # 段落に分割（空行で分割）
    paragraphs = re.split(r'\n\s*\n', text)
    paragraphs = [p.strip() for p in paragraphs if p.strip()]

    chunks = []
    current_chunk = ""

    for para in paragraphs:
        if len(current_chunk) + len(para) + 2 <= max_chunk_size:
            if current_chunk:
                current_chunk += "\n\n" + para
            else:
                current_chunk = para
        else:
            if current_chunk:
                chunks.append(current_chunk)
            # 段落が最大サイズを超える場合は文単位で分割
            if len(para) > max_chunk_size:
                sub_chunks = sentence_chunking(para, max_chunk_size)
                chunks.extend(sub_chunks[:-1])
                current_chunk = sub_chunks[-1] if sub_chunks else ""
            else:
                current_chunk = para

    if current_chunk:
        chunks.append(current_chunk)

    return chunks

セマンティックチャンキング

アルゴリズムの概要

セマンティックチャンキングは、文の意味的な類似度に基づいて分割点を決定します。隣接する文の埋め込みベクトル間の類似度が低い箇所を境界とします。

数学的定式化

文列 $\{s_1, s_2, \ldots, s_n\}$ に対し、隣接文間のコサイン類似度を計算:

$$ \text{sim}_i = \frac{E(s_i) \cdot E(s_{i+1})}{|E(s_i)| |E(s_{i+1})|} $$

類似度が閾値 $\theta$ を下回る位置で分割:

$$ \text{Split at } i \text{ if } \text{sim}_i < \theta $$

Pythonでの実装

import numpy as np
from sentence_transformers import SentenceTransformer
import re

class SemanticChunker:
    def __init__(self, model_name='all-MiniLM-L6-v2', threshold=0.5):
        """セマンティックチャンカーの初期化"""
        self.model = SentenceTransformer(model_name)
        self.threshold = threshold

    def _split_sentences(self, text):
        """文に分割"""
        sentences = re.split(r'(?<=[。！？\.\!\?])', text)
        return [s.strip() for s in sentences if s.strip()]

    def _compute_similarities(self, sentences):
        """隣接文間の類似度を計算"""
        embeddings = self.model.encode(sentences)

        similarities = []
        for i in range(len(embeddings) - 1):
            sim = np.dot(embeddings[i], embeddings[i + 1]) / (
                np.linalg.norm(embeddings[i]) * np.linalg.norm(embeddings[i + 1])
            )
            similarities.append(sim)

        return similarities

    def chunk(self, text, min_chunk_size=100):
        """セマンティックチャンキング"""
        sentences = self._split_sentences(text)

        if len(sentences) <= 1:
            return [text]

        similarities = self._compute_similarities(sentences)

        # 分割点を決定
        chunks = []
        current_chunk = sentences[0]

        for i, sim in enumerate(similarities):
            if sim < self.threshold and len(current_chunk) >= min_chunk_size:
                chunks.append(current_chunk)
                current_chunk = sentences[i + 1]
            else:
                current_chunk += sentences[i + 1]

        if current_chunk:
            chunks.append(current_chunk)

        return chunks

# 使用例
text = """機械学習は人工知能の一分野です。データからパターンを学習します。
深層学習は多層のニューラルネットワークを使います。画像認識で大きな成功を収めました。
一方で、量子コンピューティングは全く異なるアプローチです。量子力学の原理を利用します。
量子ビットは0と1の重ね合わせ状態を取ることができます。これにより並列計算が可能になります。"""

chunker = SemanticChunker(threshold=0.6)
chunks = chunker.chunk(text, min_chunk_size=50)

for i, chunk in enumerate(chunks):
    print(f"Chunk {i}: {chunk}\n")

再帰的チャンキング

アルゴリズム

大きな区切り文字から順に試し、チャンクサイズを満たすまで再帰的に分割します。

class RecursiveChunker:
    def __init__(self, chunk_size=500, overlap=50):
        """再帰的チャンカーの初期化"""
        self.chunk_size = chunk_size
        self.overlap = overlap
        # 優先度順の区切り文字
        self.separators = ["\n\n", "\n", "。", ".", " ", ""]

    def _split_text(self, text, separators):
        """再帰的にテキストを分割"""
        if not text:
            return []

        # 現在の区切り文字
        separator = separators[0]
        remaining_separators = separators[1:] if len(separators) > 1 else [""]

        # 区切り文字で分割
        if separator:
            splits = text.split(separator)
        else:
            # 文字単位で分割（最後の手段）
            return [text[i:i + self.chunk_size]
                    for i in range(0, len(text), self.chunk_size - self.overlap)]

        chunks = []
        current_chunk = ""

        for split in splits:
            piece = split + separator if separator else split

            if len(current_chunk) + len(piece) <= self.chunk_size:
                current_chunk += piece
            else:
                if current_chunk:
                    chunks.append(current_chunk.strip())

                # 分割片が大きすぎる場合は再帰
                if len(piece) > self.chunk_size:
                    sub_chunks = self._split_text(piece, remaining_separators)
                    chunks.extend(sub_chunks[:-1])
                    current_chunk = sub_chunks[-1] if sub_chunks else ""
                else:
                    current_chunk = piece

        if current_chunk:
            chunks.append(current_chunk.strip())

        return [c for c in chunks if c]

    def chunk(self, text):
        """テキストをチャンクに分割"""
        return self._split_text(text, self.separators)

# 使用例
text = """# 機械学習入門

機械学習は人工知能の一分野です。

## 教師あり学習

教師あり学習では、入力と出力のペアから学習します。
回帰と分類が主なタスクです。回帰は連続値を予測します。分類は離散ラベルを予測します。

## 教師なし学習

教師なし学習では、ラベルなしデータから構造を発見します。
クラスタリングと次元削減が代表的です。"""

chunker = RecursiveChunker(chunk_size=150, overlap=20)
chunks = chunker.chunk(text)

for i, chunk in enumerate(chunks):
    print(f"Chunk {i}:\n{chunk}\n---")

チャンキング戦略の比較

各手法の特性を比較します。

手法 実装難易度 精度 計算コスト 適用場面
固定長 低 低 低 プロトタイプ
オーバーラップ 低 中 低 汎用的
文・段落ベース 中 中 低 構造化文書
セマンティック 高 高 高 高精度が必要な場合
再帰的 中 中〜高 低 様々な形式の文書

最適なチャンクサイズの決定

経験則

一般的なガイドライン:

  • 埋め込みモデルの最大長: 512トークン（多くのモデル）
  • 推奨範囲: 100〜500トークン
  • オーバーラップ: チャンクサイズの10〜20%

評価指標

チャンキング戦略の評価には以下の指標を使用します。

検索精度:

$$ \text{Recall@}k = \frac{|\text{RelevantChunks} \cap \text{RetrievedChunks}|}{|\text{RelevantChunks}|} $$

冗長度:

$$ \text{Redundancy} = \frac{\sum_i |C_i|}{|T|} $$

まとめ

本記事では、RAGシステムにおけるテキストチャンキング戦略を解説しました。

  • 固定長チャンキング: シンプルだが境界で情報が分断される
  • オーバーラップチャンキング: 境界での情報損失を軽減
  • 文・段落ベース: 自然な単位で分割
  • セマンティックチャンキング: 意味的類似度に基づく高精度な分割
  • 再帰的チャンキング: 様々な形式に対応可能

次のステップとして、以下の記事も参考にしてください。

関連記事

前の記事

カイ二乗検定の理論と使い方を解説

次の記事

ベクトルデータベース入門 — IVF・HNSW・PQを理解する

関連タグ: LLM RAG テキスト処理 機械学習 自然言語処理