pdfmux/blog
pgvector

PDF to pgvector: an end-to-end embeddings pipeline in Python

9 min read
TL;DRExtract PDFs, chunk by heading, embed, and store in Postgres with pgvector. Full pipeline with confidence-gated ingestion and a cosine-similarity search query.

Direct answer: To get PDFs into pgvector, extract clean Markdown with pdfmux (pip install pdfmux), chunk it by heading boundary, embed each chunk, and insert into a Postgres table with a vector column via the pgvector Python client. The step people skip is confidence-gating: pdfmux returns a per-document extraction confidence score, and inserting low-confidence chunks into your index means bad extractions sit in your retrieval results indefinitely with no signal that they’re unreliable. Gate on it before the insert, not after.

If you already run Postgres — and most backends do — pgvector removes the case for a dedicated vector database for small-to-mid scale RAG. No new service to operate, no data duplicated across two stores, and you get transactional consistency between your document metadata and its embeddings for free. It stops making sense somewhere past tens of millions of vectors with heavy concurrent query load, where a purpose-built store (Qdrant, Weaviate, Pinecone) pulls ahead on raw ANN throughput. Below that, pgvector is simpler to operate and one less thing to keep in sync.

This post covers the full path: PDF in, queryable vector in.

Setup

pip install pdfmux pgvector psycopg openai watchdog
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE document_chunks (
    id bigserial PRIMARY KEY,
    source_file text NOT NULL,
    heading text,
    content text NOT NULL,
    confidence real NOT NULL,
    embedding vector(1536)
);

CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops);

vector(1536) matches OpenAI’s text-embedding-3-small. Swap the dimension if you use a different model — pgvector supports any fixed dimension, and the HNSW index needs to be built with the same distance operator you’ll query with (vector_cosine_ops for cosine, vector_l2_ops for Euclidean).

Step 1: extract with confidence

from pdfmux import process

result = process("annual-report.pdf", quality="standard")

print(f"Confidence: {result.confidence:.0%}")
print(f"Extractor used: {result.extractor_used}")
if result.warnings:
    print(f"Warnings: {result.warnings}")

quality="standard" runs the self-healing pipeline: fast-extract with PyMuPDF, audit each page, re-extract anything that fails the audit with OCR or a table-aware backend, and return one confidence score for the document. On the opendataloader-bench of 200 real-world PDFs, this pipeline scores 0.903 overall — full numbers in the benchmark writeup.

Step 2: chunk by heading

Heading-based chunking keeps each chunk semantically self-contained, which matters more for retrieval quality than raw chunk size. This is the same chunker used in PDF to Markdown for RAG pipelines — reused here rather than reinvented:

import re

def chunk_by_headings(markdown: str, max_chunk_size: int = 2000) -> list[dict]:
    sections = re.split(r'\n(?=#{1,3} )', markdown)
    chunks = []

    for section in sections:
        if len(section.strip()) < 10:
            continue
        lines = section.strip().split('\n')
        heading = lines[0].lstrip('#').strip() if lines[0].startswith('#') else None
        chunks.append({"text": section.strip(), "heading": heading})

    return chunks

For a deeper look at chunk-size tradeoffs and overlap strategy, see PDF chunking strategies for RAG.

Step 3: embed and insert, gated on confidence

This is the part worth being deliberate about. A document that extracts at 40% confidence still produces chunks — they just contain garbled or partial text. If those chunks get embedded and inserted alongside clean ones, nothing in your schema distinguishes them at query time. They’ll surface in results with the same apparent authority as a clean extraction, and the failure only shows up when a user notices the answer is wrong.

import psycopg
from pgvector.psycopg import register_vector
from openai import OpenAI

MIN_CONFIDENCE = 0.6

openai_client = OpenAI()
conn = psycopg.connect("dbname=ragdb", autocommit=True)
register_vector(conn)


def embed(text: str) -> list[float]:
    response = openai_client.embeddings.create(
        model="text-embedding-3-small", input=text
    )
    return response.data[0].embedding


def ingest(pdf_path: str):
    result = process(pdf_path, quality="standard")

    if result.confidence < MIN_CONFIDENCE:
        print(f"SKIPPED (confidence {result.confidence:.0%}): {pdf_path}")
        # Route to a re-extraction queue (try a higher quality tier)
        # or a manual-review bucket instead of silently dropping it.
        return

    chunks = chunk_by_headings(result.text)

    with conn.cursor() as cur:
        for chunk in chunks:
            embedding = embed(chunk["text"])
            cur.execute(
                """
                INSERT INTO document_chunks (source_file, heading, content, confidence, embedding)
                VALUES (%s, %s, %s, %s, %s)
                """,
                (pdf_path, chunk["heading"], chunk["text"], result.confidence, embedding),
            )

    print(f"Ingested {len(chunks)} chunks from {pdf_path} (confidence {result.confidence:.0%})")

MIN_CONFIDENCE = 0.6 is a starting point, not a rule — set it against your own tolerance for missed content versus bad content reaching the index. Documents that fail the threshold aren’t gone; pdfmux estimate can tell you upfront whether a higher quality tier (which routes more pages through OCR) is likely to help before you spend the extra time re-running it.

Step 4: query

def search(query: str, limit: int = 5):
    query_embedding = embed(query)

    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT source_file, heading, content, confidence,
                   1 - (embedding <=> %s) AS similarity
            FROM document_chunks
            ORDER BY embedding <=> %s
            LIMIT %s
            """,
            (query_embedding, query_embedding, limit),
        )
        return cur.fetchall()


for source, heading, content, confidence, similarity in search("what was Q3 revenue?"):
    print(f"[{similarity:.2f}] {source} — {heading} (extraction confidence {confidence:.0%})")

<=> is pgvector’s cosine-distance operator; 1 - distance converts it to the more intuitive cosine-similarity score. The confidence column travels with every result, so you can surface it in the UI (“this answer is sourced from a page pdfmux extracted with 72% confidence — verify against the original”) rather than presenting every retrieved chunk with equal certainty.

Bulk ingestion with pdfmux watch

For a folder that grows over time — a shared drive, an inbox, a nightly export — pdfmux watch handles the extraction side of “new file lands, get it processed”:

pdfmux watch ./incoming -o ./extracted/ --profile bulk-rag

Every new PDF dropped in ./incoming gets extracted and its result written to ./extracted/ automatically — cached by content hash, so a file that lands twice doesn’t re-extract. Point a watcher at the source directory instead and call ingest() on each new PDF directly — the extraction inside ingest() hits the same cache pdfmux watch just populated, so it returns in milliseconds rather than re-running the full pipeline:

from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

class NewPdfHandler(PatternMatchingEventHandler):
    def on_created(self, event):
        ingest(event.src_path)  # embed + insert, defined in Step 3

observer = Observer()
observer.schedule(NewPdfHandler(patterns=["*.pdf"]), "./incoming", recursive=False)
observer.start()

Combine with a cron-scheduled VACUUM ANALYZE document_chunks if the table sees heavy insert/delete churn — HNSW indexes degrade gracefully but still benefit from periodic maintenance at scale.

Handling document updates

Financial reports, contracts, and policy documents get revised. Re-ingesting a changed PDF without cleaning up the old chunks leaves stale content competing with current content in search results. Key the delete on source_file before re-inserting:

def reingest(pdf_path: str):
    with conn.cursor() as cur:
        cur.execute("DELETE FROM document_chunks WHERE source_file = %s", (pdf_path,))
    ingest(pdf_path)

pdfmux’s content-hash cache means the extraction step itself is cheap on unchanged files — the cost you’re managing here is the embedding + insert step, not re-running the extraction pipeline.

Pure similarity search gets you close, but real queries usually have a filter attached — “only this customer’s documents,” “only reports from this quarter,” “only chunks we’re confident in.” Since confidence and source_file are ordinary columns, not vector-store-specific metadata, they combine with the similarity query directly:

def search_filtered(query: str, source_prefix: str, min_confidence: float = 0.7, limit: int = 5):
    query_embedding = embed(query)

    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT source_file, heading, content, confidence,
                   1 - (embedding <=> %s) AS similarity
            FROM document_chunks
            WHERE source_file LIKE %s AND confidence >= %s
            ORDER BY embedding <=> %s
            LIMIT %s
            """,
            (query_embedding, f"{source_prefix}%", min_confidence, query_embedding, limit),
        )
        return cur.fetchall()

This is the practical argument for pgvector over a dedicated vector database at small-to-mid scale: metadata filtering is just SQL, with the same indexes, transactions, and query planner you already understand — not a separate filter DSL bolted onto the vector store’s API.

Tuning the HNSW index

The default HNSW parameters (m = 16, ef_construction = 64) are a reasonable starting point, but two knobs matter once you have real query volume:

CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- per-query recall/speed tradeoff:
SET hnsw.ef_search = 40;

Higher m and ef_construction improve recall at the cost of slower index builds and more memory — worth raising for a corpus where missing the right chunk is expensive (legal, compliance) and worth leaving at the default for a corpus where speed matters more than perfect recall (internal search over meeting notes). hnsw.ef_search is the one knob you can tune per query without rebuilding the index — set it higher for a “find this exact clause” query and lower for a “give me anything roughly relevant” one.

Estimating cost before a large ingestion run

Both extraction and embedding cost money at scale — extraction if you’re using an LLM-backed backend for hard pages, embedding always if you’re on a hosted model. Check both before pointing the pipeline at a folder of 10,000 PDFs:

pdfmux estimate ./incoming/*.pdf

This reports which backend each file would route through and the expected cost, without running the extraction. For the embedding side, text-embedding-3-small is priced per token — a rough rule of thumb is 1 token per ~4 characters of English text, so a 2,000-character chunk costs about 500 tokens. Multiply by expected chunk count (roughly document page count × 1.5 chunks/page for heading-based chunking) to get a ballpark before committing to a full run.

FAQ

Why HNSW instead of IVFFlat? HNSW builds a graph index that doesn’t require a pre-existing data sample to train on (IVFFlat needs rows in the table before you build the index, and its recall depends on choosing the right number of lists). For a pipeline where documents arrive continuously, HNSW’s build-as-you-go behavior is the better default. IVFFlat can still win on raw index-build speed for a one-time bulk load of a fixed, known corpus.

Should I store the embedding for the full document or per chunk? Per chunk. A single document-level embedding averages away the specific passage a query is actually looking for, which is why heading-based chunking is step 2 above rather than optional.

What if I need hybrid search (keyword + vector)? Postgres full-text search (tsvector/tsquery) lives in the same table alongside the vector column, so you can combine both in one query with UNION and a reciprocal-rank-fusion score, without a second search system. Out of scope for this post, but the schema above supports adding a content_tsv tsvector GENERATED ALWAYS AS (to_tsvector('english', content)) STORED column without restructuring anything.