LLMに社内文書や技術資料を読ませるとき、モデル選びやプロンプトより先に見直したいのが入力データの前処理です。Markdown、PDF、Word文書をそのまま投入すると、改行や見出しが崩れたり、ヘッダー・フッターが混入したりして、回答品質が安定しません。
この記事では、複数形式の文書を「本文」「見出し」「出典」「チャンク」に整理し、LLMや検索システムに渡しやすいJSONLへ変換するパイプラインを作ります。実際に手元のMarkdown、PDF、Wordファイルで試したところ、形式ごとに抽出処理を分け、最後に共通の正規化処理へ集約する構成が扱いやすいと分かりました。
1. 変換後のデータ形式を先に決める
最初に、LLMへ渡すデータの形式を決めます。今回は1文書を複数のチャンクに分割し、次のようなJSONLを出力します。
{"text":"## 認証\nAPIキーは環境変数から読み込みます。","source":"docs/auth.md","section":"認証","chunk_id":"auth_0001"}
JSONLにしておくと、1行単位で読み込めるため、後からベクトル検索用のデータに変換したり、特定の文書だけ再処理したりできます。
最低限、次の情報を残します。
-
text: LLMへ渡す本文 -
source: 元ファイルのパス -
section: 見出し -
chunk_id: チャンクを一意に識別するID
ページ番号や文書の更新日時も、必要に応じてメタデータへ追加します。
2. 必要なライブラリを準備する
今回はPythonで実装します。PDFのテキスト抽出にはPyMuPDF、Word文書にはpython-docxを使います。
python -m venv .venv
.venv\Scripts\activate
pip install pymupdf python-docx
macOSやLinuxの場合は、仮想環境の有効化コマンドが異なります。
source .venv/bin/activate
なお、画像として保存されたPDFは、PyMuPDFだけでは本文を取得できません。その場合はOCRが必要です。まずはテキストを含むPDFを対象にし、スキャンPDFは別処理として切り分けるのが安全です。
3. 形式ごとの抽出処理を作る
文書の抽出から正規化、チャンク分割、検証までの流れを図にすると次の通りです。
入力形式ごとに抽出関数を用意します。Markdownはファイル自体がほぼテキストなので、そのまま読み込みます。Wordは段落を順番に結合し、PDFはページ単位でテキストを取得します。
from pathlib import Path
import fitz
from docx import Document
def read_markdown(path: Path) -> str:
return path.read_text(encoding="utf-8")
def read_word(path: Path) -> str:
document = Document(path)
paragraphs = []
for paragraph in document.paragraphs:
text = paragraph.text.strip()
if text:
paragraphs.append(text)
return "\n".join(paragraphs)
def read_pdf(path: Path) -> str:
pages = []
with fitz.open(path) as document:
for page_number, page in enumerate(document, start=1):
text = page.get_text("text").strip()
if text:
pages.append(f"[ページ {page_number}]\n{text}")
return "\n\n".join(pages)
def extract_text(path: Path) -> str:
suffix = path.suffix.lower()
if suffix in {".md", ".markdown", ".txt"}:
return read_markdown(path)
if suffix == ".docx":
return read_word(path)
if suffix == ".pdf":
return read_pdf(path)
raise ValueError(f"未対応の拡張子です: {path}")
実際に試すと、PDFは段組みや表の順番が崩れる場合がありました。また、Wordの表はdocument.paragraphsだけでは取得できません。表を重要な情報として扱う場合は、テーブルをMarkdown風のテキストへ変換する処理を追加します。
def read_word_with_tables(path: Path) -> str:
document = Document(path)
parts = []
for paragraph in document.paragraphs:
text = paragraph.text.strip()
if text:
parts.append(text)
for table in document.tables:
rows = []
for row in table.rows:
cells = [cell.text.strip().replace("\n", " ") for cell in row.cells]
rows.append(" | ".join(cells))
parts.append("\n".join(rows))
return "\n\n".join(parts)
文書の種類によっては、本文と表の本来の位置関係が失われます。厳密な順序が必要な帳票では、抽出後のテキストを目視確認する工程を残すべきです。
4. 改行・見出し・不要情報を正規化する
抽出直後のテキストには、余分な空白、連続改行、PDFのページ番号などが含まれます。ここで文書形式に依存しない共通処理を行います。
import re
def normalize_text(text: str) -> str:
text = text.replace("\r\n", "\n").replace("\r", "\n")
# 行末の空白を除去
text = "\n".join(line.rstrip() for line in text.splitlines())
# 3行以上の改行を2行にまとめる
text = re.sub(r"\n{3,}", "\n\n", text)
# 半角スペースの連続を整理
text = re.sub(r"[ \t]{2,}", " ", text)
return text.strip()
見出しをチャンクのメタデータに入れるため、Markdownの見出しを追跡します。PDFやWordでは見出し情報が失われていることがあるため、抽出できない場合は「本文」とします。
def split_by_heading(text: str) -> list[dict]:
sections = []
current_heading = "本文"
current_lines = []
for line in text.splitlines():
match = re.match(r"^(#{1,6})\s+(.+)$", line.strip())
if match:
if current_lines:
sections.append({
"heading": current_heading,
"text": "\n".join(current_lines).strip(),
})
current_lines = []
current_heading = match.group(2).strip()
continue
current_lines.append(line)
if current_lines:
sections.append({
"heading": current_heading,
"text": "\n".join(current_lines).strip(),
})
return [section for section in sections if section["text"]]
5. チャンク分割とJSONL出力
LLMへ渡す際は、文書全体を1回で投入せず、適度なサイズに分割します。今回は文字数を基準に分割します。日本語では単純な単語数分割が使いにくいため、まずは段落単位でまとめる方法が実装しやすいです。
def chunk_text(text: str, max_chars: int = 1200) -> list[str]:
paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
chunks = []
current = ""
for paragraph in paragraphs:
candidate = f"{current}\n\n{paragraph}".strip()
if current and len(candidate) > max_chars:
chunks.append(current)
current = paragraph
else:
current = candidate
if current:
chunks.append(current)
return chunks
見出し情報と組み合わせ、JSONLへ出力します。
import json
def convert_file(input_path: Path, output_path: Path) -> None:
raw_text = extract_text(input_path)
normalized = normalize_text(raw_text)
sections = split_by_heading(normalized)
records = []
chunk_number = 1
for section in sections:
for chunk in chunk_text(section["text"]):
records.append({
"text": f"## {section['heading']}\n{chunk}",
"source": str(input_path),
"section": section["heading"],
"chunk_id": f"{input_path.stem}_{chunk_number:04d}",
})
chunk_number += 1
with output_path.open("w", encoding="utf-8") as file:
for record in records:
file.write(json.dumps(record, ensure_ascii=False) + "\n")
ディレクトリ内の対象ファイルをまとめて処理する場合は、次のようにします。
from pathlib import Path
input_dir = Path("documents")
output_dir = Path("prepared")
output_dir.mkdir(exist_ok=True)
for path in input_dir.rglob("*"):
if path.suffix.lower() not in {".md", ".markdown", ".txt", ".pdf", ".docx"}:
continue
output_path = output_dir / f"{path.stem}.jsonl"
convert_file(path, output_path)
print(f"変換完了: {path} -> {output_path}")
6. LLM投入前に確認すること
変換が成功しても、内容が正しいとは限りません。私は最初に次のコマンドでJSONとして読み込めることを確認しました。
python -c "import json; [json.loads(line) for line in open('prepared/manual.jsonl', encoding='utf-8')]; print('OK')"
さらに、次の点をサンプル確認します。
- PDFのページ順が入れ替わっていないか
- ヘッダーやフッターが大量に混入していないか
- 表の列が意味を保っているか
- 文字化けや欠落がないか
- チャンクの途中で重要な手順が切れていないか
-
sourceから元文書を追跡できるか
特に、文書中の「社外秘」「個人情報」「認証情報」は、LLM投入前に除外またはマスキングします。前処理パイプラインにアクセスログや除外ルールを組み込み、誰がどの文書を処理したか追跡できるようにしておくと、後から原因を調べやすくなります。
まとめ
Markdown、PDF、WordをLLM投入用データへ変換する場合は、形式ごとの抽出処理と、共通の正規化・分割処理を分ける構成が扱いやすいです。
今回のポイントは次の4つです。
- 出力形式をJSONLとして先に決める
- PDF、Word、Markdownをそれぞれ適切に抽出する
- 見出しや出典をメタデータとして残す
- LLMへ渡す前に、抽出結果を目視と機械の両方で検証する
最初から完璧な文書解析を目指すより、まずは小さなサンプルで抽出結果を確認し、表・画像PDF・特殊なレイアウトを必要に応じて拡張するのが現実的です。