長尺PDFの章ごと要約で、まる一章が要約から消えていたとき — 被覆率を計測して取りこぼしを塞ぐ運用メモ

# pip install google-genai pypdf numpy
# 環境変数: GOOGLE_API_KEY="YOUR_API_KEY"
import asyncio
import os
import statistics
from dataclasses import dataclass, field
import numpy as np
from pypdf import PdfReader
from google import genai
 
client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
SUMMARY_MODEL = "gemini-2.5-pro"       # 章要約・全体再結合
VERIFY_MODEL = "gemini-flash-latest"   # 被覆判定（安く速く）
EMBED_MODEL = "gemini-embedding-2"     # 章⇄要約の意味的近さ
 
@dataclass
class Chapter:
    idx: int
    title: str
    body: str
    summary: str = ""
    covered: bool = False
    notes: list[str] = field(default_factory=list)
 
def extract_chapters(pdf_path: str) -> list[Chapter]:
    reader = PdfReader(pdf_path)
    top = [o for o in reader.outline if not isinstance(o, list)]
    chapters: list[Chapter] = []
    for i, entry in enumerate(top):
        start = reader.get_destination_page_number(entry)
        end = (reader.get_destination_page_number(top[i + 1])
               if i + 1 < len(top) else len(reader.pages))
        body = "\n".join(reader.pages[p].extract_text() or "" for p in range(start, end))
        chapters.append(Chapter(idx=i, title=entry.title, body=body))
    return chapters
 
def flag_thin_chapters(chapters: list[Chapter]) -> None:
    """空テキスト経路を入力段で弾く"""
    lengths = [len(c.body) for c in chapters if c.body.strip()]
    if not lengths:
        return
    floor = statistics.median(lengths) * 0.1
    for c in chapters:
        if len(c.body.strip()) < floor:
            c.notes.append(f"本文が短すぎます（{len(c.body.strip())}字）。抽出漏れの疑い")

async def summarize_chapter(ch: Chapter) -> None:
    prompt = (
        f"次は技術文書の第{ch.idx}章『{ch.title}』の本文です。\n"
        "主張を3〜5項目でまとめてください。他章への参照は避けてください。\n\n"
        f"本文:\n{ch.body[:120000]}"
    )
    resp = await asyncio.to_thread(
        client.models.generate_content, model=SUMMARY_MODEL, contents=prompt)
    ch.summary = resp.text.strip()
 
async def merge_summaries(chapters: list[Chapter]) -> str:
    joined = "\n\n".join(f"[章{c.idx}] {c.title}\n{c.summary}" for c in chapters)
    prompt = (
        "以下は文書を章ごとに要約したものです。全体の主張を400字でまとめ、\n"
        "本文の最後に、言及した章を [章0, 章3, ...] の形式で必ず列挙してください。\n\n"
        f"{joined}"
    )
    resp = await asyncio.to_thread(
        client.models.generate_content, model=SUMMARY_MODEL, contents=prompt)
    return resp.text.strip()
 
def embed(texts: list[str]) -> np.ndarray:
    resp = client.models.embed_content(model=EMBED_MODEL, contents=texts)
    return np.array([e.values for e in resp.embeddings])
 
def audit_coverage(chapters: list[Chapter], final: str, threshold: float = 0.62) -> float:
    """章要約と最終要約の意味的近さで被覆を判定する"""
    live = [c for c in chapters if c.summary]
    vecs = embed([c.summary for c in live] + [final])
    final_vec = vecs[-1]
    for c, v in zip(live, vecs[:-1]):
        sim = float(v @ final_vec / (np.linalg.norm(v) * np.linalg.norm(final_vec)))
        c.covered = sim >= threshold
        if not c.covered:
            c.notes.append(f"最終要約への反映が弱い（類似度{sim:.2f}）")
    covered = sum(c.covered for c in live)
    return covered / len(live) if live else 0.0

async def summarize_pdf(pdf_path: str, max_retries: int = 2) -> dict:
    chapters = extract_chapters(pdf_path)
    flag_thin_chapters(chapters)
    sem = asyncio.Semaphore(4)  # 無料枠の429を避ける保守的な同時実行数
 
    async def bounded(c: Chapter):
        async with sem:
            await summarize_chapter(c)
 
    await asyncio.gather(*(bounded(c) for c in chapters if c.body.strip()))
 
    final = await merge_summaries(chapters)
    coverage = audit_coverage(chapters, final)
 
    attempt = 0
    while coverage < 1.0 and attempt < max_retries:
        dropped = [c for c in chapters if c.summary and not c.covered]
        emphasis = "、".join(f"第{c.idx}章「{c.title}」" for c in dropped)
        final = await merge_summaries(chapters) + (
            f"\n\n（補足指示: {emphasis} の主張を必ず本文に含めること）")
        coverage = audit_coverage(chapters, final)
        attempt += 1
 
    return {
        "coverage": coverage,
        "dropped": [c.title for c in chapters if c.summary and not c.covered],
        "thin": [c.title for c in chapters if any("短すぎ" in n for n in c.notes)],
        "summary": final,
    }
 
if __name__ == "__main__":
    report = asyncio.run(summarize_pdf("./spec.pdf"))
    print(f"被覆率: {report['coverage']:.0%}")
    if report["dropped"]:
        print("再結合から落ちた章:", report["dropped"])
    if report["thin"]:
        print("抽出が怪しい章:", report["thin"])
    print(report["summary"])