👋

ページ情報をメタデータとして持たせるチャンク分割

2025/02/17に公開

PDF

はじめに

生成AIを使った検索応答システム（RAG: Retrieval-Augmented Generation）が普及する中で、正確性を担保するために「情報の出典」を示すことがますます重要になっています。

たとえば、長大なPDFやドキュメントから情報を抽出し、回答を生成する場合、ユーザーはその回答の裏付けとなる「どこから得た情報か」を知りたくなるでしょう。特に企業のレポートや研究論文を扱う場面では、回答の信頼性を確保し、必要に応じて原文にアクセスできる仕組みが求められます。
しかし、一般的なRAGでは、検索されたテキストの断片（チャンク）だけが回答に利用されることが多く、「その情報がどのページに存在しているか」というメタデータは扱われないことがほとんどです。これでは、ユーザーが正しい情報源を確認したい場合に不便ですし、説明責任を果たすことが難しくなります。
この課題を解決するために、チャンク分割時にページ情報をメタデータとして保持する方法を実装しました。これにより、回答に加えて「この情報はPDFの何ページにあるか」を表示できるようになり、信頼性と透明性が大きく向上します。

前提

ページ番号は各pdf固有でつけられている番号ではなく、順に数えた番号となります。
つまり、pdfを開いたときに一番最初に表示されるページの番号を1とします。例えばそのページの下に「0」と記載されていても、ページ番号は1とします。

実装

言語はpython, ライブラリはpdf処理にpdfminerを、チャンク分割にlangchainを使用しました。
（pdf処理のライブラリはページごとにテキスト抽出ができるのであればなんでも構いません。
チャンク分割手法も何でも構いません。ここでは私がよく使用するRecursiveCharacterTextSplitterで実装しています。）
チャンクが始まるページ番号と、チャンクが終了するページ番号の2つを取得します。
アウトプットはチャンクと開始ページ番号と終了ページ番号の辞書のリストとなります。

[{"chunk": {チャンク}, "start_page": {開始ページ番号}, "end_page": {終了ページ番号}},
 {"chunk": {チャンク},  "start_page": {開始ページ番号}, "end_page": {終了ページ番号}},
..., ]

実装方針

PDF解析ライブラリを用いて、1ページごとにテキストを抽出します。テキストの冒頭に目印となる「###STARTJUDGE{page番号}###」という文字列を挿入します。
1で作成したテキストをチャンク分割します。その後、各チャンクに含まれている「STARTJUDGE{page番号}###」をもとにページ情報を取得します。含まれない場合は一つ前のチャンクのページ情報を引き継ぎます。
細かい処理内容はコード内のコメントを参照してください。

サンプルコード

import re
from io import StringIO

from pdfminer.pdfinterp import PDFResourceManager, PDFPageInterpreter
from pdfminer.pdfpage import PDFPage
from pdfminer.converter import TextConverter

from langchain.text_splitter import RecursiveCharacterTextSplitter



class TextSplitterWithPageMetadata(RecursiveCharacterTextSplitter):
    '''
    PDFからテキストを抽出しページ情報を保持したチャンクに分割する
    
    Attributes
    ----------
    chunk_size : int
        チャンクサイズ
    chunk_overlap : int
        重複させるチャンク数
    separators : list
        チャンク分割するポイントのリスト
    '''
    
    def extract_text_with_marker(self, pdf_data):
        # PDFリソースを準備
        resource_manager = PDFResourceManager()
        page_numbers = list(PDFPage.get_pages(pdf_data))
        full_text = ""
        
        # 各ページを順に処理
        for page_number, page in enumerate(page_numbers, start=1):
            # ページの先頭に「###STARTJUDGE{page_number}###」を追加
            full_text += f"###STARTJUDGE{page_number}###"
            
            output = StringIO()  # ページごとのテキスト出力を保持するためのストリーム
            converter = TextConverter(resource_manager, output)  # テキスト変換器
            interpreter = PDFPageInterpreter(resource_manager, converter)
            
            # テキストを抽出
            interpreter.process_page(page)
            page_text = output.getvalue()
            full_text += page_text + "\n"
            output.truncate(0)
            output.seek(0)

        # リソース解放
        converter.close()
        output.close()

        return full_text


    def split_text(self, text):
        '''
        ページ情報が含まれるテキストをチャンク分割しつつページ情報を抽出する

        Parameters
        ----------
        text : str
            チャンク分割するテキスト

        Returns
        -------
        list_info : チャンクとページ情報が格納されたリスト
        '''
        
        # チャンク分割
        chunks = super().split_text(text)
        # 前回のページ番号
        previous_page = 0
        # 現在のページ番号
        current_page = None
        list_info = []
        for chunk in chunks:
            # ページ情報を抽出（ページ番号はSTARTJUDGEの後にある）
            list_page_info = re.findall(r"###STARTJUDGE(\d+)###", chunk)
            # ###STARTJUDGE###が見つかった場合、ページ番号を更新
            if list_page_info:
                current_page = list_page_info[-1]
            # チャンクが###STARTJUDGEから始まる場合は、前回のページから開始ページが変わっているため+1
            if chunk.startswith("###STARTJUDGE"):
                start_page = int(previous_page) + 1
                print("start: ###STARTJUDGE")
            else:
                start_page = int(previous_page)
            current_page = int(current_page)
            # ページの目印を削除
            chunk = re.sub(r"###STARTJUDGE\d+###", "", chunk)
            list_info.append({"chunk": chunk, "start_page": start_page, "end_page": current_page})
            # 前回のページ番号を更新
            previous_page = current_page
        return list_info

検証

AI事業者ガイドラインのpdfで試してみます。
チャンク分割のパラメータは適当に設定しました。

サンプルコード

# PDF ファイルをバイトデータとして読み込む
with open("AI事業者ガイドライン.pdf", "rb") as f:
    pdf_bytes = f.read()
# メモリ上のバッファに格納
pdf_data = io.BytesIO(pdf_bytes)
splitter = TextSplitterWithPageMetadata(chunk_size=2048, chunk_overlap=300, separators = ["。"])
full_text = splitter.extract_text_with_marker(pdf_data)
list_info = splitter.split_text(full_text)

for idx, info in enumerate(list_info):
    print("チャンク",idx+1)
    print(f'開始ページ: {info["start_page"]}, 最初の30文字: {info["chunk"][:30]}')
    print(f'終了ページ: {info["end_page"]}, 最後の30文字: {info["chunk"][-30:]}')

チャンク 1
チャンク 1
開始ページ: 1, 最初の30文字:    AI事業者ガイドライン （第1.01版）   令和6年
終了ページ: 3, 最後の30文字: イノベーション創出及び社会課題の解決に向けても活⽤されている
チャンク 2
開始ページ: 3, 最初の30文字: 。また近年台頭してきた対話型の⽣成AIによって「AIの⺠主化
終了ページ: 3, 最後の30文字: OECD等の国際機関での議論をリードし、多くの貢献をしてきた
チャンク 3
開始ページ: 3, 最初の30文字: 。  我が国は2016年4⽉のG7⾹川・⾼松情報通信⼤⾂会合
終了ページ: 4, 最後の30文字: ねることで、実効性・正当性を重視したものとして策定されている
チャンク 4
開始ページ: 4, 最初の30文字: 。   図1. 本ガイドラインの位置づけ  AIの利⽤は、そ
終了ページ: 5, 最後の30文字: にあたって必要な取組についての基本的な考え⽅を⽰すものである
チャンク 5
開始ページ: 5, 最初の30文字: 。よって、実際のAI開発・提供・利⽤において、本ガイドライン
終了ページ: 5, 最後の30文字: AI利⽤者」の3つに⼤別され、それぞれ以下のとおり定義される
...

元のpdfと比較

2つピックアップして結果を確認します。

チャンク 2

開始ページ: 3, 最初の30文字: 。また近年台頭してきた対話型の⽣成AIによって「AIの⺠主化
終了ページ: 3, 最後の30文字: OECD等の国際機関での議論をリードし、多くの貢献をしてきた

は、元のpdfの3ページに含まれていることがわかる。

チャンク 4

開始ページ: 4, 最初の30文字: 。   図1. 本ガイドラインの位置づけ  AIの利⽤は、そ
終了ページ: 5, 最後の30文字: にあたって必要な取組についての基本的な考え⽅を⽰すものである

は、元のpdfの4ページから5ページに含まれていることがわかる。

他のチャンクについても同じように合致していることは確認しています。

まとめ

RAG の回答時にエビデンスとしてページ番号を表示するための方法を紹介しました。ページ情報を保持しながらテキストを分割することで、回答の信頼性を向上させ、ユーザーがどのページに記載されている情報なのかを簡単に確認できるようになります。

ただし、今後LLMの入力トークン数が増加すれば、大量のテキストをそのまま処理できるようになり、チャンク分割自体が不要になる可能性もあります。それでも、長い文書を扱う際に適切なメタデータを保持する手法は引き続き重要であり、用途に応じた最適なアプローチを選ぶことが求められるでしょう。

はじめに

前提

実装

実装方針

サンプルコード

検証

サンプルコード

元のpdfと比較

まとめ

Discussion