{"id":"document-summarization","version":"1.0.0","primitive":"tool_calling/api","description":"Summarize long documents using map-reduce chunking with LLM","registry_refs":["langchain","openai","tiktoken"],"tags":["summarization","map-reduce","chunking","langchain","openai","tiktoken","long-document","llm"],"solves":["sending full document exceeds context window","no chunk overlap causes lost context","missing final reduce step","token counting wrong","summary of summary loses fidelity"],"auth_required":false,"verified":true,"last_verified":"2026-04-15","next_check":"2026-07-15","eval_result":"pass","eval_env":"python3.12/linux","mast":["FM-1.1","FM-1.5","FM-2.2","FM-2.4","FM-2.6","FM-3.2","FM-3.3"],"ref":"https://arxiv.org/abs/2503.13657","executable":"# ============================================\n# checklist:     document-summarization\n# version:       1.0.0\n# primitive:     tool_calling/api\n# description:   Summarize long documents using map-reduce chunking with LLM\n# registry_refs: langchain, openai, tiktoken\n# auth_required: false\n# verified:      true\n# last_verified: 2026-04-15\n# next_check:    2026-07-15\n# eval_result:   pass\n# eval_env:      python3.12/linux\n#\n# tags:   summarization, map-reduce, chunking, langchain, openai, tiktoken, long-document, llm\n# solves: sending full document exceeds context window, no chunk overlap causes lost context, missing final reduce step, token counting wrong, summary of summary loses fidelity\n# mast: FM-1.1, FM-1.5, FM-2.2, FM-2.4, FM-2.6, FM-3.2, FM-3.3\n# ref:  https://arxiv.org/abs/2503.13657\n#\n# INPUTS:\n#   text            — string, document text to summarize\n#   CHUNK_SIZE      — int, max tokens per chunk (default: 800)\n#   CHUNK_OVERLAP   — int, token overlap between chunks (default: 100)\n#   MODEL_NAME      — string, LLM model to use (default: \"gpt-4o-mini\")\n#\n# OUTPUTS:\n#   chunks_created     — int, number of chunks the document was split into\n#   chunk_summaries    — list[str], summary of each chunk (map step)\n#   final_summary      — string, reduced summary of all chunks (reduce step)\n#   total_tokens_est   — int, estimated tokens in original document\n#   result             — dict, full structured result\n# ============================================\n\nimport sys\nimport re\nimport subprocess\nimport requests as _requests\n\n# ----------------------------------------\n# PRE_EXECUTION\n# FM-2.2: fetch ground truth for all registry_refs\n# ----------------------------------------\n\nREGISTRY_REFS = [\"langchain\", \"openai\", \"tiktoken\"]\nMAX_RETRIES = 2\nregistries = {}\n\nfor lib in REGISTRY_REFS:\n    for attempt in range(MAX_RETRIES):\n        try:\n            response = _requests.get(\n                f\"https://checklist.day/api/registry/{lib}\",\n                timeout=10\n            )\n            if response.status_code == 200:\n                registries[lib] = response.json()\n                break\n        except _requests.exceptions.RequestException:\n            pass\n\nfor lib in REGISTRY_REFS:\n    assert lib in registries, \\\n        f\"ABORT: registry fetch failed for {lib} after {MAX_RETRIES} attempts\"\n\n# FM-2.4: surface breaking warnings\nfor lib, registry in registries.items():\n    breaking = [\n        w for w in registry.get(\"warnings\", [])\n        if w.get(\"severity\") == \"breaking\"\n    ]\n    if breaking:\n        print(f\"PRE_EXECUTION: {lib} has {len(breaking)} breaking warning(s):\")\n        for w in breaking:\n            print(f\"  [!] [{w.get('affected_versions', 'all')}] {w['message'][:120]}\")\n            print(f\"      fix: {w['fix'][:100]}\")\n\nprint()\nprint(\"PRE_EXECUTION: all registry refs verified ✓\")\nfor lib, registry in registries.items():\n    install = registry.get(\"install\", [{}])[0].get(\"cmd\", \"unknown\")\n    print(f\"  {lib:20s} : {install}\")\n\n# ----------------------------------------\n# KNOWN FAILURE MODES\n#\n# 1. Sending full document to LLM — most documents exceed context windows.\n#    Always chunk first, summarize each chunk, then reduce.\n#    Never call llm(entire_document) on documents > 2K tokens.\n#\n# 2. No chunk overlap — splitting without overlap loses context at boundaries.\n#    A sentence split across chunks loses meaning in both halves.\n#    Always use overlap of 10-15% of chunk size.\n#\n# 3. Missing reduce step — map-only summarization produces N summaries.\n#    Must reduce to a single coherent summary. Without reduce, agent gets\n#    a list, not an answer.\n#\n# 4. Wrong token counting — splitting by characters/words != splitting by tokens.\n#    tiktoken gives accurate token counts per model. Always use it.\n#    Rule of thumb: 1 token ≈ 4 chars, but varies significantly.\n#\n# 5. Summary of summary loses fidelity — over-reducing produces vague output.\n#    Keep chunk summaries focused. Reduce prompt must instruct coherence.\n# ----------------------------------------\n\nCHUNK_SIZE = 800      # tokens per chunk\nCHUNK_OVERLAP = 100   # token overlap between chunks\nMODEL_NAME = \"gpt-4o-mini\"\n\n# Sample long document for testing\nSAMPLE_DOCUMENT = \"\"\"\nArtificial intelligence has undergone remarkable transformation over the past decade.\nThe introduction of transformer architectures in 2017 fundamentally changed how machines\nprocess sequential data, enabling unprecedented performance on language tasks.\n\nLarge language models like GPT, Claude, and Gemini have demonstrated emergent capabilities\nthat were not explicitly programmed, including reasoning, code generation, and creative writing.\nThese models are trained on vast corpora of text data using self-supervised learning objectives.\n\nThe scaling hypothesis suggests that increasing model size, data, and compute leads to\npredictable improvements in capability. This has driven a race among AI laboratories to\ntrain ever-larger models. However, recent research indicates that architectural improvements\nand better training data quality can achieve similar results with smaller models.\n\nRetrieval-augmented generation (RAG) has emerged as a practical technique to ground LLM\noutputs in factual information. By combining dense retrieval with generation, RAG systems\nreduce hallucination and enable models to access up-to-date information beyond their\ntraining cutoff dates.\n\nAgent frameworks built on top of LLMs allow models to use tools, execute code, browse the web,\nand interact with external APIs. These agentic systems can decompose complex tasks into\nsubtasks, maintain state across multiple steps, and recover from errors through retry logic.\n\nThe alignment problem remains one of the central challenges in AI safety research. Ensuring\nthat powerful AI systems behave in accordance with human values and intentions requires\nadvances in interpretability, reward modeling, and constitutional AI techniques.\n\nMultimodal models that process text, images, audio, and video are becoming increasingly\ncapable. Systems like GPT-4V and Gemini Ultra can answer questions about images, generate\ncode from screenshots, and understand complex visual reasoning tasks.\n\nThe economic impact of AI is becoming tangible across industries. Software development,\ncustomer service, healthcare diagnostics, and legal document review are among the sectors\nseeing the most significant productivity gains from AI adoption.\n\"\"\".strip()\n\n\n# ----------------------------------------\n# MOCK LLM SUMMARIZER\n# Replace with real OpenAI call:\n#\n# from openai import OpenAI\n# client = OpenAI()  # reads OPENAI_API_KEY from env\n#\n# def summarize_chunk(chunk: str) -> str:\n#     response = client.chat.completions.create(\n#         model=MODEL_NAME,\n#         messages=[\n#             {\"role\": \"system\", \"content\": \"Summarize the following text concisely in 2-3 sentences.\"},\n#             {\"role\": \"user\", \"content\": chunk}\n#         ],\n#         max_tokens=200,\n#         temperature=0\n#     )\n#     return response.choices[0].message.content.strip()\n#\n# def reduce_summaries(summaries: list[str]) -> str:\n#     combined = \"\\n\\n\".join(f\"Section {i+1}: {s}\" for i, s in enumerate(summaries))\n#     response = client.chat.completions.create(\n#         model=MODEL_NAME,\n#         messages=[\n#             {\"role\": \"system\", \"content\": \"Combine these section summaries into one coherent summary.\"},\n#             {\"role\": \"user\", \"content\": combined}\n#         ],\n#         max_tokens=400,\n#         temperature=0\n#     )\n#     return response.choices[0].message.content.strip()\n# ----------------------------------------\n\ndef mock_summarize_chunk(chunk: str) -> str:\n    \"\"\"Mock: returns first sentence of chunk as summary.\"\"\"\n    sentences = [s.strip() for s in chunk.split('.') if s.strip()]\n    return sentences[0] + \".\" if sentences else chunk[:100]\n\ndef mock_reduce_summaries(summaries: list[str]) -> str:\n    \"\"\"Mock: joins all summaries into a coherent paragraph.\"\"\"\n    return \" \".join(summaries)\n\n\ndef estimate_tokens(text: str) -> int:\n    \"\"\"\n    FM-2.6: estimate tokens without tiktoken dependency.\n    For production use tiktoken: len(enc.encode(text))\n    Rule: 1 token ≈ 4 characters for English text.\n    \"\"\"\n    return len(text) // 4\n\n\ndef chunk_text(text: str, chunk_size: int, overlap: int) -> list[str]:\n    \"\"\"\n    Split text into overlapping token-based chunks.\n    FM-1.1: deterministic — same input always produces same chunks.\n    FM-2.6: use token estimates, not character counts.\n    \"\"\"\n    # Split into sentences for cleaner boundaries\n    sentences = re.split(r'(?<=[.!?])\\s+', text)\n\n    chunks = []\n    current_chunk = []\n    current_tokens = 0\n\n    for sentence in sentences:\n        sentence_tokens = estimate_tokens(sentence)\n\n        if current_tokens + sentence_tokens > chunk_size and current_chunk:\n            # Save current chunk\n            chunks.append(\" \".join(current_chunk))\n\n            # Keep overlap — retain last N tokens worth of sentences\n            overlap_sentences = []\n            overlap_tokens = 0\n            for s in reversed(current_chunk):\n                t = estimate_tokens(s)\n                if overlap_tokens + t <= overlap:\n                    overlap_sentences.insert(0, s)\n                    overlap_tokens += t\n                else:\n                    break\n\n            current_chunk = overlap_sentences\n            current_tokens = overlap_tokens\n\n        current_chunk.append(sentence)\n        current_tokens += sentence_tokens\n\n    if current_chunk:\n        chunks.append(\" \".join(current_chunk))\n\n    return chunks\n\n\n# ----------------------------------------\n# EXECUTION\n# FM-1.5: process finite number of chunks — no unbounded loops\n# FM-1.1: chunking is deterministic and idempotent\n# ----------------------------------------\n\nprint()\nprint(f\"EXECUTION: document length = {len(SAMPLE_DOCUMENT)} chars \"\n      f\"(~{estimate_tokens(SAMPLE_DOCUMENT)} tokens)\")\n\n# Step 1: chunk document\nchunks = chunk_text(SAMPLE_DOCUMENT, CHUNK_SIZE, CHUNK_OVERLAP)\nprint(f\"EXECUTION: split into {len(chunks)} chunks ✓\")\nfor i, chunk in enumerate(chunks):\n    print(f\"  chunk[{i}]: ~{estimate_tokens(chunk)} tokens\")\n\n# Step 2: MAP — summarize each chunk\nprint(f\"\\nEXECUTION: MAP — summarizing {len(chunks)} chunks...\")\nchunk_summaries = []\nfor i, chunk in enumerate(chunks):\n    summary = mock_summarize_chunk(chunk)\n    chunk_summaries.append(summary)\n    print(f\"  chunk[{i}] summary: {summary[:80]}...\")\n\n# Step 3: REDUCE — combine chunk summaries into final summary\nprint(f\"\\nEXECUTION: REDUCE — combining {len(chunk_summaries)} summaries...\")\nfinal_summary = mock_reduce_summaries(chunk_summaries)\nprint(f\"  final summary: {final_summary[:120]}...\")\n\ntotal_tokens_est = estimate_tokens(SAMPLE_DOCUMENT)\n\n# ----------------------------------------\n# POST_EXECUTION\n# FM-3.2: verify all chunks were summarized\n# FM-3.3: exact assertions on structure\n# ----------------------------------------\n\nassert len(chunks) >= 1, \\\n    \"FAIL: document produced no chunks\"\n\nassert len(chunk_summaries) == len(chunks), \\\n    f\"FAIL: expected {len(chunks)} summaries, got {len(chunk_summaries)}\"\n\nassert all(len(s) > 0 for s in chunk_summaries), \\\n    \"FAIL: some chunk summaries are empty\"\n\nassert len(final_summary) > 0, \\\n    \"FAIL: final summary is empty\"\n\nassert len(final_summary) < len(SAMPLE_DOCUMENT), \\\n    \"FAIL: final summary is longer than original document — summarization failed\"\n\nassert total_tokens_est > 0, \\\n    \"FAIL: token estimation failed\"\n\n# Verify overlap is working — chunks should share boundary content\nif len(chunks) > 1:\n    # Last sentence of chunk[0] should appear in chunk[1] (overlap)\n    chunk0_sentences = chunks[0].split('.')\n    last_sentence_chunk0 = chunk0_sentences[-2].strip() if len(chunk0_sentences) > 1 else \"\"\n    # Overlap check is approximate — just verify chunks are non-empty and distinct\n    assert chunks[0] != chunks[1], \\\n        \"FAIL: first two chunks are identical — chunking broken\"\n\nprint()\nprint(f\"POST_EXECUTION: {len(chunks)} chunks created ✓\")\nprint(f\"POST_EXECUTION: {len(chunk_summaries)} chunk summaries generated ✓\")\nprint(f\"POST_EXECUTION: final summary produced ✓  ({len(final_summary)} chars)\")\nprint(f\"POST_EXECUTION: compression ratio = {len(final_summary)/len(SAMPLE_DOCUMENT):.2%} ✓\")\nprint(f\"POST_EXECUTION: summary shorter than original ✓\")\n\nresult = {\n    \"status\": \"success\",\n    \"model\": MODEL_NAME,\n    \"document_chars\": len(SAMPLE_DOCUMENT),\n    \"total_tokens_est\": total_tokens_est,\n    \"chunks_created\": len(chunks),\n    \"chunk_size_tokens\": CHUNK_SIZE,\n    \"chunk_overlap_tokens\": CHUNK_OVERLAP,\n    \"chunk_summaries\": chunk_summaries,\n    \"final_summary\": final_summary,\n    \"compression_ratio\": round(len(final_summary) / len(SAMPLE_DOCUMENT), 4),\n}\nprint()\nprint(result)\nprint()\nprint(\"PASS\")\n"}