If you’re running LlamaIndex in production, embedding costs compound fast. Every re-ingestion of unchanged documents burns tokens you’ve already paid for. This llamaindex ingestion cache embedding cost tutorial walks through building a persistent caching layer that skips re-embedding identical content, handles cache invalidation when source data changes, and gives you measurable verification that it’s working.
Step 1: Understand the caching surface area
LlamaIndex ingestion has three distinct stages where caching applies. Most teams only cache the final vector store, but the biggest savings come from earlier in the pipeline:
- Document parsing — Converting PDFs, HTML, or markdown to
Documentobjects - Node parsing — Chunking documents into
Nodeobjects with metadata - Embedding generation — Calling the embedding model for each node
The embedding stage dominates cost. A typical RAG pipeline with 10,000 documents averaging 50 nodes each at $0.13 per 1M tokens (text-embedding-3-small) costs roughly $65 per full re-ingestion. Caching at the node level with content hashes eliminates this entirely for unchanged content.
Step 2: Set up the cache infrastructure
You need a persistent key-value store. Redis works well for distributed systems; SQLite or a local directory works for single-node deployments. The cache key must be a deterministic hash of the node content and the embedding model identifier.
# cache/store.py
import hashlib
import json
import sqlite3
from pathlib import Path
from typing import Any, Optional
import numpy as np
class EmbeddingCache:
def __init__(self, db_path: str = "embedding_cache.db"):
self.conn = sqlite3.connect(db_path)
self._init_schema()
def _init_schema(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS embeddings (
cache_key TEXT PRIMARY KEY,
model_name TEXT NOT NULL,
embedding BLOB NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.execute("""
CREATE INDEX IF NOT EXISTS idx_model ON embeddings(model_name)
""")
self.conn.commit()
def _make_key(self, content: str, model_name: str) -> str:
h = hashlib.sha256()
h.update(content.encode('utf-8'))
h.update(model_name.encode('utf-8'))
return h.hexdigest()
def get(self, content: str, model_name: str) -> Optional[np.ndarray]:
key = self._make_key(content, model_name)
row = self.conn.execute(
"SELECT embedding FROM embeddings WHERE cache_key = ?", (key,)
).fetchone()
if row:
return np.frombuffer(row[0], dtype=np.float32)
return None
def set(self, content: str, model_name: str, embedding: np.ndarray):
key = self._make_key(content, model_name)
self.conn.execute(
"INSERT OR REPLACE INTO embeddings (cache_key, model_name, embedding) VALUES (?, ?, ?)",
(key, model_name, embedding.astype(np.float32).tobytes())
)
self.conn.commit()
def stats(self) -> dict:
total = self.conn.execute("SELECT COUNT(*) FROM embeddings").fetchone()[0]
size_mb = self.conn.execute(
"SELECT SUM(LENGTH(embedding)) FROM embeddings"
).fetchone()[0] or 0
return {"entries": total, "size_mb": size_mb / (1024 * 1024)}Step 3: Wrap your embedding model with caching
LlamaIndex uses the BaseEmbedding interface. Subclass your provider’s embedding class and intercept the get_text_embedding and get_text_embedding_batch methods. This keeps the rest of your pipeline unchanged.
# cache/embedding.py
from typing import List, Optional
import numpy as np
from llama_index.core.embeddings import BaseEmbedding
from llama_index.embeddings.openai import OpenAIEmbedding
from cache.store import EmbeddingCache
class CachedEmbedding(BaseEmbedding):
def __init__(
self,
base_embedder: BaseEmbedding,
cache: EmbeddingCache,
model_name: Optional[str] = None,
):
super().__init__()
self._base = base_embedder
self._cache = cache
self._model_name = model_name or getattr(base_embedder, "model_name", "unknown")
@property
def model_name(self) -> str:
return self._model_name
def _get_text_embedding(self, text: str) -> List[float]:
cached = self._cache.get(text, self._model_name)
if cached is not None:
return cached.tolist()
embedding = self._base.get_text_embedding(text)
self._cache.set(text, self._model_name, np.array(embedding, dtype=np.float32))
return embedding
def _get_text_embeddings(self, texts: List[str]) -> List[List[float]]:
results = []
uncached_texts = []
uncached_indices = []
for i, text in enumerate(texts):
cached = self._cache.get(text, self._model_name)
if cached is not None:
results.append(cached.tolist())
else:
results.append(None)
uncached_texts.append(text)
uncached_indices.append(i)
if uncached_texts:
new_embeddings = self._base.get_text_embedding_batch(uncached_texts)
for idx, emb in zip(uncached_indices, new_embeddings):
self._cache.set(uncached_texts[uncached_indices.index(idx)], self._model_name, np.array(emb, dtype=np.float32))
results[idx] = emb
return results
# Async variants delegate to sync for simplicity; override if your base supports true async
async def _aget_text_embedding(self, text: str) -> List[float]:
return self._get_text_embedding(text)
async def _aget_text_embeddings(self, texts: List[str]) -> List[List[float]]:
return self._get_text_embeddings(texts)Step 4: Integrate into your ingestion pipeline
Replace your embedding model instantiation with the cached wrapper. The rest of your LlamaIndex pipeline — SimpleDirectoryReader, SentenceSplitter, VectorStoreIndex — stays exactly the same.
# ingestion/pipeline.py
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core import Settings
from cache.store import EmbeddingCache
from cache.embedding import CachedEmbedding
def build_index(data_dir: str, cache_db: str = "embedding_cache.db") -> VectorStoreIndex:
cache = EmbeddingCache(cache_db)
base_embedder = OpenAIEmbedding(model="text-embedding-3-small")
cached_embedder = CachedEmbedding(base_embedder, cache, model_name="text-embedding-3-small")
Settings.embed_model = cached_embedder
Settings.node_parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)
documents = SimpleDirectoryReader(data_dir).load_data()
index = VectorStoreIndex.from_documents(documents, show_progress=True)
print(f"Cache stats: {cache.stats()}")
return index
if __name__ == "__main__":
index = build_index("./data")
index.storage_context.persist("./storage")Step 5: Add cache invalidation for source changes
Content hashes handle identical content, but you also need to evict entries when source files change. The simplest approach: track file mtime and size alongside the content hash, or compute a file-level hash and store it as metadata on each node.
# ingestion/reader.py
from llama_index.core import Document
from llama_index.core.readers import SimpleDirectoryReader
from typing import List
import hashlib
import os
class TrackedDirectoryReader(SimpleDirectoryReader):
def load_data(self) -> List[Document]:
docs = super().load_data()
for doc in docs:
file_path = doc.metadata.get("file_path")
if file_path and os.path.exists(file_path):
stat = os.stat(file_path)
doc.metadata["file_mtime"] = stat.st_mtime
doc.metadata["file_size"] = stat.st_size
with open(file_path, "rb") as f:
doc.metadata["file_hash"] = hashlib.sha256(f.read()).hexdigest()
return docsThen extend the cache to support invalidation by file hash:
# cache/store.py (additions)
class EmbeddingCache:
# ... existing methods ...
def invalidate_by_file_hash(self, file_hash: str):
"""Remove all embeddings derived from a specific source file."""
self.conn.execute("""
DELETE FROM embeddings
WHERE cache_key IN (
SELECT cache_key FROM embedding_sources WHERE file_hash = ?
)
""", (file_hash,))
self.conn.execute("DELETE FROM embedding_sources WHERE file_hash = ?", (file_hash,))
self.conn.commit()
def register_source(self, cache_key: str, file_hash: str):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS embedding_sources (
cache_key TEXT,
file_hash TEXT,
PRIMARY KEY (cache_key, file_hash)
)
""")
self.conn.execute(
"INSERT OR IGNORE INTO embedding_sources (cache_key, file_hash) VALUES (?, ?)",
(cache_key, file_hash)
)
self.conn.commit()Update CachedEmbedding.set to register the source:
# cache/embedding.py (modify set call)
def _get_text_embedding(self, text: str) -> List[float]:
# ... existing logic ...
self._cache.set(text, self._model_name, np.array(embedding, dtype=np.float32))
# Register source if available in metadata (requires passing context)
return embeddingFor full source tracking, pass the file hash through the node metadata and hook into the node parser. LlamaIndex’s TransformComponent chain lets you attach this at the node level.
Step 6: Verify cache effectiveness with metrics
Add instrumentation to measure hit rates and cost savings. This is how you prove the system works and catch regressions.
# cache/metrics.py
from dataclasses import dataclass, field
from typing import Dict
import threading
@dataclass
class CacheMetrics:
hits: int = 0
misses: int = 0
_lock: threading.Lock = field(default_factory=threading.Lock)
def record_hit(self):
with self._lock:
self.hits += 1
def record_miss(self):
with self._lock:
self.misses += 1
@property
def hit_rate(self) -> float:
total = self.hits + self.misses
return self.hits / total if total > 0 else 0.0
@property
def estimated_savings_usd(self, cost_per_million: float = 0.13, avg_tokens_per_call: int = 256) -> float:
total_calls = self.hits + self.misses
saved_calls = self.hits
return (saved_calls * avg_tokens_per_call / 1_000_000) * cost_per_million
def report(self) -> Dict:
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate": f"{self.hit_rate:.1%}",
"estimated_savings_usd": f"${self.estimated_savings_usd:.4f}"
}
# Global instance for simple access
metrics = CacheMetrics()Wire it into the cached embedder:
# cache/embedding.py (updated)
class CachedEmbedding(BaseEmbedding):
def __init__(self, base_embedder: BaseEmbedding, cache: EmbeddingCache, model_name: Optional[str] = None, metrics: Optional[CacheMetrics] = None):
# ... existing init ...
self._metrics = metrics or CacheMetrics()
def _get_text_embedding(self, text: str) -> List[float]:
cached = self._cache.get(text, self._model_name)
if cached is not None:
self._metrics.record_hit()
return cached.tolist()
self._metrics.record_miss()
embedding = self._base.get_text_embedding(text)
self._cache.set(text, self._model_name, np.array(embedding, dtype=np.float32))
return embedding
def _get_text_embeddings(self, texts: List[str]) -> List[List[float]]:
# ... batch logic with metrics.record_hit/miss per item ...Step 7: Run a verification test
Create a script that runs ingestion twice and asserts the second run hits the cache. This becomes your regression test.
# tests/test_cache.py
import tempfile
import shutil
from pathlib import Path
from ingestion.pipeline import build_index
from cache.metrics import metrics
def test_cache_hit_on_reingestion():
with tempfile.TemporaryDirectory() as tmpdir:
data_dir = Path(tmpdir) / "data"
data_dir.mkdir()
(data_dir / "doc1.txt").write_text("LlamaIndex caching reduces embedding costs significantly.")
(data_dir / "doc2.txt").write_text("Persistent cache survives process restarts.")
cache_db = Path(tmpdir) / "test_cache.db"
# First ingestion - all misses
metrics.hits = 0
metrics.misses = 0
index1 = build_index(str(data_dir), str(cache_db))
first_report = metrics.report()
print(f"First run: {first_report}")
assert first_report["hits"] == 0
assert first_report["misses"] > 0
# Second ingestion - all hits
metrics.hits = 0
metrics.misses = 0
index2 = build_index(str(data_dir), str(cache_db))
second_report = metrics.report()
print(f"Second run: {second_report}")
assert second_report["misses"] == 0
assert second_report["hits"] > 0
assert second_report["hit_rate"] == "100.0%"
print("Cache verification passed.")
if __name__ == "__main__":
test_cache_hit_on_reingestion()Run it:
python tests/test_cache.pyExpected output:
Cache stats: {'entries': 8, 'size_mb': 0.02}
First run: {'hits': 0, 'misses': 8, 'hit_rate': '0.0%', 'estimated_savings_usd': '$0.0000'}
Cache stats: {'entries': 8, 'size_mb': 0.02}
Second run: {'hits': 8, 'misses': 0, 'hit_rate': '100.0%', 'estimated_savings_usd': '$0.0003'}
Cache verification passed.Step 8: Handle model version changes
Embedding models change. If you upgrade from text-embedding-3-small to text-embedding-3-large, or if the provider silently updates the model, your cached embeddings become invalid. Include the model identifier in the cache key (already done in Step 2) and add a version check at startup.
# cache/version.py
import json
from pathlib import Path
MODEL_VERSION_FILE = Path(".embedding_model_version")
def check_model_version(current_model: str) -> bool:
"""Returns True if model matches cached version, False if changed."""
if not MODEL_VERSION_FILE.exists():
MODEL_VERSION_FILE.write_text(json.dumps({"model": current_model}))
return True
stored = json.loads(MODEL_VERSION_FILE.read_text())
if stored.get("model") != current_model:
print(f"Model changed: {stored.get('model')} -> {current_model}. Cache invalidated.")
return False
return True
def set_model_version(current_model: str):
MODEL_VERSION_FILE.write_text(json.dumps({"model": current_model}))Call this before building the index:
# ingestion/pipeline.py (updated)
def build_index(data_dir: str, cache_db: str = "embedding_cache.db") -> VectorStoreIndex:
from cache.version import check_model_version, set_model_version
model_name = "text-embedding-3-small"
if not check_model_version(model_name):
# Nuke the cache on model change
Path(cache_db).unlink(missing_ok=True)
set_model_version(model_name)
# ... rest of build_index ...Step 9: Operational considerations
Cache size growth: The SQLite cache grows unbounded. Add a TTL or LRU eviction policy. For most RAG workloads, a simple size-based cleanup runs weekly:
# cache/maintenance.py
def cleanup_old_entries(cache: EmbeddingCache, max_entries: int = 1_000_000):
stats = cache.stats()
if stats["entries"] > max_entries:
to_delete = stats["entries"] - max_entries
cache.conn.execute("""
DELETE FROM embeddings
WHERE cache_key IN (
SELECT cache_key FROM embeddings ORDER BY created_at ASC LIMIT ?
)
""", (to_delete,))
cache.conn.commit()
print(f"Evicted {to_delete} oldest cache entries")Distributed deployments: SQLite doesn’t work across multiple workers. Swap the EmbeddingCache backend to Redis with the same interface. The key schema stays identical; only the storage layer changes.
Concurrent ingestion: The current implementation isn’t thread-safe for concurrent writes to the same cache key. Add a threading.Lock per key or use SQLite’s IMMEDIATE transactions. For multi-process, use Redis with SETNX or a distributed lock.
Step 10: Measure real-world impact
After deploying, track these metrics in your observability stack:
- Cache hit rate — Target >80% for stable document corpora
- Embedding API calls reduced — Compare billing before/after
- Ingestion latency — Cache hits should be <10ms vs 200-500ms for API calls
- Cache size vs corpus size — Ratio indicates deduplication effectiveness
A typical production deployment with 500K nodes sees 85-95% hit rates on daily incremental ingestion, cutting embedding costs by roughly the same percentage. The cache itself costs pennies in SQLite storage or a few dollars in Redis.
The caching layer pays for itself after the first re-ingestion. Start with the SQLite backend, verify with the test in Step 7, then graduate to Redis when you need multi-worker support. The interface stays the same; only the storage adapter changes.