Document Ingestion Pipelines: Loading, Cleaning, and Preprocessing Text for RAG

Introduction

Imagine you've just been handed a folder containing 400 internal company documents — PDFs from the legal team, Word files from HR, plain text exports from Confluence, and a handful of messy CSV files containing customer support transcripts. Your goal is to build a RAG (Retrieval-Augmented Generation) system that can answer employee questions by pulling from this knowledge base. You fire up your vector database, load a few documents, and run a test query. The answers come back garbled, incomplete, or — worse — confidently wrong.

The problem almost certainly isn't your language model or your retrieval algorithm. It's the data going in. Garbage in, garbage out is one of the oldest rules in computing, and it applies with brutal force to RAG systems. What happens before the embeddings get generated — the loading, cleaning, and chunking of raw text — determines whether your retrieval system returns gold or gravel. This upstream process is called the document ingestion pipeline, and it's the unglamorous but absolutely essential foundation of every effective RAG application.

By the end of this lesson, you'll understand exactly what a document ingestion pipeline is, how to build one in Python, and how to make intelligent decisions about cleaning and preprocessing text so that the chunks your retrieval system indexes are actually useful. This isn't theoretical — you'll write real code that handles real document types.

What you'll learn:

• What a document ingestion pipeline is and why it matters for RAG quality
• How to load text from common document formats (PDF, DOCX, plain text, CSV)
• How to clean raw text by removing noise, normalizing whitespace, and stripping irrelevant content
• How to chunk documents intelligently so retrieved passages make sense in context
• How to structure metadata alongside your text so retrieval results are traceable and filterable

---

Prerequisites

You should be comfortable writing Python at a basic level — defining functions, working with strings, and installing packages with pip. You don't need prior experience with RAG, embeddings, or vector databases. This lesson focuses purely on the data preparation stage that happens before any of that.

---

What Is a Document Ingestion Pipeline, and Why Does It Exist?

A document ingestion pipeline is a sequence of processing steps that transforms raw source documents — PDFs, Word files, web pages, database exports — into clean, structured text chunks that a retrieval system can index and search.

Think of it like a food processing plant. Raw ingredients (documents) arrive in all shapes, sizes, and states of cleanliness. The plant washes, trims, cuts, and packages them into uniform portions before they ever reach the consumer. Your ingestion pipeline does the same thing for text.

In a RAG system, the pipeline feeds into a vector store — a database that stores text chunks alongside their embeddings (numerical representations of meaning). When a user asks a question, the system retrieves the most semantically relevant chunks and passes them to a language model as context. The quality of those chunks directly determines the quality of the answers.

Here's the typical flow:

Raw Documents
     ↓
[Load]  → extract raw text from each file format
     ↓
[Clean] → remove noise, normalize formatting
     ↓
[Chunk] → split into appropriately sized pieces
     ↓
[Enrich with Metadata] → tag each chunk with source, page, date, etc.
     ↓
Vector Store (embeddings + indexed chunks)

Each stage has real consequences. Skip the cleaning step and your embeddings will be polluted by headers, footers, page numbers, and boilerplate legalese. Skip thoughtful chunking and your retrieval system will either pull in too little context or retrieve meaninglessly large walls of text. Let's build each stage, one at a time.

---

Stage 1: Loading Documents from Common Formats

The first challenge is simply getting text out of whatever container it's stored in. Different file formats require different extraction libraries.

Plain Text Files

The simplest case. Python's built-in open() handles this.

def load_text_file(filepath: str) -> str:
    with open(filepath, "r", encoding="utf-8") as f:
        return f.read()

raw_text = load_text_file("support_policy.txt")

Always specify encoding="utf-8" explicitly. Files from different systems or time periods may use Latin-1 or Windows-1252 encoding, and silent encoding errors produce corrupted text that's hard to debug downstream.

PDF Files

PDFs are notoriously messy. They're designed for visual presentation, not text extraction. The pymupdf library (also imported as fitz) is fast and handles most PDFs well.

pip install pymupdf

import fitz  # pymupdf

def load_pdf(filepath: str) -> list[dict]:
    """
    Returns a list of dicts, one per page:
    {"page": int, "text": str, "source": str}
    """
    doc = fitz.open(filepath)
    pages = []
    for page_num, page in enumerate(doc):
        text = page.get_text()
        pages.append({
            "page": page_num + 1,
            "text": text,
            "source": filepath
        })
    return pages

pdf_pages = load_pdf("employee_handbook.pdf")
print(pdf_pages[0]["text"][:500])

Notice we're already capturing metadata (page, source) at load time. We'll use this later to make retrieved chunks traceable — a critical feature in enterprise RAG applications where users need to know where an answer came from.

Warning: Some PDFs are image-based (scanned documents). pymupdf will return empty strings for these pages. If you're working with scanned files, you'll need an OCR (Optical Character Recognition) step using a library like pytesseract or a cloud service like AWS Textract. That's beyond this lesson's scope, but know the symptom: if your extracted text is blank or near-blank, you're likely dealing with a scanned PDF.

Word Documents

For .docx files, use the python-docx library.

pip install python-docx

from docx import Document

def load_docx(filepath: str) -> str:
    doc = Document(filepath)
    paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
    return "\n\n".join(paragraphs)

raw_text = load_docx("onboarding_guide.docx")

The if p.text.strip() filter skips empty paragraphs — Word documents are notorious for containing dozens of blank paragraph objects between actual content.

CSV Files (Structured Text)

Customer support transcripts, FAQ tables, or knowledge base exports often come as CSVs. These need a different approach: you typically want to combine relevant columns into a single text string per row.

import csv

def load_csv_as_documents(filepath: str, text_columns: list[str]) -> list[dict]:
    """
    Combines specified columns into a single text blob per row.
    """
    documents = []
    with open(filepath, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row_num, row in enumerate(reader):
            combined_text = " | ".join(
                f"{col}: {row[col]}" for col in text_columns if row.get(col)
            )
            documents.append({
                "row": row_num + 1,
                "text": combined_text,
                "source": filepath
            })
    return documents

# Example: customer support tickets with 'subject' and 'body' columns
tickets = load_csv_as_documents("support_tickets.csv", ["subject", "body"])

---

Stage 2: Cleaning the Raw Text

Once you've extracted text, what you actually have is a mess. PDF extraction typically drags along page headers, footers, page numbers, weird Unicode characters, and OCR artifacts. Word documents include revision tracking artifacts. Even plain text files carry inconsistent whitespace and encoding quirks.

Cleaning is about removing noise that would confuse the embedding model and pollute your retrieval results.

Removing Common Boilerplate

Real extracted PDF text often looks like this:

ACME CORPORATION CONFIDENTIAL
Employee Handbook — Version 4.2
Page 12 of 89

3.1 Vacation Policy

Employees are entitled to 15 days of paid vacation per year...

ACME CORPORATION CONFIDENTIAL
Employee Handbook — Version 4.2
Page 13 of 89

That repeated header and footer adds noise to every chunk. Let's write a cleaner that handles common patterns:

import re

def clean_text(text: str) -> str:
    # Remove page number patterns like "Page 12 of 89" or "- 12 -"
    text = re.sub(r'Page\s+\d+\s+of\s+\d+', '', text, flags=re.IGNORECASE)
    text = re.sub(r'-\s*\d+\s*-', '', text)
    
    # Remove excessive whitespace and normalize line breaks
    text = re.sub(r'\n{3,}', '\n\n', text)   # collapse 3+ newlines to 2
    text = re.sub(r'[ \t]{2,}', ' ', text)   # collapse multiple spaces/tabs
    
    # Remove non-printable characters (common in PDFs)
    text = re.sub(r'[^\x20-\x7E\n]', ' ', text)
    
    # Strip leading/trailing whitespace
    text = text.strip()
    
    return text

Tip: The regex [^\x20-\x7E\n] strips everything that isn't a standard printable ASCII character or a newline. This is intentionally aggressive. If you're working with documents in languages other than English, switch to [^\u0020-\uFFFF\n] to preserve Unicode characters while still removing control characters and null bytes.

Normalizing Formatting

Legal documents and academic papers love inconsistent formatting: random ALL CAPS sections, inconsistent hyphenation across line breaks, and bullet points rendered as weird characters. A few more cleaning passes:

def normalize_formatting(text: str) -> str:
    # Fix hyphenated line breaks (word split across two lines in PDFs)
    # "confi-\ndential" → "confidential"
    text = re.sub(r'(\w)-\n(\w)', r'\1\2', text)
    
    # Replace bullet point artifacts with standard dashes
    text = re.sub(r'[•●▪▸◦]', '-', text)
    
    # Normalize smart quotes to standard quotes
    text = text.replace('\u201c', '"').replace('\u201d', '"')
    text = text.replace('\u2018', "'").replace('\u2019', "'")
    
    return text

Putting the Cleaning Together

It's good practice to compose your cleaning functions into a single pipeline function:

def preprocess_text(raw_text: str) -> str:
    text = clean_text(raw_text)
    text = normalize_formatting(text)
    return text

---

Stage 3: Chunking — The Decision That Matters Most

This is where most beginners make their biggest mistake. They either dump entire documents as single chunks (too much context, retrieval becomes imprecise) or split naively on every newline (too little context, retrieved chunks lose meaning).

Chunking is the process of splitting cleaned text into smaller pieces that will each become one retrievable unit in your vector store.

Why Chunk Size Matters

Embedding models have a context window — a maximum number of tokens (roughly 4 characters each) they can process at once. More importantly, the semantic density of a chunk affects retrieval precision. A chunk that contains exactly one coherent idea retrieves cleanly. A chunk that mixes three topics retrieves ambiguously.

The general sweet spot for most RAG applications is 300–600 tokens per chunk, with some overlap between consecutive chunks to prevent ideas from being severed at boundaries.

Fixed-Size Chunking with Overlap

The simplest approach: split text into chunks of N characters with an overlap of M characters between consecutive chunks.

def chunk_text_fixed(text: str, chunk_size: int = 1500, overlap: int = 200) -> list[str]:
    """
    Splits text into fixed-size chunks with overlap.
    chunk_size: approximate number of characters per chunk (~300-400 tokens)
    overlap: characters shared between consecutive chunks
    """
    chunks = []
    start = 0
    
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        
        # Try to end at a sentence boundary rather than mid-word
        if end < len(text):
            last_period = chunk.rfind('.')
            if last_period > chunk_size * 0.5:  # only adjust if period is in second half
                chunk = chunk[:last_period + 1]
        
        chunks.append(chunk.strip())
        start += chunk_size - overlap
    
    return [c for c in chunks if c]  # remove any empty chunks

The overlap is crucial. Without it, a sentence that spans a chunk boundary gets split in two, and neither chunk contains the complete thought. With overlap, the tail of one chunk reappears at the head of the next, preserving continuity.

Semantic Chunking by Section

For well-structured documents like handbooks or technical documentation, splitting by section headers is more natural than splitting by character count:

def chunk_by_section(text: str, min_chunk_size: int = 200) -> list[str]:
    """
    Splits text at heading-like patterns (numbered sections, markdown headers).
    """
    # Split on patterns like "3.1 ", "## ", "CHAPTER 4", etc.
    section_pattern = re.compile(
        r'\n(?=(?:\d+\.?\d*\s+[A-Z]|#{1,3}\s|CHAPTER\s+\d+|SECTION\s+\d+))',
        re.MULTILINE
    )
    
    sections = section_pattern.split(text)
    
    # Merge sections that are too short into the next section
    merged = []
    buffer = ""
    for section in sections:
        buffer += section
        if len(buffer) >= min_chunk_size:
            merged.append(buffer.strip())
            buffer = ""
    
    if buffer.strip():
        merged.append(buffer.strip())
    
    return merged

Tip: In practice, you'll often combine both strategies — use section boundaries when they exist, and fall back to fixed-size chunking for unstructured sections that are too long.

---

Stage 4: Enriching Chunks with Metadata

Every chunk that enters your vector store should travel with a metadata dictionary — structured information about where the chunk came from, what document it belongs to, what page it was on, and when the document was created.

Why does metadata matter? Two reasons:

1. Traceability. When your RAG system surfaces an answer, users need to verify it. Metadata lets you cite the source document and page number.
2. Filtering. You can narrow retrieval to specific document types, departments, or date ranges without relying purely on semantic similarity.

from datetime import datetime

def create_chunk_records(chunks: list[str], source_metadata: dict) -> list[dict]:
    """
    Wraps each text chunk in a document record with metadata.
    
    source_metadata example:
    {
        "source": "employee_handbook.pdf",
        "doc_type": "policy",
        "department": "HR",
        "last_updated": "2024-01-15"
    }
    """
    records = []
    for i, chunk in enumerate(chunks):
        record = {
            "chunk_id": f"{source_metadata['source']}::chunk_{i}",
            "text": chunk,
            "chunk_index": i,
            "total_chunks": len(chunks),
            "ingested_at": datetime.utcnow().isoformat(),
            **source_metadata  # merge in the source metadata
        }
        records.append(record)
    return records

---

Putting the Full Pipeline Together

Now let's wire every stage into a single, reusable pipeline function:

def ingest_document(filepath: str, source_metadata: dict, 
                    chunk_size: int = 1500, overlap: int = 200) -> list[dict]:
    """
    Full ingestion pipeline: load → clean → chunk → enrich with metadata.
    Returns a list of chunk records ready for embedding and indexing.
    """
    # Determine file type and load accordingly
    if filepath.endswith(".pdf"):
        pages = load_pdf(filepath)
        raw_text = "\n\n".join(p["text"] for p in pages)
    elif filepath.endswith(".docx"):
        raw_text = load_docx(filepath)
    elif filepath.endswith(".txt"):
        raw_text = load_text_file(filepath)
    else:
        raise ValueError(f"Unsupported file type: {filepath}")
    
    # Clean and normalize
    clean = preprocess_text(raw_text)
    
    # Chunk
    chunks = chunk_text_fixed(clean, chunk_size=chunk_size, overlap=overlap)
    
    # Enrich with metadata
    records = create_chunk_records(chunks, source_metadata)
    
    print(f"Ingested {filepath}: {len(records)} chunks created.")
    return records


# Example usage
handbook_records = ingest_document(
    filepath="employee_handbook.pdf",
    source_metadata={
        "source": "employee_handbook.pdf",
        "doc_type": "policy",
        "department": "HR",
        "last_updated": "2024-03-01"
    }
)

---

Hands-On Exercise

Set up your environment and work through this exercise:

Setup:

pip install pymupdf python-docx

Your task: Find any PDF on your computer (or download a public one — a Wikipedia article exported as PDF works great). Run it through the complete pipeline we built. Then answer these questions by inspecting your output:

1. Open the raw extracted text from load_pdf(). What noise do you see? Headers, page numbers, weird characters?
2. Run it through preprocess_text() and compare. What changed?
3. Try chunk_size=500 and chunk_size=2000. Print the first three chunks from each. How does the coherence of each chunk differ?
4. Add a custom field to source_metadata — something like "confidentiality": "internal". Verify it appears on every chunk record.

Bonus challenge: Write a function ingest_folder(folder_path: str) that loops through all PDFs in a folder, calls ingest_document() on each, and returns all chunk records in a single flat list.

---

Common Mistakes & Troubleshooting

"My retrieved chunks are incomplete — answers get cut off mid-sentence." Your chunk size is too small or you have no overlap. Increase overlap from 0 to at least 100-200 characters. Also check that your sentence boundary logic in chunk_text_fixed is actually finding periods — if your text uses lots of lists or headers, periods may be rare.

"Retrieval is returning chunks full of header/footer boilerplate." Your cleaning step isn't catching the specific patterns in your documents. Print 10–20 pages of raw extracted text and look for repeated patterns. Write targeted re.sub() calls for them. There's no universal boilerplate remover — you have to inspect your data.

"PDFs load but the text looks scrambled or has random character insertions." This is a PDF encoding issue. Try opening the file in a PDF reader and copying text manually — if it looks fine there, the issue is the extraction library. Try pdfplumber as an alternative to pymupdf:

pip install pdfplumber

"Some pages produce empty strings." Almost certainly a scanned/image-based PDF. Check with page.get_text() — if it returns an empty string or just whitespace for multiple consecutive pages, you need OCR.

"My chunks are all identical sizes except the last one, which is tiny." Expected behavior. The last chunk will almost always be shorter than the rest. You can merge it into the previous chunk if it falls below a minimum size threshold (e.g., less than 100 characters).

---

Summary & Next Steps

You've now built a complete document ingestion pipeline from scratch. Here's what each stage contributes:

• Loading extracts raw text from the diverse formats that real-world documents live in
• Cleaning removes noise that would corrupt embeddings and pollute retrieval results
• Chunking creates retrievable units that are sized for both embedding models and human readability
• Metadata enrichment makes chunks traceable and filterable in production systems

The pipeline you built is genuinely production-ready for small-to-medium document sets. Real enterprise pipelines add more sophistication — parallel processing for speed, deduplication to catch repeated documents, and document-level version tracking — but they all rest on the same foundation you've just built.

Where to go next:

• Embeddings & Vector Stores: Learn how to pass your chunk records to an embedding model and store them in a vector database like Chroma or Pinecone. That's the natural next step after ingestion.
• Advanced Chunking Strategies: Explore semantic chunking using sentence embeddings to find natural topic boundaries, rather than relying on character counts or headers.
• Hybrid Retrieval: Learn how to combine dense vector search (semantic similarity) with sparse keyword search (BM25) for more robust retrieval — especially important when document terminology is highly specific or technical.
• Evaluation: Learn how to measure whether your pipeline is producing chunks that actually improve answer quality, using frameworks like RAGAS.

The data doesn't get better on its own. The work you put into the ingestion pipeline is the highest-leverage investment you can make in a RAG system.