Prepare the Data

RAG Quality Depends on Your Data

A RAG pipeline is only as good as the data it indexes. You can craft the perfect prompt and select the best generation model, but if the retrieved chunk cuts a definition in half or drags in a repetitive footer instead of useful text, your answer will suffer. Preparation happens in three stages: fetch the data where it lives, strip out anything meaningless, and split it into chunks optimized for indexing.

Fetching the Data

Real-world corpora are rarely homogeneous. You’ll typically have Markdown documentation in a repository, a few pages from a public website, and a handful of regulatory PDFs. Each source needs its own loader, but they must all produce the same structure—content, source, and type—so the rest of the pipeline remains agnostic to the text’s origin.

For text files, a recursive directory walk is sufficient, and filtering by extension prevents accidentally loading binary files.

import os

def load_text_files(directory, extensions=(".txt", ".md", ".rst")):
    """Load all text files from a directory."""
    documents = []
    for root, dirs, files in os.walk(directory):
        for f in files:
            if any(f.endswith(ext) for ext in extensions):
                filepath = os.path.join(root, f)
                with open(filepath, "r", encoding="utf-8") as fh:
                    content = fh.read()
                documents.append({
                    "content": content,
                    "source": filepath,
                    "type": "text",
                })
    return documents

docs = load_text_files("./data/")
print(f"{len(docs)} documents loaded")

Web pages come in over HTTP, but what you get is HTML. The extraction below strips tags using regular expressions and collapses multiple spaces. This is enough for a prototype, but the comment reminds you that in production, BeautifulSoup will do a much better job with tables and nested lists.

import requests

def fetch_web_page(url):
    """Fetch the text content of a web page."""
    response = requests.get(url)
    response.raise_for_status()

    # Basic text extraction (use BeautifulSoup for production)
    text = response.text
    # Remove HTML tags
    import re
    text = re.sub(r'<[^>]+>', ' ', text)
    text = re.sub(r'\s+', ' ', text).strip()

    return {
        "content": text,
        "source": url,
        "type": "web",
    }

# Example
page = fetch_web_page("https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt")
print(f"Content: {len(page['content'])} characters")

PDFs deserve special treatment. Instead of concatenating the entire document, keep one entry per page and encode the page number in the source as file.pdf#page=12. When your assistant cites this passage, the user will know exactly where to verify it.

# pip install PyPDF2
from PyPDF2 import PdfReader

def load_pdf(filepath):
    """Extract text from a PDF."""
    reader = PdfReader(filepath)
    pages = []
    for i, page in enumerate(reader.pages):
        text = page.extract_text()
        if text.strip():
            pages.append({
                "content": text,
                "source": f"{filepath}#page={i+1}",
                "type": "pdf",
                "page": i + 1,
            })
    return pages

Cleaning the Data

Cleaning is critical for retrieval quality because every stray character consumes space in the chunk and muddies the resulting vector. The following function chains four treatments: it removes control characters left over from format conversions, normalizes sequences of newlines and spaces, erases pagination markers typical of PDFs, and replaces endless URLs with a simple placeholder so they don’t eat up an entire chunk.

import re

def clean_text(text):
    """Clean text for RAG indexing."""
    # Remove control characters
    text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)

    # Normalize spaces and newlines
    text = re.sub(r'\n{3,}', '\n\n', text)
    text = re.sub(r' {2,}', ' ', text)

    # Remove repetitive headers/footers (PDF)
    text = re.sub(r'Page \d+ of \d+', '', text)

    # Remove very long URLs
    text = re.sub(r'https?://\S{100,}', '[URL]', text)

    return text.strip()

# Apply cleaning
for doc in docs:
    doc["content"] = clean_text(doc["content"])

Chunking Strategies

Chunking is the most important step in preparation: chunk size and method directly impact retrieval quality. Three approaches exist, ranging from mechanical to structure-aware.

Fixed-size chunking moves through the text in regular slices. It ignores meaning entirely, making it applicable to any source, but it often cuts sentences in half. Overlap partially corrects this flaw: with overlap=200, the end of each chunk repeats at the start of the next, ensuring a split idea remains whole in at least one of the two pieces.

def chunk_fixed_size(text, chunk_size=2048, overlap=200):
    """Fixed-size chunking with overlap."""
    chunks = []
    start = 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        chunk = text[start:end]
        if chunk.strip():
            chunks.append(chunk.strip())
        start += chunk_size - overlap
    return chunks

Paragraph-based chunking, recommended for documentation, relies on the text’s natural boundaries instead. It accumulates paragraphs as long as they stay under max_size, then opens a new chunk. The min_size threshold prevents indexing an orphaned heading or isolated line, which would produce a vector with no usable content.

def chunk_by_paragraphs(text, max_size=1500, min_size=100):
    """Intelligent paragraph-based chunking."""
    paragraphs = text.split("\n\n")
    chunks = []
    current = ""

    for para in paragraphs:
        para = para.strip()
        if not para:
            continue

        if len(current) + len(para) + 2 <= max_size:
            current = current + "\n\n" + para if current else para
        else:
            if len(current) >= min_size:
                chunks.append(current)
            current = para

    if current and len(current) >= min_size:
        chunks.append(current)

    return chunks

Markdown section chunking leverages information the author already provided: their headings. Each section becomes a standalone chunk, guaranteeing that a question about “authentication” retrieves the entire section rather than a fragment. Sections that are too long fall back to paragraph chunking, ensuring no chunk exceeds the limit.

def chunk_by_markdown_sections(text, max_size=2000):
    """Chunking by Markdown headings (## and ###)."""
    sections = re.split(r'\n(?=##\s)', text)
    chunks = []

    for section in sections:
        section = section.strip()
        if not section:
            continue
        if len(section) <= max_size:
            chunks.append(section)
        else:
            # Sub-chunk overly long sections
            sub_chunks = chunk_by_paragraphs(section, max_size=max_size)
            chunks.extend(sub_chunks)

    return chunks

Complete Preparation Pipeline

Finally, chain the three operations together and choose the chunking strategy based on file type: .md files go through sections, everything else through paragraphs. Each chunk then carries its metadata—source, position in the document, and total number of chunks from that document. Without this, you won’t be able to source an answer. The last two lines display the expansion rate and average size: if this average drops well below your chunk_size, the document is more fragmented than expected, and you need to revisit your chunking strategy.

def prepare_documents(raw_docs, chunk_size=1500, overlap=200):
    """Complete pipeline: clean, chunk, enrich with metadata."""
    all_chunks = []

    for doc in raw_docs:
        # 1. Clean
        clean = clean_text(doc["content"])

        # 2. Chunk based on type
        if doc.get("source", "").endswith(".md"):
            chunks = chunk_by_markdown_sections(clean, max_size=chunk_size)
        else:
            chunks = chunk_by_paragraphs(clean, max_size=chunk_size)

        # 3. Enrich with metadata
        for i, chunk in enumerate(chunks):
            all_chunks.append({
                "text": chunk,
                "source": doc["source"],
                "chunk_index": i,
                "total_chunks": len(chunks),
            })

    return all_chunks

# Execution
chunks = prepare_documents(docs)
print(f"{len(docs)} documents -> {len(chunks)} chunks")
print(f"Average size: {sum(len(c['text']) for c in chunks) / len(chunks):.0f} characters")

Key Takeaways

• RAG quality depends directly on data quality and chunking strategy
• Clean data before indexing (remove stray characters, long URLs, duplicates)
• Paragraph or Markdown section chunking preserves semantic coherence
• A 10–15% overlap between chunks prevents splitting ideas
• Keep metadata (source, page, position) for answer traceability