#4879 Google Patents and arXiv Scraping
#4879 Google Patents and arXiv Scraping
#4879 Google Patents scraping. Can a Python app search Google Patents, scrape the response links, the iterate over the links (across maybe the first 3 pages of search results), opening each link and saving the contents to file?
A Python application can automate searching Google Patents, extracting the result URLs across multiple pages, fetching each individual patent page, and saving the contents to disk.
Technical Realities & Architecture
Client-Side Rendering (JavaScript): Google Patents (
patents.google.com) is built with Polymer/Web Components (<search-result-item>). A plainrequests.get()call on the search page will not yield the search result elements because they are rendered via JavaScript on the client side. You must use either:A headless browser driver like Playwright or Selenium.
An official or direct bulk alternative (such as Google Patents Public Datasets on BigQuery or Google Patents CSV bulk export).
Rate Limiting & Anti-Scraping: Aggressive automated requests trigger reCAPTCHAs or IP blocks. Scraping requires modest delays (
time.sleep(2–5)seconds between queries/documents) and realistic browser user-agents.Pagination Structure: Google Patents pages search queries via the
pagequery parameter (e.g.,[https://patents.google.com/?q=quantum+computing&page=0](https://patents.google.com/?q=quantum+computing&page=0),page=1,page=2).
End-to-End Implementation
The following script uses Playwright (recommended over Selenium for modern web components and reliability) to search, paginate, extract clean patent links, download each patent's text content, and store them locally as text files.
1. Setup
pip install playwright
playwright install chromium
2. Python Script
import os
import re
import time
from playwright.sync_api import sync_playwright
QUERY = "carbon capture"
NUM_PAGES = 3
OUTPUT_DIR = "./patents_output"
os.makedirs(OUTPUT_DIR, exist_ok=True)
def sanitize_filename(name: str) -> str:
"""Removes invalid characters for filenames."""
return re.sub(r'[\\/*?:"<>|]', "", name)
def run_patent_scraper():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
context = browser.new_context(
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
page = context.new_page()
collected_links = set()
# Step 1: Iterate through the first N search pages
for page_num in range(NUM_PAGES):
search_url = f"https://patents.google.com/?q={QUERY.replace(' ', '+')}&page={page_num}"
print(f"Loading search page {page_num + 1}: {search_url}")
page.goto(search_url, wait_until="networkidle")
# Wait for search result web components to render
page.wait_for_selector("search-result-item", timeout=10000)
# Extract patent links
anchors = page.query_selector_all("search-result-item state-modifier a")
for anchor in anchors:
href = anchor.get_attribute("href")
if href and "/patent/" in href:
# Clean trailing tracking/query parameters
clean_id = href.split("?")[0].split("/")[-2] if href.endswith("/en") else href.split("?")[0].split("/")[-1]
clean_url = f"https://patents.google.com/patent/{clean_id}/en"
collected_links.add((clean_id, clean_url))
time.sleep(2) # Courteous pacing between search pages
print(f"\nTotal unique patents found across {NUM_PAGES} pages: {len(collected_links)}")
# Step 2: Visit each patent link and save its contents
for idx, (patent_id, url) in enumerate(collected_links, start=1):
print(f"[{idx}/{len(collected_links)}] Fetching {patent_id}...")
try:
page.goto(url, wait_until="networkidle")
# Extract Title
title_elem = page.query_selector("meta[name='DC.title']")
title = title_elem.get_attribute("content") if title_elem else "Untitled"
# Extract Abstract
abstract_elem = page.query_selector("section[itemprop='abstract'] .abstract")
abstract = abstract_elem.inner_text() if abstract_elem else "No abstract found."
# Extract Full Description and Claims
description_elem = page.query_selector("section[itemprop='description']")
claims_elem = page.query_selector("section[itemprop='claims']")
description = description_elem.inner_text() if description_elem else ""
claims = claims_elem.inner_text() if claims_elem else ""
# Write to file
safe_id = sanitize_filename(patent_id)
file_path = os.path.join(OUTPUT_DIR, f"{safe_id}.txt")
with open(file_path, "w", encoding="utf-8") as f:
f.write(f"Patent ID: {patent_id}\n")
f.write(f"URL: {url}\n")
f.write(f"Title: {title}\n\n")
f.write("=== ABSTRACT ===\n")
f.write(abstract + "\n\n")
f.write("=== CLAIMS ===\n")
f.write(claims + "\n\n")
f.write("=== DESCRIPTION ===\n")
f.write(description + "\n")
print(f"Saved: {file_path}")
except Exception as e:
print(f"Failed to scrape {patent_id}: {e}")
time.sleep(2) # Delay between individual patent downloads
browser.close()
print("\nAll tasks completed.")
if __name__ == "__main__":
run_patent_scraper()
Key Extraction Details
| Content Target | Target Selector / Method | Description |
| Search Result Items | search-result-item | Google Patents custom Web Component containing individual hits. |
| Patent URLs | search-result-item state-modifier a | Extracts relative /patent/ID/en routes, normalized to strip query parameters. |
| Title | meta[name="DC.title"] | Uses standard Dublin Core meta tags embedded in Google Patents document headers. |
| Abstract / Claims | section[itemprop="abstract"], section[itemprop="claims"] | Semantic schema tags containing structured sections for direct text dumping. |
Repeat for arXiv
Unlike Google Patents, arXiv provides a free, public HTTP API. While you could render arxiv.org/search with a headless browser, using the official arXiv API or its standard Python client (arxiv) is vastly faster, completely bypasses Cloudflare/reCAPTCHA blocking, and gives you clean metadata, abstracts, and direct PDF/HTML download links out of the box.
Method 1: The Official API via arxiv.py (Recommended)
The standard Python wrapper arxiv handles rate limits (respecting arXiv's 3-second delay policy), query pagination (e.g., retrieving the first 3 "pages" of results, such as 30–50 papers), and file saving.
1. Setup
pip install arxiv
2. Implementation
import os
import re
import arxiv
QUERY = "diffusion models"
PAGE_SIZE = 10 # Items per page
NUM_PAGES = 3 # First 3 pages
MAX_RESULTS = PAGE_SIZE * NUM_PAGES
OUTPUT_DIR = "./arxiv_output"
os.makedirs(OUTPUT_DIR, exist_ok=True)
def sanitize_filename(name: str) -> str:
"""Strip characters incompatible with local file systems."""
return re.sub(r'[\\/*?:"<>|]', "", name)
def scrape_arxiv():
# Instantiate client with arXiv-compliant courteous delays
client = arxiv.Client(
page_size=PAGE_SIZE,
delay_seconds=3.0,
num_retries=3
)
# Configure search query
search = arxiv.Search(
query=QUERY,
max_results=MAX_RESULTS,
sort_by=arxiv.SortCriterion.Relevance
)
results = list(client.results(search))
print(f"Retrieved {len(results)} papers across {NUM_PAGES} pages.\n")
for idx, paper in enumerate(results, start=1):
clean_id = sanitize_filename(paper.get_short_id())
file_path = os.path.join(OUTPUT_DIR, f"{clean_id}.txt")
print(f"[{idx}/{len(results)}] Saving metadata for: {paper.title[:60]}...")
# Save structured metadata and full abstract
with open(file_path, "w", encoding="utf-8") as f:
f.write(f"arXiv ID: {paper.entry_id}\n")
f.write(f"Title: {paper.title}\n")
f.write(f"Authors: {', '.join(author.name for author in paper.authors)}\n")
f.write(f"Published: {paper.published.strftime('%Y-%m-%d')}\n")
f.write(f"Primary Category: {paper.primary_category}\n")
f.write(f"PDF Link: {paper.pdf_url}\n\n")
f.write("=== ABSTRACT ===\n")
f.write(paper.summary.replace("\n", " ") + "\n")
# Optional: Download the actual PDF document
# paper.download_pdf(dirpath=OUTPUT_DIR, filename=f"{clean_id}.pdf")
print(f"\nAll files saved to {OUTPUT_DIR}/")
if __name__ == "__main__":
scrape_arxiv()
Method 2: Direct HTTP Web Scraping (HTML & BeautifulSoup)
If your objective requires scraping the live search results on arxiv.org/search directly rather than the API:
import os
import re
import time
import requests
from bs4 import BeautifulSoup
QUERY = "diffusion models"
NUM_PAGES = 3
PAGE_SIZE = 25 # arXiv search defaults to 25 or 50 per page
OUTPUT_DIR = "./arxiv_scraped_output"
os.makedirs(OUTPUT_DIR, exist_ok=True)
HEADERS = {"User-Agent": "ResearchScraper/1.0 (academic; contact@example.com)"}
def clean_name(val: str) -> str:
return re.sub(r'[\\/*?:"<>|]', "", val)
def scrape_arxiv_html():
collected_urls = []
# Step 1: Iterate over search pages
for page_idx in range(NUM_PAGES):
start = page_idx * PAGE_SIZE
search_url = f"https://arxiv.org/search/?query={QUERY.replace(' ', '+')}&searchtype=all&start={start}"
print(f"Scraping search page {page_idx + 1}...")
resp = requests.get(search_url, headers=HEADERS)
if resp.status_code != 200:
print(f"Failed to fetch page {page_idx + 1}: Status {resp.status_code}")
continue
soup = BeautifulSoup(resp.text, "html.parser")
# arXiv links live inside p.list-title > a
results = soup.select("p.list-title a[href*='/abs/']")
for a in results:
url = a.get("href")
arxiv_id = url.split("/abs/")[-1]
collected_urls.append((arxiv_id, url))
time.sleep(3) # Respect rate limits
print(f"\nFound {len(collected_urls)} papers. Extracting abstract pages...\n")
# Step 2: Open each paper's /abs/ page and save contents
for idx, (arxiv_id, abs_url) in enumerate(collected_urls, start=1):
print(f"[{idx}/{len(collected_urls)}] Fetching {arxiv_id}...")
try:
page_resp = requests.get(abs_url, headers=HEADERS)
page_soup = BeautifulSoup(page_resp.text, "html.parser")
# Title
title_el = page_soup.select_one("h1.title")
title = title_el.text.replace("Title:", "").strip() if title_el else "Unknown"
# Authors
authors_el = page_soup.select("div.authors a")
authors = [a.text.strip() for a in authors_el]
# Abstract
abstract_el = page_soup.select_one("blockquote.abstract")
abstract = abstract_el.text.replace("Abstract:", "").strip() if abstract_el else ""
# Save content to file
file_name = os.path.join(OUTPUT_DIR, f"{clean_name(arxiv_id)}.txt")
with open(file_name, "w", encoding="utf-8") as f:
f.write(f"arXiv ID: {arxiv_id}\n")
f.write(f"URL: {abs_url}\n")
f.write(f"Title: {title}\n")
f.write(f"Authors: {', '.join(authors)}\n\n")
f.write("=== ABSTRACT ===\n")
f.write(abstract + "\n")
except Exception as err:
print(f"Error fetching {arxiv_id}: {err}")
time.sleep(3)
if __name__ == "__main__":
scrape_arxiv_html()
Comparison: Google Patents vs. arXiv Scraping
| Feature | Google Patents Scraper | arXiv Scraper |
| JavaScript Rendering | Required (Needs Playwright/Selenium due to Custom Elements) | Not Required (Server-rendered HTML or pure XML/JSON API) |
| Recommended Tool | Playwright / Headless Browser | arxiv library / requests + BeautifulSoup |
| Rate Limit Rules | Strict, aggressive reCAPTCHA triggers | 1 request per 3 seconds per IP (strict 3s polite interval) |
| Full Content | Full text (Claims & Description) on web page | Abstract on web page; full content in PDF/LaTeX source |
Describe how to apply LLM analysis of a set of arXiv file downloads
Applying LLM analysis across a batch of downloaded arXiv papers requires a structured pipeline: parsing academic formatting (two-column layouts, formulas, citations), managing token budgets, and selecting the right prompting or retrieval strategy.
Step 1: Content Ingestion and Normalization
Raw academic PDFs break standard text parsers because two-column layouts cause text to read horizontally across columns rather than vertically down each column.
| File Type Downloaded | Recommended Tool | Why |
| PDF Documents | marker-pdf, Docling, or PyMuPDF | Converts multi-column layouts, tables, and mathematical formulas into clean, ordered Markdown. |
| Abstract / Text Metadata | pydantic or native json | Structured extraction if you only scraped titles, authors, categories, and abstracts. |
LaTeX Source (.tar.gz) | pylatexenc / arxiv-latex-cleaner | Bypasses PDF OCR entirely by directly reading raw LaTeX sections, math, and citations. |
Step 2: Choose the LLM Processing Pattern
Depending on your analytical goal, select one of three core design architectures:
[Corpus of N Papers] │ ├──► 1. Map-Reduce (Cross-Corpus Synthesis / Literature Review) │ └─ Summarize/Extract paper-by-paper → Synthesize all extracts into one report. │ ├──► 2. Structured Extraction (Dataset / Matrix Construction) │ └─ Run Pydantic schemas over each paper → Output clean JSON/CSV records. │ └──► 3. RAG / Vector Search (Interactive Corpus Q&A) └─ Chunk & Embed → Store in Vector DB → Query with hybrid search & reranking.
Structured Information Extraction (Paper-by-Paper):
Best for: Building comparison matrices (e.g., comparing model architecture, parameter count, dataset used, benchmark scores across 50 papers).
How: Pass each parsed paper (or its abstract + methodology + conclusion) into an LLM with structured output (
toolsor JSON schema via Pydantic).
Hierarchical Map-Reduce (Literature Review & Trend Synthesis):
Best for: Synthesizing themes, identifying recurring gaps, or tracing research evolution across the collection.
How:
Map: Prompt the LLM to extract primary contributions, limitations, and theoretical claims from each individual document.
Reduce: Feed all intermediate extractions together into a large-context model (e.g., 1M+ context window) to generate a cohesive survey report.
Hybrid RAG (Interactive Corpus Q&A):
Best for: Answering specific technical questions across the paper library (e.g., "Which papers modified the attention mask for sliding window contexts?").
How: Chunk the Markdown documents by section headers (
#,##), generate dense embeddings, index them in a vector database (e.g., Chroma, LanceDB), and apply an LLM reranker over retrieved chunks.
Step 3: End-to-End Python Implementation
The following script iterates over parsed papers, applies schema-constrained JSON extraction using Pydantic, and writes a normalized research comparative matrix to disk.
import os
import glob
from typing import List, Optional
from pydantic import BaseModel, Field
from openai import OpenAI
client = OpenAI()
# 1. Define the Structured Analysis Schema
class PaperAnalysis(BaseModel):
arxiv_id: str
primary_hypothesis: str = Field(description="The core problem the authors solve")
key_methodology: str = Field(description="Specific algorithms, architectures, or datasets introduced")
benchmarks_and_metrics: List[str] = Field(description="Datasets tested on and key performance claims")
stated_limitations: List[str] = Field(description="Shortcomings or unaddressed edge cases acknowledged")
novelty_rating: int = Field(ge=1, le=5, description="1 (incremental) to 5 (paradigm shift)")
# 2. Function to Process an Individual Paper
def analyze_paper_text(filename: str, content: str) -> Optional[PaperAnalysis]:
prompt = f"""
You are an expert peer reviewer. Analyze the following arXiv document and extract
the research specifics strictly adhering to the schema.
DOCUMENT TEXT:
{content[:30000]} # Truncate or use full text depending on context budget
"""
try:
completion = client.beta.chat.completions.parse(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Extract research claims objectively."},
{"role": "user", "content": prompt}
],
response_format=PaperAnalysis,
)
return completion.choices[0].message.parsed
except Exception as e:
print(f"Error analyzing {filename}: {e}")
return None
# 3. Batch Iterate Across the Downloaded Files
def run_corpus_analysis(input_dir: str):
file_paths = glob.glob(os.path.join(input_dir, "*.txt"))
analyses: List[PaperAnalysis] = []
print(f"Found {len(file_paths)} papers to process.")
for idx, path in enumerate(file_paths, start=1):
filename = os.path.basename(path)
print(f"[{idx}/{len(file_paths)}] Processing {filename}...")
with open(path, "r", encoding="utf-8") as f:
text = f.read()
record = analyze_paper_text(filename, text)
if record:
analyses.append(record)
# 4. Optional: Synthesize the Entire Batch
synthesize_findings(analyses)
def synthesize_findings(records: List[PaperAnalysis]):
summary_prompt = "Here are structured summaries of several recent papers:\n\n"
for r in records:
summary_prompt += f"ID: {r.arxiv_id}\nProblem: {r.primary_hypothesis}\nMethod: {r.key_methodology}\nLimitations: {', '.join(r.stated_limitations)}\n---\n"
summary_prompt += "\nSynthesize these into a cohesive research overview: what are the dominant trends, common failure modes, and open research directions?"
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": summary_prompt}],
)
with open("corpus_synthesis_report.md", "w", encoding="utf-8") as f:
f.write(response.choices[0].message.content)
print("\nSynthesis report written to corpus_synthesis_report.md")
if __name__ == "__main__":
run_corpus_analysis("./arxiv_output")
Critical Traps to Avoid
Table & Equation Hallucinations: Standard PDF-to-text extractors merge adjacent table cells into continuous lines. Never ask an LLM to perform arithmetic or direct metric comparison on text generated by basic tools like
pypdf;convert through marker-pdforDoclingto preserve table Markdown formatting first.Citation Leakage: Stripping the reference section (
\bibliographyor## References) before prompting reduces token consumption by 20–35% and prevents the LLM from confusing prior work cited in the bibliography with the authors' novel contributions.Token Budgeting: For sets larger than 100 papers, process abstracts and conclusions first for screening. Filter down to the high-relevance subset before submitting full-text sections to heavier frontier models.
How do I set up and run marker-pdf to convert a folder of academic PDFs into clean Markdown with intact formulas and tables?
marker-pdf uses a pipeline of specialized deep learning models (layout detection, reading order, Surya OCR/table recognition, and Texify for math) to convert multi-column academic PDFs into clean Markdown with LaTeX equations and structured tables.
Step 1: Environment and Hardware Prerequisites
marker relies on PyTorch and performs best with an NVIDIA GPU with at least 6–8 GB of VRAM. While it runs on CPU or Apple Silicon (MPS), processing will be significantly slower.
Ensure you have Python 3.10+ installed in a dedicated virtual environment:
# Create and activate a clean environment
python -m venv marker-env
source marker-env/bin/activate # On Windows: marker-env\Scripts\activate
# Install PyTorch with GPU support first (example for CUDA 12.1/12.4)
pip install torch torchvision --index-url https://download.pytorch.org/whl/cu121
# Install marker-pdf
pip install marker-pdf
Step 2: Running via Command Line (Fastest Method)
marker provides a built-in CLI tool designed specifically for batch processing entire directories of PDFs: marker_chunk_convert.
marker_chunk_convert ./input_pdfs ./output_markdown --num_chunks 1 --max 100
Key CLI Flags:
./input_pdfs: Directory containing your downloaded.pdffiles../output_markdown: Directory where converted files and extracted images/tables will be stored.--num_chunks: Set to the number of available GPUs (e.g.,1for a single GPU). If running on multiple GPUs, increasing this enables parallel chunked execution.--workers: Number of worker processes to spawn per GPU (defaults to1or2depending on VRAM).--max: Maximum number of PDFs to process from the folder (optional; omit to convert all).
Each converted paper will create a subfolder inside ./output_markdown containing:
paper_name.md(clean Markdown with$$math and Markdown tables)paper_name_meta.json(detected languages, layout metadata, block counts)Extracted figures and rasterized charts saved as image files.
Step 3: Running via Python API (Custom Filtering & Integration)
If you want to integrate conversion directly into a data pipeline, strip unwanted sections (such as bibliographies), or catch errors per document:
import os
import glob
from marker.convert import convert_single_pdf
from marker.models import load_all_models
from marker.output import save_markdown
INPUT_DIR = "./arxiv_pdfs"
OUTPUT_DIR = "./arxiv_markdown"
os.makedirs(OUTPUT_DIR, exist_ok=True)
# 1. Load models once into memory/VRAM
print("Loading marker models...")
model_lst = load_all_models()
# 2. Iterate through all PDFs in the target folder
pdf_files = glob.glob(os.path.join(INPUT_DIR, "*.pdf"))
print(f"Found {len(pdf_files)} PDF(s) to process.\n")
for idx, pdf_path in enumerate(pdf_files, start=1):
base_name = os.path.splitext(os.path.basename(pdf_path))[0]
subfolder_out = os.path.join(OUTPUT_DIR, base_name)
print(f"[{idx}/{len(pdf_files)}] Converting {base_name}...")
try:
# Full text extraction + layout + OCR + math rendering
full_text, images, out_meta = convert_single_pdf(
pdf_path,
model_lst,
batch_multiplier=2 # Increase if you have 16GB+ VRAM for faster inference
)
# Save Markdown and extracted diagrams/figures
save_markdown(subfolder_out, base_name, full_text, images, out_meta)
print(f"Saved to {subfolder_out}/")
except Exception as e:
print(f"Failed to convert {base_name}: {e}")
print("\nBatch conversion complete.")
Key Configuration Settings & Optimization
Controlling VRAM Usage: If you encounter
CUDA Out of Memoryerrors on complex papers with many figures, setbatch_multiplier=1(or lower batch size in the CLI) or force CPU execution by setting the environment variableTORCH_DEVICE=cpu.Disabling Image Extraction: If you only need text, tables, and math equations without extracting diagrams, you can save disk space and I/O overhead by ignoring the
imagesdictionary returned byconvert_single_pdf.LaTeX Math Blocks: Inline math will automatically appear as
$inline$, and display math blocks as$$display$$.
How do I chunk converted Markdown academic papers by section headings like Abstract, Methodology, and Results for vector retrieval?
Chunking converted Markdown academic papers by section headings prevents cross-topic contamination (e.g., mixing experimental setup with introductory remarks) and attaches explicit semantic provenance to each vector embedding.
A production-grade pipeline requires two-tier hierarchical chunking:
Structural Splitting: Split by Markdown heading hierarchy (
#,##,###) to isolate sections like Abstract, Methodology, and Results.Recursive Sub-Chunking: If an extracted section exceeds your embedding model's context window (e.g., 512 or 1,024 tokens), recursively divide it by paragraphs and sentences while keeping section metadata intact.
Step 1: Install Required Dependencies
Using LangChain's dedicated text splitters is the cleanest approach, as MarkdownHeaderTextSplitter automatically propagates heading paths down into metadata.
pip install langchain-text-splitters tiktoken
Step 2: Two-Tier Section Chunking Implementation
import os
import re
from typing import List
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langchain_core.documents import Document
def clean_academic_markdown(md_text: str) -> str:
"""
Strips the references/bibliography section and normalizes
unnumbered or stylised headings commonly produced by Marker/Docling.
"""
# 1. Strip bibliography/references to save embedding tokens and avoid false matches
ref_pattern = r"(?i)(^#+\s+(?:References|Bibliography|Works\s+Cited)\b.*)"
parts = re.split(ref_pattern, md_text, flags=re.MULTILINE)
cleaned_text = parts[0] if parts else md_text
# 2. Normalize numbered headings like '## 3. Methodology' to '## Methodology'
cleaned_text = re.sub(
r"^(#{1,4})\s+(?:\d+\.?\d*\.?\s+)?",
r"\1 ",
cleaned_text,
flags=re.MULTILINE
)
return cleaned_text
def chunk_paper_by_sections(
file_path: str,
max_chunk_size: int = 1000,
chunk_overlap: int = 150
) -> List[Document]:
"""
Performs hierarchical two-tier chunking:
Tier 1: Markdown header-aware splitting (preserves section scope)
Tier 2: Character/token fallback splitting for oversized sections
"""
with open(file_path, "r", encoding="utf-8") as f:
raw_markdown = f.read()
cleaned_md = clean_academic_markdown(raw_markdown)
paper_id = os.path.splitext(os.path.basename(file_path))[0]
# Tier 1: Split on Markdown Headings
headers_to_split_on = [
("#", "title"),
("##", "section"),
("###", "subsection"),
]
markdown_splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers_to_split_on,
strip_headers=False # Keep the heading text in the chunk for embedding context
)
section_docs = markdown_splitter.split_text(cleaned_md)
# Tier 2: Sub-split large sections (e.g. extensive Experiments/Methods)
recursive_splitter = RecursiveCharacterTextSplitter(
chunk_size=max_chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", r"(?<=\.)\s+", " ", ""],
is_separator_regex=True
)
final_chunks: List[Document] = []
for doc in section_docs:
# Merge file metadata with header metadata
doc.metadata["paper_id"] = paper_id
# Canonicalize section name (defaults to "Abstract/Introduction" if before first header)
section_name = doc.metadata.get("section", doc.metadata.get("title", "Overview/Abstract"))
doc.metadata["section"] = section_name
# If section is small enough, keep as a single chunk
if len(doc.page_content) <= max_chunk_size:
final_chunks.append(doc)
else:
# Sub-divide large section while inheriting metadata
sub_chunks = recursive_splitter.split_documents([doc])
for i, sub_chunk in enumerate(sub_chunks):
sub_chunk.metadata["sub_chunk_index"] = i
final_chunks.append(sub_chunk)
return final_chunks
Step 3: Example Output & Ingestion into a Vector Store
When running this against a parsed paper:
chunks = chunk_paper_by_sections("./arxiv_markdown/2401.0001.md")
for chunk in chunks[:3]:
print("Metadata:", chunk.metadata)
print("Content preview:", chunk.page_content[:120].strip())
print("-" * 50)
Output:
Metadata: {'title': 'FlashAttention-3', 'section': 'Abstract', 'paper_id': '2401.0001'} Content preview: ## Abstract We present FlashAttention-3, an algorithm that speeds up attention computation on Hopper GPUs by 1.5-2.0x... -------------------------------------------------- Metadata: {'title': 'FlashAttention-3', 'section': 'Methodology', 'subsection': 'Warp Specialization', 'paper_id': '2401.0001', 'sub_chunk_index': 0} Content preview: ### Warp Specialization To overlap asynchronous tensor copy instructions with asynchronous matrix multiply instructions, we partition... --------------------------------------------------
Step 4: Leveraging Section Metadata in Retrieval
Attaching headings into metadata allows you to perform Metadata-Filtered Vector Retrieval in vector databases like Chroma, LanceDB, or Qdrant:
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
vectorstore = Chroma.from_documents(chunks, embedding=OpenAIEmbeddings())
# Query ONLY the Methodology sections of your corpus
method_results = vectorstore.similarity_search(
query="learning rate decay and optimizer settings",
k=5,
filter={"section": "Methodology"}
)
| Best Practice | Reason |
Keep strip_headers=False | Embedding models evaluate semantic intent better when the chunk begins with ## Methodology rather than raw floating body text. |
| Prune Bibliographies Early | Reference lists trigger high false-positive hits whenever a query mentions author names or well-known model names. |
| Preserve LaTeX Math Blocks | Do not split inside $$...$$ blocks; ensuring \n\n is the top-level separator prevents formula syntax truncation. |
How can I prevent table truncation so Markdown tables remain intact as single chunks during recursive splitting?
Standard character-based and recursive splitters treat lines indiscriminately. Because table rows often contain newlines (\n), a blind recursive splitter will sever a Markdown table mid-row or drop the header row from subsequent rows, destroying column alignments and causing hallucinations during RAG generation.
To guarantee tables remain intact, use an atomic extraction and placeholder substitution pattern:
Identify and extract full Markdown tables using regular expressions.
Replace each table with a unique sentinel token (e.g.,
<<TABLE_PLACEHOLDER_0>>).Run standard recursive text splitting on the surrounding prose.
Swap the tables back in:
If the table fits alongside surrounding text, inject it directly.
If the table is oversized, isolate it as its own dedicated
Documentchunk tagged withcontent_type="table".
Implementation: Table-Aware Splitter
import re
from typing import List, Tuple
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
# Regex to capture GFM Markdown tables (header, delimiter row, and data rows)
MARKDOWN_TABLE_REGEX = re.compile(
r"(?:^[ \t]*\|[^\n]+\|[ \t]*\n[ \t]*\|(?::?[-]+:?\|)+[ \t]*\n(?:[ \t]*\|[^\n]+\|[ \t]*(?:\n|$))+)",
re.MULTILINE
)
def extract_tables(text: str) -> Tuple[str, List[str]]:
"""
Finds all Markdown tables, stores them in an indexed list,
and replaces them in the original text with unique placeholder tokens.
"""
tables = []
def replacer(match):
table_str = match.group(0).strip()
table_idx = len(tables)
tables.append(table_str)
return f"\n\n<<TABLE_PLACEHOLDER_{table_idx}>>\n\n"
masked_text = MARKDOWN_TABLE_REGEX.sub(replacer, text)
return masked_text, tables
def chunk_with_atomic_tables(
doc: Document,
max_chunk_size: int = 1000,
chunk_overlap: int = 150
) -> List[Document]:
"""
Splits text recursively without breaking Markdown tables.
Tables are either re-inserted into prose chunks or isolated as dedicated table chunks.
"""
masked_text, tables = extract_tables(doc.page_content)
# Standard recursive splitter for prose surrounding the tables
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=max_chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", r"(?<=\.)\s+", " ", ""],
is_separator_regex=True
)
# Initial prose splits containing placeholders
raw_chunks = text_splitter.split_text(masked_text)
final_documents: List[Document] = []
placeholder_pattern = re.compile(r"<<TABLE_PLACEHOLDER_(\d+)>>")
for chunk_idx, raw_chunk in enumerate(raw_chunks):
placeholders_found = placeholder_pattern.findall(raw_chunk)
if not placeholders_found:
# Standard text chunk without tables
new_metadata = {**doc.metadata, "content_type": "text"}
final_documents.append(Document(page_content=raw_chunk.strip(), metadata=new_metadata))
continue
resolved_chunk = raw_chunk
for table_idx_str in placeholders_found:
table_idx = int(table_idx_str)
table_content = tables[table_idx]
placeholder_tag = f"<<TABLE_PLACEHOLDER_{table_idx}>>"
# Check if injecting the table keeps the chunk within budget
projected_len = len(resolved_chunk) - len(placeholder_tag) + len(table_content)
if projected_len <= max_chunk_size:
resolved_chunk = resolved_chunk.replace(placeholder_tag, table_content)
else:
# Remove placeholder from prose chunk
resolved_chunk = resolved_chunk.replace(placeholder_tag, "").strip()
# Emit the table as an independent, whole atomic chunk
table_metadata = {
**doc.metadata,
"content_type": "table",
"table_index": table_idx
}
final_documents.append(
Document(page_content=table_content, metadata=table_metadata)
)
if resolved_chunk.strip():
prose_metadata = {**doc.metadata, "content_type": "text"}
final_documents.append(
Document(page_content=resolved_chunk.strip(), metadata=prose_metadata)
)
return final_documents
Handling Ultra-Large Multi-Page Tables
If a table is larger than the embedding model's context window (e.g., a benchmark table spanning dozens of rows), it cannot remain a single chunk.
When splitting an oversized table is unavoidable:
Preserve the Schema Across All Splits: Split row-by-row, but prepend the markdown header and separator line (
| Header 1 | Header 2 |\n|---|---|) to every subsequent chunk.Flatten to Key-Value Row Objects: Convert large tables into serialized Markdown or JSON records:
Row: Model=FlashAttention-3 | Speedup=1.7x | Memory=42GB | Baseline=Flash2 Row: Model=PagedAttention | Speedup=1.3x | Memory=35GB | Baseline=vLLMThis format embeds denser semantic signals than disconnected cell pipes and retains complete contextual grounding on every row.
From <https://gemini.google.com/app/45035ea1f4252b34> Google Gemini (3.8 Flash)
Comments
Post a Comment