PDF to Qdrant: a production vector-ingestion pipeline in Python

Direct answer: Extract PDFs to Markdown with pdfmux, chunk by heading, embed each chunk, and upsert into a Qdrant collection with confidence and source_file stored as payload fields — not just the vector. That payload is what lets you filter search results by confidence at query time (min_confidence >= 0.7) and delete-and-reingest a specific document later without touching the rest of the collection. The extraction and chunking steps are identical to the pgvector pipeline; this post covers what changes when you’re indexing into Qdrant instead.

Why Qdrant over pgvector

The pgvector post makes the case for staying on Postgres if you already run it: no new service, transactional consistency, and it’s fine up to tens of millions of vectors with moderate concurrency. Qdrant is the other side of that tradeoff — a purpose-built ANN engine that pulls ahead once you need any of: horizontal scaling across nodes, payload-aware filtering built into the index rather than bolted onto SQL, quantization to shrink memory footprint at scale, or multiple named vectors per point (e.g. a dense embedding and a sparse keyword vector on the same document, for hybrid search). If your corpus is a few hundred thousand chunks with straightforward filters, either works and the choice is mostly about what’s already in your stack. Past that, Qdrant’s filtering and quantization options start to matter more than convenience.

This post covers the same full path as the pgvector one — PDF in, queryable vector out — with Qdrant-specific decisions called out where they differ.

Setup

pip install pdfmux qdrant-client openai watchdog
docker run -p 6333:6333 -v $(pwd)/qdrant_storage:/qdrant/storage qdrant/qdrant

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

client = QdrantClient(url="http://localhost:6333")

client.create_collection(
    collection_name="document_chunks",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
)

size=1536 matches OpenAI’s text-embedding-3-small, same as the pgvector setup — swap it for whatever embedding model you’re using. Qdrant creates the HNSW index automatically on collection creation; the tuning knobs are covered further down.

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}")

Same self-healing pipeline as every other ingestion path in the pdfmux docs: fast-extract, audit each page, re-extract failures with OCR or a table-aware backend, return one confidence score. 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

Reuse the same heading-based chunker as the pgvector pipeline — the chunking decision doesn’t change based on which vector store you’re loading into, so there’s no reason to re-derive it:

import re

def chunk_by_headings(markdown: str) -> 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 chunk-size tradeoffs and overlap strategy, see PDF chunking strategies for RAG.

Step 3: embed and upsert, gated on confidence

Same reasoning as the pgvector post: a document extracted at 40% confidence still produces chunks, and if those chunks land in the index next to clean ones, nothing distinguishes them at query time until a user notices a wrong answer. Gate before the upsert, not after.

import hashlib
from openai import OpenAI
from qdrant_client.models import PointStruct

MIN_CONFIDENCE = 0.6
openai_client = OpenAI()


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


def chunk_id(source_file: str, index: int) -> int:
    # Qdrant point IDs must be a u64 int or UUID — derive a stable one
    # from the source file + chunk index so re-ingesting the same file
    # produces the same IDs (see "handling document updates" below).
    digest = hashlib.sha256(f"{source_file}:{index}".encode()).hexdigest()
    return int(digest[:16], 16)


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}")
        return

    chunks = chunk_by_headings(result.text)
    points = [
        PointStruct(
            id=chunk_id(pdf_path, i),
            vector=embed(chunk["text"]),
            payload={
                "source_file": pdf_path,
                "heading": chunk["heading"],
                "content": chunk["text"],
                "confidence": result.confidence,
            },
        )
        for i, chunk in enumerate(chunks)
    ]

    client.upsert(collection_name="document_chunks", points=points)
    print(f"Ingested {len(points)} chunks from {pdf_path} (confidence {result.confidence:.0%})")

upsert batches naturally — pass a list of PointStructs and Qdrant writes them in one call, which matters once you’re ingesting more than a handful of documents at a time. For large batches (thousands of points), chunk the list into batches of a few hundred rather than one giant call, to keep memory and request size reasonable.

Step 4: query with a confidence filter

from qdrant_client.models import Filter, FieldCondition, Range

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

    results = client.query_points(
        collection_name="document_chunks",
        query=query_embedding,
        query_filter=Filter(
            must=[FieldCondition(key="confidence", range=Range(gte=min_confidence))]
        ),
        limit=limit,
    )
    return results.points


for point in search("what was Q3 revenue?"):
    p = point.payload
    print(f"[{point.score:.2f}] {p['source_file']} — {p['heading']} (confidence {p['confidence']:.0%})")

This is the part that’s structurally different from pgvector: the confidence filter runs inside the vector search as part of the same index traversal, via Qdrant’s payload index, rather than as a separate WHERE clause evaluated by the query planner. For a large collection with a selective filter, that tends to be faster — Qdrant prunes low-confidence points during the graph walk instead of retrieving a candidate set and filtering it afterward.

Payload indexing for filter performance

Filtering works without an index, but on a large collection, index the fields you filter on frequently — it turns a linear payload scan into a proper lookup:

client.create_payload_index(
    collection_name="document_chunks",
    field_name="confidence",
    field_schema="float",
)
client.create_payload_index(
    collection_name="document_chunks",
    field_name="source_file",
    field_schema="keyword",
)

Do this once per field, not per query. Skip it for a small collection (a few thousand points) where the scan cost is negligible either way.

Bulk ingestion with pdfmux watch

Identical pattern to the pgvector pipeline — point a watcher at the source directory and call ingest() on each new file:

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

class NewPdfHandler(PatternMatchingEventHandler):
    def on_created(self, event):
        ingest(event.src_path)

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

Run pdfmux watch ./incoming -o ./extracted/ --profile bulk-rag alongside it if you also want extracted output persisted to disk — the extraction inside ingest() hits the same content-hash cache, so it returns fast rather than re-running the full pipeline on a file that’s already been processed.

Handling document updates

Because chunk_id() derives a stable ID from source_file + chunk index, re-ingesting an unchanged document just overwrites the same points — upsert is idempotent on ID. But if the new version has a different chunk count than the old one (a section was added or removed), stale trailing chunks from the old version won’t get overwritten, only replaced up to the new chunk count. Delete by source_file before re-inserting to avoid that:

from qdrant_client.models import FilterSelector

def reingest(pdf_path: str):
    client.delete(
        collection_name="document_chunks",
        points_selector=FilterSelector(
            filter=Filter(must=[FieldCondition(key="source_file", match={"value": pdf_path})])
        ),
    )
    ingest(pdf_path)

Tuning HNSW for real query volume

Qdrant’s defaults (m=16, ef_construct=100) are reasonable, but the same two tradeoffs apply as any HNSW index — higher values improve recall at the cost of slower builds and more memory:

from qdrant_client.models import HnswConfigDiff

client.update_collection(
    collection_name="document_chunks",
    hnsw_config=HnswConfigDiff(m=16, ef_construct=100),
)

ef (search-time) is the knob you adjust per query without rebuilding the index — pass it via search_params on query_points when a specific query needs higher recall (a “find this exact clause” legal search) versus the default (a “roughly relevant” internal search).

Quantization for large collections

At scale, the full-precision float32 vectors are usually the largest cost in the collection. Qdrant supports scalar and binary quantization, which store a compressed representation for the fast first pass and rescore the top candidates against the original vectors:

from qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig, ScalarType

client.update_collection(
    collection_name="document_chunks",
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantizationConfig(type=ScalarType.INT8, quantile=0.99, always_ram=True)
    ),
)

This trades a small amount of recall for a large reduction in memory footprint — how much of each depends on your embedding model and corpus, so benchmark against your own recall requirements before enabling it in production rather than assuming a fixed tradeoff.

Qdrant vs pgvector — quick reference

FAQ

Do I need payload indexes on every field? No — only on fields you filter or sort by frequently. Fields you only ever return in results (like content) don’t need one.

What happens if I upsert a point with an ID that already exists? Qdrant overwrites it. That’s what makes the chunk_id() scheme above safe for re-ingestion — same file, same chunk index, same ID, so a repeat ingest() call updates in place instead of creating a duplicate.

Can I store the source PDF’s confidence at the document level instead of copying it onto every chunk? You can, but it costs you the ability to filter a single query by confidence, which is the reason this pipeline denormalizes it onto every point in the first place. A document-level lookup would mean joining chunk results back against document metadata after the vector search, in application code — the payload approach keeps it a single query.