RAG Document Indexing: PDF, DOCX, Excel Parsing for Production Pipelines

TL;DR — RAG document indexing parses PDF, DOCX, and Excel into RAG-ready chunks with metadata, deduplication, and table preservation. Unstructured-IO: "Open-source ETL for transforming complex documents into clean, structured formats for LLMs. Auto-detect file type with partition function." RAGWire: "Ingest PDFs, DOCX, XLSX, PPTX with LLM metadata extraction, SHA256 dedup, hybrid search in Qdrant." DistillCore: "7-stage pipeline: extract, classify, structure, chunk, enrich, embed, validate. Coverage checks: structuring 95%, chunking 98%, end-to-end 93%." RAGParser: "Universal parsing: PDF, DOCX, PPTX, XLSX, HTML, MD, CSV, JSON, images. Intelligent chunking: Fixed, Semantic, Adaptive." Dataloop: "Type-safe pipeline: extract → clean → chunk → upload. PyMuPDF Layout, OCR fallback, table and image extraction." Learn more with chunking strategies, chunk size, what is RAG, and build from scratch.

Unstructured-IO provides the foundation: "Unstructured is open-source ETL solution for transforming complex documents into clean, structured formats for language models. The unstructured library provides components for ingesting and pre-processing images and text documents such as PDFs, HTML, Word docs, and many more."

DistillCore describes the production pipeline: "Takes any document (PDF, DOCX, HTML, text, markdown) and runs it through an intelligent 7-stage pipeline — extracting text, classifying the document, breaking it into structured sections, chunking for RAG, enriching chunks with LLM-generated metadata, generating embeddings, and validating coverage at every stage."

Document Indexing Architecture

Document Format Support Comparison

7-Stage Pipeline Coverage Targets

Implementation

import hashlib
import os
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

class DocumentFormat(Enum):
    PDF = "pdf"
    DOCX = "docx"
    XLSX = "xlsx"
    PPTX = "pptx"
    HTML = "html"
    MARKDOWN = "md"
    TEXT = "txt"
    CSV = "csv"
    JSON = "json"
    IMAGE = "image"

@dataclass
class DocumentIndexer:
    """Index documents for RAG: extract, classify, chunk, enrich, embed."""

    def __init__(self, embedder=None, llm=None,
                 vector_store=None,
                 chunk_size: int = 512,
                 chunk_overlap: int = 50,
                 enable_dedup: bool = True,
                 enable_enrichment: bool = True):
        self.embedder = embedder
        self.llm = llm
        self.vector_store = vector_store
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.enable_dedup = enable_dedup
        self.enable_enrichment = enable_enrichment
        self._dedup_hashes: set = set()

    async def index_document(self, file_path: str,
                             metadata: dict = None) -> dict:
        """Index a single document through the 7-stage pipeline."""
        # Stage 0: Dedup check
        file_hash = self._compute_file_hash(file_path)
        if self.enable_dedup and file_hash in self._dedup_hashes:
            return {
                "status": "skipped",
                "reason": "duplicate",
                "file_hash": file_hash,
            }

        # Stage 1: Extract text
        doc_format = self._detect_format(file_path)
        extracted = await self._extract_text(file_path, doc_format)

        # Stage 2: Classify
        classification = await self._classify_document(
            extracted, file_path)

        # Stage 3: Structure into sections
        sections = self._structure_sections(extracted, doc_format)

        # Stage 4: Chunk
        chunks = self._chunk_sections(sections)

        # Stage 5: Enrich with metadata
        if self.enable_enrichment and self.llm:
            chunks = await self._enrich_chunks(chunks)

        # Stage 6: Embed
        if self.embedder:
            for chunk in chunks:
                chunk["embedding"] = await self.embedder.embed(
                    chunk["content"])

        # Stage 7: Store
        if self.vector_store:
            for chunk in chunks:
                chunk["metadata"].update({
                    "file_name": os.path.basename(file_path),
                    "file_hash": file_hash,
                    "file_type": doc_format.value,
                    **classification,
                    **(metadata or {}),
                })
            await self.vector_store.upsert(chunks)

        # Update dedup
        if self.enable_dedup:
            self._dedup_hashes.add(file_hash)
            for chunk in chunks:
                chunk_hash = self._compute_chunk_hash(
                    chunk["content"])
                self._dedup_hashes.add(chunk_hash)

        return {
            "status": "indexed",
            "file_path": file_path,
            "file_hash": file_hash,
            "format": doc_format.value,
            "classification": classification,
            "sections": len(sections),
            "chunks": len(chunks),
            "chunk_hashes": [
                self._compute_chunk_hash(c["content"])
                for c in chunks
            ],
        }

    async def index_directory(self, dir_path: str,
                              recursive: bool = True) -> list:
        """Index all documents in a directory."""
        results = []
        files = self._list_files(dir_path, recursive)
        for file_path in files:
            result = await self.index_document(file_path)
            results.append(result)
        return results

    def _compute_file_hash(self, file_path: str) -> str:
        """Compute SHA256 hash of file content."""
        h = hashlib.sha256()
        with open(file_path, 'rb') as f:
            for chunk in iter(lambda: f.read(8192), b''):
                h.update(chunk)
        return h.hexdigest()

    def _compute_chunk_hash(self, content: str) -> str:
        """Compute SHA256 hash of chunk content."""
        return hashlib.sha256(content.encode()).hexdigest()

    def _detect_format(self, file_path: str) -> DocumentFormat:
        """Detect document format from file extension."""
        ext = os.path.splitext(file_path)[1].lower().lstrip('.')
        format_map = {
            'pdf': DocumentFormat.PDF,
            'docx': DocumentFormat.DOCX,
            'xlsx': DocumentFormat.XLSX,
            'pptx': DocumentFormat.PPTX,
            'html': DocumentFormat.HTML,
            'htm': DocumentFormat.HTML,
            'md': DocumentFormat.MARKDOWN,
            'markdown': DocumentFormat.MARKDOWN,
            'txt': DocumentFormat.TEXT,
            'csv': DocumentFormat.CSV,
            'json': DocumentFormat.JSON,
            'png': DocumentFormat.IMAGE,
            'jpg': DocumentFormat.IMAGE,
            'jpeg': DocumentFormat.IMAGE,
        }
        return format_map.get(ext, DocumentFormat.TEXT)

    async def _extract_text(self, file_path: str,
                            doc_format: DocumentFormat) -> dict:
        """Extract text from document using format-specific parser."""
        # In production:
        # PDF: PyMuPDF (fitz) or pdfplumber
        # DOCX: python-docx
        # XLSX: openpyxl
        # HTML: beautifulsoup4
        # Images: pytesseract (OCR)
        # Universal: unstructured.io partition()

        extractors = {
            DocumentFormat.PDF: self._extract_pdf,
            DocumentFormat.DOCX: self._extract_docx,
            DocumentFormat.XLSX: self._extract_xlsx,
            DocumentFormat.HTML: self._extract_html,
            DocumentFormat.MARKDOWN: self._extract_text_file,
            DocumentFormat.TEXT: self._extract_text_file,
            DocumentFormat.CSV: self._extract_csv,
            DocumentFormat.IMAGE: self._extract_ocr,
        }

        extractor = extractors.get(
            doc_format, self._extract_text_file)
        return await extractor(file_path)

    async def _extract_pdf(self, file_path: str) -> dict:
        """Extract text from PDF with PyMuPDF."""
        # In production: import fitz; doc = fitz.open(file_path)
        return {
            "text": f"[PDF content from {file_path}]",
            "pages": [],
            "tables": [],
            "images": [],
            "metadata": {"page_count": 0},
        }

    async def _extract_docx(self, file_path: str) -> dict:
        """Extract text from DOCX with python-docx."""
        # In production: from docx import Document
        return {
            "text": f"[DOCX content from {file_path}]",
            "paragraphs": [],
            "tables": [],
            "metadata": {"paragraph_count": 0},
        }

    async def _extract_xlsx(self, file_path: str) -> dict:
        """Extract data from Excel with openpyxl."""
        # In production: from openpyxl import load_workbook
        return {
            "text": f"[XLSX content from {file_path}]",
            "sheets": [],
            "tables": [],
            "metadata": {"sheet_count": 0},
        }

    async def _extract_html(self, file_path: str) -> dict:
        """Extract text from HTML with beautifulsoup4."""
        # In production: from bs4 import BeautifulSoup
        return {
            "text": f"[HTML content from {file_path}]",
            "links": [],
            "tables": [],
            "metadata": {},
        }

    async def _extract_text_file(self, file_path: str) -> dict:
        """Extract plain text."""
        with open(file_path, 'r', encoding='utf-8') as f:
            text = f.read()
        return {"text": text, "metadata": {}}

    async def _extract_csv(self, file_path: str) -> dict:
        """Extract structured data from CSV."""
        import csv
        rows = []
        with open(file_path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            for row in reader:
                rows.append(row)
        text = "\n".join(
            str(row) for row in rows[:100])
        return {
            "text": text,
            "rows": rows,
            "metadata": {"row_count": len(rows)},
        }

    async def _extract_ocr(self, file_path: str) -> dict:
        """Extract text from image with OCR."""
        # In production: import pytesseract
        return {
            "text": f"[OCR content from {file_path}]",
            "metadata": {"ocr": True},
        }

    async def _classify_document(self, extracted: dict,
                                 file_path: str) -> dict:
        """Classify document type using LLM."""
        # In production: use LLM to classify
        return {
            "document_type": "unknown",
            "title": os.path.basename(file_path),
            "domain": "general",
        }

    def _structure_sections(self, extracted: dict,
                            doc_format: DocumentFormat) -> list:
        """Break document into hierarchical sections."""
        text = extracted.get("text", "")
        # Simple: split by double newlines as sections
        sections = text.split("\n\n")
        return [
            {
                "content": s.strip(),
                "section_index": i,
                "page_number": None,
                "section_header": "",
            }
            for i, s in enumerate(sections) if s.strip()
        ]

    def _chunk_sections(self, sections: list) -> list:
        """Chunk sections with overlap."""
        chunks = []
        for section in sections:
            content = section["content"]
            words = content.split()

            if len(words) <= self.chunk_size:
                chunks.append({
                    "content": content,
                    "metadata": {
                        "section_index": section["section_index"],
                        "page_number": section.get("page_number"),
                        "section_header": section.get(
                            "section_header", ""),
                        "chunk_index": len(chunks),
                        "char_start": 0,
                        "char_end": len(content),
                    },
                })
            else:
                # Sliding window chunking
                for i in range(0, len(words),
                               self.chunk_size - self.chunk_overlap):
                    chunk_words = words[i:i + self.chunk_size]
                    chunk_text = " ".join(chunk_words)
                    chunks.append({
                        "content": chunk_text,
                        "metadata": {
                            "section_index": section["section_index"],
                            "page_number": section.get("page_number"),
                            "section_header": section.get(
                                "section_header", ""),
                            "chunk_index": len(chunks),
                            "char_start": i,
                            "char_end": i + len(chunk_text),
                        },
                    })
        return chunks

    async def _enrich_chunks(self, chunks: list) -> list:
        """Enrich chunks with LLM-generated metadata."""
        # In production: use LLM to tag topic, key concepts, relevance
        for chunk in chunks:
            chunk["metadata"]["topic"] = "auto-classified"
            chunk["metadata"]["key_concepts"] = []
            chunk["metadata"]["relevance_score"] = 0.8
        return chunks

    def _list_files(self, dir_path: str,
                    recursive: bool) -> list:
        """List all supported files in directory."""
        supported_exts = {'.pdf', '.docx', '.xlsx', '.pptx',
                          '.html', '.htm', '.md', '.txt',
                          '.csv', '.json', '.png', '.jpg', '.jpeg'}
        files = []
        if recursive:
            for root, dirs, filenames in os.walk(dir_path):
                for f in filenames:
                    if os.path.splitext(f)[1].lower() in supported_exts:
                        files.append(os.path.join(root, f))
        else:
            for f in os.listdir(dir_path):
                fp = os.path.join(dir_path, f)
                if os.path.isfile(fp) and \
                   os.path.splitext(f)[1].lower() in supported_exts:
                    files.append(fp)
        return files

RAG Document Indexing Checklist

• [ ] Use unstructured.io for universal document parsing with auto-detection
• [ ] unstructured.io partition() auto-detects file type and routes to parser
• [ ] Install format-specific extras: pip install unstructured[docx,pptx]
• [ ] Use PyMuPDF for high-performance PDF text extraction
• [ ] Use pdfplumber for PDF table extraction
• [ ] Use python-docx for DOCX parsing (text, tables, images, comments)
• [ ] Use openpyxl for Excel parsing (sheets, formulas, charts, named ranges)
• [ ] Use python-pptx for PowerPoint parsing (slides, notes, tables, images)
• [ ] Use beautifulsoup4 for HTML parsing (structure, links, tables)
• [ ] Use pytesseract for OCR on images and scanned PDFs
• [ ] Use RAGParser for universal parsing with LangChain/LlamaIndex integration
• [ ] Use RAGWire for production toolkit with Qdrant, dedup, hybrid search
• [ ] Use DistillCore for 7-stage pipeline with coverage validation
• [ ] 7-stage pipeline: extract, classify, structure, chunk, enrich, embed, validate
• [ ] Stage 1 Extract: pull text from PDF (with OCR fallback), DOCX, HTML, TXT, MD
• [ ] Stage 2 Classify: LLM identifies document type, title, domain-specific metadata
• [ ] Stage 3 Structure: LLM breaks document into hierarchical sections with page ranges
• [ ] Stage 4 Chunk: section-aware splitting with 4 strategies (paragraph, sentence, fixed, LLM)
• [ ] Stage 5 Enrich: LLM tags each chunk with topic, key concepts, relevance
• [ ] Stage 6 Embed: generate vector embeddings (OpenAI, Ollama, local, Cohere)
• [ ] Stage 7 Validate: coverage checks (structuring 95%, chunking 98%, end-to-end 93%)
• [ ] Implement SHA256 deduplication at both file and chunk level
• [ ] Compute SHA256 hash of file content at ingestion start
• [ ] Check if hash exists in dedup database before processing
• [ ] Only embed and store new chunks — skip duplicates
• [ ] Track file_hash and chunk_hash in vector database metadata
• [ ] SHA256 dedup enables safe re-ingestion without duplicate embeddings
• [ ] Preserve table structure as markdown for embedding
• [ ] Extract tables with pdfplumber (PDF) or openpyxl (Excel)
• [ ] Convert tables to markdown table format
• [ ] Preserve row/column structure in chunks
• [ ] Add table caption or surrounding context
• [ ] Store table metadata: sheet name, page number, row range
• [ ] Extract rich metadata: file_name, file_hash, file_type
• [ ] Extract document_type, author, creation_date, modification_date
• [ ] LLM-extract: company, fiscal_period, domain
• [ ] LLM-enrich: topic, key_concepts, relevance_score
• [ ] Store page_number, section_header, chunk_index for citations
• [ ] Store char_start, char_end for span-level citation grounding
• [ ] Support directory ingestion with recursive scan
• [ ] Use metadata filters for multi-company, multi-year, multi-domain retrieval
• [ ] Enable sparse vectors (use_sparse: true) with fastembed for keyword matching
• [ ] Tune chunk_size, chunk_overlap, top_k, search_type per document type
• [ ] Store API keys in environment variables, reference as ${VAR} in YAML
• [ ] Use Qdrant Cloud or persistent local volume for production
• [ ] Keep force_recreate: false after initial setup to avoid collection resets
• [ ] LLM-assisted filter extraction grounded in values stored in vector DB
• [ ] Type-safe pipeline: ExtractedData dataclass for clean data flow
• [ ] Configurable error modes: 'stop' or 'continue' on processing errors
• [ ] Read chunking strategies for Stage 4
• [ ] Read chunk size for optimal token count
• [ ] Read what is RAG for architecture
• [ ] Build RAG from scratch with indexing
• [ ] Add hybrid search after indexing
• [ ] Add reranking for retrieval precision
• [ ] Choose embedding model for Stage 6
• [ ] Evaluate with RAG metrics
• [ ] Test: SHA256 dedup skips already-indexed files
• [ ] Test: tables preserved as markdown in chunks
• [ ] Test: metadata enables filtered retrieval
• [ ] Test: 7-stage pipeline achieves coverage targets
• [ ] Test: directory ingestion processes all supported formats
• [ ] Document supported formats, parsers, metadata schema, and dedup strategy

FAQ

How do you index PDF, DOCX, and Excel files for RAG?

Index documents for RAG by extracting text with format-specific parsers, chunking, enriching with metadata, embedding, and storing in a vector database. Unstructured-IO: "Unstructured is open-source ETL solution for transforming complex documents into clean, structured formats for language models. The unstructured library provides components for ingesting and pre-processing images and text documents such as PDFs, HTML, Word docs. Use the partition function to detect file type and route to the appropriate partitioning function." RAGWire: "Ingest PDFs, DOCX, XLSX, PPTX, Markdown, and text files. Extract structured metadata with LLM. Store dense and sparse vectors in Qdrant. SHA256 deduplication at file and chunk level." DistillCore: "7-stage pipeline: extract, classify, structure, chunk, enrich, embed, validate. Supports PDF (with OCR fallback), DOCX, HTML, TXT, MD, Excel." Steps: (1) Extract text with format-specific parser. (2) Classify document type with LLM. (3) Structure into hierarchical sections. (4) Chunk with section-aware splitting. (5) Enrich with LLM-generated metadata. (6) Embed with chosen model. (7) Store in vector database.

What is the best library for parsing documents for RAG?

Use unstructured.io for universal document parsing, PyMuPDF for high-performance PDF extraction, and format-specific libraries for specialized needs. Unstructured-IO: "Unstructured is open-source ETL for transforming complex documents into clean, structured formats for LLMs. Supports PDFs, HTML, Word docs, and many more. Use partition function to auto-detect file type. Install extras per doc type: pip install unstructured[docx,pptx]." RAGParser: "Comprehensive document parser for RAG applications. Supports PDF, DOCX, PPTX, XLSX, HTML, MD, CSV, JSON, images. Intelligent chunking strategies (Fixed, Semantic, Adaptive). Rich metadata extraction. Framework integration with LangChain and LlamaIndex. Dependencies: PyMuPDF, pdfplumber, python-docx, python-pptx, openpyxl, beautifulsoup4." RAGWire: "Document loading via MarkItDown for PDF, DOCX, XLSX, PPTX, TXT, MD. LLM metadata extraction for company, doc type, fiscal period." Choice: (1) Universal: unstructured.io (all formats, auto-detection). (2) PDF-focused: PyMuPDF + pdfplumber. (3) Production toolkit: RAGWire or DistillCore. (4) Lightweight: RAGParser.

How do you handle tables in PDF and Excel for RAG?

Extract tables with format-specific parsers and preserve table structure as markdown or structured text for embedding. RAGParser: "Content structure preservation: maintains headers, tables, images, and formatting context. PDF features: text, images, tables, metadata, OCR. Excel features: sheets, formulas, charts, named ranges." DistillCore: "Structure: LLM breaks document into hierarchical sections with page ranges. Chunk: section-aware splitting with 4 strategies." Dataloop: "PDF: ML-enhanced text extraction with PyMuPDF Layout, optional OCR. DOCX: document processing with tables and images. XLSX: spreadsheet processing with tables and images." Table handling: (1) Extract tables with pdfplumber (PDF) or openpyxl (Excel). (2) Convert to markdown table format for embedding. (3) Preserve row/column structure. (4) Add table caption or context. (5) Chunk table with surrounding text for context. (6) Store table metadata (sheet name, page number, row range).

How do you implement SHA256 deduplication in RAG indexing?

Implement SHA256 deduplication by hashing file content and chunk content to prevent re-indexing duplicates. RAGWire: "SHA256 Deduplication at both file and chunk level. Re-run ingestion safely with SHA256 deduplication. Directory ingestion: ingest an entire folder with one call, with optional recursive scan." Implementation: (1) Compute SHA256 hash of file content at ingestion start. (2) Check if hash exists in dedup database. (3) If exists, skip file (already indexed). (4) If new, compute SHA256 hash of each chunk. (5) Check chunk hashes against existing chunks. (6) Only embed and store new chunks. (7) Update dedup database with new hashes. (8) Track file_hash and chunk_hash in vector database metadata. Benefits: prevents duplicate embeddings, saves compute cost, enables safe re-ingestion, handles file updates (only re-embed changed chunks).

What metadata should you extract during RAG document indexing?

Extract rich metadata during indexing to enable filtered retrieval and provenance tracking. RAGWire: "LLM Metadata Extraction: extracts company, doc type, fiscal period using your LLM; fully customisable via YAML. Metadata filters for high-precision retrieval over multi-company, multi-year, or multi-domain collections." DistillCore: "Classify: LLM identifies document type, title, and domain-specific metadata. Enrich: LLM tags each chunk with topic, key concepts, and relevance." RAGParser: "Rich metadata including author, creation date, structure info. Content structure preservation: maintains headers, tables, images, and formatting context." Essential metadata: (1) file_name, file_hash (SHA256), file_type (pdf, docx, xlsx). (2) document_type (contract, report, invoice, manual). (3) author, creation_date, modification_date. (4) page_number, section_header, chunk_index. (5) company, fiscal_period, domain (LLM-extracted). (6) topic, key_concepts, relevance_score (LLM-enriched). (7) table_data, image_references. (8) char_start, char_end for citation grounding.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →