When a Whole Chapter Vanished From My Long-PDF Summary — Field Notes on Auditing Coverage

# pip install google-genai pypdf numpy
# env: 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"       # chapter + merge summaries
VERIFY_MODEL = "gemini-flash-latest"   # coverage checks (cheap, fast)
EMBED_MODEL = "gemini-embedding-2"     # chapter vs. summary proximity
 
@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:
    """Catch the empty-text path at the input stage."""
    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"body too short ({len(c.body.strip())} chars); suspect extraction gap")

async def summarize_chapter(ch: Chapter) -> None:
    prompt = (
        f"The following is the body of Chapter {ch.idx}, '{ch.title}', from a technical document.\n"
        "Summarize its claims in 3-5 bullet points. Avoid references to other chapters.\n\n"
        f"Body:\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"[ch{c.idx}] {c.title}\n{c.summary}" for c in chapters)
    prompt = (
        "Below are per-chapter summaries of one document. Summarize the overall argument in "
        "400 words, and at the very end list the chapters you referenced as [ch0, ch3, ...].\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:
    """Judge coverage by semantic proximity of chapter summaries to the final one."""
    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"weak reflection in final summary (similarity {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)  # conservative concurrency to dodge free-tier 429s
 
    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"Chapter {c.idx} '{c.title}'" for c in dropped)
        final = await merge_summaries(chapters) + (
            f"\n\n(Extra instruction: the claims of {emphasis} must appear in the body.)")
        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("too short" in n for n in c.notes)],
        "summary": final,
    }
 
if __name__ == "__main__":
    report = asyncio.run(summarize_pdf("./spec.pdf"))
    print(f"coverage: {report['coverage']:.0%}")
    if report["dropped"]:
        print("dropped in merge:", report["dropped"])
    if report["thin"]:
        print("suspect extraction:", report["thin"])
    print(report["summary"])