Embedding Pipeline

The chunks that came back twice

Monday morning, a support lead forwards a screenshot. The help-center assistant answered "Can I get a deleted project back after a month?" with a thin, half-right answer. The retrieval trace shows why: the four chunks in the prompt are really two chunks, each one twice. Nobody touched the retrieval code. The nightly indexing job ran, as always.

Here is that job, shrunk to six help-center pages of a project-management app. It embeds one chunk per call and uploads each vector to Qdrant. The small open model all-MiniLM-L6-v2 stands in for the embedding service, so everything runs offline.

import hashlib, re, uuid
import numpy as np
from qdrant_client import QdrantClient, models
from sentence_transformers import SentenceTransformer

DOCS = {  # doc_id -> Markdown; the doc_id is the source system's id, stable across edits
    "retention": "## Deleted projects\nDeleted projects stay in the trash for 30 days. An owner can "
                 "restore them from Settings > Trash. After 30 days they are purged and support "
                 "cannot bring them back.\n## Audit log retention\nBusiness workspaces keep audit "
                 "logs for 400 days, Team workspaces for 90 days.",
    "security":  "## Two-factor authentication\nAdmins can require two-factor authentication for "
                 "every member. Members without it are signed out at their next request.\n"
                 "## Session timeout\nIdle sessions end after 12 hours. Business workspaces can "
                 "shorten this to 15 minutes.",
    "billing":   "## Plans\nThere are three plans: Free, Team and Business, billed per member.\n"
                 "## Invoices\nInvoices are emailed to the billing owner at the start of each "
                 "billing period and can be downloaded as PDF.\n## Refunds\nYearly plans can be "
                 "refunded within 14 days of purchase. Monthly plans are not refunded.",
    "export":    "## CSV export\nAny project can be exported to CSV from the project menu.\n"
                 "## Full workspace export\nOwners can export the whole workspace as a ZIP of "
                 "JSON files. Large exports are emailed as a download link within 24 hours.",
    "api":       "## API keys\nCreate API keys in Settings > API. A key has the permissions of the "
                 "member who created it.\n## Rate limits\nEach key may send 100 requests per "
                 "minute. Extra requests get HTTP 429 with a Retry-After header.",
    "sso":       "## SAML single sign-on\nBusiness workspaces can require SAML single sign-on with "
                 "any SAML 2.0 identity provider.\n## SCIM provisioning\nSCIM creates and removes "
                 "members automatically when they join or leave your identity provider.",
}

def split_sections(text):
    """One chunk per '## ' section, heading included. Smart Chunking covers better splitters."""
    return [part.strip() for part in re.split(r"(?m)^(?=## )", text) if part.strip()]

model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")  # 384-d, unit-length output
client = QdrantClient(url="http://localhost:6333")                     # or QdrantClient(":memory:")
print(sum(len(split_sections(t)) for t in DOCS.values()), "chunks in", len(DOCS), "documents")
# 13 chunks in 6 documents

The nightly job, as it was written: two scheduled tasks, one that recreates the collection and one that loads it. When a task fails, the scheduler retries that task.

COLL = "help_naive"

def recreate():                          # task 1: start from an empty collection
    if client.collection_exists(COLL):
        client.delete_collection(COLL)
    client.create_collection(COLL, vectors_config=models.VectorParams(size=384, distance=models.Distance.COSINE))

calls = 0
def embed_one(text):                     # stands in for an embedding API
    global calls
    calls += 1
    if calls == 8:
        raise RuntimeError("429 Too Many Requests")   # the job has no retry
    return model.encode(text)

def load(docs):                          # task 2: embed and upload every chunk
    for doc_id, text in docs.items():
        for piece in split_sections(text):
            client.upload_collection(COLL, vectors=[embed_one(piece)], wait=True,
                                     payload=[{"doc_id": doc_id, "text": piece}])

recreate()
try:
    load(DOCS)
except RuntimeError as err:
    print("load failed:", err, "| points:", client.count(COLL).count)
load(DOCS)                               # the scheduler retries the failed task
print("retry finished | points:", client.count(COLL).count)

q = model.encode("Can I get a deleted project back after a month?")
for hit in client.query_points(COLL, query=q.tolist(), limit=4).points:
    print(f"{hit.score:.3f}  {hit.payload['doc_id']:<9} {hit.payload['text'][:40]!r}")

Predict first: the load task dies on the eighth embedding call, and the scheduler retries it. How many points does the collection hold afterwards, and how many different chunks are in the top 4?

load failed: 429 Too Many Requests | points: 7
retry finished | points: 20
0.729  retention '## Deleted projects\nDeleted projects sta'
0.729  retention '## Deleted projects\nDeleted projects sta'
0.357  billing   '## Refunds\nYearly plans can be refunded '
0.357  billing   '## Refunds\nYearly plans can be refunded '

The crash was not the bug. A job that runs every night for years will crash, time out, hit rate limits and get retried. The bug was a job that could not survive that: one request per chunk, no retry, random ids and no memory of earlier runs. The nightly rebuild hides some of this, at a price: it re-embeds every page every night, and search sees a half-built index for the whole run (7 of 13 chunks when this one crashed). Replace the rebuild with an incremental sync, and removed sections and deleted pages become the job's problem. And nobody recorded which model made the vectors. Each of those is one move:

This lesson assumes Vector Embeddings (model contracts, batching on one machine, the migration recipe) and Classic RAG Pipeline (chunk ids, the stale-chunk fix for one page). Data Pipeline & Indexing maps all of ingestion; this lesson zooms in on embedding and indexing. Splitting documents belongs to Smart Chunking; here a heading splitter is enough.

Before any move: what must be true after every run

After every run, whatever happened during it:

1. Complete: every chunk of every current document is in the index exactly once.
2. Current: no chunk of a deleted document, or of a replaced version, is retrievable.
3. One space: every vector was made under the model contract recorded on the index, and the query side embeds under the same contract.
4. Idempotent: a second run over unchanged sources embeds nothing and writes nothing.
5. Bounded and recoverable: the run fits its time window and budget, and whatever a crash leaves half-done, the next run repairs.

Three identity rules, shared with Data Pipeline & Indexing and Data Freshness & Lifecycle, make that possible:

• A document id is stable across edits: the source system's id or a canonical URL. With a version, timestamp or content hash in it, an edited document can never replace its old chunks.
• A point id is deterministic, so a repeated write overwrites instead of duplicating. This lesson derives it from the document id and the chunk's position, uuid.uuid5(uuid.NAMESPACE_URL, f"{doc_id}:{chunk_index}"); a version number can be part of it too. Deriving it from the document id and a hash of the chunk text also works, and leaves an unchanged chunk's point alone when its document is edited (identical chunks within one document then need a tie-breaker). Either way, each update ends by deleting the document's points that the new version did not write. Qdrant accepts only unsigned integers and UUIDs as ids; sending "billing:0" gets a 400 error (checked on Qdrant 1.19.1).
• Versions and content hashes go in the payload, where the pipeline can compare them.

The part of ingestion this lesson builds:

source listing
  --> diff fingerprints (text + chunker settings + model) against the manifest
  --> delete brake: stop if too many documents vanished at once
  --> changed docs, one bounded slice at a time:
        chunk --> cache lookup (contract + text)
                   hit |       | miss
                       |       v
                       |   pack by tokens, embed with retries,
                       |   store each batch in the cache as it returns
                       v       v
        per doc: upsert its chunk slots (deterministic ids),
                 delete its slots past the new end,
                 then record its fingerprint in the manifest
  --> gone docs: delete all their points, then drop them from the manifest

Move 1: Count tokens, then find the binding limit

Smell: the full re-index "should take a weekend", and nobody wrote down why.

Four ceilings cap an embedding run, and the lowest one wins: the provider's tokens per minute, its requests per minute, the requests you keep in flight divided by the seconds each one takes, and, self-hosted, the GPU's tokens per second. (The vector database's write rate rarely binds once writes are batched; Advanced Qdrant covers bulk loads.) Count in tokens, with the model's own tokenizer.

Each request has limits too. OpenAI's embeddings reference (and its Python SDK's docstring) allows 8,192 tokens per input, no empty strings, 2,048 inputs and 300,000 tokens per request; Vector Embeddings lists other providers' caps. A packer that respects both request limits and sets aside inputs no request can carry:

import math
import tiktoken

enc = tiktoken.encoding_for_model("text-embedding-3-small")     # cl100k_base
MAX_INPUT, MAX_INPUTS, MAX_REQUEST = 8_192, 2_048, 300_000     # OpenAI's documented limits

def pack(texts, max_inputs=256, max_tokens=MAX_REQUEST):
    """Batches that fit both per-request limits; inputs no request can carry go to `rejected`."""
    batches, rejected, batch, used = [], [], [], 0
    for text in texts:
        n = len(enc.encode(text))
        if n == 0 or n > MAX_INPUT:
            rejected.append((text[:30], n))
            continue
        if batch and (len(batch) == max_inputs or used + n > max_tokens):
            batches.append(batch)
            batch, used = [], 0
        batch.append(text)
        used += n
    return batches + [batch] if batch else batches, rejected

chunks = [c for t in DOCS.values() for c in split_sections(t)]
words, tokens = sum(len(c.split()) for c in chunks), sum(len(enc.encode(c)) for c in chunks)
print(f"{len(chunks)} chunks: {words} words, {tokens} tokens ({tokens / words:.2f} tokens per word)")

texts = ["refund " * 349] * 5_000 + ["table row " * 4_600] * 3 + [""]   # 350, 9,201 and 0 tokens
batches, rejected = pack(texts, max_inputs=2_048)
print(len(batches), "requests;", [len(b) for b in batches], "| rejected:", rejected)

13 chunks: 256 words, 328 tokens (1.28 tokens per word)
6 requests; [857, 857, 857, 857, 857, 715] | rejected: [('table row table row table row ', 9201), ('table row table row table row ', 9201), ('table row table row table row ', 9201), ('', 0)]

On this text a word is 1.28 tokens, so a budget made from word counts runs short. With 350-token chunks the token cap binds first (857 × 350 = 299,950 tokens per request), long before the 2,048-input cap. The three 9,201-token "chunks" (a table dumped whole) and the empty string would fail any request they joined, so they go to a dead-letter list with the reason; the fix belongs in the chunker.

The maximum is a ceiling, not a target. A failed 2,048-input request re-sends 2,048 inputs, and finding a bad input inside it costs more requests (Move 2). A few hundred inputs per request keeps retries cheap; measure throughput at two or three sizes.

The arithmetic of a full run

A few lines tell you which limit binds. Request rate times seconds per request is the number of requests in flight (Little's law), which is how many workers you need:

def plan(chunks, avg_tokens, tpm, rpm, per_request, latency_s):
    """Hours for a full embedding run, and the concurrency needed to reach them."""
    tokens = chunks * avg_tokens
    requests = math.ceil(chunks / per_request)
    minutes = max(tokens / tpm, requests / rpm)            # whichever limit binds
    in_flight = requests / (minutes * 60) * latency_s      # Little's law: rate x time per request
    binding = "tokens/min" if tokens / tpm >= requests / rpm else "requests/min"
    return f"{requests:>9,} requests, {minutes / 60:5.1f} h, bound by {binding}, {in_flight:4.1f} in flight"

# Assumptions: 5M chunks of 350 tokens; limits of 1M tokens and 3,000 requests per minute
print("1 input per request:  ", plan(5_000_000, 350, 1_000_000, 3_000, 1, 0.25))
print("256 inputs per request:", plan(5_000_000, 350, 1_000_000, 3_000, 256, 2.0))
print("same, 10M tokens/min:  ", plan(5_000_000, 350, 10_000_000, 3_000, 256, 2.0))

1 input per request:   5,000,000 requests,  29.2 h, bound by tokens/min, 11.9 in flight
256 inputs per request:    19,532 requests,  29.2 h, bound by tokens/min,  0.4 in flight
same, 10M tokens/min:      19,532 requests,   2.9 h, bound by tokens/min,  3.7 in flight

The corpus (5 million chunks × 350 tokens = 1.75 billion tokens), the limits and the latencies are assumptions; use your account's limits page and time a real request. One input per request matches the 29.2 hours only with about 12 requests always in flight, and it uses 2,857 of the 3,000 requests per minute, so any other traffic on the account gets 429s; the opening's loop, with one request in flight, would take 5,000,000 × 0.25 s, about 14.5 days. 256 inputs per request leaves the token limit as the only one that matters, and one worker reaches it; more workers only collect 429s. A 10× higher token limit needs about 4 requests in flight. Batching does not make a token cheaper or raise the token limit. It makes tokens the binding constraint, reachable with one worker.

Your own GPU instead

Self-hosted, the ceiling is the GPU's tokens per second. Measure it with chunk-sized texts:

import random, time

def tokens_per_second(name, texts, device=None, batch_size=32):
    m = SentenceTransformer(name, device=device)           # device=None picks a GPU if there is one
    n_tokens = sum(len(m.tokenizer(t)["input_ids"]) for t in texts)
    m.encode(texts[:batch_size], batch_size=batch_size)     # warm-up
    start = time.perf_counter()
    m.encode(texts, batch_size=batch_size)
    return n_tokens / (time.perf_counter() - start), n_tokens / len(texts)

random.seed(0)
words = " ".join(DOCS.values()).split()
texts = [" ".join(random.choices(words, k=230)) for _ in range(1_000)]   # chunk-sized texts
for name, device in [("BAAI/bge-small-en-v1.5", None), ("BAAI/bge-small-en-v1.5", "cpu")]:
    tps, avg = tokens_per_second(name, texts, device)
    print(f"{name} on {device or 'default device'}: {avg:.0f} tokens per text, {tps:,.0f} tokens/s")

BAAI/bge-small-en-v1.5 on default device: 310 tokens per text, 41,441 tokens/s
BAAI/bge-small-en-v1.5 on cpu: 310 tokens per text, 5,502 tokens/s

That ran on an Apple M2 Max laptop (default device: its GPU, through PyTorch's MPS backend) while other heavy jobs shared the machine. Other runs on the same laptop gave about 27,700 to 89,000 tokens per second on the GPU and 3,000 to 14,300 on the CPU, and with the machine idle the block printed about 123,000 and 18,000. Load alone moved the result several times over, so measure on the machine you will use, under realistic load. At 41,000 tokens per second, the 1.75 billion tokens above take about 11.9 hours; at 123,000, about 4 hours. Model size dominates: on the same laptop at the same time, Qwen3-Embedding-0.6B (about 18 times bge-small's parameters) ran more than 20 times slower.

Price it

At OpenAI's prices in September 2026 (USD 0.02 per million tokens for text-embedding-3-small, USD 0.13 for -large), the 1.75 billion tokens cost USD 35 or USD 227.50. OpenAI's Batch API gives a "50% cost discount compared to synchronous APIs", with a completion window that "can only be set to 24h", at most 50,000 embedding inputs per batch (100 batches here), and rate limits separate from the synchronous ones: USD 17.50 or USD 113.75, for results within a day. Good for backfills; wrong for the query path or a tight freshness target. Self-hosted, the bill is GPU hours plus the engineering to run them. Storage, which recurs monthly, is sized in Vector Embeddings.

Your turn: your corpus grows to 30 million chunks of 300 tokens. Your limits are 5 million tokens and 10,000 requests per minute, and a 512-input request takes about 3 seconds. How long does a full run take, how many requests must be in flight, and does the request limit matter?

Move 2: Retry what is transient, isolate what is not

Smell: the logs are full of 429s, the run dies at the first 503, and one chunk fails its whole batch every night.

Data Pipeline & Indexing sorts failures into transient and permanent; here are the parts specific to embedding APIs. Not every error deserves the same answer:

Exponential backoff doubles the wait after each failure, up to a cap; jitter randomizes it, so workers that failed together do not come back together. Marc Brooker's Exponential Backoff And Jitter (AWS Architecture Blog, 2015) compared the variants: "the no-jitter exponential backoff approach is the clear loser". Its "full jitter" waits a random time between 0 and min(cap, base * 2 ** attempt). When the server sends Retry-After, OpenAI's rate-limit guide says to treat it as a minimum and add a small random delay.

import random, time

class APIError(Exception):
    def __init__(self, status, retry_after=None):
        super().__init__(f"HTTP {status}")
        self.status, self.retry_after = status, retry_after

RETRY = {408, 409, 429, 500, 502, 503, 504}    # transient: wait, then send the same batch again

def with_retries(call, batch, max_attempts=6, base=1.0, cap=60.0, sleep=time.sleep):
    for attempt in range(1, max_attempts + 1):
        try:
            return call(batch)
        except APIError as err:
            if err.status not in RETRY or attempt == max_attempts:
                raise
            if err.retry_after is not None:                 # the server's minimum, plus jitter
                wait = err.retry_after + random.uniform(0, 1)
            else:                                           # full jitter
                wait = random.uniform(0, min(cap, base * 2 ** (attempt - 1)))
            print(f"  attempt {attempt}: {err}, sleeping {wait:.2f} s")
            sleep(wait)

def embed_or_isolate(call, batch, dead_letters, **kw):
    """A 400 blames the batch, not the network: split it until the bad input is alone."""
    try:
        return with_retries(call, batch, **kw)
    except APIError as err:
        if err.status != 400:
            raise              # 401/403/404: halt the job; retries ran out: requeue the batch
        if len(batch) == 1:
            dead_letters.append((batch[0], str(err)))
            return [None]
        mid = len(batch) // 2
        return (embed_or_isolate(call, batch[:mid], dead_letters, **kw)
                + embed_or_isolate(call, batch[mid:], dead_letters, **kw))

script = [APIError(429, retry_after=2), APIError(503)]      # what the fake API answers first
calls = []
def fake_api(batch):
    calls.append(len(batch))
    if script:
        raise script.pop(0)
    if any(t.startswith("BAD") for t in batch):              # e.g. over the provider's token count
        raise APIError(400)
    return [f"vec:{t}" for t in batch]

random.seed(0)
dead = []
out = embed_or_isolate(fake_api, ["c0", "c1", "c2", "c3", "c4", "BAD c5", "c6", "c7"], dead,
                       sleep=lambda s: None)
print("requests sent:", len(calls), "sizes:", calls)
print("vectors:", sum(v is not None for v in out), "| dead letters:", dead)

Predict first: how many requests does this batch of 8 cost, and in what sizes?

  attempt 1: HTTP 429, sleeping 2.84 s
  attempt 2: HTTP 503, sleeping 1.52 s
requests sent: 9 sizes: [8, 8, 8, 4, 4, 2, 1, 1, 2]
vectors: 7 | dead letters: [('BAD c5', 'HTTP 400')]

When retries run out on a transient error, the provider is down, not the data: put the batch back on the queue with a delay, or dead-letter it with the reason "retries exhausted" and re-drive it automatically once the provider recovers (Data Pipeline & Indexing does the latter). Either way it must not wait for a human the way a bad input does.

The real call, and the retries you already have

With OpenAI's Python SDK the adapter is short (it needs pip install openai and an OPENAI_API_KEY; checked against openai 3.19.2 with a local mock server):

import openai
from openai import OpenAI

EMBED_MODEL = "text-embedding-3-small"      # pick a current model; it goes into the contract (Move 6)
oa = OpenAI(max_retries=0)                   # one retry layer, ours; the SDK default is 2 retries

def openai_call(batch):
    try:
        resp = oa.embeddings.create(model=EMBED_MODEL, input=batch)
    except openai.APIStatusError as err:     # 4xx/5xx, with the HTTP response attached
        ra = err.response.headers.get("retry-after", "")
        raise APIError(err.status_code, float(ra) if ra.replace(".", "", 1).isdigit() else None)
    except openai.APIConnectionError:        # network failures and timeouts
        raise APIError(408)
    return [d.embedding for d in sorted(resp.data, key=lambda d: d.index)]

⚠️ Retries multiply. By default the SDK already retries 408, 409, 429, 5xx and connection errors twice, with its own backoff, honouring Retry-After up to two minutes (read in the installed package; against the mock server a persistent 429 cost 3 requests). Wrap your 6 attempts around that and one batch can send 18 requests, and OpenAI's guide notes that "unsuccessful requests contribute to your per-minute limit". Keep one retry layer: the SDK's with no wrapper, or yours with max_retries=0. Sorting by index protects the order of the vectors for free.

Many workers: pace, don't bounce

One worker rarely needs more. Eight backfill workers sharing one account do. A small simulator (below) sends 400 batches of 40,000 tokens to a toy provider allowing 1 million tokens and 300 requests per minute over a sliding 60-second window, counting rejected requests toward the request limit:

fixed 1 s, no backoff        stuck: 25 of 400 batches after 2 h, 57,583 rejected
exponential, no jitter       400 batches in   995 s,    720 rejected
exponential, full jitter     400 batches in   998 s,    518 rejected
Retry-After + jitter         400 batches in   906 s,    244 rejected
pacer at 95% + Retry-After   400 batches in  1009 s,      0 rejected

The limit allows 25 batches a minute, so the best possible finish is just after 900 s.

• No backoff is the disaster: eight workers retrying every second send 8 requests per second against a limit of 5, rejected requests count, and the window never drains. After 25 batches, nothing succeeds for two hours: a retry storm feeding itself.
• Any exponential backoff finished within about 11% of the floor. Here jitter did not change the finish time; it cut rejections by about a quarter.
• Honouring Retry-After was fastest, with the fewest rejections: the server knows when capacity returns.
• A shared pacer (a limiter every worker asks before sending) at 95% of the token limit had zero rejections and finished about 11% later than the Retry-After rule: the price of 5% headroom and of skipping the window's initial burst.

Headroom is the point. If your query path embeds questions with the same account, a backfill using 100% of the limit makes live searches fail. Pace below the limit, keep Retry-After handling as the fallback, and alert on the rejected share, which a healthy pacer keeps near zero. When nearly every request fails, pause all workers at once (a circuit breaker); the pacer is the natural place for that switch.

import heapq, random

def simulate(wait_rule, pace=False, workers=8, batches=400, batch_tokens=40_000,
             tpm=1_000_000, rpm=300, latency=1.0, seed=1, give_up_at=7200):
    """Toy provider: TPM and RPM over a sliding 60 s window; rejected requests count toward RPM."""
    rng, sent, accepted = random.Random(seed), [], []          # sent: times; accepted: times
    todo, done, rejected, next_slot, t_end = list(range(batches)), 0, 0, 0.0, 0.0
    events = [(0.0, w, None, 0) for w in range(workers)]       # (time, worker, batch, attempt)
    while events:
        now, w, b, attempt = heapq.heappop(events)
        if now > give_up_at:
            break
        if b is None:                                          # worker is free: take a batch
            if not todo:
                continue
            b, attempt = todo.pop(), 0
            if pace and next_slot > now:                       # shared pacer: wait for a slot
                heapq.heappush(events, (next_slot, w, b, attempt))
                next_slot += batch_tokens / (0.95 * tpm / 60)
                continue
            next_slot = max(next_slot, now) + batch_tokens / (0.95 * tpm / 60)
        sent = [t for t in sent if t > now - 60] + [now]
        accepted = [t for t in accepted if t > now - 60]
        if len(sent) > rpm:
            retry_after = sent[-rpm - 1] + 60 - now
        elif (len(accepted) + 1) * batch_tokens > tpm:
            retry_after = accepted[len(accepted) + 1 - tpm // batch_tokens - 1] + 60 - now
        else:
            accepted.append(now)
            done, t_end = done + 1, now + latency
            heapq.heappush(events, (t_end, w, None, 0))
            continue
        rejected += 1
        heapq.heappush(events, (now + wait_rule(attempt + 1, retry_after, rng), w, b, attempt + 1))
    return done, t_end, rejected

RULES = {
    "fixed 1 s, no backoff":      lambda a, ra, rng: 1.0,
    "exponential, no jitter":     lambda a, ra, rng: min(60, 2 ** (a - 1)),
    "exponential, full jitter":   lambda a, ra, rng: rng.uniform(0, min(60, 2 ** (a - 1))),
    "Retry-After + jitter":       lambda a, ra, rng: ra + rng.uniform(0, 1),
}
def report(name, done, t_end, rejected):
    if done < 400:
        print(f"{name:28} stuck: {done} of 400 batches after 2 h, {rejected:,} rejected")
    else:
        print(f"{name:28} {done} batches in {t_end:5.0f} s, {rejected:6,} rejected")
for name, rule in RULES.items():
    report(name, *simulate(rule))
report("pacer at 95% + Retry-After", *simulate(RULES["Retry-After + jitter"], pace=True))

Your turn: four workers each wrap their own 5-attempt retry around an SDK that keeps its default 2 retries. The provider has a 10-minute outage. How many requests can one batch send, what does the whole fleet do to the provider when it comes back, and what do you change?

Move 3: Make every write idempotent

Smell: a re-run grows the index, and the top-k shows the same chunk twice.

Assume every batch will sometimes be processed twice (a timeout on a write that actually succeeded, a restart, a queue that redelivers) and make "twice" harmless:

1. Deterministic point ids. The same chunk slot gets the same id on every run, so a second write overwrites the first. Qdrant's docs say it plainly: "All APIs in Qdrant, including point loading, are idempotent… points with the same id will be overwritten when re-uploaded."
2. Record progress only after the write succeeded. Recorded before, a crash leaves silently missing chunks; recorded after, a harmless repeat.
3. Keep what you paid for, one slice at a time. Work through the changed documents in bounded slices and store each batch of vectors the moment it returns, so a crash loses at most the batch in flight and memory stays flat however large the run.

Here is the whole job. It extends the sync from Data Pipeline & Indexing with an embedding cache, bounded slices and a model contract. Its numbered comments mark the steps: this move is about the slices and steps (3) and (4), Move 4 about (1) and (2), Move 5 about the deletes in (3) and (5), and Move 6 about the Embedder's contract.

import json, os, sqlite3
from collections import Counter

def sha(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def point_id(doc_id, i):                    # same document slot -> same point id, on every run
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"{doc_id}:{i}"))

CHUNKER = {"splitter": "sections", "version": 1, "header": ""}   # everything that shapes chunk text

def make_chunks(text, chunker):
    return [chunker["header"] + c for c in split_sections(text)]

class Embedder:
    """Everything that changes a vector lives in the contract; the tag names that exact space."""
    def __init__(self, contract):
        self.contract = contract
        self.tag = sha(json.dumps(contract, sort_keys=True))[:12]
        self.model = SentenceTransformer(contract["model"], revision=contract["revision"])
    def documents(self, texts):             # swap in an API call with pack() + retries (Moves 1-2)
        return self.model.encode([self.contract["doc_prefix"] + t for t in texts], batch_size=64,
                                 normalize_embeddings=self.contract["normalized"])
    def query(self, text):
        return self.model.encode(self.contract["query_prefix"] + text,
                                 normalize_embeddings=self.contract["normalized"])

class Store:
    """One SQLite file: the embedding cache and a manifest per collection. Survives crashes."""
    def __init__(self, path):
        self.db = sqlite3.connect(path)
        self.db.execute("CREATE TABLE IF NOT EXISTS cache (key TEXT PRIMARY KEY, vec BLOB)")
        self.db.execute("CREATE TABLE IF NOT EXISTS manifest "
                        "(coll TEXT, doc_id TEXT, fingerprint TEXT, PRIMARY KEY (coll, doc_id))")
    def cached(self, tag, texts):
        rows = {t: self.db.execute("SELECT vec FROM cache WHERE key = ?", (sha(tag + t),)).fetchone()
                for t in texts}
        return {t: np.frombuffer(r[0], dtype=np.float32) for t, r in rows.items() if r}
    def remember(self, tag, texts, vecs):
        self.db.executemany("INSERT OR REPLACE INTO cache VALUES (?, ?)",
                            [(sha(tag + t), v.astype(np.float32).tobytes()) for t, v in zip(texts, vecs)])
        self.db.commit()
    def manifest(self, coll):
        return dict(self.db.execute("SELECT doc_id, fingerprint FROM manifest WHERE coll = ?", (coll,)))
    def mark(self, coll, doc_id, fp):                   # fp None: the document is gone
        if fp is None:
            self.db.execute("DELETE FROM manifest WHERE coll = ? AND doc_id = ?", (coll, doc_id))
        else:
            self.db.execute("INSERT OR REPLACE INTO manifest VALUES (?, ?, ?)", (coll, doc_id, fp))
        self.db.commit()

def fingerprint(text, chunker, emb):
    """Changes whenever the document's chunks or vectors would: its text, the chunker, the model."""
    return sha(json.dumps([sha(text), chunker, emb.tag], sort_keys=True))

def diff(source, known, chunker, emb):
    """Documents to (re)index, and documents that vanished from the source."""
    changed = [d for d, text in source.items() if known.get(d) != fingerprint(text, chunker, emb)]
    return changed, [d for d in known if d not in source]

def doc_points(doc_id, from_index=0):
    return models.FilterSelector(filter=models.Filter(must=[
        models.FieldCondition(key="doc_id", match=models.MatchValue(value=doc_id)),
        models.FieldCondition(key="chunk_index", range=models.Range(gte=from_index))]))

def sync(source, coll, emb, store, chunker=CHUNKER, slice_docs=500, batch=64,
         crash_after=None, max_delete_share=0.2):
    stats = Counter(dict.fromkeys(["docs changed", "docs deleted", "chunks embedded",
                                   "chunks from cache", "points upserted"], 0))
    known = store.manifest(coll)
    changed, gone = diff(source, known, chunker, emb)             # (1) diff
    if len(gone) > max_delete_share * max(len(known), 1):       # an empty listing is not a purge
        raise RuntimeError(f"refusing to delete {len(gone)} of {len(known)} documents")
    stats["docs changed"], written = len(changed), 0
    for s in range(0, len(changed), slice_docs):                  # a bounded slice of documents
        chunks = {d: make_chunks(source[d], chunker) for d in changed[s:s + slice_docs]}
        needed = sorted({c for cs in chunks.values() for c in cs})
        vecs = store.cached(emb.tag, needed)                      # (2) embed only cache misses,
        missing = [t for t in needed if t not in vecs]
        for b in range(0, len(missing), batch):                   #     caching each batch on return
            store.remember(emb.tag, missing[b:b + batch], emb.documents(missing[b:b + batch]))
        vecs.update(store.cached(emb.tag, missing))
        stats["chunks embedded"] += len(missing)
        stats["chunks from cache"] += len(needed) - len(missing)
        for doc_id, cs in chunks.items():                         # (3) write, one document at a time
            if written == crash_after:
                raise RuntimeError("worker killed")               # test hook
            client.upsert(coll, wait=True, points=[models.PointStruct(
                id=point_id(doc_id, i), vector=vecs[c].tolist(),
                payload={"doc_id": doc_id, "chunk_index": i, "text": c, "content_hash": sha(c),
                         "doc_hash": sha(source[doc_id]), "embed": emb.tag}) for i, c in enumerate(cs)])
            client.delete(coll, points_selector=doc_points(doc_id, from_index=len(cs)), wait=True)
            store.mark(coll, doc_id, fingerprint(source[doc_id], chunker, emb))  # (4) after both writes
            stats["points upserted"] += len(cs)
            written += 1
    for doc_id in gone:                                           # (5) documents that disappeared
        client.delete(coll, points_selector=doc_points(doc_id), wait=True)
        store.mark(coll, doc_id, None)
        stats["docs deleted"] += 1
    return dict(stats)

Now the opening's incident again, with the same crash:

MINI = {"model": "sentence-transformers/all-MiniLM-L6-v2",
        "revision": "1110a243fdf4706b3f48f1d95db1a4f5529b4d41",       # pinned commit
        "dim": 384, "normalized": True, "doc_prefix": "", "query_prefix": ""}
mini = Embedder(MINI)

def create(coll, emb):
    if client.collection_exists(coll):
        client.delete_collection(coll)
    client.create_collection(coll, metadata={**emb.contract, "tag": emb.tag},
        vectors_config=models.VectorParams(size=emb.contract["dim"], distance=models.Distance.COSINE))
    client.create_payload_index(coll, "doc_id", models.PayloadSchemaType.KEYWORD)     # before
    client.create_payload_index(coll, "chunk_index", models.PayloadSchemaType.INTEGER)  # loading

if os.path.exists("pipeline.db"):
    os.remove("pipeline.db")
store = Store("pipeline.db")
create("help_v1", mini)
try:
    sync(DOCS, "help_v1", mini, store, slice_docs=2, crash_after=3)   # tiny slices, to see them
except RuntimeError as err:
    print("run 1:", err, "| points:", client.count("help_v1").count)
print("run 2:", sync(DOCS, "help_v1", mini, store), "| points:", client.count("help_v1").count)
print("run 3:", sync(DOCS, "help_v1", mini, store), "| points:", client.count("help_v1").count)

run 1: worker killed | points: 7
run 2: {'docs changed': 3, 'docs deleted': 0, 'chunks embedded': 4, 'chunks from cache': 2, 'points upserted': 6} | points: 13
run 3: {'docs changed': 0, 'docs deleted': 0, 'chunks embedded': 0, 'chunks from cache': 0, 'points upserted': 0} | points: 13

Run 1 died after three documents, like the opening job, halfway through its second slice (billing and export). Run 2 found those three documents recorded and re-used the two export vectors paid for just before the crash, because each batch went into the cache as it returned. It embedded only the four chunks of the two documents run 1 never reached, and wrote the last three documents. Run 3 had nothing to do. Thirteen points, not twenty. Delivery is still at least once: a crash between a document's writes and its manifest update means one harmless repeat.

The slices keep a run of any size small in memory: only one slice's texts and vectors are held at a time. Two documents per slice only makes them visible here; use hundreds or thousands, enough to fill several request batches. sync still keeps the source listing and the manifest in dicts; at tens of millions of documents, stream both sorted by doc_id and merge them instead, and let each slice fetch its own texts.

Two details in create matter at scale. The payload indexes exist before any point is loaded, so delete filters stay cheap and filtered search keeps its recall (Advanced Qdrant explains why order matters). And wait=True returns only after Qdrant has applied the write. wait=False is faster, but after a crash the manifest can claim writes that never landed; then the audit in "Watch the pipeline" is what catches the gap.

Idempotent is not the same as order-safe

Two workers, or a late retry, can deliver an older version after a newer one, and a plain upsert lets it win. Qdrant 1.16 and later can make a write conditional: update_filter lets it through only if the stored point matches.

def guarded_upsert(coll, point, version):
    """Overwrite only an older copy, so a late retry cannot roll a chunk back."""
    point.payload["version"] = version
    client.upsert(coll, points=[point], wait=True, update_filter=models.Filter(must=[
        models.FieldCondition(key="version", range=models.Range(lt=version))]))

create("guard_demo", mini)
vec = mini.query("rate limits").tolist()
for version, text in [(7, "v7: 100 requests per minute"), (6, "v6, a late retry: 60 per minute")]:
    guarded_upsert("guard_demo", models.PointStruct(id=1, vector=vec, payload={"text": text}), version)
    print(f"after writing v{version}:", client.retrieve("guard_demo", [1])[0].payload["text"])
client.delete_collection("guard_demo")

after writing v7: v7: 100 requests per minute
after writing v6: v7: 100 requests per minute

The late v6 was ignored. Two more behaviours, both checked on Qdrant 1.19.1, limit the guard:

• A stored point without a version field never matches the condition, so a guarded write to it is silently skipped. Backfill the field before turning the guard on.
• A point that does not exist is inserted regardless of the filter (as the docs say), so a late write after a delete brings the chunk back. The same page offers update_mode=models.UpdateMode.UPDATE_ONLY (Qdrant 1.17+), which never inserts, so that late write is dropped. So is the first write of every new document or new chunk slot, which makes it fit only writes known to update an existing point.

The structural fix covers both: route all work for one doc_id through one queue partition, so its updates apply in order. The guard is the second line of defence.

Your turn: a teammate's version of the job saves last_doc_id as its checkpoint before writing each document, "so a restart doesn't redo work". It crashes during a write. What does the index look like after the restart, and how would you notice?

Move 4: Embed only what changed

Smell: the nightly run costs the same whether 3 pages changed or 3,000.

sync skips work at two levels:

• Documents. The manifest stores each document's fingerprint: a hash of its text, the chunker settings (splitter, version, header template) and the model contract's tag. Everything that decides the document's chunks and vectors is in it. This lesson's pages carry no permissions, but a real source's access-control fields go into the fingerprint too, or a permission change never reaches the index (Data Pipeline & Indexing shows the leak). A permission-only change still needs no embedding: when only the ACL differs (the doc_hash on its points, the chunker and the model tag are unchanged), write the new ACL to all of the document's points with set_payload (Data Freshness & Lifecycle). A document with an unchanged fingerprint (including one "saved" with no change) is skipped before chunking.
• Chunks. The cache key is sha256(contract tag + exact chunk text). It hits when an edited document keeps most sections, when boilerplate repeats across documents, when a chunker change leaves a chunk's text as it was, and when you rebuild a collection with the same model.

Watch both levels work when the chunker changes. Add a header line to every chunk, run again the next night, then roll the header back:

HEADED = {**CHUNKER, "version": 2, "header": "Source: help center\n"}
print("header added:", sync(DOCS, "help_v1", mini, store, chunker=HEADED))
print("next night:  ", sync(DOCS, "help_v1", mini, store, chunker=HEADED))
print("rolled back: ", sync(DOCS, "help_v1", mini, store))

header added: {'docs changed': 6, 'docs deleted': 0, 'chunks embedded': 13, 'chunks from cache': 0, 'points upserted': 13}
next night:   {'docs changed': 0, 'docs deleted': 0, 'chunks embedded': 0, 'chunks from cache': 0, 'points upserted': 0}
rolled back:  {'docs changed': 6, 'docs deleted': 0, 'chunks embedded': 0, 'chunks from cache': 13, 'points upserted': 13}

The header changed every chunk's text, so the first run re-chunked all six documents and embedded all 13 chunks: a chunker or header change costs one full re-embed, so price it before you merge it. The next night was back to zero. The rollback re-chunked all six documents again but embedded nothing, because the header-less texts were still in the cache; it only rewrote the 13 points. Had the fingerprint hashed the text alone, the header change would have done nothing at all that night, and only documents edited later would ever have received the header: an index mixing two chunkers for good, with no error. The model tag is there for the same reason: a manifest entry must never call a document current when it was indexed under another contract.

Rolling a chunker change through in place, as here, suits small changes such as a header or a cleanup fix: the embedding space stays the same, documents switch over during one run, and rolling back is one more run that the cache makes cheap. A new chunking strategy is a bigger decision: prove it on labeled queries in a second collection and switch with an alias, like a model change (Move 7; Advanced Qdrant recommends the same).

Two rules decide whether all this skipping helps or hurts.

Put everything that changes the vector in the key, and nothing else. The tag hashes the whole contract: model, revision, dimensions, normalization, prefixes. Leave the model out and, after an upgrade, the cache quietly serves old-model vectors into the new index: the same-dimension mix from Vector Embeddings, made by your own cache. Put a timestamp or document id in the key and identical text never hits.

Keep run-specific values out of the text and the chunker settings. An "indexed at" date, a run id or unordered iteration in a chunk's text makes every chunk of every changed document miss the cache. Put the same value into the chunker settings (a header rendered with today's date when the job starts) and every fingerprint changes too: every document looks changed on every run, and everything is re-embedded every night. Run-specific values belong in the payload.

Where an edit lands decides which chunk texts change:

def fixed_windows(text, size=40):
    words = text.split()
    return [" ".join(words[i:i + size]) for i in range(0, len(words), size)]

page = "\n".join(DOCS.values())                      # the whole help center as one long page
edited = page.replace("from Settings > Trash.", "from Settings > Trash. Admins can restore them too.")
for name, split in [("fixed 40-word windows", fixed_windows), ("one chunk per section", split_sections)]:
    before, after = {sha(c) for c in split(page)}, split(edited)
    misses = sum(sha(c) not in before for c in after)
    print(f"{name:22} {misses:2} of {len(after)} chunks need a new embedding")

fixed 40-word windows   7 of 7 chunks need a new embedding
one chunk per section   1 of 13 chunks need a new embedding

One sentence added near the top shifts every later fixed-window boundary by five words, so every window changes and misses the cache. Section boundaries stay at the headings, so only the edited section changes. Chunkers whose boundaries follow the document's structure keep the cache warm; Smart Chunking covers the choice itself.

Finding out what changed

The manifest diff lists the whole source every run. Cheaper signals each miss something: a "modified since" timestamp query never reports deletes (and trips on clock skew and imports that keep old timestamps); events or change-data capture are fast but can be lost in an outage. A common combination is events for speed plus a periodic full diff as the safety net. Data Freshness & Lifecycle covers change detection in depth.

When an LLM writes part of the chunk

Context Augmentation describes contextual enrichment at index time: an LLM writes a 50–100-token context situating each chunk in its document, embedded with the chunk. Here it costs three things. More tokens: on 350-token chunks, 14–29% more to embed, and longer payloads and lexical entries. Whole-document churn: the LLM reads the whole document, so an edit rewrites every chunk's context, and every chunk of that document misses the cache, even those whose own text did not change. Non-determinism: the same prompt can return different text. The enrichment prompt and model shape the chunk text, so they belong in the chunker settings, and therefore in the fingerprint. Store each generated context keyed by the chunk's and the document's hash, so unchanged documents never call the LLM again, and decide on purpose whether unchanged chunks of an edited document keep their old context (cheaper, slightly stale). Measure that choice on labeled queries.

Cache size. Cached vectors are as large as the raw vectors: 5 million chunks × 1,024 dimensions × 4 bytes = 20.48 GB, plus about 0.32 GB of 64-character keys. Disk is cheap next to re-embedding; drop a model's entries when it is retired. Other cache layers are in Caching Strategies.

Your turn: your index holds 5 million chunks from 1 million documents, and about 3,000 documents change per night. You release chunker version 2, which passed your labeled-query check, changes the text of about 10% of the chunks (the ones with tables) and leaves the rest byte-identical. What do the first two nights after the release re-chunk, embed and write? What would they do if the fingerprint hashed only each document's text?

Move 5: Sync, don't append

Smell: a section removed from a page still answers questions, and a retired page still shows up.

Classic RAG Pipeline showed the fix for one page: upsert its new chunks, then delete its points not in the new set. sync does that for every changed document in step (3), and removes vanished documents in step (5). Now the billing page loses its Invoices section and gains a sentence in Plans, and the export page is retired:

source = dict(DOCS)
source["billing"] = ("## Plans\nThere are three plans: Free, Team and Business, billed per member. "
                     "Yearly billing saves two months.\n"
                     "## Refunds\nYearly plans can be refunded within 14 days of purchase. "
                     "Monthly plans are not refunded.")          # Invoices section removed
del source["export"]                                             # the export page was retired
print(sync(source, "help_v1", mini, store), "| points:", client.count("help_v1").count)
for hit in client.query_points("help_v1", query=mini.query("How do I download my invoices?").tolist(),
                               limit=2).points:
    print(f"{hit.score:.3f}  {hit.payload['doc_id']}#{hit.payload['chunk_index']}  {hit.payload['text'][:28]!r}")
try:
    sync({}, "help_v1", mini, store)                             # the connector returned nothing
except RuntimeError as err:
    print(err)

Predict first: how many chunks are embedded, how many come from the cache, and how many points are left?

{'docs changed': 1, 'docs deleted': 1, 'chunks embedded': 1, 'chunks from cache': 1, 'points upserted': 2} | points: 10
0.198  billing#0  '## Plans\nThere are three pla'
0.081  api#0  '## API keys\nCreate API keys '
refusing to delete 5 of 5 documents

The last line is the delete brake. A connector that fails authentication and returns an empty listing looks exactly like "every document was deleted", and a faithful sync would empty the index. Refusing to delete more than a set share per run (20% here; set it from your normal churn) turns that outage into an alert, and a real mass deletion into a human decision.

Order matters in step (3): upserting first and deleting the tail afterwards means a document is never missing. Deleting all its points first and then upserting also ends in the right state, but leaves the document unsearchable between the two writes, so write first.

Hybrid retrieval is the same sync. A lexical or sparse index next to the dense one must be updated by the same job, or deleted chunks come back through the other leg. In Qdrant, put the sparse vector on the same point so one upsert or delete covers both; with two engines, write both under the same chunk ids and reconcile them (Hybrid Retrieval Systems).

Run often enough for your freshness target. In Hybrid Retrieval Systems' example of a nightly rebuild at 02:00, a chunk deleted at 10:00 stays retrievable for 16 hours. An incremental sync every 5 minutes cuts that to about 5 minutes plus the run time, for a listing (or an event feed) every 5 minutes. Deleting data everywhere else it lives belongs to Data Freshness & Lifecycle, along with freshness targets. That includes this pipeline's own embedding cache: embeddings can be partly inverted back to text (Vector Embeddings cites the evidence), so a deletion that must be complete removes the document's cache entries too.

Your turn: the billing page is split into two new pages, plans and refunds, with new source ids and the same section texts. What does the next sync embed, write and delete?

Move 6: Stamp every vector with its space

Smell: someone asks which model built the production index, and nobody can say for sure.

Vector Embeddings showed why it matters: a query embedded by another model of the same dimension still returns results, just worse ones, with no error. So make the contract visible where each side can check it. The pipeline records it in three places:

• the collection's metadata, set in create (Qdrant 1.16 and later supports collection metadata);
• a tag on every point (embed in the payload), so you can count vectors from any other space;
• the cache key (Move 4).

The contract holds the model id and its pinned revision (SentenceTransformer(name, revision=...) loads exactly that commit; for an API model, record the name and the date you validated it), plus dimensions, normalization and both prefixes. The chunker settings are not part of the contract: they change the texts, not the space, so they stay out of the cache key. They are recorded per document instead, in the fingerprint (Move 4), so a chunker change re-chunks every document on the next run. As the Vector Embeddings recipe does, also keep a readable chunker version in the payload, so the points a chunker release touched can be selected by filter. The query side checks the contract before it searches:

BGE = {"model": "BAAI/bge-small-en-v1.5", "revision": "5c38ec7c405ec4b44b94cc5a9bb96e735b38267a",
       "dim": 384, "normalized": True, "doc_prefix": "",
       "query_prefix": "Represent this sentence for searching relevant passages: "}
bge = Embedder(BGE)

def check_contract(coll, emb):
    """Query side, at startup and after every switch: same space, or no search at all."""
    meta = client.get_collection(coll).config.metadata or {}
    if meta.get("tag") != emb.tag:
        raise RuntimeError(f"{coll} was built with {meta.get('model')} ({meta.get('tag')}), "
                           f"but this service embeds queries with {emb.contract['model']} ({emb.tag})")

check_contract("help_v1", mini)                      # passes silently
try:
    check_contract("help_v1", bge)                   # same 384 dimensions, different space
except RuntimeError as err:
    print(err)
foreign = client.count("help_v1", count_filter=models.Filter(must_not=[
    models.FieldCondition(key="embed", match=models.MatchValue(value=mini.tag))])).count
print("points from another space:", foreign)

help_v1 was built with sentence-transformers/all-MiniLM-L6-v2 (a68ba07af56c), but this service embeds queries with BAAI/bge-small-en-v1.5 (093eed7873e0)
points from another space: 0

bge-small-en-v1.5 also makes 384-dimensional vectors, so without the check nothing would complain; with it, the mistake surfaces at startup with both model names. The foreign-point count should always be 0 (give embed a keyword payload index so it stays cheap). The query prefix in BGE comes from the bge-small-en-v1.5 model card, which suggests that instruction for queries that search passages, and none for passages; it changes the query vector, so it belongs in the contract.

Move 7: Re-embed at scale, then switch

Smell: a better model has been approved. The index holds 20 million chunks, pages change every minute, and search cannot be down for a day.

Vector Embeddings gives the recipe: a new index beside the old one, evaluation, the query encoder and index switched together, the old index kept for rollback. Advanced Qdrant shows aliases and bulk loads. At scale four things get harder: the time, the changes made during the backfill, the proof, and the switch.

1. Size it before you promise a date

20 million chunks of 400 tokens is 8 billion tokens (the "ten times bigger" exercise in Vector Embeddings). The limits and GPU speed are assumptions to replace with yours:

While both indexes exist you also store the new vectors: at 1,024 dimensions, 20,000,000 × 1,024 × 4 B = 81.92 GB of raw float32 (245.76 GB at 3,072). The Batch API's separate limits are worth more than its discount: that backfill does not compete with the query path for tokens per minute.

2. Backfill, then catch up, with the same sync

The backfill reads the source as it was at the start (moment T0) and takes days, while pages are edited and deleted and the live sync keeps the old collection current. sync already has what a days-long run needs: a per-collection manifest, bounded slices, and a checkpoint for everything it pays for (each embedded batch goes into the cache as it returns, each written document into the manifest). So the migration needs no special code. The backfill is a first sync into an empty collection, a crash on day 3 costs one re-run that starts where it stopped, and the catch-up is an ordinary incremental sync. The same diff, without acting on it, tells you how far a collection is behind the current source:

create("help_v2", bge)
print("backfill:", sync(DOCS, "help_v2", bge, store))      # DOCS = the source as read at T0
print("behind:  ", diff(source, store.manifest("help_v2"), CHUNKER, bge))    # vs today's source
print("catch-up:", sync(source, "help_v2", bge, store))    # an ordinary incremental run
for coll, emb in [("help_v1", mini), ("help_v2", bge)]:
    print(coll, "behind:", diff(source, store.manifest(coll), CHUNKER, emb),
          "| points:", client.count(coll).count)

backfill: {'docs changed': 6, 'docs deleted': 0, 'chunks embedded': 13, 'chunks from cache': 0, 'points upserted': 13}
behind:   (['billing'], ['export'])
catch-up: {'docs changed': 1, 'docs deleted': 1, 'chunks embedded': 1, 'chunks from cache': 1, 'points upserted': 2}
help_v1 behind: ([], []) | points: 10
help_v2 behind: ([], []) | points: 10

After the backfill, the new collection still had the old billing page and the retired export page; switching then would bring a deleted page back. The catch-up fixed exactly the two documents that changed after T0, and both collections now match the current source. On a real corpus, repeat the catch-up until it keeps up, then run both syncs every cycle until the old collection is dropped.

Dual-writes are safe only with one writer per document. The recipe in Vector Embeddings catches up before the switch and allows sending new and changed documents to both indexes instead, provided they share one per-document queue with the re-embedding. Here is why that condition matters:

• Safe when each document still has one writer: live changes and backfill work for the same doc_id go through one queue partition, and the backfill reads each document's current version when its turn comes. Nothing older can then land after something newer, and a document deleted before its turn is simply skipped.
• Risky when a days-long backfill replays a T0 snapshot with plain upserts while live changes are written beside it. The backfill overwrites fresh edits with its older copies and re-creates pages deleted after T0, and nothing reports either. Insert-only backfill writes (update_mode=models.UpdateMode.INSERT_ONLY, Qdrant 1.17+, which Qdrant's docs suggest for exactly this conflict) stop the overwrites but not the resurrections: a missing point still gets inserted (both checked on Qdrant 1.19.1).

Either way, compare the new collection with the current source before the switch. In the risky setup, compare what the index itself holds (each document's stored doc_hash and chunk count), not only a manifest: two writers can leave a manifest entry that says "current" above points the backfill overwrote. Backfill-then-catch-up avoids all of this. Each collection has one writer at a time, so its manifest stays true and there is nothing to race.

3. Prove it before you switch

• Parity: an empty diff for both collections against the same source listing, equal per-document chunk counts (the audit in "Watch the pipeline"), and zero points from another space.
• Vector sanity: the right dimension, no NaN or infinite values, and norms of 1 if the model normalizes.
• Quality: run the labeled queries against (old model, old collection) and (new model, new collection), per slice (language, product area, query type), comparing recall@k: relevant chunks in the top k over all relevant chunks, averaged over queries. Retrieval Metrics covers metrics and significance. Check p95 query latency with the new model too.

4. Switch the encoder and the index in one step

def point_alias(alias, coll):
    ops = [models.CreateAliasOperation(create_alias=models.CreateAlias(collection_name=coll, alias_name=alias))]
    if any(a.alias_name == alias for a in client.get_aliases().aliases):
        ops.insert(0, models.DeleteAliasOperation(delete_alias=models.DeleteAlias(alias_name=alias)))
    client.update_collection_aliases(change_aliases_operations=ops)   # one atomic change

point_alias("help", "help_v1")
point_alias("help", "help_v2")                     # the switch
try:
    check_contract("help", mini)                   # a query service that was not redeployed
except RuntimeError as err:
    print(err)
check_contract("help", bge)
hit = client.query_points("help", query=bge.query("Can I restore a deleted project?").tolist(), limit=1).points[0]
print(f"{hit.score:.3f}  {hit.payload['doc_id']}#{hit.payload['chunk_index']}")

help was built with BAAI/bge-small-en-v1.5 (093eed7873e0), but this service embeds queries with sentence-transformers/all-MiniLM-L6-v2 (a68ba07af56c)
0.777  retention#0

The alias moved atomically, but an alias only switches the index. A query service still on the old model now fails its contract check loudly instead of degrading quietly. In practice, give the query service one configuration value naming both collection and encoder and flip it in one deployment, or have it read the contract of whatever the alias points to and load the matching encoder. The alias still serves readers that don't embed queries, and rollback is one more atomic change.

5. Keep the old index current until you drop it

Keep syncing the old collection through the rollback window: a rollback to an index that stopped updating last Tuesday brings back last Tuesday's pages, deleted ones included. When the window closes, drop the old collection, its sync and its cache entries.

T0                backfill done     caught up                     switch          window ends
|-- backfill v2 --|-- catch-up v2 --|-- sync both; parity, eval --|-- sync both --|-> drop v1
|                     queries: model A on v1                      | model B on v2 |

Your turn: your account allows 2 million tokens per minute for the embedding model, and live query embedding peaks at 300,000 tokens per minute on the same account. How long does the 8-billion-token backfill take if it leaves room for that peak, and what would the Batch API change?

Watch the pipeline

A pipeline fails quietly: nothing crashes, the index just drifts from the source. The counters sync returns are the first signal; the second is a per-document audit of index against source. Qdrant's facet API counts points per payload value (on a server it needs a keyword index on the field):

def audit(coll, source):
    """Documents whose chunk count in the index differs from what the source produces."""
    got = {h.value: h.count for h in client.facet(coll, key="doc_id", limit=100_000, exact=True).hits}
    want = {d: len(make_chunks(t, CHUNKER)) for d, t in source.items()}
    return {d: (got.get(d, 0), want.get(d, 0)) for d in got.keys() | want.keys()
            if got.get(d, 0) != want.get(d, 0)}

print("help_v2:", audit("help_v2", source))
client.create_payload_index("help_naive", "doc_id", models.PayloadSchemaType.KEYWORD)
print("help_naive:", dict(sorted(audit("help_naive", DOCS).items())))

help_v2: {}
help_naive: {'billing': (6, 3), 'retention': (4, 2), 'security': (4, 2)}

The opening's collection fails on exactly the three documents written twice. On millions of documents, audit in slices (facet_filter) or a random sample each night.

When you are done with the examples, remove them:

client.update_collection_aliases(change_aliases_operations=[
    models.DeleteAliasOperation(delete_alias=models.DeleteAlias(alias_name="help"))])
for coll in ("help_naive", "help_v1", "help_v2"):
    client.delete_collection(coll)
os.remove("pipeline.db")

Final round: no label on the incident

Real incidents don't say which move they need. For each one, name the likely cause, the first check, and the fix.

1. The nightly run used to embed about 40,000 chunks. The night a teammate added a header line to every chunk, "Source: help center, indexed 2026-09-25", it embedded all 8 million, and it has done so every night since. Every run now reports every document as changed.
2. The day after a model migration, a page deleted two weeks ago is back in search results. The team backfilled from a nightly export of the source and switched when the backfill finished.
3. Eight backfill workers hit a burst of 429s at 09:00. By 09:05 the job has done less than one worker did alone, and the provider's dashboard shows more rejected requests than accepted ones.
4. The collection holds 4% more points than the sum of the documents' chunk counts. The audit shows the extra points only on documents that got shorter recently.
5. One long page has been missing from search for a month. Every night the dead-letter list contains the same entry: "HTTP 400: input too long".

Cheat sheet: symptom → move → cost

Before moving on, write down for your own corpus: chunk count and average length in the model's tokens; the limit that binds a full run, and its duration; what your pipeline does on a 429, a 400 and a 401; the point id scheme and when progress is recorded; the cache key and what goes into each document's fingerprint; what happens to a document that shrinks or disappears; where the model contract lives; and the steps, duration and cost of your next migration.

How would you measure it on your own data? Sync twice over an unchanged source: the second run must embed and write nothing. Edit five documents, delete two, add one: the counters and the audit must show exactly that. Kill the job halfway and re-run: the audit must come back empty, and the re-run must embed only what the first run never reached. Bump the chunker version: the next run must re-chunk every document once and embed only chunks whose text changed. Before any migration, run parity and your labeled queries on both collections, per slice.

Next: Data Freshness & Lifecycle for change detection, freshness targets and deletes beyond the index; Smart Chunking for the splitter this pipeline calls; Caching Strategies for the other cache layers; and Production Deployment & Optimization for rolling out index versions behind evaluation gates.