Capsule 06: Cost and Performance Optimization for RAG

Capsule description

This capsule helps you maintain the balance between user experience and operational cost in RAG systems. The goal is to improve efficiency without compromising answer quality. You will learn concrete levers (embedding cache, result cache, batch operations, index tuning with PQ, model selection, top_k optimization), production-ready Python code, real cost numbers, and exercises with detailed solutions.

Estimated time: 35-45 minutes

Where does cost come from in RAG?

A typical RAG pipeline has three main cost sources:

The most cost-effective optimization is usually reducing repeated calls: duplicate embeddings and already-generated answers. In this capsule you focus on embedding, retrieval, and infra costs; LLM token optimization is another topic.

1. Embedding cache (don't re-embed existing documents)

The problem

Every time you re-ingest a document, you call the embeddings API again. If you update 1000 documents daily and each has ~500 tokens, that's 500K tokens/day just on re-embedding. At $0.02/1M tokens, that's $0.01/day = **$0.30/month** just on unnecessary re-embedding.

The solution

Store the content hash alongside the embedding. If the document did not change, don't re-embed.

import hashlib
import json
from typing import Optional

def content_hash(content: str, metadata: Optional[dict] = None) -> str:
    """Generate a unique hash to detect whether the document changed."""
    payload = content
    if metadata:
        payload += json.dumps(metadata, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()

def should_reembed(
    doc_id: str,
    content: str,
    metadata: Optional[dict],
    cache: dict  # {doc_id: {"hash": str, "embedding": list}}
) -> bool:
    """Returns True only if the document changed or is not in the cache."""
    new_hash = content_hash(content, metadata)
    if doc_id not in cache:
        return True
    return cache[doc_id]["hash"] != new_hash

# Use in the ingestion pipeline
def ingest_with_embedding_cache(
    doc_id: str,
    content: str,
    metadata: Optional[dict],
    embedding_cache: dict,
    embed_fn,
    vector_db
):
    if not should_reembed(doc_id, content, metadata, embedding_cache):
        # Retrieve the existing embedding from the DB (or a separate store)
        embedding = vector_db.get_embedding_by_id(doc_id)
        if embedding is not None:
            vector_db.upsert(doc_id, embedding, metadata)
            return  # No API call
    embedding = embed_fn(content)
    embedding_cache[doc_id] = {"hash": content_hash(content, metadata), "embedding": embedding}
    vector_db.upsert(doc_id, embedding, metadata)

Cost saved

• Without cache: 1000 docs/day × 500 tokens × $0.02/1M = $0.01/day ≈ $0.30/month.
• With cache (70% unchanged): 300 docs/day × 500 tokens × $0.02/1M = $0.09/month.
• Savings: ~$0.21/month just on ingestion embeddings.

2. Result cache with TTL (answers and retrieval)

The problem

Many queries are repeated or very similar. Without cache, each one pays for the query embedding + vector search. Example:

• Without cache: 1000 queries/day × $0.00002/query (embedding) = $0.60/month just on query embeddings.
• If you include retrieval and LLM cost, a full query can cost ~$0.001-0.01.

The solution

Result cache with TTL. Key: hash of the normalized query. If the content does not change, you reuse the answer.

import redis
import json
import hashlib
import pickle
from datetime import timedelta
from typing import Any, Optional

class RAGResultCache:
    """RAG result cache with TTL."""
    
    def __init__(self, redis_url: str = "redis://localhost:6379", ttl_seconds: int = 3600):
        self.client = redis.from_url(redis_url)
        self.ttl = ttl_seconds
    
    def _cache_key(self, query: str, top_k: int, filters: Optional[dict] = None) -> str:
        payload = f"{query.strip().lower()}|{top_k}|{json.dumps(filters or {}, sort_keys=True)}"
        return f"rag:result:{hashlib.sha256(payload.encode()).hexdigest()}"
    
    def get(self, query: str, top_k: int, filters: Optional[dict] = None) -> Optional[dict]:
        key = self._cache_key(query, top_k, filters)
        data = self.client.get(key)
        if data is None:
            return None
        return pickle.loads(data)
    
    def set(self, query: str, top_k: int, result: dict, filters: Optional[dict] = None) -> None:
        key = self._cache_key(query, top_k, filters)
        self.client.setex(key, self.ttl, pickle.dumps(result))

# Use in an endpoint
def rag_query(query: str, top_k: int = 5):
    cache = RAGResultCache(ttl_seconds=1800)  # 30 min
    cached = cache.get(query, top_k)
    if cached:
        return cached
    result = do_full_rag_pipeline(query, top_k)
    cache.set(query, top_k, result)
    return result

Concrete numbers

• Without cache: 1000 queries/day × 30 days × $0.00002/query (embedding) = $0.60/month.
• With 30% cache hit rate: 70% × 30,000 = 21,000 paid queries = $0.42/month.
• With 60% cache hit rate: 40% × 30,000 = 12,000 paid queries = $0.24/month.

If each full query (embedding + retrieval + LLM) costs ~$0.005:

• Without cache: 30,000 × $0.005 = $150/month.
• With 30% cache: 21,000 × $0.005 = $105/month (savings $45/month).
• With 60% cache: 12,000 × $0.005 = $60/month (savings $90/month).

3. Batch operations in ingestion

The problem

Calling the embeddings API document by document is slow and sometimes more expensive. Many providers offer batch discounts and limit requests/second.

The solution

Group documents into batches (e.g. 100) and embed in a single call.

def batch_embed(texts: list[str], embed_fn, batch_size: int = 100) -> list[list[float]]:
    """Embed in batches to reduce calls and latency."""
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        embeddings = embed_fn(batch)  # API accepts an array
        results.extend(embeddings)
    return results

# ChromaDB accepts add with documents in batch
def batch_ingest(chroma_collection, documents: list[dict], embed_fn, batch_size: int = 100):
    for i in range(0, len(documents), batch_size):
        batch = documents[i : i + batch_size]
        ids = [d["id"] for d in batch]
        texts = [d["text"] for d in batch]
        metadatas = [d.get("metadata", {}) for d in batch]
        embeddings = batch_embed(texts, embed_fn, batch_size)
        chroma_collection.add(ids=ids, embeddings=embeddings, metadatas=metadatas)

Impact

• Fewer round-trips → lower latency.
• Some providers charge per request; batching reduces the number of requests.
• Typically 2-5x faster in bulk ingestion.

4. Index tuning with Product Quantization (PQ)

What is PQ?

Product Quantization compresses vectors into lower-dimensionality subspaces. It reduces memory usage and speeds up search, with a small trade-off in recall.

When to use it

• Large indexes (>100K vectors).
• Memory constraints.
• Critical latency.

# Qdrant with PQ
from qdrant_client.models import VectorParams, Distance, QuantizationConfig, ScalarQuantization

# Configure the collection with compression
client.create_collection(
    collection_name="docs",
    vectors_config=VectorParams(
        size=1536,
        distance=Distance.COSINE,
        on_disk=True,
    ),
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantization(
            type="int8",
            quantile=0.99,
            always_ram=True,
        )
    ),
)

Typical impact

• Memory: ~4x reduction (float32 → int8).
• Speed: 1.5-2x faster on many workloads.
• Recall: 1-3% loss on typical benchmarks; recoverable with a higher top_k in some cases.

5. Model selection per environment

Simple rule

Code for switching per environment

import os

def get_embedding_model(env: str = None):
    env = env or os.getenv("ENV", "development")
    if env == "production":
        return "text-embedding-3-small"  # OpenAI
    return "all-MiniLM-L6-v2"  # Local, free

def get_llm_model(env: str = None):
    env = env or os.getenv("ENV", "development")
    if env == "production":
        return "gpt-4o-mini"
    return "gpt-3.5-turbo"

Savings

• Local in dev: $0 vs ~$0.02/1M tokens.
• On a team of 5 developers doing 50K tokens/day in dev: ~$30/month saved.

6. top_k optimization

The problem

High top_k = more retrieved documents = more tokens to the LLM = higher cost and latency. Low top_k = risk of losing relevant context.

Strategy

Tune per use case:

Controlled experiment

Hypothesis: lowering top_k from 8 to 5 reduces latency without losing quality.

1. Run an internal A/B for 3-5 days.
2. Compare latency (p50, p95), cost per query, and quality feedback.
3. Keep the change only if it meets thresholds (e.g. recall >95%, p95 <500ms).

def optimize_top_k(current: int, candidate: int, rag_fn, eval_queries: list) -> dict:
    """Compare current vs candidate top_k on approximate recall and latency."""
    results = {"current": [], "candidate": []}
    for q in eval_queries:
        r_curr = rag_fn(q, top_k=current)
        r_cand = rag_fn(q, top_k=candidate)
        results["current"].append({"latency_ms": r_curr["latency_ms"], "docs": r_curr["doc_ids"]})
        results["candidate"].append({"latency_ms": r_cand["latency_ms"], "docs": r_cand["doc_ids"]})
    # Compute metrics and decide
    return results

7. Cache layer and cost tracking in Python

Full implementation of a layer that combines query embedding cache, result cache, and cost tracking:

import hashlib
import time
import redis
import pickle
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class CostTracker:
    """Cost tracking per operation."""
    embedding_calls: int = 0
    embedding_tokens: int = 0
    cache_hits: int = 0
    cache_misses: int = 0
    
    def cost_embedding(self) -> float:
        # $0.02/1M tokens (text-embedding-3-small)
        return (self.embedding_tokens / 1_000_000) * 0.02
    
    def cache_hit_rate(self) -> float:
        total = self.cache_hits + self.cache_misses
        return (self.cache_hits / total * 100) if total > 0 else 0.0

class RAGCachingLayer:
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        result_ttl: int = 1800,
        embed_fn=None,
        vector_db=None,
    ):
        self.redis = redis.from_url(redis_url)
        self.result_ttl = result_ttl
        self.embed_fn = embed_fn
        self.vector_db = vector_db
        self.cost_tracker = CostTracker()
    
    def _query_embedding_key(self, query: str) -> str:
        return f"rag:embed:{hashlib.sha256(query.encode()).hexdigest()}"
    
    def _result_key(self, query: str, top_k: int) -> str:
        return f"rag:result:{hashlib.sha256(f'{query}|{top_k}'.encode()).hexdigest()}"
    
    def get_or_create_embedding(self, query: str) -> list[float]:
        key = self._query_embedding_key(query)
        cached = self.redis.get(key)
        if cached:
            self.cost_tracker.cache_hits += 1
            return pickle.loads(cached)
        self.cost_tracker.cache_misses += 1
        emb = self.embed_fn(query)
        self.cost_tracker.embedding_calls += 1
        self.cost_tracker.embedding_tokens += len(query.split()) * 2  # approx
        self.redis.setex(key, self.result_ttl * 2, pickle.dumps(emb))  # longer TTL for embeddings
        return emb
    
    def get_or_create_result(self, query: str, top_k: int, retrieve_fn) -> dict:
        key = self._result_key(query, top_k)
        cached = self.redis.get(key)
        if cached:
            self.cost_tracker.cache_hits += 1
            return pickle.loads(cached)
        self.cost_tracker.cache_misses += 1
        embedding = self.get_or_create_embedding(query)
        start = time.perf_counter()
        docs = self.vector_db.query(embedding, top_k=top_k)
        result = {"docs": docs, "latency_ms": (time.perf_counter() - start) * 1000}
        self.redis.setex(key, self.result_ttl, pickle.dumps(result))
        return result
    
    def cost_summary(self) -> dict:
        return {
            "embedding_cost_usd": round(self.cost_tracker.cost_embedding(), 6),
            "cache_hit_rate_pct": round(self.cost_tracker.cache_hit_rate(), 2),
            "embedding_calls": self.cost_tracker.embedding_calls,
            "cache_hits": self.cost_tracker.cache_hits,
            "cache_misses": self.cost_tracker.cache_misses,
        }

Usage example and numbers

# 1000 queries/day for 30 days
# Without cache: 30,000 embedding calls
# With 30% cache: 21,000 embedding calls

queries_per_day = 1000
days = 30
cost_per_embedding = 0.00002  # approx for a short query

without_cache = queries_per_day * days * cost_per_embedding
with_30_cache = queries_per_day * days * 0.7 * cost_per_embedding

print(f"No cache: ${without_cache:.2f}/month")
print(f"With 30% cache: ${with_30_cache:.2f}/month (savings ${without_cache - with_30_cache:.2f})")
# No cache: $0.60/month
# With 30% cache: $0.42/month (savings $0.18)

Mini decision framework

1. Identify the biggest cost: embeddings, LLM generation, or infra.
2. Apply one optimization at a time: embedding cache, result cache, top_k, etc.
3. Measure the impact on latency, recall, and cost.
4. Keep only the changes with a clear benefit and no quality degradation.

Metrics to decide whether an optimization stays

• Improvement in latency p95 (quantitative target, e.g. <500ms).
• Reduction in cost per query.
• Impact on retrieval accuracy (recall, MRR).
• Impact on operational complexity (maintenance, deps).

Troubleshooting

1. "I optimized cost and quality dropped"

Symptom: After reducing top_k or enabling aggressive cache, answers get worse.

Solution: Restore the previous configuration. Look for another, less aggressive lever (e.g. embedding cache but not full result cache, or raise top_k by one point).

2. "We see no cache impact"

Symptom: Very low hit rate (<10%) even though there are repeated queries.

Solution: Review the TTL (it may be too short), the cache key (do you normalize the query? do you include top_k/filters?), and the real query pattern. Consider semantic caching if queries vary a lot in wording.

3. "Latency went up when I enabled Redis"

Symptom: You added cache and p95 got worse.

Solution: Verify that Redis is in the same region/VPC as the app. Review serialization (pickle vs msgpack). If the payload is large, consider compression.

4. "Too many optimization initiatives in parallel"

Symptom: You don't know which change caused which effect.

Solution: Prioritize by impact × effort. Implement one optimization at a time, measure for 3-5 days, and document. Then move to the next.

5. "The cache grows uncontrolled"

Symptom: Redis uses more memory than expected.

Solution: Review the TTL (make sure it's not 0 or very high). Use maxmemory and the allkeys-lru policy. Monitor redis-cli INFO memory.

Exercises

Exercise 1: Calculate savings with result cache

You have 2000 queries/day. Each query costs $0.00002 in embedding. Without cache you pay 100%. With a 40% hit rate, how much do you save per month?

queries_per_day = 2000
days = 30
cost_per_query = 0.00002
hit_rate = 0.40

total_queries = queries_per_day * days  # 60,000
without_cache = total_queries * cost_per_query  # $1.20
with_cache = total_queries * (1 - hit_rate) * cost_per_query  # 36,000 * 0.00002 = $0.72
savings = without_cache - with_cache  # $0.48/month

print(f"No cache: ${without_cache:.2f}/month")
print(f"With 40% cache: ${with_cache:.2f}/month")
print(f"Savings: ${savings:.2f}/month ({savings/without_cache*100:.0f}%)")

Exercise 2: Implement should_reembed with a persistent store

Implement a version of should_reembed that uses Redis to persist each document's hash, instead of an in-memory dictionary.

import redis
import hashlib
import json

def content_hash(content: str, metadata: dict = None) -> str:
    payload = content + (json.dumps(metadata or {}, sort_keys=True) if metadata else "")
    return hashlib.sha256(payload.encode()).hexdigest()

def should_reembed_redis(doc_id: str, content: str, metadata: dict, r: redis.Redis) -> bool:
    key = f"doc_hash:{doc_id}"
    new_hash = content_hash(content, metadata)
    stored = r.get(key)
    if stored is None:
        return True  # Never seen
    return stored.decode() != new_hash

def update_doc_hash(doc_id: str, content: str, metadata: dict, r: redis.Redis, ttl: int = 86400 * 90):
    key = f"doc_hash:{doc_id}"
    h = content_hash(content, metadata)
    r.setex(key, ttl, h)

Exercise 3: Batch embedding with error handling

Improve the batch_embed function so that, if a batch fails, it retries with smaller sub-batches instead of failing everything.

def batch_embed_resilient(texts: list[str], embed_fn, batch_size: int = 100) -> list[list[float]]:
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        try:
            embeddings = embed_fn(batch)
            results.extend(embeddings)
        except Exception as e:
            if len(batch) == 1:
                raise
            mid = len(batch) // 2
            left = batch_embed_resilient(batch[:mid], embed_fn, batch_size)
            right = batch_embed_resilient(batch[mid:], embed_fn, batch_size)
            results.extend(left)
            results.extend(right)
    return results

Exercise 4: Cache hit rate and cost tracking

Write a function that, given a CostTracker with cache_hits and cache_misses, returns the hit rate in % and the estimated embedding cost saved, assuming $0.00002 per avoided embedding.

def cache_stats(tracker: CostTracker, cost_per_embedding: float = 0.00002) -> dict:
    total = tracker.cache_hits + tracker.cache_misses
    hit_rate = (tracker.cache_hits / total * 100) if total > 0 else 0.0
    saved_cost = tracker.cache_hits * cost_per_embedding
    return {
        "hit_rate_pct": round(hit_rate, 2),
        "saved_embeddings": tracker.cache_hits,
        "saved_cost_usd": round(saved_cost, 4),
    }

Exercise 5: Decide top_k based on recall

You have 20 evaluation queries. With top_k=8 you retrieve the correct document in 18 of 20. With top_k=5 you retrieve it in 16 of 20. Which top_k would you choose if the cost per additional document matters?

def choose_top_k(results: list[tuple[int, float]]) -> int:
    """results = [(top_k, recall), ...]"""
    # Goal: recall >= 0.85 with the minimum top_k
    for top_k, recall in sorted(results, key=lambda x: x[0]):
        if recall >= 0.85:
            return top_k
    return results[-1][0]  # fallback to the highest

Exercise 6: Project costs with and without cache over 6 months

Starting from 500 queries/day with 10% monthly growth, project the embedding cost (at $0.00002/query) over 6 months: (a) without cache, (b) with a stable 30% cache.

base_queries = 500
growth = 0.10
cost_per = 0.00002
cache_rate = 0.30
months = 6

without, with_cache = 0, 0
q = base_queries
for m in range(months):
    days = 30
    monthly = q * days
    without += monthly * cost_per
    with_cache += monthly * (1 - cache_rate) * cost_per
    q *= (1 + growth)

print(f"No cache (6 months): ${without:.2f}")
print(f"With 30% cache (6 months): ${with_cache:.2f}")
print(f"Savings: ${without - with_cache:.2f}")

Summary

• Embedding cache: Don't re-embed unchanged documents; use a content hash and a store (Redis or DB).
• Result cache with TTL: Cache RAG answers per query; a 30-60% hit rate is common and reduces cost and latency.
• Batch operations: Group embeddings into batches for fewer calls and better throughput.
• Index tuning (PQ): Product Quantization reduces memory and can improve speed with a small trade-off in recall.
• Model per environment: Dev with local/free models; prod with managed models + aggressive cache.
• top_k: Tune per use case (3-5 for FAQ, 5-8 for docs, 8-12 for research); measure recall before lowering it.
• Optimizing well means measuring: One optimization at a time, clear metrics, keep only what provides net benefit.
• Cost tracking: Implement CostTracker and cache_hit_rate to make data-driven decisions.

Additional resources

1. AWS Caching Best Practices — Production caching patterns.
2. OpenAI Embeddings Pricing — Current embedding prices.
3. Qdrant Quantization — PQ and other techniques in Qdrant.
4. ChromaDB Batch Operations — Efficient ingestion.
5. Redis Caching Patterns — TTL, eviction, best practices.
6. Semantic Caching for RAG — Cache by semantic similarity.
7. LangChain Caching — Caching in LLM pipelines.
8. Cost Optimization & Caching Guide — Internal AI cost optimization guide.

Estimated time: 35-45 minutes
Next: 07-zero-downtime-migration.md