RAG Data Pipeline: From Raw Documents to Searchable Chunks

Every RAG demo looks the same: load a PDF, split it into chunks, embed it, ask a question, get a magically correct answer. Then you point the same pipeline at your company's actual documentation — scanned PDFs, PowerPoint decks exported as images, Confluence pages with nested tables, Slack exports with broken threading — and the retrieval quality collapses. The model isn't the problem. The rag data pipeline that feeds it is. Most teams spend a weekend on the retriever and prompt, then months quietly patching the ingestion layer because nobody designed it properly in the first place. This article walks through that ingestion layer in detail: how to go from raw, messy documents to clean, well-structured, searchable chunks that actually make your retrieval step useful.

This isn't a theoretical concern. Teams routinely ship a RAG assistant that performs beautifully in a demo with ten hand-picked PDFs, then watch accuracy fall off a cliff the moment real users start asking real questions against the full document corpus. The gap almost never traces back to the language model itself — it traces back to a chunk boundary that split a warranty clause from its exception, an OCR pass that silently produced blank text on a scanned invoice, or a duplicate document that's quietly outvoting the current version in the vector store. Fixing that requires treating ingestion as a first-class engineering discipline rather than a preprocessing afterthought, which is what the rest of this article covers stage by stage.

Why the pipeline matters more than the model

It's tempting to think of RAG as "embeddings plus a vector database plus an LLM." In practice, the embedding model and the LLM are the easy, replaceable parts. You can swap text-embedding-3-large for a different model in an afternoon. What you can't swap out easily is bad data that's already been chunked wrong and indexed. If a chunk splits a table in half, or cuts a legal clause mid-sentence, no amount of prompt engineering recovers the lost context. The retriever will faithfully return the broken chunk, and the LLM will confidently hallucinate a bridge between two unrelated fragments.

A rag data pipeline is really a data engineering problem wearing an AI costume. It has the same failure modes as any ETL system: bad parsing, silent data loss, schema drift, and no observability into what happened between "raw file" and "row in the database." Treat it that way — with real tests, real logging, and real versioning — and retrieval quality stops being a mystery.

Stage 1: Ingestion and format normalization

The first job is just getting bytes off disk (or from an API) and into a normalized text representation, without losing structure you'll need later. This is where most homegrown pipelines cut corners.

Documents typically arrive in wildly different shapes:

• PDFs — some are real text, some are scanned images requiring OCR, some are a mix per page
• Office documents — Word, PowerPoint, Excel, each with their own layout quirks
• HTML — marketing pages, internal wikis, help centers
• Markdown — README files, engineering docs
• Structured exports — Notion, Confluence, Zendesk, which come with metadata worth preserving

A pragmatic ingestion layer routes each file type to an appropriate parser rather than trying to force one library to do everything. For PDFs, a combination of pdfplumber for born-digital text and a fallback OCR path for scanned pages is a reasonable default:

import pdfplumber
import pytesseract
from pdf2image import convert_from_path

def extract_pdf_text(file_path: str) -> list[dict]:
    pages = []
    with pdfplumber.open(file_path) as pdf:
        for i, page in enumerate(pdf.pages):
            text = page.extract_text() or ""
            if len(text.strip()) < 20:
                # Likely a scanned page — fall back to OCR
                images = convert_from_path(file_path, first_page=i + 1, last_page=i + 1)
                text = pytesseract.image_to_string(images[0])
            pages.append({
                "page_number": i + 1,
                "text": text,
                "source": file_path,
            })
    return pages

Notice that the function returns a list of page-level dictionaries, not one giant string. That's deliberate. Page numbers, section headers, and source file paths are metadata you want to carry forward through every later stage — they're what let you cite "page 14 of the vendor contract" instead of "somewhere in some document." Losing this information at ingestion means it's gone forever; you can't reconstruct it during chunking.

A second, often-skipped step here is text cleanup: dehyphenating words broken across line wraps ("infor-\nmation" to "information"), collapsing repeated whitespace, and stripping headers/footers that repeat on every page (page numbers, confidentiality banners). These artifacts don't just look ugly — they pollute embeddings and waste tokens.

Stage 2: Structure-aware parsing, not just text extraction

A raw text blob throws away the single most useful signal in most documents: structure. Headings, bullet lists, tables, and code blocks all carry meaning about how content relates to its neighbors. A pipeline that flattens everything into plain text before chunking is discarding information it will never get back.

Instead, parse documents into a structured intermediate representation — usually a tree or a flat list of typed blocks (heading, paragraph, table, list-item, code) — before you touch chunking logic. Tools like unstructured do this reasonably well out of the box, and for HTML or Markdown you can roll your own with BeautifulSoup or a Markdown AST parser:

import re

def parse_markdown_blocks(md_text: str) -> list[dict]:
    blocks = []
    current_heading_path = []
    lines = md_text.split("\n")
    buffer = []

    def flush_buffer():
        if buffer:
            blocks.append({
                "type": "paragraph",
                "text": "\n".join(buffer).strip(),
                "heading_path": list(current_heading_path),
            })
            buffer.clear()

    for line in lines:
        heading_match = re.match(r"^(#{1,4})\s+(.*)", line)
        if heading_match:
            flush_buffer()
            level = len(heading_match.group(1))
            title = heading_match.group(2).strip()
            current_heading_path = current_heading_path[: level - 1] + [title]
        elif line.strip() == "":
            flush_buffer()
        else:
            buffer.append(line)
    flush_buffer()
    return blocks

The heading_path field is the payoff here — every paragraph now knows it lives under, say, ["Billing", "Refund Policy", "International Orders"]. That path becomes searchable metadata and, more importantly, it becomes context you can prepend to a chunk before embedding, so the chunk is no longer an orphaned paragraph but a paragraph that knows where it came from.

Tables deserve special handling. Splitting a table row-by-row destroys the column headers that give each cell meaning. A far better approach is to serialize each table as a self-contained unit — either as Markdown table syntax or as a list of "row described in natural language" sentences — so a chunk boundary never lands mid-table.

Stage 3: Chunking strategy

This is the stage most people associate with "the RAG pipeline," and for good reason — it's where information density and context boundaries are decided permanently. Get chunk size wrong and you either drown the retriever in irrelevant text or you starve the LLM of the context it needs to answer correctly.

There's no universal chunk size. The right size depends on the retrieval task:

• FAQ-style content — small chunks (150–300 tokens) work well because each answer is self-contained
• Narrative or technical documentation — medium chunks (300–600 tokens) that respect paragraph and section boundaries
• Legal or contractual text — chunk by clause or numbered section, never by fixed token count, because clause boundaries carry legal meaning
• Code — chunk by function or class, not by line count, so a chunk never splits a function signature from its body

A naive fixed-size splitter looks simple but causes real damage:

def naive_chunk(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunks.append(" ".join(words[start:end]))
        start = end - overlap
    return chunks

This works, technically, but it is blind to sentence and section boundaries — it will cheerfully cut a sentence in half if the word count lands there. A better approach is recursive, boundary-aware splitting: try to split on section breaks first, then paragraphs, then sentences, only falling back to raw token counts as a last resort.

import tiktoken

encoder = tiktoken.get_encoding("cl100k_base")

def recursive_split(text: str, max_tokens: int = 400, overlap_tokens: int = 40) -> list[str]:
    separators = ["\n\n", "\n", ". ", " "]
    def split_by(sep, chunk):
        return chunk.split(sep) if sep else list(chunk)

    def token_len(s: str) -> int:
        return len(encoder.encode(s))

    def _split(chunk: str, seps: list[str]) -> list[str]:
        if token_len(chunk) <= max_tokens or not seps:
            return [chunk]
        sep, rest = seps[0], seps[1:]
        pieces = split_by(sep, chunk)
        results, buffer = [], ""
        for piece in pieces:
            candidate = buffer + sep + piece if buffer else piece
            if token_len(candidate) <= max_tokens:
                buffer = candidate
            else:
                if buffer:
                    results.append(buffer)
                buffer = piece
        if buffer:
            results.append(buffer)
        final = []
        for r in results:
            if token_len(r) > max_tokens:
                final.extend(_split(r, rest))
            else:
                final.append(r)
        return final

    return _split(text, separators)

The overlap_tokens parameter matters more than people expect. A small overlap (10–15% of chunk size) between adjacent chunks means a sentence that spans a boundary still appears intact in at least one chunk. Too much overlap and you're paying for redundant embeddings and diluting retrieval precision; too little and you reintroduce the exact boundary problem you were trying to solve.

Stage 4: Metadata enrichment

A chunk without metadata is a chunk the retriever can only match on semantic similarity — which means it has no way to filter by recency, source authority, document type, or access permissions. Every chunk that goes into your vector store should carry a metadata payload, not just raw text.

At minimum, attach:

• source_id — the originating document's stable identifier
• source_type — pdf, confluence, slack, api-doc, etc.
• heading_path — the structural breadcrumb from Stage 2
• page_number or section_id — for citation
• created_at / updated_at — for recency-weighted retrieval
• access_level — critical if different users should see different documents

def build_chunk_record(chunk_text: str, doc_meta: dict, chunk_index: int) -> dict:
    return {
        "id": f"{doc_meta['source_id']}::chunk-{chunk_index}",
        "text": chunk_text,
        "metadata": {
            "source_id": doc_meta["source_id"],
            "source_type": doc_meta["source_type"],
            "heading_path": doc_meta.get("heading_path", []),
            "page_number": doc_meta.get("page_number"),
            "updated_at": doc_meta.get("updated_at"),
            "access_level": doc_meta.get("access_level", "internal"),
        },
    }

This metadata is what makes hybrid filtering possible later: "only search chunks from the last 90 days," or "only search documents this user is authorized to see." Bolting access control onto a vector store after the fact is painful; baking access_level into every chunk from day one is nearly free.

A metadata field people forget: a lightweight summary or a synthetic "context header" prepended to each chunk before embedding. Something as simple as prefixing the chunk with its heading path ("Document: Vendor Contract 2026 > Section 4: Termination Clauses") measurably improves retrieval, because the embedding model now encodes both the local content and where it sits structurally.

Stage 5: Embeddings and the vector store

Once chunks and metadata exist, embedding is mechanically simple but operationally easy to get wrong. Two mistakes show up constantly:

1. Embedding the wrong text. If you prepend a context header to the chunk for retrieval purposes, decide deliberately whether that header should also be shown to the LLM at generation time or stripped back out. Store both the "embed text" and the "display text" if they differ.
2. Batching without backpressure. Embedding APIs rate-limit, and a large backfill job with no retry/backoff logic will silently drop chunks on transient failures.

import time

def embed_batch(chunks: list[dict], embed_fn, batch_size: int = 64, max_retries: int = 3) -> list[dict]:
    results = []
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        texts = [c["text"] for c in batch]
        for attempt in range(max_retries):
            try:
                vectors = embed_fn(texts)
                break
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                time.sleep(2 ** attempt)
        for chunk, vector in zip(batch, vectors):
            results.append({**chunk, "embedding": vector})
    return results

On the storage side, whether you use Postgres with pgvector, a dedicated vector database, or a managed search service, the same principle applies: store the metadata alongside the vector, not in a separate system you have to join at query time. Query-time joins across two databases are a common source of production latency spikes that never show up in a local notebook demo.

Stage 6: Deduplication and freshness

Real document sets are messy in a way toy datasets never are. The same policy document gets uploaded three times with minor edits. A Confluence page gets updated weekly, and if your pipeline just appends new chunks every time it re-crawls, you end up with five near-duplicate versions of the same content competing for the same query — the retriever starts surfacing stale, contradictory information alongside the current version.

Two practical defenses:

• Content hashing. Hash each chunk's normalized text (lowercased, whitespace-collapsed) before insertion. If the hash already exists for that source_id, skip re-embedding — you just saved an API call and avoided a duplicate.
• Upsert by source, not append. When a source document changes, delete or mark-stale all of its previous chunks before inserting the new set, keyed by source_id. Never let a re-ingested document coexist indefinitely with its outdated predecessor.

import hashlib

def content_hash(text: str) -> str:
    normalized = " ".join(text.lower().split())
    return hashlib.sha256(normalized.encode()).hexdigest()

def upsert_document_chunks(source_id: str, new_chunks: list[dict], vector_store):
    vector_store.delete(filter={"source_id": source_id})
    for chunk in new_chunks:
        chunk["metadata"]["content_hash"] = content_hash(chunk["text"])
    vector_store.upsert(new_chunks)

This sounds obvious written down, but it's the single most common production bug in RAG systems that have been running for more than a few months: nobody built a re-ingestion path, so the index only ever grows, and answer quality degrades slowly and invisibly as stale content accumulates.

Stage 7: Evaluation before you trust it

A rag data pipeline isn't done when chunks land in the vector store — it's done when you can prove retrieval actually works for the queries your users will ask. This means building a small, honest evaluation set early, not after launch.

A minimal but effective approach:

1. Collect 30–50 real questions your users are likely to ask
2. For each, manually identify which document(s) and chunk(s) should be retrieved
3. Run retrieval and check whether the correct chunk appears in the top-k results
4. Track this number over time as a recall@k metric

def evaluate_recall_at_k(eval_set: list[dict], retriever, k: int = 5) -> float:
    hits = 0
    for item in eval_set:
        query = item["question"]
        expected_source = item["expected_source_id"]
        results = retriever.search(query, top_k=k)
        retrieved_sources = {r["metadata"]["source_id"] for r in results}
        if expected_source in retrieved_sources:
            hits += 1
    return hits / len(eval_set)

This is unglamorous work, but it's the only reliable way to know whether a chunking change actually improved things or just moved the problem around. Teams that skip this step end up debugging RAG quality by vibes — reading a handful of chat transcripts and guessing — which doesn't scale and doesn't catch regressions.

Stage 8: Handling multi-modal and semi-structured sources

Most write-ups about retrieval quietly assume every source is a clean text document. Real organizations don't work that way. Support tickets live in Zendesk with threaded replies. Product specs are Google Slides exported as PDFs where half the content is inside images. Engineering runbooks are Notion pages with toggle lists and embedded diagrams. Each of these needs its own normalization logic, and pretending one parser handles them all is how teams end up with a retriever that's excellent at answering questions about the one document type they tested with and useless everywhere else.

For image-heavy slides, running a vision-capable model over each slide to generate a text description before chunking is often more reliable than OCR alone, especially when the slide's meaning depends on a chart or diagram rather than literal text. For threaded conversations like Slack or support tickets, preserve the thread as a unit rather than splitting individual messages into separate chunks — a customer's problem statement and the agent's resolution are only useful together. A common pattern is to reconstruct each thread into a single "conversation transcript" block, tag it with participants and timestamps, and only then run it through the same chunking logic used for documents.

def flatten_thread(messages: list[dict]) -> str:
    lines = []
    for msg in sorted(messages, key=lambda m: m["timestamp"]):
        speaker = msg.get("author", "unknown")
        lines.append(f"{speaker}: {msg['text']}")
    return "\n".join(lines)

It's a small function, but the decision it encodes — conversation as unit, not message as unit — is the difference between a support bot that understands resolutions and one that returns disconnected fragments of customer complaints.

Stage 9: Observability and pipeline versioning

The last piece that separates a pipeline that works in a demo from one that survives in production is observability. When retrieval quality drops for a specific query next month, you need to be able to answer: which parser produced this chunk, which chunking version created these boundaries, and when was this document last re-embedded? Without that trail, debugging a bad answer becomes archaeology.

Treat your chunking logic like you'd treat a database migration — version it. Tag every chunk with the pipeline version that produced it (chunker_version: "v3-recursive-400tok"), so that when you improve the splitter, you can identify and selectively re-process only the documents that were chunked with an older, worse version, instead of guessing whether a full re-index is necessary. Log ingestion runs with counts: documents processed, chunks produced, chunks skipped as duplicates, OCR fallbacks triggered. A sudden spike in OCR fallbacks, for instance, is often the first sign that a document source has changed format upstream — a vendor started sending scanned images instead of native PDFs, and nobody would have noticed without that count being logged somewhere visible.

import logging

logger = logging.getLogger("rag_pipeline")

def log_ingestion_run(source_id: str, stats: dict) -> None:
    logger.info(
        "ingestion_run source_id=%s chunks_created=%d duplicates_skipped=%d ocr_fallbacks=%d chunker_version=%s",
        source_id,
        stats["chunks_created"],
        stats["duplicates_skipped"],
        stats["ocr_fallbacks"],
        stats["chunker_version"],
    )

None of this is glamorous, and none of it shows up in a quick demo. But it's exactly the kind of infrastructure that determines whether a RAG system is still trustworthy six months after launch, once the initial document set has been re-uploaded, edited, deleted, and re-uploaded again by a dozen different teams.

Common failure patterns worth watching for

A few patterns show up repeatedly across pipelines built without enough care at the ingestion stage:

• Tables silently converted to garbled text with no column alignment, making numeric answers unreliable
• Headers and footers ("Confidential — Internal Use Only," repeated on every page) polluting every chunk's embedding
• Chunk boundaries that split a heading from its content, so the chunk reads like an answer with no question
• No document-level access control, so a chatbot serving external customers can retrieve internal-only content
• Silent OCR failures on scanned documents that produce mostly whitespace, indexed as if they were valid text

Each of these is invisible in a demo with two or three clean documents and becomes a real support ticket at production scale with thousands of files from a dozen different sources. The uncomfortable pattern behind all five is the same: they're all data quality problems that existed before the LLM ever saw a token, and no amount of prompt tuning downstream can undo them. This is exactly why experienced teams spend disproportionate time on ingestion relative to prompting — the return on an hour spent fixing table parsing is almost always higher than an hour spent tweaking the system prompt, because a parsing fix improves every future query against that document, while a prompt tweak only ever addresses the query in front of you.

It's also worth building the habit of sampling raw chunks by hand, regularly, rather than trusting the pipeline blindly once it's running. Pull twenty random chunks from the vector store every couple of weeks and actually read them. It's a low-tech check, but it catches problems no automated eval will flag — a chunk that's technically well-formed but semantically meaningless because it lost its heading context, or a table that got serialized in a way a human would never parse correctly either.

Closing thoughts

The unglamorous truth about retrieval-augmented generation is that most of the leverage lives upstream of the model. A well-designed rag data pipeline — one that parses structure instead of flattening it, chunks with boundaries in mind instead of fixed word counts, enriches every chunk with metadata, and actively manages duplicates and staleness — will outperform a fancier retriever sitting on top of sloppy ingestion, every time. Treat this layer with the same engineering discipline you'd apply to any data pipeline feeding a production system, because that's exactly what it is. If you want a guided, hands-on walkthrough of building this end-to-end alongside retrieval and generation, our course Introduction to RAG covers exactly this territory, from raw document ingestion through evaluation, with real datasets instead of toy examples.