8. 実践データソース別の索引システム構築

索引構築パイプライン（Ingestion Pipeline）の重要性

前章までで、BM25 とベクトル検索を組み合わせたハイブリッド検索やリランキングのアルゴリズムを解説しました。しかし、実際の業務システムで RAG を安定運用するためには、「元データをどのように前処理・チャンキングし、高精度な索引（インデックス）として永続化するか」というインジェスト設計が極めて重要になります。

対象となるデータソースの特性によって、求められる前処理やインデックス化のアプローチは大きく異なります。

データソース種別	主な特性と課題	推奨されるインデックス化戦略
IRIS バンドルデータ	構造化 Markdown、YAML フロントマター、コンセプト親子関係	フロントマターからメタデータを抽出、見出し構造を保持した階層チャンキング
LOTUS ノートデータ	継続的に推敲されるノート、階層ディレクトリ、頻繁な更新	更新日時やファイルハッシュによる差分検知、増分インデックス更新
大規模テキスト（行単位）	数十万〜数百万行のログや対話履歴、メモリ枯渇リスク	ジェネレータによるストリーミング読み込み、行番号保持、バッチ登録
複数 PDF ドキュメント	レイアウト構造、ページ境界、改行ノイズ、表やヘッダー	ページ単位抽出、不要な改行やヘッダーの正規化、ページ番号メタデータ付与

本章では、これら4つの実践的なデータソースを対象に、Python による具体的な索引構築パイプラインの実装手順を解説します。

必要なパッケージのセットアップ

各データソースの前処理や解析に必要なライブラリをインストールします。

pip install langchain langchain-community langchain-chroma langchain-ollama rank-bm25 janome pypdf pyyaml

共通の前処理として利用する日本語形態素解析トークナイザーを定義しておきます。

from janome.tokenizer import Tokenizer

tokenizer = Tokenizer()


def japanese_tokenizer(text: str) -> list[str]:
    """日本語テキストを形態素解析して自立語を抽出するトークナイザー."""
    tokens = []
    for token in tokenizer.tokenize(text):
        pos = token.part_of_speech.split(",")[0]
        if pos in ["名詞", "動詞", "形容詞"]:
            tokens.append(token.base_form)
    return tokens

IRIS のナレッジバンドル（data/bundles/<bundle-name>/wiki/）は、YAML フロントマターでメタデータ（タグ、参照ソース、親コンセプトなど）が厳密に管理された Markdown ドキュメント群です。単にテキストとして分割するのではなく、フロントマターの属性を検索メタデータとして抽出し、見出し構造（H1, H2, H3）を意識してチャンキングすることで、親ドキュメントの文脈を失わない高精度なインデックスを構築できます。

実装手順

Markdown ファイルから YAML フロントマターと本文を分離して解析します。
MarkdownHeaderTextSplitter で見出し階層を保持しながら本文を分割します。
フロントマターのメタデータを各チャンクに引き継ぎ、BM25 およびベクトルストアへ登録します。

from pathlib import Path
from typing import Any
import yaml

from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from langchain_text_splitters import MarkdownHeaderTextSplitter


def parse_frontmatter_markdown(file_path: Path) -> tuple[dict[str, Any], str]:
    """YAMLフロントマターと本文を抽出する."""
    content = file_path.read_text(encoding="utf-8")
    metadata: dict[str, Any] = {}
    body = content

    if content.startswith("---"):
        parts = content.split("---", 2)
        if len(parts) >= 3:
            try:
                metadata = yaml.safe_load(parts[1]) or {}
            except Exception:
                metadata = {}
            body = parts[2].strip()

    metadata["source_file"] = str(file_path.as_posix())
    metadata["doc_id"] = file_path.stem
    return metadata, body


def index_iris_bundle(
    bundle_wiki_dir: Path,
) -> tuple[BM25Retriever, list[Document]]:
    """IRISバンドルのwikiディレクトリを走査して索引を構築する."""
    headers_to_split_on = [
        ("#", "Header_1"),
        ("##", "Header_2"),
        ("###", "Header_3"),
    ]
    markdown_splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on
    )
    all_chunks: list[Document] = []

    for md_file in bundle_wiki_dir.glob("*.md"):
        meta, body = parse_frontmatter_markdown(md_file)
        split_docs = markdown_splitter.split_text(body)

        for chunk in split_docs:
            chunk_metadata = meta.copy()
            chunk_metadata.update(chunk.metadata)
            all_chunks.append(
                Document(
                    page_content=chunk.page_content, metadata=chunk_metadata
                )
            )

    retriever = BM25Retriever.from_documents(
        documents=all_chunks,
        preprocess_func=japanese_tokenizer,
        k=3,
    )
    return retriever, all_chunks

2. LOTUS ノートデータの索引化と差分更新

特徴と課題

LOTUS（data/lotus/<bundle-name>/）は、日報やドラフト、アイデアメモなどの「生きたドキュメント」が格納される階層構造のディレクトリです。ファイルが頻繁に新規作成・推敲・更新されるため、毎回全件を再インデックスすると大きな計算コストがかかります。ファイルの最終更新日時（mtime）やハッシュ値を追跡し、変更があったノートのみを増分更新する仕組み が有効です。

実装手順

インデックス対象のファイルパスとハッシュ値を管理台帳（JSON / メモリ）に記録します。
更新または新規追加されたファイルのみを読み込み、チャンキングしてインデックスに追加します。
削除されたファイルのチャンクはインデックスから除外します。

import hashlib
import json
from pathlib import Path
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter


def compute_file_hash(path: Path) -> str:
    """ファイルのMD5ハッシュ値を計算する."""
    hasher = hashlib.md5()
    hasher.update(path.read_bytes())
    return hasher.hexdigest()


class LotusIncrementalIndexer:
    """LOTUSノート用の差分インデックス更新マネージャー."""

    def __init__(self, root_dir: Path, ledger_path: Path):
        self.root_dir = root_dir
        self.ledger_path = ledger_path
        self.ledger: dict[str, str] = self._load_ledger()
        self.documents: list[Document] = []
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=400, chunk_overlap=50
        )

    def _load_ledger(self) -> dict[str, str]:
        if self.ledger_path.exists():
            return json.loads(self.ledger_path.read_text(encoding="utf-8"))
        return {}

    def _save_ledger(self) -> None:
        self.ledger_path.parent.mkdir(parents=True, exist_ok=True)
        self.ledger_path.write_text(
            json.dumps(self.ledger, indent=2, ensure_ascii=False),
            encoding="utf-8",
        )

    def scan_and_build(self) -> BM25Retriever:
        """差分を検知してBM25Retrieverを構築する."""
        current_files: dict[str, Path] = {}
        for p in self.root_dir.rglob("*.md"):
            if not p.name.startswith("."):
                current_files[str(p.as_posix())] = p

        all_docs: list[Document] = []
        new_ledger: dict[str, str] = {}

        for rel_path, file_path in current_files.items():
            file_hash = compute_file_hash(file_path)
            new_ledger[rel_path] = file_hash

            # 本文読み込みとチャンキング
            content = file_path.read_text(encoding="utf-8")
            chunks = self.splitter.split_text(content)
            for i, chunk_text in enumerate(chunks):
                all_docs.append(
                    Document(
                        page_content=chunk_text,
                        metadata={
                            "source": rel_path,
                            "chunk_id": f"{file_path.stem}_{i}",
                            "file_name": file_path.name,
                        },
                    )
                )

        self.ledger = new_ledger
        self._save_ledger()
        self.documents = all_docs

        return BM25Retriever.from_documents(
            documents=all_docs,
            preprocess_func=japanese_tokenizer,
            k=3,
        )

3. 大規模テキストデータ（行単位データ）のストリーミング索引化

特徴と課題

数百万行におよぶアクセスログ、チャット履歴、辞書データ、TSV/CSV などの大容量ファイルを扱う場合、ファイル全体を一度にメモリへ読み込むとメモリ枯渇（OOM: Out of Memory）を引き起こします。このようなデータには、ジェネレータを用いたストリーミング処理 と、行番号やファイルオフセットをメタデータとして保持する設計が必須です。

実装手順

ファイルを 1 行ずつ読み込むジェネレータを作成し、メモリ使用量を最小限に抑えます。
一定件数（バッチサイズ）ごとに Document を生成し、BM25 またはベクトルストアへ流し込みます。
メタデータに line_number を保持することで、検索ヒット時に前後の文脈をピンポイントで逆引きできるようにします。

from collections.abc import Generator
from pathlib import Path
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document


def stream_line_documents(
    file_path: Path, batch_size: int = 1000
) -> Generator[list[Document], None, None]:
    """大容量テキストファイルをストリーミングで1行ずつ読み込みバッチ化する."""
    batch: list[Document] = []
    with file_path.open("r", encoding="utf-8") as f:
        for line_num, line in enumerate(f, start=1):
            cleaned_line = line.strip()
            if not cleaned_line:
                continue

            doc = Document(
                page_content=cleaned_line,
                metadata={
                    "source": file_path.name,
                    "line_number": line_num,
                },
            )
            batch.append(doc)

            if len(batch) >= batch_size:
                yield batch
                batch = []

    if batch:
        yield batch


def build_line_index(
    file_path: Path, max_lines: int | None = None
) -> BM25Retriever:
    """行単位テキストからBM25Retrieverを構築する."""
    all_docs: list[Document] = []
    count = 0

    for doc_batch in stream_line_documents(file_path, batch_size=5000):
        all_docs.extend(doc_batch)
        count += len(doc_batch)
        if max_lines and count >= max_lines:
            break

    return BM25Retriever.from_documents(
        documents=all_docs,
        preprocess_func=japanese_tokenizer,
        k=5,
    )

4. 複数 PDF ドキュメントの索引化パイプライン

特徴と課題

業務規程集や仕様書、製品マニュアルなど、複数の PDF ファイルを横断して検索したいユースケースは非常に多く存在します。しかし、PDF は以下のような特有のノイズを含んでいるため、事前のクレンジングが検索精度を大きく左右します。

行末の不要な改行（文の途中で改行が入る）
ページ番号やヘッダー・フッターの混入
表や記号のレイアウト崩れ

実装手順

pypdf（または PyMuPDF）で指定ディレクトリ内の全 PDF を走査し、ページごとにテキストを抽出します。
日本語の文中にある不自然な改行を結合し、テキストを正規化します。
RecursiveCharacterTextSplitter で分割し、ページ番号とファイル名をメタデータとして付与します。

from pathlib import Path
import re
from langchain_community.document_loaders import PyPDFLoader
from langchain_community.retrievers import BM25Retriever
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter


def clean_pdf_text(text: str) -> str:
    """PDF抽出テキストの改行ノイズを正規化する."""
    # 日本語文字に挟まれた単一改行を結合（文末の句読点以外）
    cleaned = re.sub(r"([^\n。！？])\n([^\n])", r"\1\2", text)
    # 連続する複数の空白を1つの空白に置換
    cleaned = re.sub(r"[ \t]+", " ", cleaned)
    return cleaned.strip()


def index_multiple_pdfs(pdf_dir: Path) -> BM25Retriever:
    """ディレクトリ配下の複数PDFファイルを走査してインデックスを構築する."""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=80,
        separators=["\n\n", "。", "！", "？", "\n", " ", ""],
    )
    all_chunks: list[Document] = []

    for pdf_path in pdf_dir.glob("*.pdf"):
        loader = PyPDFLoader(str(pdf_path))
        pages = loader.load()

        for page in pages:
            normalized_text = clean_pdf_text(page.page_content)
            if not normalized_text:
                continue

            page_doc = Document(
                page_content=normalized_text, metadata=page.metadata
            )
            chunks = text_splitter.split_documents([page_doc])
            all_chunks.extend(chunks)

    return BM25Retriever.from_documents(
        documents=all_chunks,
        preprocess_func=japanese_tokenizer,
        k=3,
    )

5. インデックスの永続化と検索システムの統合

なぜインデックスの永続化が必要か

BM25Retriever は内部でメモリ上に転置インデックス（単語と文書の対応表）を保持しているため、Python プロセスを終了するとインデックス情報が消失します。 Web サーバーの起動時や検索リクエストのたびに膨大なドキュメントを再トークナイズすると、起動や応答に多大な時間がかかってしまいます。そのため、構築した BM25 インデックスおよびベクトルストアをディスクに保存（シリアライズ）し、次回以降は瞬時に再ロードできる仕組みを導入します。

インデックスの保存とロードの実装

from pathlib import Path
import pickle
from langchain_community.retrievers import BM25Retriever


def save_bm25_retriever(retriever: BM25Retriever, save_path: Path) -> None:
    """BM25Retrieverをpickleでディスクにシリアライズ保存する."""
    save_path.parent.mkdir(parents=True, exist_ok=True)
    with save_path.open("wb") as f:
        pickle.dump(retriever, f)


def load_bm25_retriever(save_path: Path) -> BM25Retriever:
    """保存されたBM25Retrieverをディスクから高速に復元する."""
    with save_path.open("rb") as f:
        retriever = pickle.load(f)
    return retriever

検索システム（EnsembleRetriever）への接続

永続化した BM25 インデックスと、ディスク保存された Chroma ベクトルストアを組み合わせることで、高速起動可能なハイブリッド検索システムが完成します。

from pathlib import Path
from langchain.retrievers import EnsembleRetriever
from langchain_chroma import Chroma
from langchain_ollama import OllamaEmbeddings


def create_production_hybrid_retriever(
    bm25_index_path: Path,
    chroma_persist_dir: Path,
) -> EnsembleRetriever:
    """永続化されたインデックスをロードして即座にハイブリッドリトリーバーを起動する."""
    # 1. 保存済みBM25Retrieverの復元
    bm25_retriever = load_bm25_retriever(bm25_index_path)
    bm25_retriever.k = 3

    # 2. 保存済みChromaベクトルストアのロード（Ollama bge-m3）
    embeddings = OllamaEmbeddings(model="bge-m3")
    vectorstore = Chroma(
        persist_directory=str(chroma_persist_dir),
        embedding_function=embeddings,
    )
    chroma_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

    # 3. EnsembleRetrieverで統合
    hybrid_retriever = EnsembleRetriever(
        retrievers=[bm25_retriever, chroma_retriever],
        weights=[0.5, 0.5],
    )
    return hybrid_retriever

このように、データソースの性質（構造化 Markdown、推敲ノート、大容量行テキスト、PDF）に応じた前処理・チャンキングを適用し、インデックスを永続化することで、実運用環境に耐えうる頑健で高精度な RAG 索引システムを構築できます。続く実践チュートリアルでは、これまでの実装技術を結集し、SQLite3 FTS5 による高速 BM25 検索と Chroma によるベクトル検索、FlashRank によるリランキング、そして Ollama（gemma4:12b）による推論を組み合わせた、完全なローカル BM25+RAG パイプラインの構築手順をハンズオン形式で体験します。

Keyboard shortcuts