長文の技術記事や議事録を一度にLLMへ渡して要約すると、重要な話題が抜けたり、出力が途中で切れたりします。さらに、記事全体だけを見て章立てやタグ付けを行うと、局所的な内容と全体のテーマが混ざり、結果が安定しません。
そこで今回は、長文を分割して段階的に処理する「階層型要約パイプライン」を実装します。
具体的には、次の順番で処理します。
- 入力文を意味のまとまりごとに分割する
- 各チャンクを要約する
- チャンク要約を章ごとにまとめる
- 章要約から記事全体の要約を作る
- 全体要約と章情報からタグを生成する
私が実際に試した構成では、最初から完成形を要求するよりも、各段階で出力形式を検証するほうが扱いやすく、失敗箇所も特定しやすくなりました。
階層型にする理由
階層型要約パイプライン全体の流れを図にすると次の通りです。
単純な要約は、次のような構造です。
全文 → 要約
一方、階層型では情報量を段階的に減らします。
全文
├─ チャンク1 → 要約1
├─ チャンク2 → 要約2
├─ チャンク3 → 要約3
↓
章ごとの要約
↓
記事全体の要約・タグ
この方式の利点は、LLMに渡す入力サイズを制御しやすいことです。チャンク要約の時点で「事実」「主張」「コード例」「注意点」を抽出しておけば、後段の要約で細部が消えにくくなります。
また、チャンク単位で再実行できるため、APIエラーや出力形式の不備があっても全文を処理し直す必要がありません。
入力の分割と章候補の作成
まずは見出しを優先して分割します。既存のMarkdown記事であれば、## 見出しを章候補として利用できます。見出しがない文章では、空行や文字数を基準に分割します。
以下は、Markdownを章候補に分ける最小限のコードです。
from dataclasses import dataclass
import re
@dataclass
class Section:
title: str
text: str
def split_markdown(markdown: str, max_chars: int = 5000) -> list[Section]:
parts = re.split(r"(?m)^##\s+(.+?)\s*$", markdown)
sections = []
# 見出し前の導入文
if parts[0].strip():
sections.append(Section("導入", parts[0].strip()))
for i in range(1, len(parts), 2):
title = parts[i].strip()
text = parts[i + 1].strip()
# 章が大きすぎる場合は段落単位で分割する
paragraphs = []
current = ""
for paragraph in text.split("\n\n"):
if current and len(current) + len(paragraph) > max_chars:
paragraphs.append(current)
current = ""
current += ("\n\n" if current else "") + paragraph
if current:
paragraphs.append(current)
for index, paragraph in enumerate(paragraphs, start=1):
suffix = f"（{index}）" if len(paragraphs) > 1 else ""
sections.append(Section(title + suffix, paragraph))
return sections
実際にはトークン数で制限するほうが正確ですが、最初の検証では文字数でも十分に挙動を確認できます。日本語では文字数とトークン数が完全には一致しないため、本番運用では使用するモデルのトークナイザーで上限を計測します。
構造化された要約を生成する
要約文だけを返させると、後段で情報を利用しにくくなります。そこで、チャンクごとにJSON形式で出力させます。
今回使うデータ構造は次のようにしました。
from dataclasses import dataclass, asdict
@dataclass
class ChunkSummary:
section: str
summary: str
key_points: list[str]
terms: list[str]
cautions: list[str]
def build_prompt(section: Section) -> str:
return f"""
あなたは技術文書の編集者です。
以下の文章を、後段の章立て・全体要約に利用できるよう整理してください。
必ずJSONだけを返してください。
形式:
見出し: {section.title}
本文:
{section.text}
""".strip()
LLM呼び出し部分は利用する環境に合わせて差し替えられるよう、関数として分離します。
import json
def call_llm(prompt: str) -> str:
# 実際の環境では、利用するLLMのSDKやHTTPクライアントを実装する
raise NotImplementedError
def summarize_section(section: Section) -> ChunkSummary:
raw = call_llm(build_prompt(section))
data = json.loads(raw)
required = {"summary", "key_points", "terms", "cautions"}
missing = required - data.keys()
if missing:
raise ValueError(f"必須キーがありません: {missing}")
return ChunkSummary(
section=section.title,
summary=data["summary"],
key_points=data["key_points"],
terms=data["terms"],
cautions=data["cautions"],
)
JSONが壊れるケースもあるため、実運用ではリトライ時に「前回の出力をJSONに修正してください」と伝える処理を追加します。ただし、無制限にリトライするとコストが増えるので、回数は2〜3回程度に制限しました。
章要約から全体要約とタグを作る
チャンク要約を作ったら、それらを連結して章または記事全体の入力にします。元の長文ではなく、すでに圧縮された要約を渡すのがポイントです。
def build_global_prompt(items: list[ChunkSummary]) -> str:
source = "\n\n".join(
f"### {item.section}\n"
f"要約: {item.summary}\n"
f"重要点: {'、'.join(item.key_points)}\n"
f"用語: {'、'.join(item.terms)}"
for item in items
)
return f"""
以下は長文記事を分割して要約した結果です。
内容の重複を整理し、記事全体の構成を作成してください。
JSONだけを返してください。
形式:
],
"abstract": "記事全体の要約。300文字以内",
"tags": ["Qiita向けタグを3〜5個"]
}}
入力:
{source}
""".strip()
def build_article_result(items: list[ChunkSummary]) -> dict:
raw = call_llm(build_global_prompt(items))
result = json.loads(raw)
if not 3 <= len(result["tags"]) <= 5:
raise ValueError("タグ数は3〜5個にしてください")
return result
タグ付けでは、単に頻出語を抽出するのではなく、「記事の主題を表すタグ」を選ばせるようにしています。例えば本文に一度しか登場しない技術でも、記事全体の中心であればタグ候補になります。
一方で、タグの表記揺れは残ります。生成AI と 生成ＡＩ のような表記を統一する辞書を用意しておくと、検索性が向上します。
CLI化して検証する
処理をコマンドとして実行できるようにすると、入力ファイルを変えた検証が簡単になります。
from pathlib import Path
import json
import sys
def main() -> None:
if len(sys.argv) != 2:
raise SystemExit("使い方: python pipeline.py article.md")
input_path = Path(sys.argv[1])
markdown = input_path.read_text(encoding="utf-8")
sections = split_markdown(markdown)
summaries = [summarize_section(section) for section in sections]
result = build_article_result(summaries)
output = {
"sections": [asdict(item) for item in summaries],
"article": result,
}
output_path = input_path.with_suffix(".summary.json")
output_path.write_text(
json.dumps(output, ensure_ascii=False, indent=2),
encoding="utf-8",
)
print(f"出力しました: {output_path}")
if __name__ == "__main__":
main()
実行方法は次のとおりです。
python pipeline.py sample.md
検証では、次の観点を確認しました。
- 各チャンクの要約が入力内容から逸脱していないか
- 同じ内容が複数章に重複していないか
- 章タイトルが内容を正しく表しているか
- タグが一般的すぎたり、細かすぎたりしないか
- JSONとして毎回読み込めるか
特に重要だったのは、最終結果だけでなく中間JSONも保存することです。最終要約が不自然なとき、チャンク要約の問題なのか、全体要約の問題なのかを切り分けられます。
まとめ
長文の章立て・要約・タグ付けでは、全文を一度に処理するよりも、チャンク、章、記事全体という階層に分けるほうが安定します。
実装上の要点は次の3つです。
- 入力を意味のまとまりごとに分割する
- 中間結果をJSONで固定し、後段で再利用する
- 各段階の出力を保存して検証可能にする
まずは文字数ベースの分割と簡単なJSON検証から始め、運用時にトークン計測、リトライ、キャッシュ、タグ表記の正規化を追加するのがおすすめです。これだけでも、長文処理の再現性とデバッグ性は大きく改善できます。