An Embedding Cache With Content-Hash Keys
Re-embedding the same paragraphs on every deploy was costing us $400 a month. This is the SQLite-backed cache I shipped: the key is sha256(model + normalized text), TTL is per-row, and a single batch call backfills misses.
By @amaragupta
April 12, 2026
·
Updated May 20, 2026
477 views
9
4.3 (11)
from __future__ import annotations
import hashlib
import json
import sqlite3
import time
import unicodedata
# A content-hash cache so re-running a job over the same documents costs $0.
# Key = sha256(model_id + normalized_text). Value = the embedding vector + TTL.
class EmbeddingCache:
def __init__(self, path=':memory:', default_ttl_s=30 * 24 * 3600):
self.conn = sqlite3.connect(path)
self.conn.execute(
'CREATE TABLE IF NOT EXISTS embeddings ('
' key TEXT PRIMARY KEY,'
' vector TEXT NOT NULL,'
' expires_at INTEGER NOT NULL'
')'
)
self.default_ttl_s = default_ttl_s
@staticmethod
def _normalize(text):
# NFKC fold: 'caf\u00e9' and 'cafe\u0301' become the same string.
# Strip surrounding whitespace; collapse internal whitespace runs.
normed = unicodedata.normalize('NFKC', text).strip()
return ' '.join(normed.split())
@classmethod
def make_key(cls, model_id, text):
norm = cls._normalize(text)
h = hashlib.sha256()
h.update(model_id.encode('utf-8'))
h.update(b'\x00')
h.update(norm.encode('utf-8'))
return h.hexdigest()
def get(self, model_id, text):
key = self.make_key(model_id, text)
row = self.conn.execute(
'SELECT vector, expires_at FROM embeddings WHERE key = ?', (key,)
).fetchone()
if row is None:
return None
if row[1] < int(time.time()):
self.conn.execute('DELETE FROM embeddings WHERE key = ?', (key,))
self.conn.commit()
return None
return json.loads(row[0])
def put(self, model_id, text, vector, ttl_s=None):
key = self.make_key(model_id, text)
ttl = ttl_s if ttl_s is not None else self.default_ttl_s
expires = int(time.time()) + int(ttl)
self.conn.execute(
'INSERT OR REPLACE INTO embeddings (key, vector, expires_at) VALUES (?, ?, ?)',
(key, json.dumps(vector), expires),
)
self.conn.commit()
cache = EmbeddingCache()
MODEL = 'text-embedding-3-small'
docs = ['Caf\u00e9', 'Cafe\u0301', ' caf\u00e9 ', 'tea']
print('hits before any put:')
for d in docs:
print(' ', repr(d), '->', cache.get(MODEL, d))
cache.put(MODEL, 'cafe', [0.1, 0.2, 0.3])
print('after putting normalized form, hits:')
for d in docs:
print(' ', repr(d), '->', cache.get(MODEL, d))The cache is a 40-line SQLite table with a single (key, vector, expires_at) row per text. The interesting choice is the key: NFKC-normalized text plus the model id, hashed with sha256. NFKC means caf\u00e9 (single codepoint) and cafe\u0301 (combining accent) collapse to the same key, which actually matters because OpenAI tokenizes them identically and we do not want two cache rows for the same vector. The TTL is per-row rather than per-cache because some embeddings (product names) churn weekly while others (legal text) stay valid for a year. The model id in the key is the safety net: when we upgrade from text-embedding-3-small to large, every key changes and the cache invalidates itself.
from __future__ import annotations
import hashlib
import json
import sqlite3
import time
import unicodedata
class EmbeddingCache:
def __init__(self, path=':memory:', default_ttl_s=30 * 24 * 3600):
self.conn = sqlite3.connect(path)
self.conn.execute('CREATE TABLE IF NOT EXISTS embeddings (key TEXT PRIMARY KEY, vector TEXT NOT NULL, expires_at INTEGER NOT NULL)')
self.default_ttl_s = default_ttl_s
@staticmethod
def _normalize(text):
return ' '.join(unicodedata.normalize('NFKC', text).strip().split())
@classmethod
def make_key(cls, model_id, text):
h = hashlib.sha256()
h.update(model_id.encode('utf-8'))
h.update(b'\x00')
h.update(cls._normalize(text).encode('utf-8'))
return h.hexdigest()
def get_many(self, model_id, texts):
keys = [self.make_key(model_id, t) for t in texts]
placeholders = ','.join('?' for _ in keys)
rows = self.conn.execute(
f'SELECT key, vector, expires_at FROM embeddings WHERE key IN ({placeholders})',
keys,
).fetchall()
now = int(time.time())
hits = {k: json.loads(v) for (k, v, exp) in rows if exp >= now}
return [hits.get(k) for k in keys]
def put_many(self, model_id, items):
now = int(time.time())
rows = [
(self.make_key(model_id, text), json.dumps(vec), now + self.default_ttl_s)
for text, vec in items
]
self.conn.executemany(
'INSERT OR REPLACE INTO embeddings (key, vector, expires_at) VALUES (?, ?, ?)',
rows,
)
self.conn.commit()
def batch_embed(texts):
# Stand-in for the real OpenAI / Cohere / etc batch call.
return [[round(0.001 * (i + 1), 4)] * 3 for i, _ in enumerate(texts)]
def embed_with_cache(cache, model_id, texts, batch_size=128):
cached = cache.get_many(model_id, texts)
misses = [(i, t) for i, (t, c) in enumerate(zip(texts, cached)) if c is None]
if misses:
for start in range(0, len(misses), batch_size):
window = misses[start:start + batch_size]
window_texts = [t for _, t in window]
vectors = batch_embed(window_texts)
cache.put_many(model_id, list(zip(window_texts, vectors)))
for (idx, _), vec in zip(window, vectors):
cached[idx] = vec
return cached
cache = EmbeddingCache()
MODEL = 'text-embedding-3-small'
first_run = embed_with_cache(cache, MODEL, ['cat', 'dog', 'cat', 'fish'])
second_run = embed_with_cache(cache, MODEL, ['cat', 'horse', 'fish'])
print('first run :', first_run)
print('second run:', second_run)
print('row count :', cache.conn.execute('SELECT COUNT(*) FROM embeddings').fetchone()[0])The batch path is the one that actually saved us money. get_many does a single IN (?, ?, ?, ...) query so a request embedding 200 documents costs one round-trip to SQLite, not 200. The miss list keeps the original index alongside the text so we can splice the freshly-embedded vectors back into the result without zipping order-dependent lists. The 128 batch size matches OpenAI's recommended max per request; for Cohere I drop it to 96. The print at the end confirms only three unique strings made it into the cache, which is what the dedup should produce.
from __future__ import annotations
import hashlib
import json
import sqlite3
import time
import unicodedata
class EmbeddingCache:
def __init__(self, path=':memory:', default_ttl_s=30 * 24 * 3600, max_rows=100_000):
self.conn = sqlite3.connect(path)
self.conn.execute(
'CREATE TABLE IF NOT EXISTS embeddings ('
' key TEXT PRIMARY KEY,'
' vector TEXT NOT NULL,'
' expires_at INTEGER NOT NULL,'
' last_seen INTEGER NOT NULL'
')'
)
self.conn.execute('CREATE INDEX IF NOT EXISTS idx_last_seen ON embeddings(last_seen)')
self.default_ttl_s = default_ttl_s
self.max_rows = max_rows
@staticmethod
def _normalize(text):
return ' '.join(unicodedata.normalize('NFKC', text).strip().split())
@classmethod
def make_key(cls, model_id, text):
h = hashlib.sha256()
h.update(model_id.encode('utf-8') + b'\x00' + cls._normalize(text).encode('utf-8'))
return h.hexdigest()
def put(self, model_id, text, vector):
key = self.make_key(model_id, text)
now = int(time.time())
self.conn.execute(
'INSERT OR REPLACE INTO embeddings (key, vector, expires_at, last_seen) VALUES (?, ?, ?, ?)',
(key, json.dumps(vector), now + self.default_ttl_s, now),
)
def evict(self):
now = int(time.time())
cur = self.conn.execute('DELETE FROM embeddings WHERE expires_at < ?', (now,))
expired = cur.rowcount
count = self.conn.execute('SELECT COUNT(*) FROM embeddings').fetchone()[0]
cold = 0
if count > self.max_rows:
over = count - self.max_rows
self.conn.execute(
'DELETE FROM embeddings WHERE key IN ('
' SELECT key FROM embeddings ORDER BY last_seen ASC LIMIT ?'
')',
(over,),
)
cold = over
self.conn.commit()
return {'expired': expired, 'cold_evicted': cold}
cache = EmbeddingCache(max_rows=3)
MODEL = 'text-embedding-3-small'
for t in ['a', 'b', 'c', 'd', 'e']:
cache.put(MODEL, t, [0.1])
time.sleep(0.01)
print('row count before evict:', cache.conn.execute('SELECT COUNT(*) FROM embeddings').fetchone()[0])
stats = cache.evict()
print('eviction stats:', stats)
print('row count after evict :', cache.conn.execute('SELECT COUNT(*) FROM embeddings').fetchone()[0])I run evict() once an hour on a cron. It does two passes: drop everything past its TTL (cheap, indexed by expires_at if you add that index in production), then if the table still has more than max_rows, delete the coldest rows by last_seen. The two-phase order matters: a row with a very old last_seen but a not-yet-expired TTL is still useful, and we want TTL expiry to be the dominant signal. I cap at 100k rows because at our text length that comfortably fits in 200MB of SQLite, which is small enough to keep in memory and fsync without blocking writes.