【FastAPIで作る社内ナレッジ検索(RAG) 第1回】社内文書を読み込んで、検索しやすい断片に分ける

社内ナレッジ検索(RAG)を作るとき、最初につまずきやすいのが「文書をどう切り分けて保存するか」です。この連載の第0回では動く最小のAPIを作りました。第1回では、社内文書(Markdown・テキスト)を読み込み、検索しやすい断片(チャンク)に分ける処理をPythonで作ります。

先に結論をお伝えします。文書は、見出しと段落の区切りを守って、300〜500文字程度の断片に分けるのが基本です。丸ごと1つで保存すると、質問と関係ない部分まで混ざって検索の精度が落ちます。逆に細かく切りすぎると、前後の文脈が失われて、根拠として使えなくなります。この回では、その中間を取る「見出しを超えない、段落を割らない」分け方を実装して、テストで確かめます。

前回の記事:【第0回】全体像と、動く最小APIを作る

社内文書を断片(チャンク)に分ける理由

RAG(＝検索で見つけた情報を材料に、AIに答えを作らせる方式)では、質問が来るたびに「関係する断片だけ」を探してAIに渡します。この考え方は、2020年の研究論文で示された方式がもとになっています。

📰 出典：Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks(arXiv)

断片に分ける理由は、次の3つです。

• 検索の精度:「有給休暇の申請方法」を探したいのに、休暇規程全体が1件だと、慶弔休暇や半日休暇の話まで一緒に評価されてしまいます。
• 出典の分かりやすさ:回答の根拠を「休暇規程 > 年次有給休暇 > 申請方法」のように示すには、断片ごとに見出しの情報を持たせる必要があります。
• AIに渡す量の節約:AIに渡す文章が多いほど、利用料金と回答までの時間が増えやすくなります。必要な部分だけ渡せれば無駄が減ります。

なお、断片の大きさに唯一の正解はありません。この記事の「400文字」は、サンプル文書で動きを確かめやすい値として筆者が選んだ目安で、公式の推奨値ではありません。実際の文書で検索の結果を見ながら調整します。

この回で作るもの:読み込みと分割の2段構え

処理は「読み込む」と「分ける」の2つに分けます。役割を分けておくと、第7回でPDFを追加するときに、読み込み側だけを増やせば済みます。

部品	ファイル	役割
読み込み	app/ingest/loader.py	フォルダ内の.md/.txtを読み、文字コードと改行をそろえる
分割	app/ingest/chunker.py	見出しごとに区切り、段落を詰めて、長すぎる段落は句点で切る
データの型	app/ingest/models.py	文書(Document)と断片(Chunk)の形を決める

分割のルールは次のとおりです。

1. Markdownの見出し(#)ごとに区切る。見出しをまたいで1つの断片にしない
2. 見出しは階層でつなぐ(例:休暇規程 > 年次有給休暇 > 申請方法)
3. 段落(空行で区切られた文章のかたまり)は、できるだけ割らずに、上限文字数まで詰める
4. 1段落だけで上限を超える場合は、句点「。」の位置で切る
5. 句点が無い長文だけは、やむを得ず文字数で切る

実装

断片のデータ型(app/ingest/models.py)

from dataclasses import dataclass


@dataclass(frozen=True)
class Document:
    """読み込んだ社内文書1つ分。source は出典表示に使うファイル名。"""

    source: str
    text: str


@dataclass(frozen=True)
class Chunk:
    """検索の単位になる文書の断片。heading は「見出しの階層」で、出典表示に使う。"""

    source: str
    heading: str
    index: int
    text: str

断片には、本文だけでなく、出典の表示に使うファイル名(source)と見出しの階層(heading)を持たせます。第4回で「どの文書のどこか」を答えに添えるための準備です。

文書の読み込み(app/ingest/loader.py)

from pathlib import Path

from app.ingest.models import Document

SUPPORTED_SUFFIXES = {".md", ".txt"}


def load_documents(directory: Path) -> list[Document]:
    """ディレクトリ配下の Markdown / テキストを読み込む（並び順は固定）。"""
    documents: list[Document] = []
    for path in sorted(directory.rglob("*")):
        if not path.is_file() or path.suffix.lower() not in SUPPORTED_SUFFIXES:
            continue
        text = path.read_text(encoding="utf-8").replace("\r\n", "\n").strip()
        if text:
            documents.append(Document(source=path.relative_to(directory).as_posix(), text=text))
    return documents

sorted で並び順を固定しているのは、実行のたびに順番が変わると、テストの結果や後で付ける番号がぶれるためです。Windowsの改行(\r\n)は\nにそろえます。文字コードはUTF-8を前提にしています。

見出しごとに区切る(app/ingest/chunker.py)

import re

from app.ingest.models import Chunk, Document

HEADING = re.compile(r"^(#{1,6})\s+(.+?)\s*$")


def split_sections(text: str) -> list[tuple[str, str]]:
    """Markdown を見出しごとの (見出しの階層, 本文) に分ける。見出しが無ければ全体で1つ。"""
    sections: list[tuple[str, str]] = []
    path: list[tuple[int, str]] = []
    body: list[str] = []

    def flush() -> None:
        content = "\n".join(body).strip()
        if content:
            sections.append((" > ".join(title for _, title in path), content))
        body.clear()

    for line in text.split("\n"):
        match = HEADING.match(line)
        if match:
            flush()
            level = len(match.group(1))
            while path and path[-1][0] >= level:
                path.pop()
            path.append((level, match.group(2)))
        else:
            body.append(line)
    flush()
    return sections

見出しの深さ(#の数)を見て、同じ深さ以上の見出しが来たら階層を巻き戻します。これで「### 申請方法」の次に「### 半日休暇」が来たとき、階層が「休暇規程 > 年次有給休暇 > 半日休暇」となり、「申請方法」が残りません。

段落を詰めて、長すぎる場合は句点で切る(app/ingest/chunker.py)

def split_long_paragraph(paragraph: str, max_chars: int) -> list[str]:
    """長すぎる段落を、句点「。」の位置で max_chars 以内に分ける。"""
    pieces: list[str] = []
    current = ""
    for sentence in re.findall(r"[^。]+。?", paragraph):
        while len(sentence) > max_chars:  # 句点が無い長文は、やむを得ず文字数で切る
            if current:
                pieces.append(current)
                current = ""
            pieces.append(sentence[:max_chars])
            sentence = sentence[max_chars:]
        if current and len(current) + len(sentence) > max_chars:
            pieces.append(current)
            current = ""
        current += sentence
    if current:
        pieces.append(current)
    return pieces


def pack_paragraphs(body: str, max_chars: int) -> list[str]:
    """段落を max_chars を超えない範囲でまとめる。段落の途中では切らないのが基本。"""
    paragraphs = [p.strip() for p in re.split(r"\n\s*\n", body) if p.strip()]
    units: list[str] = []
    for paragraph in paragraphs:
        if len(paragraph) > max_chars:
            units.extend(split_long_paragraph(paragraph, max_chars))
        else:
            units.append(paragraph)

    chunks: list[str] = []
    current = ""
    for unit in units:
        if current and len(current) + len(unit) + 2 > max_chars:
            chunks.append(current)
            current = unit
        else:
            current = f"{current}\n\n{unit}" if current else unit
    if current:
        chunks.append(current)
    return chunks


def chunk_document(document: Document, max_chars: int = 400) -> list[Chunk]:
    """1つの文書を、見出しの範囲を超えない断片（チャンク）に分ける。"""
    chunks: list[Chunk] = []
    for heading, body in split_sections(document.text):
        for text in pack_paragraphs(body, max_chars):
            chunks.append(
                Chunk(source=document.source, heading=heading, index=len(chunks), text=text)
            )
    return chunks

正規表現は標準ライブラリだけで書いています。日本語の文章を単語に分けるような専用のライブラリは、この回では使いません。まずは「見出し・段落・句点」という文書の構造だけで、十分に実用的な断片が作れます。

動作確認の方法

自動テスト(tests/test_ingest.py)

    assert split_sections(text) == [
        ("規程 > 有給 > 申請", "3営業日前まで。"),
        ("規程 > 有給 > 半日", "半日単位で取得可。"),
    ]


def test_chunks_never_exceed_max_chars():
    text = "。".join(f"これは{i}番目の文です" for i in range(60)) + "。"
    for piece in pack_paragraphs(text, 100):
        assert len(piece) <= 100


def test_chunks_do_not_cross_headings():
    doc = Document(source="a.md", text="## A\n\nいち\n\n## B\n\nに")
    chunks = chunk_document(doc)
    assert [(c.heading, c.text) for c in chunks] == [("A", "いち"), ("B", "に")]
    assert [c.index for c in chunks] == [0, 1]


def test_text_without_heading_is_one_section():
    chunks = chunk_document(Document(source="a.txt", text="段落1\n\n段落2"))
    assert len(chunks) == 1 and chunks[0].heading == ""

「どの断片も上限を超えない」「見出しをまたがない」ことをテストで固定しています。後の回で分割ルールを変えたとき、意図せず挙動が変わっても気づけます。

pip install -e ".[dev]"
ruff check .
pytest

筆者の環境(Python 3.13)では、第0回の3件と今回の5件を合わせて、テスト8件とruffのチェックがすべて成功しました。

サンプル文書で分割結果を見る

連載用に作った架空の休暇規程(sample_docs/vacation.md)と経費精算ルール(sample_docs/expense.txt)を用意しました。次のコマンドで、断片の一覧を表示できます。

python -m app.ingest sample_docs

実行結果は次のとおりです。

expense.txt: 1件
  [0] (見出しなし) / 159文字 / 経費精算ルール（サンプル）  経費は、支払いから30日以内に…
vacation.md: 4件
  [0] 休暇規程（サンプル） / 38文字 / この文書は連載用に作った架空の規程です。実在の会社・制度とは…
  [1] 休暇規程（サンプル） > 年次有給休暇 > 申請方法 / 105文字 / 年次有給休暇は、原則として3営業日前までに勤怠システムから申…
  [2] 休暇規程（サンプル） > 年次有給休暇 > 半日休暇 / 52文字 / 午前休・午後休として半日単位で取得できます。午前休は始業から…
  [3] 休暇規程（サンプル） > 特別休暇 > 慶弔休暇 / 76文字 / 本人の結婚、配偶者の出産、親族の逝去の際に取得できます。日数…
合計 5件

出力には、見出しの階層と文字数が1件ずつ表示されます。「休暇規程(サンプル) > 年次有給休暇 > 申請方法」のように、出典に使える見出しが付いていることを確認できます。この回では、断片を保存するところまでは進みません。保存は第2回で、PostgreSQLとpgvectorを使って行います。

つまずきやすい点・セキュリティ上の注意

• pip installが失敗する:フォルダを増やした(今回はsample_docs)ところ、pip install -e . が「複数のトップレベルパッケージが見つかった」という趣旨のエラーで失敗しました。pyproject.tomlに [tool.setuptools.packages.find] で include = ["app*"] と書き、アプリのフォルダだけを対象にして解決しています。
• 表の多いMarkdownやスキャンした文書:今回の分け方は、文章が中心の文書を想定しています。表は行の途中で切れることがあります。PDFやスキャン文書は第7回で扱います。
• 文書に含まれる個人情報:断片は後で外部のAIサービスに送られる可能性があります。何を取り込んでよいかは、取り込む前に決めておきます。サンプル文書は架空の内容です。
• この段階では権限を考慮していません:どの部署の誰が読めるかの区別は、第6回で入れます。

発注者向けメモ:文書の取り込みを頼むときの確認点

社内ナレッジ検索は、AIの部分に目が行きがちですが、文書の取り込みと分割の作り込みが回答の品質を大きく左右します。開発会社に依頼するときは、次の点を確認しておくと安心です。

• ☐ 取り込む文書の形式(Word・Excel・PDF・紙のスキャンなど)と、おおよその量を整理した
• ☐ 見出しや章立てが整っている文書か、そうでない文書かを把握した
• ☐ 取り込んではいけない文書(個人情報・取引先との秘密保持契約の対象など)を決めた
• ☐ 文書が更新されたときに、誰が、どの頻度で取り込み直すかを決めた
• ☐ 改訂前の古い版の規程が、検索に混ざらない方針を決めた

開発会社への質問例:

• 「文書はどんな単位で分割し、その大きさはどう決めますか。あとで調整できますか」
• 「回答の根拠として、文書名と見出しまで表示できますか」
• 「Word・Excel・PDFの取り込みは、それぞれ対応範囲と精度の見込みを教えてください」
• 「規程を改訂したとき、古い内容が検索に残らない仕組みはありますか」

工数が増えやすいのは、表や図が多い文書、見出しの付け方が文書ごとに違う場合、そして紙をスキャンした文書です。最初は「見出しが整ったWord・Markdown」など対象を絞ると、見積もりも品質確認も進めやすくなります。

まとめと次回予告

第1回では、社内文書を読み込み、見出しと段落を守りながら断片に分ける処理を作りました。断片には出典表示用の見出し階層を持たせ、テストで上限文字数と見出しの境界を固定しています。

次回の第2回では、PostgreSQLとpgvectorをDocker Composeで動かし、今回作った断片を保存する土台を作ります(タイトル案:「PostgreSQL + pgvectorを Docker Composeで動かす」)。

この連載の記事一覧

この記事は連載「FastAPIで作る社内ナレッジ検索(RAG)」の1回です。連載のほかの回は次のとおりです（連載の一覧ページ）。

• 【FastAPIで作る社内ナレッジ検索(RAG) 第0回】全体像と、動く最小APIを作る
• 【FastAPIで作る社内ナレッジ検索(RAG) 第1回】社内文書を読み込んで、検索しやすい断片に分ける（この記事）
• 【FastAPIで作る社内ナレッジ検索(RAG) 第2回】PostgreSQLとpgvectorをDocker Composeで動かし、文書の断片を保存する
• 【FastAPIで作る社内ナレッジ検索(RAG) 第3回】質問に近い社内文書を探す(ベクトル検索)を実装する
• 【FastAPIで作る社内ナレッジ検索(RAG) 第4回】見つけた社内文書を根拠に、出典つきで生成AIに回答させる
• 【FastAPIで作る社内ナレッジ検索(RAG) 第5回】答えが社内文書に無いときは「見つかりません」と返す仕組みを作る
• 【FastAPIで作る社内ナレッジ検索(RAG) 第6回】部署ごとに見てよい文書だけを検索する、アクセス制御を実装する
• 【FastAPIで作る社内ナレッジ検索(RAG) 第7回】社内のPDFを取り込み、ページ番号つきの出典で答えさせる
• 【FastAPIで作る社内ナレッジ検索(RAG) 第8回】回答の品質を評価ケースで測り、直したつもりの後退を自動テストで見つける