Pipelines for AI: keeping a RAG index in sync

A vector index is a copy of your documents, and copies go stale. This section is the ingestion job that keeps one honest as the source changes: working out which documents actually changed, replacing their chunks, removing the ones that have been deleted, and keeping permissions attached to every chunk. Deletion is the half that is usually missing, and it is the half that has an assistant confidently citing a policy withdrawn last month to someone who was never allowed to read it.

Only process what changed#

import hashlib

def sync(source_docs, index):
    seen = set()
    for doc in source_docs:                       # id, text, updated_at, acl
        seen.add(doc.id)
        digest = hashlib.sha256(doc.text.encode()).hexdigest()
        if index.stored_hash(doc.id) == digest:
            continue                              # unchanged: skip parsing and embedding
        chunks = chunk(parse(doc))
        index.replace_document(doc.id, embed(chunks), digest=digest, acl=doc.acl)

    for stale_id in index.document_ids() - seen:  # gone from the source
        index.delete_document(stale_id)

The hash is what makes this affordable. Embedding is the expensive step, and on a corpus of any size almost nothing changes between runs, so skipping unchanged documents takes a nightly sync from hours to minutes. Hash the content you actually embed, after parsing and normalising, not the raw bytes: a PDF re-exported with a new timestamp has different bytes and identical text, and hashing the bytes re-embeds the whole corpus every time someone touches the export job.

import hashlib

def fingerprint(text, acl, chunk_size, model):
    """Everything that, if changed, means the stored chunks are wrong."""
    parts = [text, ",".join(sorted(acl)), str(chunk_size), model]
    return hashlib.sha256("|".join(parts).encode()).hexdigest()

doc = dict(text="Refunds are issued within 14 days.", acl=["support", "billing"])
a = fingerprint(doc["text"], doc["acl"], 512, "embed-v3")
b = fingerprint(doc["text"], ["billing", "support"], 512, "embed-v3")
c = fingerprint(doc["text"], ["billing"], 512, "embed-v3")
d = fingerprint(doc["text"], doc["acl"], 256, "embed-v3")
print(a == b, a == c, a == d)      # True False False

Replacing a document means deleting its chunks first#

A document that shrinks from twelve chunks to eight leaves four orphans behind if you only overwrite by chunk id. They are still in the index, still retrievable, and they contain text that was removed from the source, which is the same incident as failing to delete a document, arrived at more subtly. Delete every chunk belonging to the document, then insert the new set.

def replace_document(index, doc_id, chunks, digest, acl):
    index.delete_where(doc_id=doc_id)              # all of them, by document
    for i, chunk in enumerate(chunks):
        index.insert(
            chunk_id=f"{doc_id}:{i}",
            vector=embed(chunk),
            text=chunk,
            doc_id=doc_id,                          # the key deletion needs
            digest=digest,
            acl=acl,
        )

Deletions#

If a document is deleted or unpublished at the source but its chunks stay in the index, the assistant keeps quoting it. For a withdrawn policy or a document a user lost access to, that is a real incident. Every sync must compare the full set of source ids with the index and remove the difference.

The trap in that set difference is a partial source listing. If the crawl fails halfway and returns two hundred of two thousand documents, the difference is eighteen hundred documents and the sync cheerfully empties the index. Guard it: require the listing to have completed, and refuse to delete more than a small share of the index in one run without a human saying so.

def safe_delete(index_ids, source_ids, listing_complete, max_share=0.05):
    if not listing_complete:
        raise RuntimeError("source listing incomplete, refusing to delete")
    missing = index_ids - source_ids
    if len(missing) > max_share * len(index_ids):
        raise RuntimeError(
            f"{len(missing)} of {len(index_ids)} documents missing, needs review"
        )
    return missing

print(safe_delete({"a", "b", "c", "d"}, {"a", "b", "c"}, True, max_share=0.5))  # {'d'}
try:
    safe_delete({"a", "b", "c", "d"}, {"a"}, True, max_share=0.5)
except RuntimeError as e:
    print(e)                          # 3 of 4 documents missing, needs review

When parsing fails#

Some documents will not parse cleanly: scanned pages, broken encodings, huge tables. Do not index garbage. Send them to a quarantine list with the error, alert when it grows, and review it regularly. The reason to alert on the size rather than on each failure is that the interesting signal is a change: three files that never parse are a known annoyance, and three hundred means the source system changed its export format last night.

def sync_one(doc, index, quarantine):
    try:
        chunks = chunk(parse(doc))
    except Exception as e:
        quarantine[doc.id] = f"{type(e).__name__}: {e}"
        return "quarantined"
    if not chunks or sum(len(c) for c in chunks) < 40:
        quarantine[doc.id] = "parsed to nothing, probably a scan"
        return "quarantined"
    replace_document(index, doc.id, chunks, doc.digest, doc.acl)
    return "indexed"

Permissions travel with the chunk#

Store who may see each document on every chunk, and filter on it at query time. When permissions change at the source, the sync must update them even if the text did not change, which is why the permission list is inside the fingerprint above rather than beside it.

Filter before the search, not after. Retrieving the top twenty and then discarding the ones this user may not see returns four results for one person and twenty for another, and on a document set where most of it is restricted it returns nothing at all while the index plainly contains the answer. Every serious vector store takes a metadata filter that is applied during the search; use it, and test it with an account that should see almost nothing.

New embedding model#

Changing the embedding model means re-embedding everything into a new index, comparing retrieval quality, and switching over, as with any migration. Store the model name with each vector so a mixed index can never happen silently. A mixed index does not error; it returns nearest neighbours in a space where half the points were placed by a different function, and the results are plausible and wrong.

1  build a second index with the new model, from the same source
2  replay a fixed set of real queries against both
3  compare recall at 10 on the same judgements, and eyeball the disagreements
4  switch reads to the new index, keep the old one for a week
5  delete the old index, once nobody has asked for a rollback

Cost it before you agree to it. A million chunks at roughly five hundred tokens each is five hundred million tokens through an embedding model, plus the storage for a second full index while both exist, plus whatever the re-crawl costs at the source. None of that is prohibitive and all of it is a surprise if the first estimate appears after the migration has started.