Document Ingestion & Parsing: Getting Knowledge Into Your Pipeline

Answer:

RAG systems ingest documents from multiple source types, each requiring a different extraction strategy:

The extraction method must match the source: a web scraper on a PDF returns nothing; a PDF parser on a JS-rendered site misses all dynamic content.

# PDF extraction example
import pymupdf  # or: import fitz (legacy alias, same package)

def extract_pdf_text(path: str) -> list[dict]:
    doc = pymupdf.open(path)
    pages = []
    for page_num, page in enumerate(doc):
        pages.append({
            "text": page.get_text(),
            "metadata": {
                "source": path,
                "page": page_num + 1,
                "total_pages": len(doc)
            }
        })
    return pages

Production note: Always validate that text was actually extracted (len > 0) before proceeding. A blank extraction means the PDF is scanned — route it to the OCR branch.

Answer:

The three libraries target different priorities:

Decision rule:

Does the PDF have tables you must extract as structured data?
  └─ Yes → pdfplumber

Is throughput critical (>10k docs/hour)?
  └─ Yes → pymupdf

Need pure Python (no C deps, e.g., restricted environments)?
  └─ Yes → pypdf

Is the PDF scanned (no text layer)?
  └─ Yes → None of the above; use OCR (pytesseract, easyocr, or paddleocr)

PyMuPDF import note: Use import pymupdf (version ≥ 1.24.0) or import fitz (legacy alias). Both refer to the same installed package.

# Fastest path: PyMuPDF
import pymupdf

doc = pymupdf.open("report.pdf")
text = "\n".join(page.get_text() for page in doc)

# Table path: pdfplumber
import pdfplumber

with pdfplumber.open("report.pdf") as pdf:
    for page in pdf.pages:
        tables = page.extract_tables()  # returns list[list[list[str]]]

Answer:

Every downstream component — embeddings, vector search, reranking, LLM generation — operates on the text produced by ingestion. If that text is corrupted, the LLM has no way to recover it.

Bad ingestion produces:
  ├─ Corrupt text ────────► Meaningless embeddings → wrong retrieval
  ├─ Missing metadata ────► No pre-filtering → low precision
  ├─ Duplicate content ───► Biased retrieval → same answer retrieved 3x
  └─ Lost table data ─────► LLM answers from missing information

Concrete failure examples:

The principle is simple: garbage in, garbage out. No retrieval or generation optimization compensates for corrupted text.

Answer:

Cleaning transforms raw extracted text into structured, usable input. Apply these in order:

Step 1 — Fix encoding issues

import ftfy

text = ftfy.fix_text(raw_text)
# Converts: "’" → "'"
# Fixes mojibake, surrogate pairs, wrong codecs

Step 2 — Strip HTML/XML tags

from bs4 import BeautifulSoup

text = BeautifulSoup(text, "html.parser").get_text(separator=" ")

Step 3 — Remove boilerplate (headers, footers, watermarks)

import re

# Remove lines that are short repeated patterns (headers/footers)
lines = text.split("\n")
text = "\n".join(line for line in lines if len(line.strip()) > 20)

Step 4 — Normalize whitespace

text = re.sub(r'\s+', ' ', text).strip()

Step 5 — Remove PII (if required)

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

results = analyzer.analyze(text=text, language="en")
text = anonymizer.anonymize(text=text, analyzer_results=results).text
# Replaces: "John Smith, SSN 123-45-6789" → "<PERSON>, SSN <US_SSN>"

Step 6 — Deduplication (document level, before adding to index)

import hashlib

def content_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

Critical: Do NOT discard metadata (source, page, section) during cleaning — these fields enable precision pre-filtering at retrieval time.

Answer:

Metadata unlocks pre-filtering — narrowing the search space before vector similarity runs. This improves precision and reduces hallucination.

Minimum viable metadata schema:

{
    "source": "Q4-2024-annual-report.pdf",
    "page": 12,
    "section": "Revenue Analysis",
    "doc_type": "financial_report",
    "department": "finance",
    "date": "2024-12-01",
    "language": "en"
}

How metadata enables precision retrieval:

User query: "2024 finance report revenue"

Without metadata → vector search across all 500k chunks
  └─ Returns: legal, HR, tech, and finance docs mixed

With metadata pre-filter:
  doc_type = "financial_report" AND date >= "2024-01-01"
  └─ Narrows to 8,000 finance chunks → much higher precision

Where to extract metadata:

• source: full file path or URL
• page: from the parser (PDF page number, HTML URL anchor)
• section: from document structure (headings, chapter titles)
• doc_type: classify at ingestion time (regex pattern or classifier)
• date: from file metadata or document headers

Performance note: Use pre-filtering (apply metadata filter before vector search), not post-filtering, to guarantee the top-K result pool is not depleted before ranking.

Answer:

Detection pattern:

def validate_extracted_text(text: str, source: str) -> bool:
    if len(text.strip()) < 50:
        print(f"WARNING: Near-empty extraction from {source}")
        return False
    # Check for common mojibake signatures
    if "â€" in text or "é" in text:
        print(f"WARNING: Possible encoding issue in {source}")
    return True

Answer:

Multilingual ingestion requires decisions at three levels:

1. Encoding normalization (always do this first)

import ftfy
import unicodedata

def normalize_text(text: str) -> str:
    text = ftfy.fix_text(text)
    # NFC normalization: canonical composition (é as single char, not e + combining accent)
    text = unicodedata.normalize("NFC", text)
    return text

2. Language detection

from langdetect import detect, LangDetectException

def get_language(text: str) -> str:
    try:
        return detect(text)  # returns ISO 639-1 code: "en", "fr", "de", etc.
    except LangDetectException:
        return "unknown"

langdetect supports 55 languages. For higher accuracy at the cost of more memory, lingua-py supports 75 languages with better confidence scoring.

3. Embedding model selection

Language affects which embedding model to use:

Key warning: An English-only embedding model will degrade silently on French or German documents — cosine similarity scores drop, retrieval recall falls, and no error is thrown.

Metadata to store: Add "language": lang_code to each document's metadata. This enables per-language pre-filtering and model routing at query time.

Answer:

A production OCR pipeline adds preprocessing, engine selection, confidence filtering, and fallback logic:

Scanned PDF Page
  │
  ▼
Convert to image (if PDF)
  │  pdf2image.convert_from_path()
  ▼
Image preprocessing
  ├─ Deskew (cv2.warpAffine on detected rotation angle)
  ├─ Binarize (cv2.threshold with OTSU method)
  └─ Denoise (cv2.fastNlMeansDenoising)
  │
  ▼
OCR engine
  ├─ pytesseract → get_data() returns per-word confidence scores
  ├─ easyocr    → readtext() returns (bbox, text, confidence) tuples
  └─ paddleocr  → best for structured layouts; returns bounding boxes
  │
  ▼
Confidence filtering
  └─ If mean confidence < threshold (e.g., 60 for tesseract, 0.6 for easyocr):
       → Flag page for human review or discard
  │
  ▼
Post-OCR text cleaning (ftfy, whitespace normalization)

Pytesseract confidence filtering:

import pytesseract
from PIL import Image

def ocr_with_confidence(image_path: str, min_conf: int = 60) -> str:
    img = Image.open(image_path)
    data = pytesseract.image_to_data(img, output_type=pytesseract.Output.DICT)

    words = [
        data["text"][i]
        for i in range(len(data["text"]))
        if int(data["conf"][i]) >= min_conf and data["text"][i].strip()
    ]
    return " ".join(words)

EasyOCR example:

import easyocr

reader = easyocr.Reader(["en"], gpu=False)

def ocr_easyocr(image_path: str, min_conf: float = 0.6) -> str:
    results = reader.readtext(image_path)
    # Each result: (bbox, text, confidence)
    return " ".join(text for _, text, conf in results if conf >= min_conf)

Engine selection guide:

• pytesseract: best for clean, high-resolution typed text; free, no GPU needed
• easyocr: balanced accuracy + speed; 80+ languages; works on GPU or CPU
• paddleocr: best for structured layouts (tables, forms); superior bounding-box accuracy for multilingual docs

Answer:

Tables are among the hardest PDF elements to extract — most parsers read cells left-to-right, destroying structure. The two reliable approaches are:

Approach 1: pdfplumber (best for digital PDFs with text layer)

import pdfplumber

def extract_tables_as_markdown(pdf_path: str) -> list[dict]:
    results = []
    with pdfplumber.open(pdf_path) as pdf:
        for page_num, page in enumerate(pdf.pages):
            tables = page.extract_tables()  # returns list[list[list[str | None]]]
            for table in tables:
                if not table:
                    continue
                md = table_to_markdown(table)
                results.append({
                    "text": md,
                    "metadata": {"source": pdf_path, "page": page_num + 1, "type": "table"}
                })
    return results

def table_to_markdown(table: list) -> str:
    if not table or not table[0]:
        return ""
    header = "| " + " | ".join(str(c or "") for c in table[0]) + " |"
    separator = "| " + " | ".join("---" for _ in table[0]) + " |"
    rows = [
        "| " + " | ".join(str(c or "") for c in row) + " |"
        for row in table[1:]
    ]
    return "\n".join([header, separator] + rows)

Approach 2: camelot-py (for complex lattice or stream tables)

pip install camelot-py[cv]

camelot uses two modes: lattice (tables with visible borders) and stream (borderless, whitespace-separated). It returns tables as pandas DataFrames.

Why Markdown output?

LLMs understand Markdown tables better than raw CSV or space-delimited text. Converting tables to Markdown before embedding preserves column headers in the vector representation and makes the LLM's task easier during generation.

Key consideration: Chunk tables as a single unit — never split a table across chunks. A table split mid-row loses its meaning entirely.

Answer:

Full re-embedding on every update is prohibitively expensive at scale. The standard pattern uses content hashing for change detection:

import hashlib
from datetime import datetime

def compute_content_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

class IncrementalIngester:
    def __init__(self, fingerprint_store: dict, vector_db):
        # fingerprint_store: {doc_id: content_hash}
        self.fingerprints = fingerprint_store
        self.vector_db = vector_db

    def ingest(self, doc_id: str, text: str, metadata: dict):
        new_hash = compute_content_hash(text)
        existing_hash = self.fingerprints.get(doc_id)

        if existing_hash == new_hash:
            return  # Unchanged — skip re-embedding

        if existing_hash:
            # Updated — delete old vectors, insert new
            self.vector_db.delete(filter={"doc_id": doc_id})

        # New or updated document — embed and store
        chunks = self.chunk(text)
        self.vector_db.upsert(chunks, metadata=metadata)
        self.fingerprints[doc_id] = new_hash

    def delete(self, doc_id: str):
        self.vector_db.delete(filter={"doc_id": doc_id})
        self.fingerprints.pop(doc_id, None)

Handling deletions — two approaches:

Async ingestion at scale:

For high-volume pipelines, push documents into a queue (e.g., Kafka, SQS) and process with worker threads. This decouples ingestion latency from the upstream source and allows retries on failure.

See also: 04-stale_index_problem.md for a full treatment of index staleness patterns.

Answer:

Duplicate content in the index causes retrieval to return the same information multiple times, wasting context window and skewing relevance scores. Three levels of deduplication:

Level 1: Exact deduplication (cheapest)

import hashlib

seen_hashes: set[str] = set()

def is_duplicate(text: str) -> bool:
    h = hashlib.sha256(text.encode("utf-8")).hexdigest()
    if h in seen_hashes:
        return True
    seen_hashes.add(h)
    return False

Level 2: Near-duplicate detection (MinHash + LSH)

Exact hashing misses documents that are 95% identical (e.g., v1 and v2 of a policy with minor edits). Use datasketch for probabilistic near-duplicate detection:

from datasketch import MinHash, MinHashLSH

# Build LSH index during ingestion
lsh = MinHashLSH(threshold=0.85, num_perm=128)

def get_minhash(text: str, num_perm: int = 128) -> MinHash:
    m = MinHash(num_perm=num_perm)
    for shingle in {text[i:i+5] for i in range(len(text) - 4)}:
        m.update(shingle.encode("utf-8"))
    return m

def is_near_duplicate(doc_id: str, text: str) -> bool:
    m = get_minhash(text)
    result = lsh.query(m)  # returns doc_ids with Jaccard similarity >= threshold
    if result:
        return True
    lsh.insert(doc_id, m)
    return False

datasketch requires Python ≥ 3.9 and NumPy ≥ 1.11.

Level 3: Semantic deduplication (most expensive)

After embedding, compute cosine similarity between new document's embedding and existing embeddings. Discard if similarity exceeds threshold (e.g., 0.97). Use only for small corpora or as a final pass — it requires an embedding call per document.

When to use each:

Exact hash        → Always; zero cost; catches identical re-uploads
MinHash LSH       → When corpus has many document versions / copies
Semantic          → When documents are paraphrases of each other (e.g., translated duplicates)

Answer:

Production ingestion pipelines must isolate failures so a single bad document doesn't block the entire batch.

Error taxonomy:

Idempotent pipeline design:

def ingest_document(doc_id: str, path: str) -> None:
    try:
        text = extract_text(path)          # may raise ParseError
        if not text.strip():
            text = ocr_fallback(path)      # scanned PDF fallback
        text = clean_text(text)
        chunks = chunk_text(text)
        embed_and_store(doc_id, chunks)    # may raise EmbeddingError
    except ParseError as e:
        send_to_dead_letter(doc_id, path, reason=str(e))
    except EmbeddingError as e:
        retry_queue.push(doc_id, path, attempt=1)
    except Exception as e:
        send_to_dead_letter(doc_id, path, reason=f"unexpected: {e}")

Dead-letter queue pattern:

• Store failed documents with: doc_id, source_path, failure_reason, timestamp, attempt_count
• Alert on queue depth > threshold (e.g., >1% of daily volume)
• Reprocess after root cause is fixed (make processing idempotent so re-runs are safe)

Key principle: A partial failure should never silently corrupt the index. Either the document is ingested correctly, or it goes to the dead-letter queue with a logged reason.