Overview
A RAG index is only as good as its freshness. Documents change — prices update, policies revise, code evolves — and any chunk from a deleted or outdated document that remains in the vector store can be retrieved and cited, producing confident-sounding wrong answers. The failure mode is insidious: the system looks healthy on your golden test set (which uses current documents) but silently degrades as the production corpus drifts from what is actually indexed.
There are three fundamental update strategies: full re-index (delete everything and rebuild), incremental update (re-embed and replace only changed documents), and versioned indices (maintain multiple index snapshots for atomic cutover). Full re-index is the safest for correctness but expensive at scale. Incremental updates are efficient but require careful tracking of which chunks belong to which document so deletions propagate correctly. Versioned indices add operational complexity but enable zero-downtime schema migrations or embedding model upgrades.
Content hashing is the core mechanism for incremental updates: store an MD5 or SHA-256 of each document's text alongside its vector metadata. At refresh time, compute the current hash and compare it to the stored one. Skip unchanged documents entirely. This simple check reduces the re-embedding cost of a daily refresh by 80–95% for knowledge bases where only a small fraction of documents change each day.
Update Strategies
1. Full Re-index (Simplest)
Delete all vectors, re-embed all documents. Works for small datasets, inefficient for large.
2. Incremental Updates (Recommended)
Only update changed documents. Track hashes, timestamps, or version numbers.
3. Versioned Indices
Create new index for each batch of updates. Switch atomically. Keep old index for rollback.
Incremental Indexing
def update_index(documents, vector_db):
"""Update only changed documents."""
import hashlib
for doc in documents:
# Compute hash of current doc
current_hash = hashlib.md5(doc.content.encode()).hexdigest()
# Check if already indexed
stored = vector_db.get_metadata(doc.id)
if stored and stored.get('content_hash') == current_hash:
continue # No change, skip
# Document changed or new
# 1. Delete old chunks for this doc
vector_db.delete_by_metadata({"source_id": doc.id})
# 2. Chunk and embed
chunks = chunk(doc.content)
embeddings = embed_batch(chunks)
# 3. Add to DB with metadata
for chunk, emb in zip(chunks, embeddings):
vector_db.upsert(
id=f"{doc.id}_{chunk.index}",
embedding=emb,
metadata={
"source_id": doc.id,
"content_hash": current_hash,
"updated_at": time.time()
}
)
print(f"Updated {doc.id}")Handling Deletes
def delete_document(doc_id, vector_db):
"""Remove all chunks for a document."""
# Delete by metadata
vector_db.delete_by_metadata({
"source_id": doc_id
})
print(f"Deleted all chunks for {doc_id}")
# For compliance/privacy
def soft_delete(doc_id, vector_db):
"""Mark as deleted but keep history."""
vector_db.update_metadata(
ids=[f"{doc_id}_*"],
metadata={"deleted": True}
)
def purge_old_versions(doc_id, keep_versions=3):
"""Delete old versions, keep recent ones."""
versions = vector_db.get_versions(doc_id)
if len(versions) > keep_versions:
for old in versions[:-keep_versions]:
vector_db.delete_version(old)Document Versioning
class VersionedDocument:
def __init__(self, doc_id, content, version=1):
self.id = doc_id
self.content = content
self.version = version
self.created_at = time.time()
def upsert(self, vector_db):
"""Store with version tracking."""
# Chunk
chunks = chunk(self.content)
embeddings = embed_batch(chunks)
# Add with version
for chunk, emb in zip(chunks, embeddings):
vector_db.upsert(
id=f"{self.id}_v{self.version}_{chunk.index}",
embedding=emb,
metadata={
"doc_id": self.id,
"version": self.version,
"created_at": self.created_at
}
)
def rollback(self, vector_db, target_version):
"""Switch back to previous version."""
# Mark new version as deleted
vector_db.update_metadata(
filter={"doc_id": self.id, "version": self.version},
metadata={"active": False}
)
# Mark old version as active
vector_db.update_metadata(
filter={"doc_id": self.id, "version": target_version},
metadata={"active": True}
)Real-time vs Batch Updates
Real-time (Event-driven)
When doc changes, immediately update vectors. Fast, fresh data. Complex infrastructure.
# On doc change event
async def on_doc_updated(event):
doc = await fetch_document(event.doc_id)
await vector_db.update(doc) # Real-timeBatch (Time-based)
Update every hour/day. Simpler, but delayed. Good for non-critical data.
# Scheduled job (every hour)
@scheduler.scheduled_job('cron', minute=0)
async def refresh_index():
docs = await fetch_changed_docs(since=last_refresh)
await update_index(docs)Notes
Delete semantics differ across vector DBs — plan per-provider
Pinecone deletes by vector ID (you must track IDs per document). Qdrant supports filter-based deletion (delete all vectors where metadata.source_id == "doc123"). FAISS has no native delete — you must rebuild the index or use a soft-delete filter. Plan your deletion strategy based on your chosen vector store before writing ingestion code; retrofitting delete logic after the fact is painful.
Soft deletes require filtering at query time
Marking a vector as deleted with metadata ({"deleted": true}) rather than removing it is safer for compliance and rollback, but every query must now include a filter to exclude soft-deleted vectors. This adds a small overhead to every search and requires your vector DB to support metadata filtering efficiently (Qdrant and Weaviate handle this well; FAISS does not support it natively).
Real-time pipelines need retry logic and a dead-letter queue
Real-time ingestion pipelines that call the embedding API on each document change will silently drop updates when API rate limits are hit or the embedding service is temporarily unavailable. Add exponential backoff retry logic and route failed events to a dead-letter queue (SQS, Redis list) for reprocessing. Without this, your index accumulates gaps that are invisible until users report missing answers.
Full re-index serves as a consistency check, not just a fallback
Even if incremental updates are running correctly, running a full re-index monthly catches accumulated drift: vectors that were not deleted when they should have been, chunks that failed to update silently, or schema mismatches introduced by an embedding model version change. Schedule full re-indexes during off-peak hours and track the delta (vectors added, updated, deleted) to catch anomalies before they affect users.
RAG Data Refresh FAQ
How do I update a RAG vector store when documents change?
Implement upsert logic: generate the new embedding, then call the vector store's upsert API with the same document ID. The old vector is replaced without requiring a full re-index.
How do I handle document deletions in RAG?
Track document IDs in a metadata store alongside your vector DB. When a document is deleted from the source, call the vector store's delete API with its ID. Without this, stale chunks remain retrievable indefinitely.
What is a real-time RAG ingestion pipeline?
A real-time pipeline listens to document change events (webhooks, database triggers, file watchers), immediately embeds the changed content, and upserts into the vector store. It keeps the index fresh with sub-minute latency.
How do I avoid duplicate chunks during data refresh?
Use a content hash or document ID as the vector's metadata key. Before inserting, check if the hash already exists. If it matches, skip; if it differs, delete the old vectors and insert the new ones.
How often should I refresh my RAG index?
It depends on how fast your source data changes. For live databases, near-real-time ingestion is best. For document repositories that change weekly, a nightly batch job is sufficient. Static archives rarely need refreshing.