はじめに
生成AIの学習の一環としてQiitaに記事投稿しているのですが、前回書いたRAGの記事が思った以上に反響があったので今回はRAGについてもう少し詳しく書きたいと思います。
筆者が前回投稿したRAGの記事は以下。良かったらこっちも見ていってください。
前回の記事では「RAGについて知る」,「とりあえず動かしてRAGを実感する」ことにフォーカスしましたが、実際にはそれだけでは上手くいかないです。
- 思ったより回答の精度低いな
- 関連ドキュメントは検索できるのに良い感じの回答してくれないな
そう感じることになると思います。
筆者も色々試してみて、RAGを使った回答の精度を上げるには、どう「チャンク」を作るのかが非常に重要だと感じました。
今回はチャンクの重要性と実装の具体例を紹介します。
チャンク化とは、テキストやファイル等をいくつかのまとまりに分割すること
概要
本記事では、専門書（PDF形式・約400ページ）を対象に、高精度な回答を返すチャット機能を構築するプロセスを紹介します。
RAGの精度を左右する最大の要因は、「LLMにとって文脈が理解しやすい単位(チャンク)でデータを渡せるか」 です。
単なる文字数ベースの分割（固定長チャンク）ではなく、書籍の構造（章・節・項）を維持した「構造的チャンク化」の実装方法を解説します。
構造的チャンク化の実装方法
構造的なチャンクを作りベクトルDB作成までの流れ
- PDFを読み込みLlamaParseで解析
- 解析したデータをチャンク化
- Chroma（ベクトルDB）に格納
各項目の詳細を以下を解説していきます。
1. PDFを読み込みLlamaParseで解析
RAGの性能を高めるため、まずは対象ドキュメントを正しい状態で読み込む必要があります。
通常PDFの場合はPDFMinerLoaderを利用しますが、それだとページ切り替わりの箇所、表データを正しく読み取ることできずLLMが文脈を理解できなくなるというリスクがあります。
そこで今回は、LlamaIndexが提供するLlamaParseを利用します。
LlamaParseとは
LlamaIndexが提供するLlamaCloudのコンポーネント。
PDF、パワーポイント、Wordドキュメント、スプレッドシートなどの複雑なドキュメントを構造化されたデータに解析することができる。
PDFは、「表は表として（Markdown形式のテーブルで）」「見出しは見出しとして」 解析することができる。
LlamaIndexは、大規模言語モデル（LLM）アプリケーションを構築するためのオープンソースのデータ・オーケストレーション・フレームワークです。
実装例
from dotenv import load_dotenv
from llama_parse import LlamaParse
# 1. APIキーの設定
load_dotenv()
file_path = "Sample.pdf"
parser = LlamaParse(
result_type="markdown",
num_workers=4, # 並列処理の数
verbose=True,
language="ja" # 日本語の書籍の場合は指定
)
print(f"解析開始: {file_path}")
llama_docs = parser.load_data(file_path)
print(f"解析完了: {len(llama_docs)} ページ")
LlamaParseを利用するにはLlamaにアクセスしてAPIキーを取得する必要がある。
取得したAPIキーは以下のような.envファイルを作成して管理しpythonで設定値を取得する。
LLAMA_CLOUD_API_KEY="xxxxxxxxx"
2.解析したデータをチャンク化
このデータのチャンク化がとても重要です。
チャンクの作り方によって検索（RAG）性能、処理速度、トークン消費量が大きく変わります。
書籍は章や節ごとのまとまりでチャンク化します。
実装例
上記の実装で解析したllama_docsを利用してチャンク化を行う。
import re
from typing import List, Optional
from langchain_classic.schema import Document
# 階層的なヘッダー構造に基づいた分割
# 「第X部」「第X章」「X.Y」「X.Y.Z」のパターンを認識して分割
langchain_docs = []
for doc_idx, doc in enumerate(llama_docs):
# ページ番号を取得（LlamaParseのメタデータから）
page_number = doc.metadata.get("page", doc_idx + 1) if hasattr(doc, 'metadata') else doc_idx + 1
# 階層的なヘッダーで分割
split_docs = split_by_hierarchical_headers(doc.text, page_number, file_path)
langchain_docs.extend(split_docs)
# 分割結果のサマリーを表示
print(f"\n分割結果サマリー:")
print(f" 総チャンク数: {len(langchain_docs)}")
上記実装例のsplit_by_hierarchical_headersメソッドの詳細は以下。
Markdown形式のテキストから章や節等のヘッダーを読み取り、分割する。
分割後はメタデータとして以下の項目のリストを作成する。
メタデータを付与することで検索時に絞り込みが可能になり、処理速度や精度向上につながる
| 項目名 | 表している内容 |
|---|---|
| source | 元のファイル名（Sample.pdf） |
| page | PDF上の物理的なページ番号 |
| chunk_id | セクションを識別する固有ID |
| header_level | ヘッダーの重要度（1=部, 2=章...） |
| header_text | そのセクションの見出し文 |
| part | 該当する「部」（例: "I"） |
| chapter | 該当する「章」（例: "9"） |
| section | 該当する「節」（例: "9.1"） |
| subsection | 該当する「小節」（例: "9.1.2"） |
| text_length | チャンク内の文字数 |
def split_by_hierarchical_headers(text: str, page_number: int, file_path: str) -> List[Document]:
"""
テキストを階層的なヘッダーで分割してDocumentのリストを返す
Args:
text: Markdown形式のテキスト
page_number: ページ番号
file_path: ソースファイルパス
Returns:
Document オブジェクトのリスト
"""
documents = []
# ヘッダー行を抽出（#で始まる行）
lines = text.split('\n')
header_indices = []
for i, line in enumerate(lines):
if line.strip().startswith('# '):
header_text = line.strip()[2:].strip()
level, part, chapter, section, subsection = classify_header_level(header_text)
header_indices.append({
'index': i,
'text': header_text,
'level': level,
'part': part,
'chapter': chapter,
'section': section,
'subsection': subsection
})
# ヘッダー間でテキストを分割
for idx, header_info in enumerate(header_indices):
start_idx = header_info['index']
# 次のヘッダーまで、またはテキストの最後まで
end_idx = header_indices[idx + 1]['index'] if idx + 1 < len(header_indices) else len(lines)
# このセクションの内容を取得（ヘッダー行を含む）
section_lines = lines[start_idx:end_idx]
content = '\n'.join(section_lines).strip()
# 空のコンテンツはスキップ
if not content or len(content) < 10:
continue
# chunk_idを生成
chunk_id_parts = []
if header_info['part']:
chunk_id_parts.append(f"part{header_info['part']}")
if header_info['chapter']:
chunk_id_parts.append(f"ch{header_info['chapter']}")
if header_info['section']:
chunk_id_parts.append(f"sec{header_info['section'].replace('.', '_')}")
if header_info['subsection']:
chunk_id_parts.append(f"subsec{header_info['subsection'].replace('.', '_')}")
chunk_id = "_".join(chunk_id_parts) if chunk_id_parts else f"other_p{page_number}"
# メタデータを作成
metadata = {
"source": file_path,
"page": page_number,
"chunk_id": chunk_id,
"header_level": header_info['level'],
"header_text": header_info['text'],
"part": header_info['part'] or "",
"chapter": header_info['chapter'] or "",
"section": header_info['section'] or "",
"subsection": header_info['subsection'] or "",
"text_length": len(content),
}
# Documentオブジェクトを作成
doc = Document(page_content=content, metadata=metadata)
documents.append(doc)
return documents
classify_header_levelメソッドの詳細は以下。
正規表現を用いて階層レベルと部、章、節などの情報を抽出する。
def classify_header_level(header_text: str) -> tuple[int, Optional[str], Optional[str], Optional[str], Optional[str]]:
"""
ヘッダーテキストから階層レベルと部・章・節・小節情報を抽出
Returns:
(level, part, chapter, section, subsection)
- level: 0=その他, 1=部, 2=章, 3=節, 4=小節
- part: 第X部 (例: "I", "II", "III")
- chapter: 第X章 (例: "1", "2", "3")
- section: X.Y (例: "1.1", "2.3")
- subsection: X.Y.Z (例: "1.1.1", "2.3.4")
"""
# 第X部パターン
part_match = re.match(r'第\s*([IVX]+)\s*部', header_text)
if part_match:
return (1, part_match.group(1), None, None, None)
# 第X章パターン
chapter_match = re.match(r'第\s*(\d+)\s*章', header_text)
if chapter_match:
return (2, None, chapter_match.group(1), None, None)
# X.Y.Zパターン（小節）
subsection_match = re.match(r'(\d+)\.(\d+)\.(\d+)', header_text)
if subsection_match:
section_num = f"{subsection_match.group(1)}.{subsection_match.group(2)}"
subsection_num = f"{subsection_match.group(1)}.{subsection_match.group(2)}.{subsection_match.group(3)}"
return (4, None, subsection_match.group(1), section_num, subsection_num)
# X.Yパターン（節）
section_match = re.match(r'(\d+)\.(\d+)', header_text)
if section_match:
section_num = f"{section_match.group(1)}.{section_match.group(2)}"
return (3, None, section_match.group(1), section_num, None)
# その他
return (0, None, None, None, None)
3.Chroma（VectorDB）に格納
上記でチャンク化した内容をベクトルDBに保存します。
今回はベクトルDBとしてChromaを利用します。
ベクトルDBには登録対象のドキュメント、埋め込み処理を行うモデル、メタデータを設定する。
ChromaとはAIアプリケーション向けに設計されたオープンソースのベクトルデータベースです。
データの「意味」を理解して検索するベクトル検索やテキスト・画像・音声といったさまざまなデータ形式を扱えるマルチモーダル検索機能がある。
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
# Vector DB (Chroma) への格納（collection_metadataを追加）
collection_metadata = {
"book_title": "sampleBook",
"total_chunks": len(langchain_docs),
"file_path": file_path
}
vectorstore = Chroma.from_documents(
documents=langchain_docs,
embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
persist_directory="./chroma_db", # ベクトルDBをメモリではなく永続化するために格納先ディレクトリを設定
collection_metadata=collection_metadata
)
print(f"\n✅ 成功: {len(langchain_docs)} 個のチャンクをChromaDBに格納しました。")
OpenAIEmbeddingsを利用するにはOpenAIにアクセスしてAPIキーを取得する必要がある。
最後に
今回はチャンク化の重要性とその実装方法について記載しました。
ただ単にベクトルDBに保存するのではなく、 どう解析するか、どうチャンク化するか 。
この部分をどれだけ力を入れて考え実装できるかが、今回のような検索ツール作成では重要だと思います。
（うーん、、生成AIって便利な分やっぱり奥がとても深いな、、）