Incremental ingestion in LlamaIndex with document hashes

Incremental ingestion is the difference between a pipeline that scales and one that burns your embedding budget reprocessing the same PDFs every night. This llamaindex incremental ingestion tutorial walks through a production-ready pattern: compute a content hash for each document, store it alongside your vectors, and only re-embed what actually changed. You’ll finish with a runnable pipeline that handles new, modified, and deleted files correctly.

Step 1: Set up the document store and vector store

LlamaIndex separates document metadata from vector storage. For incremental ingestion you need both: a DocumentStore to track hashes and a VectorStore for embeddings. The simplest local setup uses the built-in SimpleDocumentStore and SimpleVectorStore, but the pattern is identical with Pinecone, Weaviate, or PGVector.

# ingest_setup.py
from llama_index.core import (
    SimpleDirectoryReader,
    VectorStoreIndex,
    StorageContext,
    Document,
)
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.vector_stores import SimpleVectorStore
from llama_index.core.ingestion import IngestionPipeline, DocstoreStrategy
from llama_index.embeddings.openai import OpenAIEmbedding
import hashlib
import json
import os

PERSIST_DIR = "./storage_incremental"
DATA_DIR = "./data"

def get_content_hash(text: str) -> str:
    """Stable hash of document text content."""
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def load_existing_hashes(docstore: SimpleDocumentStore) -> dict[str, str]:
    """Return mapping of doc_id -> content_hash from persisted docstore."""
    hashes = {}
    for doc_id, doc in docstore.docs.items():
        # Hash is stored in metadata if we put it there
        if "content_hash" in doc.metadata:
            hashes[doc_id] = doc.metadata["content_hash"]
    return hashes

Run this once to create the persistence directory structure. The SimpleDocumentStore persists as JSON; SimpleVectorStore persists as a pickle file. Both survive process restarts.

Step 2: Build the ingestion pipeline with upsert strategy

The IngestionPipeline class handles the heavy lifting. The key is DocstoreStrategy.UPSERTS — it tells LlamaIndex to compare incoming documents against the docstore by doc_id, compute a hash, and only run transformations (splitting, embedding) when the hash differs.

# ingest_pipeline.py
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.extractors import TitleExtractor
from llama_index.embeddings.openai import OpenAIEmbedding

def build_pipeline(
    docstore: SimpleDocumentStore,
    vector_store: SimpleVectorStore,
    embed_model: OpenAIEmbedding,
) -> IngestionPipeline:
    return IngestionPipeline(
        transformations=[
            SentenceSplitter(chunk_size=512, chunk_overlap=50),
            TitleExtractor(),
            embed_model,
        ],
        docstore=docstore,
        vector_store=vector_store,
        docstore_strategy=DocstoreStrategy.UPSERTS,
    )

DocstoreStrategy.UPSERTS does three things automatically:

1. Looks up each incoming doc_id in the docstore
2. Compares the stored content_hash (if any) against the new document’s hash
3. Skips transformations for unchanged documents; runs them for new or modified ones

You still need to attach the hash to each Document before passing it to the pipeline. That happens in the next step.

Step 3: Load files, compute hashes, and attach metadata

SimpleDirectoryReader assigns a doc_id based on the file path by default. That’s stable across runs, which is exactly what you want. Compute the hash of the full document text and stash it in metadata["content_hash"] before ingestion.

# ingest_run.py
from llama_index.core import SimpleDirectoryReader
from ingest_setup import get_content_hash, load_existing_hashes, PERSIST_DIR, DATA_DIR
from ingest_pipeline import build_pipeline
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.vector_stores import SimpleVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import StorageContext

def run_ingestion():
    # Load or create stores
    if os.path.exists(PERSIST_DIR):
        storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
        docstore = storage_context.docstore
        vector_store = storage_context.vector_store
    else:
        docstore = SimpleDocumentStore()
        vector_store = SimpleVectorStore()

    embed_model = OpenAIEmbedding(model="text-embedding-3-small")
    pipeline = build_pipeline(docstore, vector_store, embed_model)

    # Load raw documents
    reader = SimpleDirectoryReader(DATA_DIR, recursive=True)
    documents = reader.load_data()

    # Attach content hash to each document
    for doc in documents:
        doc.metadata["content_hash"] = get_content_hash(doc.text)

    # Run pipeline — UPSERTS handles the rest
    nodes = pipeline.run(documents=documents)

    # Persist everything
    storage_context = StorageContext.from_defaults(
        docstore=docstore, vector_store=vector_store
    )
    storage_context.persist(persist_dir=PERSIST_DIR)

    print(f"Ingested {len(nodes)} nodes from {len(documents)} documents")
    return nodes

if __name__ == "__main__":
    run_ingestion()

Run python ingest_run.py the first time — it embeds everything. Run it again without changing files — it prints Ingested 0 nodes because every hash matches. That’s the verification signal you’re looking for.

Step 4: Handle deleted files

DocstoreStrategy.UPSERTS only processes documents you pass to it. If a file disappears from DATA_DIR, its entry lingers in the docstore and vector store forever unless you explicitly clean it up. Add a reconciliation step that compares the docstore’s known doc_ids against the current filesystem.

# ingest_cleanup.py
import os
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core import StorageContext

def cleanup_deleted_documents(
    docstore: SimpleDocumentStore,
    vector_store: SimpleVectorStore,
    data_dir: str,
) -> int:
    """Remove docstore and vector entries for files that no longer exist."""
    # Current file paths -> doc_ids (SimpleDirectoryReader uses relative path as doc_id)
    current_doc_ids = set()
    for root, _, files in os.walk(data_dir):
        for f in files:
            rel_path = os.path.relpath(os.path.join(root, f), data_dir)
            current_doc_ids.add(rel_path)

    stored_doc_ids = set(docstore.docs.keys())
    deleted_doc_ids = stored_doc_ids - current_doc_ids

    if not deleted_doc_ids:
        return 0

    # Delete from docstore
    for doc_id in deleted_doc_ids:
        docstore.delete_document(doc_id)

    # Delete from vector store — requires knowing node_ids associated with doc_id
    # SimpleVectorStore stores nodes by node_id; we must find nodes referencing the doc_id
    nodes_to_delete = []
    for node_id, node in vector_store._data.items():  # internal dict
        if node.metadata.get("doc_id") in deleted_doc_ids:
            nodes_to_delete.append(node_id)

    for node_id in nodes_to_delete:
        vector_store.delete(node_id)

    print(f"Cleaned up {len(deleted_doc_ids)} deleted documents, {len(nodes_to_delete)} nodes")
    return len(deleted_doc_ids)

Call this at the start of run_ingestion() before loading new documents. The vector store deletion uses a private _data attribute on SimpleVectorStore; with a production vector store (Pinecone, Weaviate, etc.) you’d use their native delete-by-metadata-filter API instead.

Step 5: Verify incremental behavior end to end

Create a test script that proves the pipeline only re-embeds changed files. This is your regression guard — commit it to the repo.

# test_incremental.py
import tempfile
import shutil
from pathlib import Path
from ingest_run import run_ingestion
from ingest_cleanup import cleanup_deleted_documents
from llama_index.core import SimpleDirectoryReader
from llama_index.core.storage.docstore import SimpleDocumentStore
from llama_index.core.vector_stores import SimpleVectorStore
from llama_index.core import StorageContext
from ingest_setup import PERSIST_DIR, get_content_hash

def test_incremental_ingestion():
    with tempfile.TemporaryDirectory() as tmpdir:
        data_dir = Path(tmpdir) / "data"
        data_dir.mkdir()
        persist_dir = Path(tmpdir) / "storage"

        # Write two files
        (data_dir / "doc1.txt").write_text("Alpha content for first document.")
        (data_dir / "doc2.txt").write_text("Beta content for second document.")

        # Monkey-patch constants for this test
        import ingest_run, ingest_setup, ingest_pipeline
        ingest_run.DATA_DIR = str(data_dir)
        ingest_setup.DATA_DIR = str(data_dir)
        ingest_setup.PERSIST_DIR = str(persist_dir)

        # First run — full ingestion
        nodes1 = run_ingestion()
        assert len(nodes1) > 0, "First run should produce nodes"

        # Second run — no changes
        nodes2 = run_ingestion()
        assert len(nodes2) == 0, "Second run should produce zero new nodes"

        # Modify one file
        (data_dir / "doc1.txt").write_text("Alpha content for first document UPDATED.")
        nodes3 = run_ingestion()
        assert len(nodes3) > 0, "Third run should re-embed modified doc"

        # Delete one file
        (data_dir / "doc2.txt").unlink()
        # Need to re-run cleanup + ingestion
        storage_context = StorageContext.from_defaults(persist_dir=str(persist_dir))
        cleanup_deleted_documents(
            storage_context.docstore, storage_context.vector_store, str(data_dir)
        )
        nodes4 = run_ingestion()
        # doc2 gone, doc1 unchanged -> 0 new nodes
        assert len(nodes4) == 0, "Fourth run should produce zero new nodes"

        print("All incremental ingestion tests passed.")

if __name__ == "__main__":
    test_incremental_ingestion()

Run python test_incremental.py. It should exit cleanly. If any assertion fails, the hash comparison or cleanup logic has a bug.

Step 6: Scale to production vector stores

The local SimpleVectorStore is fine for prototypes. In production you’ll swap it for a managed vector database. The ingestion pipeline code barely changes — only the vector store initialization.

# production_stores.py
from llama_index.vector_stores.pinecone import PineconeVectorStore
from llama_index.vector_stores.weaviate import WeaviateVectorStore
from llama_index.core.vector_stores import VectorStore
import os

def get_vector_store() -> VectorStore:
    provider = os.getenv("VECTOR_STORE", "pinecone").lower()
    if provider == "pinecone":
        from pinecone import Pinecone
        pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])
        index = pc.Index(os.environ["PINECONE_INDEX"])
        return PineconeVectorStore(pinecone_index=index)
    elif provider == "weaviate":
        import weaviate
        client = weaviate.connect_to_wcs(
            cluster_url=os.environ["WEAVIATE_URL"],
            auth_credentials=weaviate.auth.AuthApiKey(os.environ["WEAVIATE_API_KEY"]),
        )
        return WeaviateVectorStore(weaviate_client=client, index_name="LlamaIndex")
    else:
        raise ValueError(f"Unknown vector store: {provider}")

The docstore can stay local (JSON on disk) or move to Redis / Postgres via RedisDocumentStore or PostgresDocumentStore. The IngestionPipeline doesn’t care — it only requires the BaseDocumentStore and VectorStore interfaces.

One operational note: managed vector stores often charge per write. Incremental ingestion directly reduces that bill. If you’re routing traffic through a gateway like n4n.ai that meters per-token usage, you’ll see the savings in the usage dashboard immediately — fewer embedding calls means lower cost and lower latency on the ingestion path.

Step 7: Monitor and alert on ingestion health

Add structured logging so you can graph ingestion volume over time. A sudden spike in “nodes embedded” usually means a hash mismatch bug or a downstream system rewriting files unnecessarily.

# ingest_logging.py
import logging
import json
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ingestion")

def log_ingestion_metrics(
    total_docs: int,
    new_nodes: int,
    skipped_docs: int,
    deleted_docs: int,
    duration_seconds: float,
):
    metrics = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "total_documents": total_docs,
        "nodes_created": new_nodes,
        "documents_skipped": skipped_docs,
        "documents_deleted": deleted_docs,
        "duration_seconds": round(duration_seconds, 2),
    }
    logger.info(json.dumps(metrics))

Wrap run_ingestion() with timing and call log_ingestion_metrics. Ship these logs to your observability stack (Datadog, Grafana Loki, CloudWatch) and alert on nodes_created > 2 * baseline for more than two consecutive runs.

Common pitfalls

Hash instability from metadata. If you include timestamps or file stats in the hash input, the hash changes on every run even when content is identical. Hash only the document text (or the normalized text you actually embed).

Chunk-level hashing. The pipeline hashes at the document level, not the chunk level. If you modify one paragraph of a 50-page PDF, the whole document re-embeds. That’s usually the right trade-off — chunk-level hashing adds complexity and rarely saves enough to justify it. If you need finer granularity, split large documents into separate files before ingestion.

Docstore and vector store drift. If you delete from one but not the other (crash mid-run, manual cleanup), you get orphaned vectors or ghost documents. Run a periodic reconciliation job that scans both stores and repairs mismatches.

Embedding model changes. Switching from text-embedding-3-small to text-embedding-3-large invalidates every vector. The hash won’t catch this because the text didn’t change. Include the model name in the hash or maintain a model_version field in metadata and force a full re-ingest on model upgrades.

Verification checklist

Before declaring the pipeline production-ready, confirm each of these:

1. Idempotency — Run the ingestion twice with zero file changes. Second run must produce zero new nodes and zero log errors.
2. Modification detection — Edit one file. Next run re-embeds only that file’s chunks.
3. Deletion cleanup — Remove a file. Run cleanup + ingestion. The docstore and vector store no longer contain its nodes.
4. Large file handling — Drop a 100 MB PDF in the data directory. Pipeline completes without OOM (adjust chunk_size if needed).
5. Concurrent safety — If two ingestion processes start simultaneously, the docstore’s JSON file can corrupt. Use a file lock (fcntl or portalocker) or move the docstore to Redis/Postgres for multi-process safety.

Once those five pass, you have an ingestion pipeline that scales linearly with actual data change rate, not with cron schedule frequency. That’s the engineering outcome this tutorial delivers.