よくある現象:RAGシステムが最強のLLMを使用し、プロンプトが何度も何度もチューニングされているにもかかわらず、クイズはまだうまく機能せず、解答は文脈的に不完全であったり、事実誤認を含んでいたりする。
エンジニアは検索アルゴリズムを検討し、最適化した。 Embedding
しかし、データがベクトル・ライブラリに入る前の重要なステップを見落としていることが多い。
不適切なチャンキングは、情報が欠落した「不良データ」の束をモデルに与えているに等しい。モデルの推論能力がいかに高くても、断片的な知識から完全な答えを導き出すことはできない。チャンキングの質は、RAGシステムの性能の下限を直接決定する。
この記事では、漠然とした理論について語るのではなく、RAGシステムの強固な基盤を作るための様々なチャンキング戦略について、実際のコードとエンジニアリングの経験に焦点を当てる。
なぜ塊なのか?
チャンキングの必要性は、2つの核となる制限から生じている:
- モデルコンテキストウィンドウラージ・ランゲージ・モデル(LLM)は、無限の長さのテキストを一度に処理することはできません。チャンキングとは、長い文書をモデルが処理できる断片にスライスするプロセスです。
- 検索効率とノイズ検索中、テキストのブロックに無関係な情報(ノイズ)が多すぎると、核となる信号が希薄になり、検索エンジンがユーザーの意図に正確に一致することが難しくなる。
理想的なチャンキングは文脈的完全性とともに情報密度そのバランスを見つけること。chunk_size
歌で応える chunk_overlap
は、この均衡を制御する基本的なパラメータである。chunk_overlap
隣接するブロック間で繰り返されるテキストの一部を保持することで、ブロック境界を越えた意味の連続性が確保されている。
基本チャンキング戦略
固定長チャンキング
これは最も単純な方法で、あらかじめ設定された文字数でカットする。テキストの論理構造を考慮しないので実装は簡単だが、意味的整合性が失われる傾向がある。
- コア・アイデア: 固定文字数による
chunk_size
スライスされたテキスト。 - 適用シナリオ構造が弱いプレーンテキスト、または意味的要求が低い前処理段階。
from langchain_text_splitters import CharacterTextSplitter
sample_text = (
"LangChain was created by Harrison Chase in 2022. It provides a framework for developing applications "
"powered by language models. The library is known for its modularity and ease of use. "
"One of its key components is the TextSplitter class, which helps in document chunking."
)
text_splitter = CharacterTextSplitter(
separator=" ", # Split on spaces
chunk_size=100, # Size of each chunk
chunk_overlap=20, # Overlap between chunks
length_function=len,
)
docs = text_splitter.create_documents([sample_text])
for i, doc in enumerate(docs):
print(f"--- Chunk {i+1} ---")
print(doc.page_content)
再帰的文字チャンキング
LangChain
推奨される一般的なポリシー。これは、あらかじめ定義された文字のリストによって設定される(たとえば ["\n\n", "\n", " ", ""]
)は再帰的なセグメンテーションを行い、パラグラフやセンテンスといった論理的な単位の保持を優先させようとする。
- コア・アイデアセパレータの階層リストによる再帰的スライス。
- 適用シナリオテキストタイプの大部分に適した一般的な戦略。
from langchain_text_splitters import RecursiveCharacterTextSplitter
# Using the same sample_text from the previous example
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=100,
chunk_overlap=20,
# Default separators are ["\n\n", "\n", " ", ""]
)
docs = text_splitter.create_documents([sample_text])
for i, doc in enumerate(docs):
print(f"--- Chunk {i+1} ---")
print(doc.page_content)
パラメーター・チューニング固定長や再帰的なチャンキングではchunk_size
歌で応える chunk_overlap
セッティングが重要だ。
chunk_size
各ブロックのサイズを決める。ブロックが小さすぎると文脈情報が不足し、ブロックが大きすぎるとノイズが多くなってAPI
通話料金。この値は通常Embedding
モデルへの入力token
選択できる制限、共通256
,512
,1024
等価である。BERT
同型512
token
コンテキストウィンドウ。chunk_overlap
隣接するブロック間の重複文字数を決定します。適度なオーバーラップを設定する。chunk_size
の10%-20%)は、ブロック境界で完全な意味ユニットが切断されるのを効果的に防ぐことができ、これは意味的連続性を確保する鍵である。
文ベースのチャンキング
最小単位としての文の組み合わせは、最も基本的な意味的整合性を保証する。
- コア・アイデアテキストをセンテンスに分割し、センテンスをチャンクにまとめる。
- 適用シナリオ例:法律文書、ニュース報道など。
import nltk
try:
nltk.data.find('tokenizers/punkt')
except nltk.downloader.DownloadError:
nltk.download('punkt')
from nltk.tokenize import sent_tokenize
def chunk_by_sentences(text, max_chars=500, overlap_sentences=1):
sentences = sent_tokenize(text)
chunks = []
current_chunk = ""
for i, sentence in enumerate(sentences):
if len(current_chunk) + len(sentence) <= max_chars:
current_chunk += " " + sentence
else:
chunks.append(current_chunk.strip())
# Create overlap
start_index = max(0, i - overlap_sentences)
current_chunk = " ".join(sentences[start_index:i+1])
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
long_text = "This is the first sentence. This is the second sentence, which is a bit longer. Now we have a third one. The fourth sentence follows. Finally, the fifth sentence concludes this paragraph."
chunks = chunk_by_sentences(long_text, max_chars=100)
for i, chunk in enumerate(chunks):
print(f"--- Chunk {i+1} ---")
print(chunk)
銘記する中国人を相手にする場合nltk.tokenize.sent_tokenize
デフォルトの英語モデルは失敗します。中国語に適したセグメンテーション方法を使用する必要があります。例えば、中国語の句読点(.)) に基づくか、あるいは spaCy
もしかしたら HanLP
などのライブラリーがある。
構造を意識したチャンキング
文書固有の構造情報(見出しやリストなど)をチャンクの境界として使うこのアプローチは論理的で、文脈をよりよく保存する。
マークダウン・テキスト・チャンキング
- コア・アイデアをベースにしている。
Markdown
のタイトル階層でブロックの境界を定義する。 - 適用シナリオフォーマル
Markdown
以下のような書類GitHub
README
技術文書。
from langchain_text_splitters import MarkdownHeaderTextSplitter
markdown_document = """
# Chapter 1: The Beginning
## Section 1.1: The Old World
This is the story of a time long past.
## Section 1.2: A New Hope
A new hero emerges.
# Chapter 2: The Journey
## Section 2.1: The Call to Adventure
The hero receives a mysterious call.
"""
headers_to_split_on = [
("#", "Header 1"),
("##", "Header 2"),
]
markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_header_splits = markdown_splitter.split_text(markdown_document)
for split in md_header_splits:
print(f"Metadata: {split.metadata}")
print(split.page_content)
print("-" * 20)
チャンキング
- コア・アイデア対話の話者またはラウンドに基づくチャンキング。
- 適用シナリオ顧客サービス対話、インタビュー記録、会議議事録。
dialogue = [
"Alice: Hi, I'm having trouble with my order.",
"Bot: I can help with that. What's your order number?",
"Alice: It's 12345.",
"Alice: I haven't received any shipping updates.",
"Bot: Let me check... It seems your order was shipped yesterday.",
"Alice: Oh, great! Thank you.",
]
def chunk_dialogue(dialogue_lines, max_turns_per_chunk=3):
chunks = []
for i in range(0, len(dialogue_lines), max_turns_per_chunk):
chunk = "\n".join(dialogue_lines[i:i + max_turns_per_chunk])
chunks.append(chunk)
return chunks
chunks = chunk_dialogue(dialogue)
for i, chunk in enumerate(chunks):
print(f"--- Chunk {i+1} ---")
print(chunk)
意味的・主題的チャンキング
この種のアプローチは、テキストの物理的な構造を超えて、その意味論に基づいてコンテンツを切り刻む。
セマンティックチャンキング
- コア・アイデア隣接する文や段落のベクトルの類似度を計算し、意味が急激に変化する(類似度が低い)位置でスライスする。
- 適用シナリオ知識ベースや研究論文など、精度の高い意味的結合を必要とする文書。
import os
from langchain_experimental.text_splitter import SemanticChunker
from langchain_huggingface import HuggingFaceEmbeddings
os.environ["TOKENIZERS_PARALLELISM"] = "false"
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
# LangChain's SemanticChunker offers different threshold types:
# "percentile": Threshold based on the percentile of similarity score differences. Good for adaptability.
# "standard_deviation": Threshold based on standard deviation of similarity scores.
# "interquartile": Uses the interquartile range, robust to outliers.
# "gradient": Looks for sharp changes in similarity, useful for detecting abrupt topic shifts.
text_splitter = SemanticChunker(
embeddings,
breakpoint_threshold_type="percentile",
breakpoint_threshold_amount=95 # A higher percentile means it only breaks on very significant semantic shifts.
)
long_text = (
"The Wright brothers, Orville and Wilbur, were two American aviation pioneers "
"generally credited with inventing, building, and flying the world's first successful motor-operated airplane. "
"They made the first controlled, sustained flight of a powered, heavier-than-air aircraft on December 17, 1903. "
"In the following years, they continued to develop their aircraft. "
"Switching topics completely, let's talk about cooking. "
"A good pizza starts with a perfect dough, which needs yeast, flour, water, and salt. "
"The sauce is typically tomato-based, seasoned with herbs like oregano and basil. "
"Toppings can vary from simple mozzarella to a wide range of meats and vegetables. "
"Finally, let's consider the solar system. "
"It is a gravitationally bound system of the Sun and the objects that orbit it. "
"The largest objects are the eight planets, in order from the Sun: Mercury, Venus, Earth, Mars, Jupiter, Saturn, Uranus, and Neptune."
)
docs = text_splitter.create_documents([long_text])
for i, doc in enumerate(docs):
print(f"--- Chunk {i+1} ---")
print(doc.page_content)
print()
パラメーター・チューニング::SemanticChunker
プログラムの有効性は、以下の点に大きく依存する。 breakpoint_threshold_amount
この閾値は「意味的変化感度」をコントロールする。この閾値は「意味的変化感度」を制御する。低いしきい値では小さくまとまった塊が大量に生成され、高いしきい値ではトピックに大きな変化があった場合のみカットされる。文書の内容に応じて実験を繰り返す必要がある。
テーマに基づいたチャンキング
- コア・アイデアテーマ別モデルを使う(例
LDA
)や、マクロテーマの変化に応じて文書をスライスして切り刻むクラスタリング・アルゴリズムがある。 - 適用シナリオ長い、複数の主題を扱った報告書や書籍。
import numpy as np
import re
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.decomposition import LatentDirichletAllocation
import nltk
from nltk.corpus import stopwords
try:
stopwords.words('english')
except LookupError:
nltk.download('stopwords')
def lda_topic_chunking(text: str, n_topics: int = 3) -> list[str]:
# 1. Preprocessing: Treat each paragraph as a "document"
paragraphs = [p.strip() for p in text.split('\n\n') if p.strip()]
if len(paragraphs) <= 1:
return [text]
cleaned_paragraphs = [re.sub(r'[^a-zA-Z\s]', '', p).lower() for p in paragraphs]
# 2. Bag of Words + Stopword Removal
vectorizer = CountVectorizer(min_df=1, stop_words=stopwords.words('english'))
X = vectorizer.fit_transform(cleaned_paragraphs)
if X.shape == 0:
return paragraphs
# 3. LDA Topic Modeling
lda = LatentDirichletAllocation(n_components=n_topics, random_state=42)
lda.fit(X)
# 4. Determine dominant topic for each paragraph
topic_dist = lda.transform(X)
dominant_topics = np.argmax(topic_dist, axis=1)
# 5. Chunking based on topic changes
chunks = []
current_chunk_paragraphs = []
current_topic = dominant_topics
for i, paragraph in enumerate(paragraphs):
if dominant_topics[i] == current_topic:
current_chunk_paragraphs.append(paragraph)
else:
chunks.append("\n\n".join(current_chunk_paragraphs))
current_chunk_paragraphs = [paragraph]
current_topic = dominant_topics[i]
chunks.append("\n\n".join(current_chunk_paragraphs))
return chunks
銘記するトピックベースのチャンキングは、テキストの長さ、トピックの区別、前処理ステップに非常に敏感であり、あらかじめ設定されたトピック数を必要とする。この方法は、トピックの境界が明確な長い文書の探索ツールとして適している。
高度なチャンキング戦略
スモールからビッグまで
- コア・アイデア小さなチャンク(例:文)を使って高精度の検索を行い、そのチャンクを含む元のチャンク(例:段落)をコンテキストとして
LLM
.小さなチャンクの高い検索精度と、大きなチャンクの豊富なコンテキストを兼ね備えている。 - 適用シナリオ高い検索精度と豊富な生成コンテキストを必要とする複雑なQ&Aシナリオ。
ある LangChain
真ん中だ。ParentDocumentRetriever
はこのアイデアを実装している。バックグラウンドで2つの並列処理フローを管理する:
- 文書を大きな「親ブロック」に分割する。
- それぞれの親ブロックは、さらに小さな「子ブロック」に分割される。
- サブブロックに対してのみベクトル・インデックスを作成する。
- 検索は、まず関連するサブブロックを見つけ、次に別の
docstore
対応する親ブロックを抽出すると、次のように返される。LLM
.
# from langchain.embeddings import OpenAIEmbeddings
# from langchain_text_splitters import RecursiveCharacterTextSplitter
# from langchain.retrievers import ParentDocumentRetriever
# from langchain_community.document_loaders import TextLoader
# from langchain_chroma import Chroma
# from langchain.storage import InMemoryStore
# Assume 'docs' are loaded documents
# parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
# child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
# vectorstore = Chroma(embedding_function=OpenAIEmbeddings(), collection_name="split_parents")
# store = InMemoryStore() # This store holds the parent documents
# retriever = ParentDocumentRetriever(
# vectorstore=vectorstore,
# docstore=store,
# child_splitter=child_splitter,
# parent_splitter=parent_splitter,
# )
# retriever.add_documents(docs)
# sub_docs = vectorstore.similarity_search("query") # Retrieves small chunks
# retrieved_docs = retriever.get_relevant_documents("query") # Retrieves large parent chunks
# print(retrieved_docs.page_content)
エージェント・チャンキング
- コア・アイデアを使用している。
LLM
Agent
人間の読解プロセスをシミュレートし、チャンクの境界を動的に決定する。例えば、キューLLM
テキストを複数の「自己完結した知識のかたまり」に分解する。 - 適用シナリオ実験的なプロジェクトや、非常に複雑で構造化されていないテキストを扱う場合。非常にコストがかかり、安定性が実証される必要がある。
import textwrap
# from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List
class KnowledgeChunk(BaseModel):
chunk_title: str = Field(description="A concise title for this knowledge chunk.")
chunk_text: str = Field(description="A self-contained text extracted and synthesized from the original paragraph.")
representative_question: str = Field(description="A typical question that can be answered by this chunk.")
class ChunkList(BaseModel):
chunks: List[KnowledgeChunk]
parser = PydanticOutputParser(pydantic_object=ChunkList)
prompt_template = """
[ROLE]: You are a top-tier document analyst. Your task is to decompose complex text into a set of core, self-contained "Knowledge Chunks".
[TASK]: Read the provided text, identify the distinct core concepts, and create a knowledge chunk for each.
[RULES]:
1. Self-Contained: Each chunk must be understandable on its own.
2. Single Concept: Each chunk should focus on only one core idea.
3. Extract and Restructure: Pull all relevant sentences for a concept and combine them into a coherent paragraph.
4. Follow Format: Strictly adhere to the JSON format instructions below.
{format_instructions}
[TEXT TO PROCESS]:
{paragraph_text}
"""
prompt = PromptTemplate(
template=prompt_template,
input_variables=["paragraph_text"],
partial_variables={"format_instructions": parser.get_format_instructions()},
)
# The following part is a simulation, as it requires a running LLM model.
# model = ChatOpenAI(model="gpt-4", temperature=0.0)
# chain = prompt | model | parser
# result = chain.invoke({"paragraph_text": document_text})
ミックス・チャンキング:効率と品質のバランス
実際には、単一の戦略ですべての状況に対処することは難しい。ミックス・チャンキングは非常に実用的なテクニックである。
- コア・アイデアマクロ戦略(構造化チャンキングなど)による粗い粒度のスライシングの後、より細かい戦略(再帰的チャンキングなど)を用いて、特大チャンクの二次スライシングを行う。
- 適用シナリオ複雑な構造を持ち、内容の密度が不均一な文書を処理します。
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langchain_core.documents import Document
markdown_document = """
# Chapter 1: Company Profile
Our company was founded in 2017...
## 1.1 Development History
The company has experienced rapid growth...
# Chapter 2: Core Technology
This chapter describes our core technologies in detail. Our framework is based on advanced distributed computing concepts... (A very long paragraph with multiple sentences describing different technical aspects like CNNs, Transformers, data pipelines, etc.)
## 2.1 Technical Principles
Our principles combine statistics, machine learning...
# Chapter 3: Future Outlook
Looking ahead, we will continue to invest in AI...
"""
def hybrid_chunking(
markdown_document: str,
coarse_chunk_threshold: int = 400,
fine_chunk_size: int = 100,
fine_chunk_overlap: int = 20
) -> list[Document]:
# 1. Coarse-grained splitting by structure
headers_to_split_on = [("#", "Header 1"), ("##", "Header 2")]
markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
coarse_chunks = markdown_splitter.split_text(markdown_document)
# 2. Fine-grained recursive splitter for oversized chunks
fine_splitter = RecursiveCharacterTextSplitter(
chunk_size=fine_chunk_size,
chunk_overlap=fine_chunk_overlap
)
final_chunks = []
for chunk in coarse_chunks:
if len(chunk.page_content) > coarse_chunk_threshold:
# If a chunk is too large, split it further
finer_chunks = fine_splitter.split_documents([chunk])
final_chunks.extend(finer_chunks)
else:
final_chunks.append(chunk)
return final_chunks
final_chunks = hybrid_chunking(markdown_document)
for i, chunk in enumerate(final_chunks):
print(f"--- Final Chunk {i+1} (Length: {len(chunk.page_content)}) ---")
print(f"Metadata: {chunk.metadata}")
print(chunk.page_content)
print("-" * 80)
最適なチャンキング戦略を選ぶには?
数多くの戦略を前にしては、一つ一つ試すよりも、合理的な道を選ぶことの方が重要である。以下のような階層的な意思決定のフレームワークに従うことが推奨される。
ステップ1:ベースライン戦略から始める
- デフォルト・オプション::
RecursiveCharacterTextSplitter
.これは、どのようなテキストが処理されているかにかかわらず、最も安全なスタート地点である。パフォーマンスのベースラインを確立するために使用してください。
ステップ2:構造化された特徴を調べる
- 優先オプション構造を考慮したチャンキング。文書に明示的な構造 (
Markdown
タイトルHTML
タブ)に切り替えてMarkdownHeaderTextSplitter
などの方法がある。これは、最もコストが低く、最も明白な利益をもたらす最適化である。
ステップ3:精度がボトルネックになる場合
- 高度なオプション意味的チャンキングまたは小-大チャンキング。もし、基本戦略や構造化戦略の検索結果がまだ満足のいくものでない場合は、より高次元の意味情報が必要であることを示している。
SemanticChunker
ブロック内で高度なセマンティック一貫性が要求されるシナリオ向け。ParentDocumentRetriever
(小さな塊から大きな塊まで):検索精度と、より明確な理解を提供する必要性の両方に適している。LLM
完全な文脈を提供する複雑なQ&Aシナリオ。
ステップ4:極めて複雑な文書への対応
- アドバンスド・プラクティスハイブリッド・チャンキング。複雑な構造を持ち、コンテンツの密度が不均一なドキュメントでは、ハイブリッド・チャンキングがコストと効果のバランスを取るためのベストプラクティスです。
次の表は、論じられたすべてのチャンキング戦略をまとめたものである。
チャンキング戦略 | コア・ロジック | バンテージ | デメリットとコスト |
---|---|---|---|
固定長チャンキング | 固定文字数または token ぶすう |
シンプルで迅速な導入 | セマンティクスを簡単に破壊し、最も効果的ではない |
再帰的チャンキング | 定義済みのセパレータ(段落、センテンス)による再帰的スライス | 多用途性、より良い構造の維持 | 不規則な文書にかなり有効 |
文ベースのチャンキング | 文章を最小単位とし、チャンクにまとめる | 文章の完全性を確保する | 単文の文脈では不十分な場合がある。 |
構造化されたチャンキング | 文書固有の構造(見出しなど)を利用して、スライスする。 | 論理的で文脈が明確 | 汎用的ではなく、文書フォーマットに強く依存する |
セマンティックチャンキング | 局所的な意味類似度の変化に基づくスライシング | ブロック内の概念の凝集性が高く、検索精度が高い。 | 高い計算コスト(Embedding 計算)、モデルの品質に頼る |
テーマに基づいたチャンキング | トピックモデルを用いたグローバルなトピック境界によるスライシング | ブロック内の情報は相関性が高い | 複雑な実装、データやパラメータに敏感、不安定な結果 |
ハイブリッドチャンキング | マクロクルード+マイクロセグメンテーション | 効率と品質と実用性のバランス | より複雑な実装ロジック |
小大塊 | 小さなブロックは検索用、大きなブロックは生成用 | 高精度検索と豊富なコンテキストの組み合わせ | パイプラインの複雑さ、2組のインデックスを管理する必要性、ストレージコストの倍増 |
プロキシ・チャンキング | AI エージェント・ダイナミクス分析とスライシング・ドキュメント |
理論的には最適 | 実験的で非常にコストがかかる(API コール)に大きな遅延がある |