Cost: Embedding Budgets and Call Patterns

~22 min · ops, cost

Level 0Scout

0 XP0/41 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

Where the bill comes from

Ingestion embeddings — paid once per chunk, every time you re-embed.
Query embeddings — paid once per user query.
Storage — vector size × chunks × replication. Usually small unless you are at tens of millions of rows.
LLM calls — by far the largest line item once retrieval works. RAG does not save you from this.

Three cost levers that matter

Cache embeddings by content hash. Re-ingesting the same chunk should cost zero. Hash the text, look up the vector, only call the model on miss.
Batch. Hosted embedding APIs charge the same per-token whether you send 1 chunk or 100, but a batched call has 1 round-trip instead of 100. Batch sizes of 32–128 are typical.
Pick a smaller model when your eval allows. A 384-dim local model can match a 3072-dim hosted model on factoid retrieval and costs nothing per call. The eval set tells you whether the gap actually matters for your queries.

Code

Content-hash cache·python

import hashlib, json
from pathlib import Path

CACHE = Path('embedding_cache.jsonl')

def cached_embed(texts: list[str]) -> list[list[float]]:
    cache: dict[str, list[float]] = {}
    if CACHE.exists():
        for line in CACHE.read_text().splitlines():
            row = json.loads(line)
            cache[row['hash']] = row['vec']

    results, misses, miss_hashes = [], [], []
    for t in texts:
        h = hashlib.sha256(t.encode()).hexdigest()
        if h in cache:
            results.append(cache[h])
        else:
            results.append(None)
            misses.append(t)
            miss_hashes.append(h)

    if misses:
        new_vecs = model.encode(misses, normalize_embeddings=True).tolist()
        with CACHE.open('a') as f:
            for h, v in zip(miss_hashes, new_vecs):
                f.write(json.dumps({'hash': h, 'vec': v}) + '\n')
        miss_iter = iter(new_vecs)
        results = [r if r is not None else next(miss_iter) for r in results]
    return results

Batched OpenAI embedding·python

from openai import OpenAI
from itertools import islice

client = OpenAI()

def batched(it, n):
    it = iter(it)
    while batch := list(islice(it, n)):
        yield batch

def embed_corpus(texts, batch_size=64):
    out = []
    for batch in batched(texts, batch_size):
        resp = client.embeddings.create(model='text-embedding-3-large', input=batch)
        out.extend([d.embedding for d in resp.data])
    return out

External links

Exercise

Add the content-hash cache to your ingestion pipeline. Run a full re-ingest of your corpus twice. Confirm the second run hits 100% cache rate and zero embedding API cost. Log the savings.

Progress

Progress is local-only — sign in to sync across devices.

← 📚 RAG System DesignQuiz · 4 questions Next →Latency: Batching, Async, and Where the Time Goes

Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)

No comments yet — be the first.