Most retrieval-augmented generation (RAG) tutorials end at the same point: load some documents, split them into chunks, embed them, store them in a vector database and ask questions. That is enough for a demo. It is not enough for a system that a business relies on, because real documents change every day.

Policies get revised, product pages get updated, a contract gets replaced and an old one should no longer be quoted. If your index does not reflect those changes, your assistant will confidently answer with yesterday's facts. In my experience, stale or duplicated chunks cause more wrong answers than the language model itself.

The quick fix teams usually try is to rebuild the whole index every night. That works until the corpus grows, embedding bills climb and a nightly job fails halfway, leaving a half-built index in production. In this article I walk through a simpler pattern: an incremental pipeline that only re-embeds what changed, removes what was deleted and keeps a record of where every chunk came from.

The design in one picture

Incremental RAG ingestion pipeline: chunk registry with stable IDs and content hashes

The key idea is a chunk registry: a small table that stores, for every chunk, a stable ID and a hash of its content. On each run, we compute the chunks again, compare hashes and only touch what is different.

Step 1: Chunk by structure, with stable IDs

Chunking by fixed character counts creates two problems. It cuts sentences and clauses in half, and the chunk boundaries move every time someone edits the start of a document. If boundaries move, every chunk after the edit looks "new" and you re-embed the whole document.

Chunking by headings solves both. A section is a natural unit of meaning, and an edit in one section does not change the others.

import hashlib import re from dataclasses import dataclass @dataclass class Chunk: doc_id: str chunk_id: str text: str content_hash: str source_url: str section: str HEADING = re.compile(r"^(#{1,3})\s+(.*)$", re.MULTILINE) def chunk_markdown(doc_id: str, text: str, source_url: str, max_chars: int = 2000): """Split a markdown document by headings; split long sections by paragraph.""" positions = [m.start() for m in HEADING.finditer(text)] or [0] if positions[0] != 0: positions.insert(0, 0) positions.append(len(text)) chunks = [] for start, end in zip(positions, positions[1:]): section = text[start:end].strip() if not section: continue title = section.splitlines()[0].lstrip("# ").strip() parts, buf = [], "" for para in section.split("\n\n"): if buf and len(buf) + len(para) > max_chars: parts.append(buf) buf = "" buf = f"{buf}\n\n{para}" if buf else para if buf: parts.append(buf) for i, part in enumerate(parts): # Stable ID: document + section title + position inside the section  chunk_id = hashlib.sha1(f"{doc_id}|{title}|{i}".encode()).hexdigest() content_hash = hashlib.sha256(part.encode()).hexdigest() chunks.append(Chunk(doc_id, chunk_id, part, content_hash, source_url, title)) return chunks

Two hashes do two different jobs. The chunk_id identifies where a chunk lives (document, section, position). The content_hash tells us whether its text changed.

Step 2: Add document context to each chunk

A chunk that reads "Refunds are processed within 14 days" is ambiguous on its own. Which product? Which region? Which version? Retrieval improves noticeably when each chunk carries a short header with the document title and section path. Anthropic described a more advanced version of this idea as contextual retrieval. A plain header is a good, cheap start:

def with_context(chunk: Chunk, doc_title: str) -> str: return f"Document: {doc_title}\nSection: {chunk.section}\n\n{chunk.text}"

Embed the contextualised text, but hash the raw text. That way, renaming a document does not force a full re-embed unless you want it to.

Step 3: Compare against the registry

The registry can live in the same Postgres database as your vectors if you use pgvector, which keeps the update atomic. One table is enough:

CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE IF NOT EXISTS rag_chunks ( chunk_id TEXT PRIMARY KEY, doc_id TEXT NOT NULL, content_hash TEXT NOT NULL, source_url TEXT NOT NULL, body TEXT NOT NULL, embedding vector(384) NOT NULL, updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX IF NOT EXISTS rag_chunks_doc ON rag_chunks (doc_id);

For each document we load the existing hashes and sort the new chunks into three groups:

def diff_chunks(new_chunks, existing: dict[str, str]): """existing maps chunk_id -> content_hash for one document.""" new_ids = {c.chunk_id for c in new_chunks} to_upsert = [c for c in new_chunks if existing.get(c.chunk_id) != c.content_hash] to_delete = [cid for cid in existing if cid not in new_ids] unchanged = len(new_chunks) - len(to_upsert) return to_upsert, to_delete, unchanged

Step 4: Embed only what changed, in one transaction 

import psycopg from pgvector.psycopg import register_vector from sentence_transformers import SentenceTransformer model = SentenceTransformer("all-MiniLM-L6-v2") # 384 dimensions  def sync_document(conn, doc_id, doc_title, text, source_url): new_chunks = chunk_markdown(doc_id, text, source_url) rows = conn.execute( "SELECT chunk_id, content_hash FROM rag_chunks WHERE doc_id = %s", (doc_id,) ).fetchall() to_upsert, to_delete, unchanged = diff_chunks(new_chunks, dict(rows)) vectors = model.encode([with_context(c, doc_title) for c in to_upsert]) if to_upsert else [] with conn.transaction(): for c, v in zip(to_upsert, vectors): conn.execute( """INSERT INTO rag_chunks (chunk_id, doc_id, content_hash, source_url, body, embedding) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (chunk_id) DO UPDATE SET content_hash = EXCLUDED.content_hash, body = EXCLUDED.body, embedding = EXCLUDED.embedding, source_url = EXCLUDED.source_url, updated_at = now()""", (c.chunk_id, c.doc_id, c.content_hash, c.source_url, c.text, v), ) if to_delete: conn.execute("DELETE FROM rag_chunks WHERE chunk_id = ANY(%s)", (to_delete,)) return {"upserted": len(to_upsert), "deleted": len(to_delete), "unchanged": unchanged} with psycopg.connect("postgresql://localhost/rag") as conn: register_vector(conn) print(sync_document(conn, "refund-policy", "Refund Policy (EU)", open("refund_policy.md").read(), "https://intranet/refund-policy"))

Because the upserts and deletes run in one transaction, a failure halfway leaves the previous version of the document intact rather than a mix of old and new chunks.

Step 5: Handle deleted documents

The diff above handles deleted sections. Deleted documents need one more check: compare the list of document IDs in your source system with the IDs in the table and remove the ones that no longer exist. Skipping this step is the most common reason old policies keep showing up in answers.

def remove_missing_documents(conn, live_doc_ids: set[str]): indexed = {r[0] for r in conn.execute("SELECT DISTINCT doc_id FROM rag_chunks")} gone = list(indexed - live_doc_ids) if gone: conn.execute("DELETE FROM rag_chunks WHERE doc_id = ANY(%s)", (gone,)) return gone

What this changes in practice

On a typical business corpus, only a small share of chunks changes on any given day, so most runs embed very little. A nightly full rebuild re-embeds every chunk whether it changed or not, so its cost grows with the size of the corpus. The incremental run grows with the amount of change instead, which is usually a small fraction.

Two other benefits matter as much as cost:

  • Traceability. Every answer can cite source_url and updated_at, so users can check the source and see how recent it is.
  • Debuggability. When an answer is wrong, you can see exactly which chunk version was retrieved and when it was last updated.

Things to watch

  • Changing the embedding model invalidates every vector. Store the model name in the table and do a full rebuild into a new table when you switch, then swap.
  • Changing the chunking logic changes chunk IDs. Treat it the same way as a model change.
  • Very large sections still need a size limit, as in the paragraph splitting above, or retrieval quality drops.
  • Test retrieval on its own. Keep a small set of real questions with known source sections, and check after each pipeline change that the right chunk still appears in the top results.

Conclusion

A RAG system is only as current as its index. Chunk by structure, give every chunk a stable ID and a content hash, re-embed only what changed and delete what disappeared. It is a few dozen lines of code, and it removes a whole class of confidently wrong answers.

I'm Yaman Kavishwar, founder of VectovateAI, where we build RAG systems, AI agents and data platforms for businesses. If you've solved index freshness differently, I'd like to hear how in the comments.