Build a Document Ingestion Pipeline for AI Search

Most RAG failures get diagnosed as retrieval problems, but they originate upstream. Before a vector store can return the right chunk for a query, that chunk has to exist in a form that is actually searchable: cleanly parsed, correctly split, and tagged with enough metadata to be filtered and audited. A weak ingestion pipeline produces a weak index no matter what retrieval strategy you layer on top.

This guide walks through building a document ingestion pipeline in Python that handles real-world document formats, survives messy input, and produces chunks your retrieval layer can actually use. By the end you will have a staged pipeline that parses PDFs and Word files, cleans the extracted text, splits it into retrievable chunks, attaches audit metadata, and writes embeddings to pgvector.

What you will build#

By the end of this guide you will have a document ingestion pipeline that:

• Parses PDFs, Word documents, and plain text using unstructured
• Cleans and normalises extracted text (whitespace, encoding, boilerplate)
• Splits documents into retrievable chunks with RecursiveCharacterTextSplitter
• Attaches source, page number, document title, and ingestion timestamp as metadata
• Generates embeddings and writes chunks to a vector store (pgvector)
• Saves intermediate artifacts per stage for debugging

This pipeline feeds directly into a retrieval layer. If you want the retrieval and generation side, read Build a RAG Search Pipeline which picks up where this guide leaves off.

Prerequisites#

The unstructured library handles format-specific parsing so you do not have to maintain separate PDF, DOCX, and HTML parsers. Its partition functions return typed Element objects with category labels (Title, NarrativeText, ListItem), which helps during cleaning (unstructured.io partitioning docs).

Architecture overview#

The pipeline is a linear flow of five stages. Source files enter on the left, and a parse, clean, chunk, enrich, and embed sequence turns them into vector rows in pgvector. Each stage takes the previous stage's output as input, which is exactly what makes per-stage inspection possible.

Save a JSON artifact after stages 2 and 4. When retrieval quality drops, you want to inspect what the index actually contains, not re-run the full pipeline to find out.

Step-by-step implementation#

from pathlib import Path
from unstructured.partition.pdf import partition_pdf
from unstructured.partition.docx import partition_docx
from unstructured.partition.text import partition_text

def parse_document(file_path: str) -> list[dict]:
    path = Path(file_path)
    suffix = path.suffix.lower()

    if suffix == ".pdf":
        elements = partition_pdf(filename=file_path, strategy="hi_res")
    elif suffix in (".docx", ".doc"):
        elements = partition_docx(filename=file_path)
    elif suffix == ".txt":
        elements = partition_text(filename=file_path)
    else:
        raise ValueError(f"Unsupported file type: {suffix}")

    return [
        {
            "text": str(el),
            "category": el.category,
            "page": getattr(el.metadata, "page_number", None),
            "source": str(path),
        }
        for el in elements
        if str(el).strip()
    ]

import re
import unicodedata

BOILERPLATE_PATTERNS = [
    r"^\s*Page \d+ of \d+\s*$",
    r"^\s*CONFIDENTIAL\s*$",
    r"^\s*\d+\s*$",        # Standalone page numbers
]

def clean_text(text: str) -> str:
    # Normalise Unicode
    text = unicodedata.normalize("NFKC", text)
    # Collapse excessive whitespace
    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    # Strip control characters (except newlines and tabs)
    text = re.sub(r"[^\S\n\t ]+", "", text)
    return text.strip()

def filter_elements(elements: list[dict]) -> list[dict]:
    cleaned = []
    for el in elements:
        text = clean_text(el["text"])
        if len(text) < 20:
            continue  # Skip near-empty fragments
        if any(re.match(p, text) for p in BOILERPLATE_PATTERNS):
            continue  # Skip structural noise
        cleaned.append({**el, "text": text})
    return cleaned

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_elements(elements: list[dict], chunk_size: int = 512, chunk_overlap: int = 64) -> list[dict]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = []
    for el in elements:
        sub_texts = splitter.split_text(el["text"])
        for i, sub_text in enumerate(sub_texts):
            chunks.append({
                "text": sub_text,
                "source": el["source"],
                "page": el["page"],
                "category": el["category"],
                "sub_index": i,
            })
    return chunks

from datetime import datetime, timezone

def enrich_chunks(chunks: list[dict], doc_title: str, doc_id: str) -> list[dict]:
    ingested_at = datetime.now(timezone.utc).isoformat()
    enriched = []
    for i, chunk in enumerate(chunks):
        enriched.append({
            **chunk,
            "doc_id": doc_id,
            "doc_title": doc_title,
            "chunk_index": i,
            "ingested_at": ingested_at,
            "char_count": len(chunk["text"]),
        })
    return enriched

-- Deletion query during re-ingestion
DELETE FROM document_chunks WHERE doc_id = 'contracts-v3-final.pdf';

import openai
import psycopg2
from pgvector.psycopg2 import register_vector

client = openai.AsyncOpenAI()

async def embed_texts(texts: list[str], batch_size: int = 100) -> list[list[float]]:
    all_embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        response = await client.embeddings.create(
            input=batch,
            model="text-embedding-3-small",
        )
        all_embeddings.extend([item.embedding for item in response.data])
    return all_embeddings

def store_chunks(conn, chunks: list[dict], embeddings: list[list[float]]) -> None:
    with conn.cursor() as cur:
        records = [
            (
                c["text"],
                c["source"],
                c["doc_id"],
                c["doc_title"],
                c["page"],
                c["chunk_index"],
                c["category"],
                c["ingested_at"],
                emb,
            )
            for c, emb in zip(chunks, embeddings)
        ]
        cur.executemany(
            """
            INSERT INTO document_chunks
              (content, source, doc_id, doc_title, page, chunk_index,
               category, ingested_at, embedding)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
            """,
            records,
        )
    conn.commit()

import json
from pathlib import Path

def save_artifact(data: list[dict], stage: str, doc_id: str) -> None:
    out_dir = Path("pipeline_artifacts") / doc_id
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / f"{stage}.json"
    with open(out_path, "w") as f:
        json.dump(data, f, indent=2, default=str)
    print(f"Saved {len(data)} records → {out_path}")

Test and verify the result#

Run the pipeline end-to-end on a single document and inspect the artifact files before indexing at scale.

import asyncio

async def ingest_document(file_path: str, doc_id: str, doc_title: str, dsn: str):
    # Stage 1: Parse
    elements = parse_document(file_path)
    print(f"Parsed {len(elements)} elements")

    # Stage 2: Clean
    cleaned = filter_elements(elements)
    save_artifact(cleaned, "02_cleaned", doc_id)
    print(f"Cleaned to {len(cleaned)} elements")

    # Stage 3: Chunk
    chunks = chunk_elements(cleaned)
    print(f"Chunked into {len(chunks)} segments")

    # Stage 4: Enrich
    enriched = enrich_chunks(chunks, doc_title=doc_title, doc_id=doc_id)
    save_artifact(enriched, "04_enriched", doc_id)

    # Stage 5: Embed + store
    texts = [c["text"] for c in enriched]
    embeddings = await embed_texts(texts)

    conn = psycopg2.connect(dsn)
    register_vector(conn)
    store_chunks(conn, enriched, embeddings)
    conn.close()

    print(f"Ingested {len(enriched)} chunks for '{doc_title}'")

asyncio.run(ingest_document(
    file_path="contracts/agreement_v3.pdf",
    doc_id="contracts-v3-final",
    doc_title="Service Agreement v3",
    dsn="postgresql://user:pass@localhost:5432/ragdb",
))

Check pipeline_artifacts/contracts-v3-final/04_enriched.json to verify chunk content, boundaries, and metadata before querying.

Common mistakes#

Parsing HTML instead of source PDFs. HTML is a rendering artifact, not the document. If you have access to the original PDF or DOCX, parse that instead. HTML parsing often loses table structure, footnotes, and heading hierarchy.

Not persisting doc_id for deletion. When you update a document, you need to remove its old chunks. Without a stable doc_id, you cannot safely delete without wiping the whole index.

Chunking before cleaning. Boilerplate and whitespace in a chunk waste context window space and add noise to the embedding vector. Always clean first, then chunk.

Skipping the metadata page field. Page numbers are the single most useful field for human verification of retrieved results. When a user asks "which page says X?", this field makes the answer instant. Without it, you must re-read the whole document.

Assuming all text elements are equally useful. A parsed PDF produces Header, Footer, Title, NarrativeText, Table, and ListItem elements. Footers and headers are usually noise; narrative text and table content are usually signal. Filter by category before chunking.

Production considerations#

Re-ingestion strategy. Build re-ingestion as a first-class operation, not an afterthought. When a source document changes, delete by doc_id, re-parse, and re-index. Keep a separate ingestion_log table recording which files have been processed and when, so you can detect stale documents.

Scale and parallelism. A single-threaded ingestion loop stalls on I/O, since network calls to the embedding API and disk reads are the bottlenecks. Use asyncio.gather for parallel embedding batches. For tens of thousands of documents, wrap the pipeline in a task queue (Celery, RQ, or a cloud queue) to distribute work across workers.

Idempotency. Re-running ingestion on the same document should not create duplicate chunks. Hash the file content and skip files whose hash has not changed since the last ingestion run:

import hashlib

def file_hash(path: str) -> str:
    with open(path, "rb") as f:
        return hashlib.sha256(f.read()).hexdigest()

Embedding model versioning. If you switch embedding models, all existing vectors become incompatible with new query vectors. Document the model name in the document_chunks table as a column. When you upgrade, re-embed all chunks, or run both old and new indexes during the migration window.

Monitoring pipeline health. Track chunks-per-document, average chunk character count, and embedding API error rate per ingestion run. A sudden drop in chunks-per-document usually signals a parsing failure on a new document format. For a fuller treatment of AI system observability, see observability for AI systems.

Key Takeaways#

Document ingestion is the foundation of retrieval quality. Every downstream problem (wrong answers, missed context, irrelevant chunks) traces back to how documents were parsed, cleaned, chunked, and tagged before they reached the vector store.

The most important structural decision is treating ingestion as a staged pipeline with inspectable artifacts at each step. When retrieval quality degrades, you need to know whether the problem is in parsing, chunking, or embedding, not run the whole pipeline again to find out.

Metadata is not optional. doc_id, source, page, and ingested_at are the minimum fields needed for deletion, filtering, and auditing. Add them at enrichment time, not as an afterthought after the index is full.

FAQ#

Which file formats does unstructured support?

unstructured supports PDF, DOCX, PPTX, XLSX, HTML, Markdown, plain text, EML, and more. Install pip install "unstructured[all-docs]" for full format coverage. Individual format extras like [pdf] or [docx] are available for lighter installs (unstructured partitioning docs).

How do I handle very large documents (500+ pages)?

Process large documents in page batches rather than loading the full content into memory. The unstructured partition_pdf function returns elements per page, so you can process and store pages incrementally. This also makes partial re-ingestion possible when only a few pages change.

Should I store the full document text alongside the chunks?

Storing the full raw text separately is useful for debugging and for generating answers with longer context. However, do not index the full document as a single vector, because the embedding will average out the semantics and retrieval will be poor. Keep full-text storage separate from the chunk vector table.

How do I handle scanned PDFs with no extractable text?

Set strategy="hi_res" in partition_pdf to trigger OCR via Tesseract. This requires the unstructured[pdf] extra and system-level Tesseract installation. OCR quality depends on scan resolution; expect degraded chunk quality for low-DPI scans.

Can I run this pipeline without OpenAI for embeddings?

Yes. Sentence Transformers from HuggingFace can run locally and produce dense vectors for any of the supported dimensions in pgvector. The tradeoff is local compute requirements and typically lower out-of-the-box multilingual quality compared to hosted models. Ensure the vector(dimensions) column in your database schema matches the output size of your chosen model.