Handling Updates

Documents change. The return policy gets updated. The product spec adds a new feature. A blog post gets corrected. If your RAG system does not reflect these changes, it serves outdated information to users who trust it. This lesson covers strategies for updating, deleting, and versioning indexed documents so your retrieval stays current without requiring a full rebuild every time something changes.

What you'll learn

• Upsert (INSERT ON CONFLICT UPDATE) replaces old embeddings with new ones in a single atomic operation
• Soft delete with a status column avoids losing data while keeping queries clean
• Cache invalidation on update ensures users see the new content immediately, not the cached old version

The problem

Your documentation site has 500 pages. You update the return policy page. The old version is still in the vector store. Users search for "return policy" and get the outdated information. The naive fix is to re-index the entire 500-page corpus, which takes 5 minutes and costs money for 499 embeddings you already had. The correct fix is to update only the changed document: re-embed its new content, replace the old vector, and invalidate any cached search results that might contain the old version.

Build it

Step 1: Upsert a single document

When a document changes, do not rebuild the entire index. Use an atomic upsert operation to replace the specific embedding and update its metadata inline.

def upsert_document(doc_id, new_content, metadata=None):
    """Update or insert a document. Replaces old embedding atomically."""
    embedding = generate_embedding(new_content)

    cursor.execute("""
        INSERT INTO documents (id, content, embedding, metadata, updated_at)
        VALUES (%s, %s, %s, %s, NOW())
        ON CONFLICT (id) DO UPDATE SET
            content = EXCLUDED.content,
            embedding = EXCLUDED.embedding,
            metadata = EXCLUDED.metadata,
            updated_at = NOW(),
            version = documents.version + 1
    """, (doc_id, new_content, embedding, json.dumps(metadata or {})))

    # Invalidate any cached search results that might contain this document
    invalidate_cache_for_document(doc_id)

    return True

Step 2: Soft delete with status tracking

Avoid permanently deleting records to maintain data integrity and allow for recovery. Implement a soft delete strategy by updating a status column.

def soft_delete(doc_id):
    """Mark as deleted without removing. Recoverable and auditable."""
    cursor.execute("""
        UPDATE documents 
        SET status = 'deleted', 
            deleted_at = NOW(), 
            updated_at = NOW()
        WHERE id = %s
    """, (doc_id,))

def restore_document(doc_id):
    """Restore a soft-deleted document."""
    cursor.execute(
        "UPDATE documents SET status = 'active', deleted_at = NULL WHERE id = %s",
        (doc_id,)
    )

def hard_delete_old_deletions(days=90):
    """Permanently remove documents deleted more than N days ago."""
    cursor.execute(
        "DELETE FROM documents WHERE status = 'deleted' AND deleted_at < NOW() - INTERVAL '%s days'",
        (days,)
    )

Step 3: Version tracking for rollback

Version control for knowledge bases enables recovering from bad updates. This logic fetches historical versions and restores previous states via the upsert mechanism.

def get_document_version(doc_id):
    cursor.execute("SELECT version, updated_at FROM documents WHERE id = %s", (doc_id,))
    return cursor.fetchone()

def rollback_to_version(doc_id, target_version):
    """Roll back to a previous version from the audit log."""
    cursor.execute("""
        SELECT content, metadata FROM document_versions
        WHERE doc_id = %s AND version = %s
    """, (doc_id, target_version))
    row = cursor.fetchone()
    if row:
        upsert_document(doc_id, row[0], row[1])

Step 4: Cache invalidation on update

A document update is useless if users still receive the old version from memory. Ensure you invalidate any relevant cache entries whenever a record is modified.

import redis
cache = redis.Redis(decode_responses=True)

def invalidate_cache_for_document(doc_id):
    """Invalidate any cached search results that might contain this document."""
    # Pattern-based invalidation: delete all cached searches
    # In production, use a more targeted approach based on query patterns
    keys = cache.keys("search:*")
    if keys:
        cache.delete(*keys)

    # Also invalidate the embedding cache for this document
    cache.delete(f"embed:{doc_id}")

What goes wrong

Mistake	How you notice it	The fix
Full rebuild for one changed document	Re-indexing 10K docs because one changed	Always use incremental updates. Only re-embed what changed
Stale results after update	Search returns old content for minutes after update	Invalidate cache on update. Use short TTLs for frequently changing content
Orphaned embeddings after hard delete	Search returns documents that no longer exist in the source	Always filter by status = 'active' in search queries. Run periodic cleanup of old deleted documents
Concurrent updates cause race conditions	Two updates to the same document, one wins silently	Use ON CONFLICT with version incrementing. The second update sees the new version and can retry

Confirm it worked

Execute an end-to-end test of the update lifecycle. Verify that modifying a document correctly updates the index and that subsequent searches return the new content.

# Index a document, then update it
index_documents([{"id": "doc-1", "content": "Old return policy: 30 days"}])
results = vector_search("return policy")
assert "30 days" in results[0]["content"]

# Update the document
upsert_document("doc-1", "New return policy: 60 days")
results = vector_search("return policy")
assert "60 days" in results[0]["content"]
assert "30 days" not in results[0]["content"]

Next: Caching