
Imagine you've built a RAG (Retrieval-Augmented Generation) system for your company's internal knowledge base. It works beautifully on day one. Users ask questions, the system fetches the right documents, and the AI generates accurate, grounded answers. Then three months pass. Policies get updated, product specs change, and two entire departments reorganize. Your vector store — the indexed collection of documents your RAG system searches through — is still serving answers based on January's reality. Users start getting confident, well-written responses that are quietly, dangerously wrong.
This is the indexing maintenance problem, and it's one of the most underestimated challenges in production RAG systems. Building your first index is the easy part. The real engineering work is keeping that index accurate, efficient, and consistent as your underlying documents evolve. Whether documents are added daily, updated weekly, or occasionally deleted, your vector store needs a strategy for staying in sync — not just a one-time build script you run and forget.
By the end of this lesson, you'll understand how vector stores work under the hood, how to build one correctly from the start, and how to implement three practical strategies for keeping it updated as your document corpus changes. You'll walk away with working Python code, an understanding of the tradeoffs between approaches, and the intuition to choose the right strategy for your situation.
What you'll learn:
Before we can talk about updating a vector store, we need to understand what we're actually building. A vector store is a database that stores documents not as text, but as embeddings — numerical representations of meaning.
Here's the intuition: when you embed the sentence "How do I reset my password?", an embedding model converts it into a list of numbers (a vector) like [0.12, -0.84, 0.33, ...] with hundreds of dimensions. The key property is that sentences with similar meaning produce vectors that are geometrically close to each other in that high-dimensional space. "I forgot my login credentials" would produce a vector very near the password-reset vector, even though the words are completely different.
When a user asks a question in your RAG system, that question gets embedded into the same vector space, and the vector store finds the stored document chunks whose vectors are nearest — a process called approximate nearest neighbor search (ANN search). Those retrieved chunks become the context your language model uses to generate an answer.
The critical implication: the vector store doesn't store your documents as text you can easily "edit." Each stored item is a chunk of text plus its embedding vector plus some metadata. Changing a document means you need to remove the old chunks, re-chunk the updated document, generate new embeddings, and insert the new chunks. This is why indexing strategy matters so much.
Let's look at what actually goes into a well-designed vector store before we worry about updating it. Every entry in your index should contain four things:
Getting this structure right at the start is what makes updates tractable later. Let's look at a concrete example using ChromaDB (a popular open-source vector store) and sentence-transformers for embeddings.
First, install the dependencies:
pip install chromadb sentence-transformers
Now let's set up a basic index with proper structure:
import chromadb
from sentence_transformers import SentenceTransformer
import hashlib
import json
from datetime import datetime
# Initialize the embedding model
embedder = SentenceTransformer("all-MiniLM-L6-v2")
# Initialize ChromaDB with persistent storage
client = chromadb.PersistentClient(path="./knowledge_base_store")
collection = client.get_or_create_collection(
name="company_knowledge_base",
metadata={"hnsw:space": "cosine"} # Use cosine similarity for text
)
def compute_document_hash(text: str) -> str:
"""Generate a fingerprint of the document's content."""
return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
def chunk_document(doc_id: str, title: str, text: str,
chunk_size: int = 400, overlap: int = 50) -> list[dict]:
"""
Split a document into overlapping chunks with full metadata.
Overlap ensures that sentences spanning chunk boundaries
don't lose their context entirely.
"""
words = text.split()
chunks = []
start = 0
chunk_index = 0
while start < len(words):
end = min(start + chunk_size, len(words))
chunk_text = " ".join(words[start:end])
chunks.append({
"chunk_id": f"{doc_id}_chunk_{chunk_index}",
"doc_id": doc_id,
"title": title,
"text": chunk_text,
"chunk_index": chunk_index,
"doc_hash": compute_document_hash(text), # Hash of full document
})
chunk_index += 1
start += chunk_size - overlap # Step forward, keeping some overlap
return chunks
Notice the doc_hash field. This is the fingerprint of the entire source document. When the document changes, this hash changes, which is how we'll detect that re-indexing is needed. The doc_id is the stable identifier for the source — it doesn't change even when the document's content does.
Chunking — splitting documents into smaller pieces — is one of the highest-leverage decisions in your entire RAG system. Chunks that are too small lose their context; chunks that are too large dilute the specific information the embedding captures, and retrieval quality drops.
The naive approach is fixed-size character chunking. Don't do this. It will split sentences mid-thought, destroying the semantic coherence the embedding model needs to work well.
Here are three better approaches, in order of sophistication:
Word-based chunking with overlap (what we implemented above): Split by word count, not character count, with an overlap window so context bleeds between adjacent chunks. Good default. Use 300–500 words per chunk, 10–15% overlap.
Sentence-aware chunking: Split at sentence boundaries, then accumulate sentences until you hit a token budget. This preserves grammatical units. Libraries like nltk or spacy can handle sentence detection:
import nltk
nltk.download("punkt", quiet=True)
from nltk.tokenize import sent_tokenize
def chunk_by_sentences(doc_id: str, title: str, text: str,
max_words_per_chunk: int = 350) -> list[dict]:
sentences = sent_tokenize(text)
chunks = []
current_chunk_sentences = []
current_word_count = 0
chunk_index = 0
doc_hash = compute_document_hash(text)
for sentence in sentences:
sentence_word_count = len(sentence.split())
# If adding this sentence exceeds our budget, save the current chunk
if current_word_count + sentence_word_count > max_words_per_chunk and current_chunk_sentences:
chunk_text = " ".join(current_chunk_sentences)
chunks.append({
"chunk_id": f"{doc_id}_chunk_{chunk_index}",
"doc_id": doc_id,
"title": title,
"text": chunk_text,
"chunk_index": chunk_index,
"doc_hash": doc_hash,
})
# Carry over the last sentence for overlap context
current_chunk_sentences = [current_chunk_sentences[-1]]
current_word_count = len(current_chunk_sentences[0].split())
chunk_index += 1
current_chunk_sentences.append(sentence)
current_word_count += sentence_word_count
# Don't forget the final chunk
if current_chunk_sentences:
chunks.append({
"chunk_id": f"{doc_id}_chunk_{chunk_index}",
"doc_id": doc_id,
"title": title,
"text": " ".join(current_chunk_sentences),
"chunk_index": chunk_index,
"doc_hash": doc_hash,
})
return chunks
Hierarchical chunking: Create both large "parent" chunks and small "child" chunks. Retrieve using small chunks (more precise semantic matching) but return the parent chunk as context to the LLM (more complete information). This is the approach used by LlamaIndex's "parent document retriever." It's powerful but more complex to implement and maintain.
Tip: For most production use cases, sentence-aware chunking with 300–400 word chunks is the sweet spot. Start there and iterate based on retrieval quality metrics.
Now let's put everything together and actually index a document collection. We'll use a realistic scenario: a company's HR policy documents stored as a dictionary (in production, these would come from a database, S3 bucket, or document management system).
def index_document(collection, doc_id: str, title: str, text: str):
"""Index a single document: chunk it, embed it, store it."""
chunks = chunk_by_sentences(doc_id, title, text)
if not chunks:
print(f"Warning: Document {doc_id} produced no chunks. Skipping.")
return
# Generate embeddings for all chunks at once (batching is faster)
chunk_texts = [c["text"] for c in chunks]
embeddings = embedder.encode(chunk_texts, show_progress_bar=False).tolist()
# Prepare data for ChromaDB
ids = [c["chunk_id"] for c in chunks]
metadatas = [
{
"doc_id": c["doc_id"],
"title": c["title"],
"chunk_index": c["chunk_index"],
"doc_hash": c["doc_hash"],
"indexed_at": datetime.utcnow().isoformat(),
}
for c in chunks
]
# Insert into the vector store
collection.add(
ids=ids,
embeddings=embeddings,
documents=chunk_texts,
metadatas=metadatas,
)
print(f"Indexed '{title}' → {len(chunks)} chunks")
# Sample document corpus
documents = {
"hr_policy_001": {
"title": "Remote Work Policy",
"text": """Employees may work remotely up to three days per week with manager approval.
Remote work arrangements must be formally documented using Form HR-44. Employees are
expected to be available during core hours of 10am to 3pm in their local timezone.
All remote work must comply with the data security guidelines outlined in IT Policy 7.
Requests for fully remote arrangements require VP-level approval and a six-month
performance review history on file..."""
},
"hr_policy_002": {
"title": "Expense Reimbursement Policy",
"text": """Business expenses up to $75 may be submitted without a receipt. Expenses
between $75 and $500 require itemized receipts submitted within 30 days of purchase.
Expenses exceeding $500 require pre-approval from a department director. All expense
reports must be submitted through the Concur portal. International travel expenses
must be converted to USD using the exchange rate on the date of purchase..."""
},
}
# Initial bulk indexing
for doc_id, doc_data in documents.items():
index_document(collection, doc_id, doc_data["title"], doc_data["text"])
print(f"\nTotal chunks in index: {collection.count()}")
The simplest update strategy is to delete everything and re-index from scratch. This sounds wasteful, and it is — but it's also perfectly correct and often the right choice.
def full_rebuild(collection, all_documents: dict):
"""
Delete the entire collection and re-index all documents.
Use when: document count is manageable and you want guaranteed consistency.
"""
# Delete all existing entries
existing_ids = collection.get()["ids"]
if existing_ids:
collection.delete(ids=existing_ids)
print(f"Cleared {len(existing_ids)} existing chunks")
# Re-index everything
for doc_id, doc_data in all_documents.items():
index_document(collection, doc_id, doc_data["title"], doc_data["text"])
print(f"Full rebuild complete. Total chunks: {collection.count()}")
When full rebuild makes sense:
When it doesn't:
This is the workhorse strategy for most production systems. The idea is simple: before re-indexing a document, check whether it has actually changed. If the hash is the same, skip it. If the hash differs, delete the old chunks and insert new ones.
This approach uses the document hash we stored in the metadata as a change fingerprint.
def get_indexed_hashes(collection) -> dict[str, str]:
"""
Return a mapping of {doc_id: doc_hash} for all currently indexed documents.
This tells us what version of each document we have in the store.
"""
results = collection.get(include=["metadatas"])
indexed = {}
for metadata in results["metadatas"]:
doc_id = metadata["doc_id"]
doc_hash = metadata["doc_hash"]
indexed[doc_id] = doc_hash # If a doc has multiple chunks, last one wins (same hash)
return indexed
def delete_document_chunks(collection, doc_id: str):
"""Remove all chunks belonging to a specific document."""
results = collection.get(
where={"doc_id": doc_id},
include=["metadatas"]
)
if results["ids"]:
collection.delete(ids=results["ids"])
print(f" Deleted {len(results['ids'])} old chunks for doc '{doc_id}'")
def incremental_upsert(collection, updated_documents: dict):
"""
Sync only the documents that have changed.
Args:
updated_documents: The current authoritative set of documents.
"""
indexed_hashes = get_indexed_hashes(collection)
stats = {"added": 0, "updated": 0, "skipped": 0, "deleted": 0}
# Process current documents: add new ones, update changed ones
for doc_id, doc_data in updated_documents.items():
current_hash = compute_document_hash(doc_data["text"])
if doc_id not in indexed_hashes:
# New document — index it fresh
index_document(collection, doc_id, doc_data["title"], doc_data["text"])
stats["added"] += 1
elif indexed_hashes[doc_id] != current_hash:
# Document has changed — delete old chunks, insert new ones
print(f"Document '{doc_id}' changed. Re-indexing...")
delete_document_chunks(collection, doc_id)
index_document(collection, doc_id, doc_data["title"], doc_data["text"])
stats["updated"] += 1
else:
# Document unchanged — skip it
stats["skipped"] += 1
# Handle deletions: documents in index that no longer exist in source
current_doc_ids = set(updated_documents.keys())
indexed_doc_ids = set(indexed_hashes.keys())
deleted_docs = indexed_doc_ids - current_doc_ids
for doc_id in deleted_docs:
print(f"Document '{doc_id}' removed from source. Deleting from index...")
delete_document_chunks(collection, doc_id)
stats["deleted"] += 1
print(f"\nSync complete: {stats}")
Let's see this in action. Suppose the expense policy gets updated:
# Simulate a document update
documents["hr_policy_002"]["text"] = """Business expenses up to $100 may be submitted
without a receipt (updated from previous $75 threshold). Expenses between $100 and $750
require itemized receipts submitted within 45 days of purchase. Expenses exceeding $750
require pre-approval from a department director..."""
# Add a new document
documents["hr_policy_003"] = {
"title": "Parental Leave Policy",
"text": """Full-time employees are eligible for 16 weeks of paid parental leave
following the birth, adoption, or foster placement of a child. Leave must be taken
within 12 months of the qualifying event..."""
}
# Run incremental sync
incremental_upsert(collection, documents)
Warning: The deletion step — removing documents that no longer exist in the source — is easy to forget and critically important. A RAG system that serves answers from deleted policies is actively dangerous. Always reconcile both directions: additions and removals.
The most sophisticated approach is to trigger re-indexing based on events rather than periodic scans. When your document management system creates, updates, or deletes a document, it emits an event that your indexing pipeline consumes.
This requires an event queue (like Kafka, AWS SQS, or even a simple database queue), but it gives you near-real-time index freshness with minimal wasted computation.
from enum import Enum
from dataclasses import dataclass
class DocumentEventType(Enum):
CREATED = "created"
UPDATED = "updated"
DELETED = "deleted"
@dataclass
class DocumentEvent:
event_type: DocumentEventType
doc_id: str
title: str = ""
text: str = ""
def process_document_event(collection, event: DocumentEvent):
"""
Handle a single document change event.
In production, this would be called by a queue consumer.
"""
if event.event_type == DocumentEventType.CREATED:
print(f"[EVENT] New document: {event.doc_id}")
index_document(collection, event.doc_id, event.title, event.text)
elif event.event_type == DocumentEventType.UPDATED:
print(f"[EVENT] Updated document: {event.doc_id}")
delete_document_chunks(collection, event.doc_id)
index_document(collection, event.doc_id, event.title, event.text)
elif event.event_type == DocumentEventType.DELETED:
print(f"[EVENT] Deleted document: {event.doc_id}")
delete_document_chunks(collection, event.doc_id)
# Simulate receiving events from a document management system
events = [
DocumentEvent(
event_type=DocumentEventType.UPDATED,
doc_id="hr_policy_001",
title="Remote Work Policy",
text="Updated remote work policy text allowing four days remote per week..."
),
DocumentEvent(
event_type=DocumentEventType.DELETED,
doc_id="hr_policy_002",
title="Expense Reimbursement Policy",
text=""
),
]
for event in events:
process_document_event(collection, event)
Event-driven indexing is the right choice when you have a well-defined system of record for your documents (SharePoint, Confluence, a CMS with webhooks, a database with change data capture). It's overkill for simpler setups, but it's the only architecture that truly scales.
Now it's your turn to put these concepts together. Work through the following exercise step by step.
Setup: Create a new Python file and initialize a fresh ChromaDB collection called "product_docs".
Step 1: Create a dictionary of three product documentation articles — a "Getting Started" guide, an "API Reference," and a "Troubleshooting" guide. Each should be at least 3–4 sentences of realistic-looking content.
Step 2: Index all three documents using the index_document function. Print the total chunk count after indexing.
Step 3: Simulate two changes: update the "Getting Started" guide with new content (add a new paragraph), and delete the "Troubleshooting" guide entirely from your source dictionary.
Step 4: Run incremental_upsert with your modified document dictionary. Verify that:
Step 5: Perform a test query to confirm the index reflects reality:
query = "How do I get started with the product?"
query_embedding = embedder.encode([query]).tolist()
results = collection.query(
query_embeddings=query_embedding,
n_results=3,
include=["documents", "metadatas", "distances"]
)
for doc, meta, dist in zip(results["documents"][0],
results["metadatas"][0],
results["distances"][0]):
print(f"[{meta['title']}] similarity: {1 - dist:.3f}")
print(f" {doc[:150]}...\n")
Challenge extension: Add a last_modified timestamp to each document and modify incremental_upsert to re-index a document if either its hash OR its timestamp is newer than what's stored in the index metadata.
Mistake 1: Using chunk position as the chunk ID
If you name your chunks chunk_0, chunk_1, etc. without including the doc_id, you'll have ID collisions across documents and no way to delete all chunks for a specific document. Always namespace chunk IDs: {doc_id}_chunk_{index}.
Mistake 2: Forgetting to handle deletions The incremental upsert only adding new/changed documents is a common first implementation. Documents that get deleted from your source will linger in the index indefinitely, serving stale answers. Always compute the set difference and delete orphaned document chunks.
Mistake 3: Re-embedding unchanged documents Generating embeddings is computationally expensive (and costs money if you're using an API like OpenAI's embedding endpoint). Always implement hash-based change detection before triggering re-indexing. Running a full rebuild nightly on a 100K-document corpus when only 50 documents changed is a waste of both time and budget.
Mistake 4: Chunk size that ignores your embedding model's token limit
Most embedding models have a maximum token limit (often 512 tokens for all-MiniLM-L6-v2). If your chunks exceed this limit, the model silently truncates them. Always verify your chunk sizes against your model's documentation and include a token count check in your chunking logic.
Mistake 5: No source-of-truth reconciliation If your indexing pipeline crashes halfway through an update, your vector store might have some documents at their old version and some at the new version. Build a reconciliation check — compare the hashes stored in your index metadata against the current source document hashes — and run it on a schedule to detect drift.
Tip: Store a separate "index manifest" — a simple JSON file or database table mapping
{doc_id: doc_hash, last_indexed_at: timestamp}— outside your vector store. This makes reconciliation, auditing, and debugging much easier than querying the vector store's own metadata.
You've just built the foundation of a production-grade RAG indexing system. Here's what you can now do:
The most important mindset shift from this lesson: a vector store is not a static artifact you build once. It's a living index that must stay synchronized with your source documents. How well your RAG system performs six months from now depends almost entirely on how well you manage that synchronization.
Where to go next:
The indexing layer is your RAG system's foundation. Get it right, and everything built on top of it becomes dramatically more reliable.
Learning Path: RAG & AI Agents