長文の技術記事や議事録を一度に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検証から始め、運用時にトークン計測、リトライ、キャッシュ、タグ表記の正規化を追加するのがおすすめです。これだけでも、長文処理の再現性とデバッグ性は大きく改善できます。