Production: Scaling and Caching

Objectives

• Optimize the performance of an embedding system in production
• Implement caching to reduce costs and latency
• Manage scaling and high availability

Production Challenges

During development, calling the OpenAI API for every request isn’t an issue: you might make ten calls a day. In production, that same code hits three walls simultaneously. First, cost: every call is billed, and a full re-indexing of a corpus is paid for in full every time it runs. Second, latency: a network round trip of 100 to 300 ms per request adds to search time and generation time, and the user feels the sum of all three. Finally, rate limits: OpenAI limits the number of calls per minute—a limit you’ll never see in development but will hit on launch day.

All three problems largely have the same solution: don’t call the API when it’s not necessary.

Caching Embeddings

An embedding is deterministic: the same text with the same model always returns the same vector. This makes caching highly effective here, far more so than for a generative LLM. For an application with moderate traffic, the standard library’s lru_cache decorator is sufficient and takes just two lines. The return value is converted to a tuple because lru_cache requires hashable values; that’s the only wrinkle to know.

from functools import lru_cache
from openai import OpenAI
import hashlib
import json

client = OpenAI()

@lru_cache(maxsize=10_000)
def embedding_cache(text: str, model: str = "text-embedding-3-large") -> tuple:
    """LRU cache for embeddings (tuple because it's hashable)."""
    response = client.embeddings.create(input=text, model=model)
    return tuple(response.data[0].embedding)

# Usage - the 2nd call is instant
emb1 = embedding_cache("Hello world")  # API call
emb2 = embedding_cache("Hello world")  # From cache

This cache disappears on any restart and isn’t shared across processes: as soon as you run two instances behind a load balancer, each rebuilds its own. Redis solves both problems. The key is a hash of the model-text pair, which prevents a model change from serving old vectors, and a thirty-day TTL caps storage growth.

The get_embeddings_batch method deserves close attention, as it’s what makes the difference during ingestion. It doesn’t just check the cache text by text: it separates the misses, sends them in a single API call, then reinserts each result into its original position using indices_a_calculer. On a batch of a thousand documents where nine hundred are already known, you go from one call per document to a single call for the remaining hundred.

import redis
import numpy as np
import json

class EmbeddingCacheRedis:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.client = OpenAI()
        self.ttl = 86400 * 30  # 30 days

    def _cache_key(self, text: str, model: str) -> str:
        h = hashlib.md5(f"{model}:{text}".encode()).hexdigest()
        return f"emb:{h}"

    def get_embedding(
        self, text: str, model: str = "text-embedding-3-large"
    ) -> list[float]:
        key = self._cache_key(text, model)

        # Check cache
        cached = self.redis.get(key)
        if cached:
            return json.loads(cached)

        # Generate and cache
        response = self.client.embeddings.create(
            input=text, model=model
        )
        embedding = response.data[0].embedding

        self.redis.setex(key, self.ttl, json.dumps(embedding))
        return embedding

    def get_embeddings_batch(
        self, texts: list[str], model: str = "text-embedding-3-large"
    ) -> list[list[float]]:
        """Batch with partial caching."""
        results = [None] * len(texts)
        to_calculate = []
        indices_to_calculate = []

        # Check cache for each text
        for i, text in enumerate(texts):
            key = self._cache_key(text, model)
            cached = self.redis.get(key)
            if cached:
                results[i] = json.loads(cached)
            else:
                to_calculate.append(text)
                indices_to_calculate.append(i)

        # Calculate misses in a single call
        if to_calculate:
            response = self.client.embeddings.create(
                input=to_calculate, model=model
            )
            for resp_item in sorted(response.data, key=lambda x: x.index):
                idx = indices_to_calculate[resp_item.index]
                text = to_calculate[resp_item.index]
                embedding = resp_item.embedding

                results[idx] = embedding
                key = self._cache_key(text, model)
                self.redis.setex(key, self.ttl, json.dumps(embedding))

        print(f"Cache: {len(texts) - len(to_calculate)}/{len(texts)} hits")
        return results

# Usage
cache = EmbeddingCacheRedis()
emb = cache.get_embedding("cache test")

Handling Rate Limits

Caching reduces traffic but doesn’t eliminate it, and a traffic spike will eventually trigger a RateLimitError. The correct response isn’t to retry immediately—which worsens congestion—but to wait, doubling the delay each time: one second, two, four, eight, sixteen. Five attempts thus cover about thirty seconds, which absorbs the vast majority of temporary spikes, and the final attempt re-raises the exception so the incident is reported instead of being silently swallowed.

import time
from openai import RateLimitError

class RobustEmbeddingClient:
    def __init__(self, max_retries: int = 5):
        self.client = OpenAI()
        self.max_retries = max_retries

    def create(
        self, input: str | list[str], model: str = "text-embedding-3-large",
        **kwargs
    ):
        """Call with exponential retry."""
        for attempt in range(self.max_retries):
            try:
                return self.client.embeddings.create(
                    input=input, model=model, **kwargs
                )
            except RateLimitError as e:
                if attempt == self.max_retries - 1:
                    raise
                wait = 2 ** attempt
                print(f"Rate limit hit, pausing for {wait}s...")
                time.sleep(wait)

Asynchronous Processing

For ingestion pipelines, the bottleneck isn’t computation but network wait: your program spends most of its time doing nothing. Asynchronous processing lets you keep multiple requests in flight. The semaphore intentionally limits this parallelism, because launching a hundred simultaneous calls only triggers the rate limits you just learned to handle—three to five concurrent requests is a healthy setting. Texts are first split into batches of one hundred, and the starting index of each batch travels with the task so that as_completed, which returns results out of order, can still place them correctly.

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def generate_embeddings_async(
    texts: list[str],
    model: str = "text-embedding-3-large",
    max_concurrent: int = 5
) -> list[list[float]]:
    """Generate embeddings in parallel with a semaphore."""
    semaphore = asyncio.Semaphore(max_concurrent)
    results = [None] * len(texts)

    async def process_batch(batch_idx, batch):
        async with semaphore:
            response = await async_client.embeddings.create(
                input=batch, model=model
            )
            return batch_idx, [
                d.embedding
                for d in sorted(response.data, key=lambda x: x.index)
            ]

    # Split into batches of 100
    tasks = []
    batch_size = 100
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        tasks.append(process_batch(i, batch))

    for coro in asyncio.as_completed(tasks):
        batch_idx, embeddings = await coro
        for j, emb in enumerate(embeddings):
            results[batch_idx + j] = emb

    return results

# Usage
embeddings = asyncio.run(
    generate_embeddings_async(my_texts, max_concurrent=3)
)

Optimizing Storage

Then there’s the infrastructure bill. Ten thousand vectors of 3,072 dimensions in float32 occupy about 122 MB; a million occupies twelve gigabytes, which must be held in memory for search to remain fast. Quantization to int8 maps each coordinate to 256 integer values, dividing the footprint by four to around 30 MB in the example below. The loss of precision exists but remains marginal compared to the similarity gaps you’re working with.

import numpy as np

def quantize_int8(embeddings: np.ndarray) -> tuple:
    """Quantize embeddings to int8 (4x less memory)."""
    min_val = embeddings.min(axis=1, keepdims=True)
    max_val = embeddings.max(axis=1, keepdims=True)
    scale = (max_val - min_val) / 255.0

    quantized = ((embeddings - min_val) / scale).astype(np.int8)
    return quantized, min_val, scale

def dequantize_int8(quantized, min_val, scale):
    """Restore embeddings from int8."""
    return quantized.astype(np.float32) * scale + min_val

# Size comparison
embeddings = np.random.randn(10000, 3072).astype(np.float32)
print(f"float32 : {embeddings.nbytes / 1e6:.1f} MB")  # ~122 MB

quantized, min_val, scale = quantize_int8(embeddings)
print(f"int8 :    {quantized.nbytes / 1e6:.1f} MB")    # ~30 MB

However, the simplest lever remains the API’s dimensions parameter, which produces shorter vectors directly without any post-processing on your part: 3,072 dimensions weigh about 24 KB per vector, 1,024 dimensions drop to 8 KB, and 256 dimensions to 2 KB.

# 3072 dim -> ~24 KB per vector
# 1024 dim -> ~8 KB per vector  (API reduction)
# 256 dim  -> ~2 KB per vector  (API reduction)

response = client.embeddings.create(
    input="text",
    model="text-embedding-3-large",
    dimensions=1024  # 3x less storage
)

Production Architecture

Assembled, these pieces give a lean pipeline: Redis cache in front, robust client for calls that get through anyway. Ingestion takes the batch route, search caches even the query embedding—and that last point isn’t trivial, because user queries repeat far more than you’d imagine.

class ProductionPipeline:
    """RAG pipeline optimized for production."""

    def __init__(self):
        self.cache = EmbeddingCacheRedis()
        self.robust_client = RobustEmbeddingClient()

    def ingest(self, documents: list[dict]):
        """Ingestion with cache and retry."""
        texts = [d["text"] for d in documents]
        embeddings = self.cache.get_embeddings_batch(texts)
        # Store in vector database...

    def search(self, query: str, k: int = 5):
        """Search with query caching."""
        query_emb = self.cache.get_embedding(query)
        # Search in vector database...
        return query_emb

Production Checklist

Two lines in this table produce no performance gain yet are the most important. Monitoring the cache hit rate is your economic health indicator: if it collapses, your bill triples before anyone notices. And backing up embeddings saves you, on the day of an incident, from having to repay and re-wait for the full encoding of the corpus.

Summary

• Redis cache reduces costs and latency by 90%+
• Exponential retry handles OpenAI rate limits
• Asynchronous processing speeds up ingestion of large corpora
• int8 quantization reduces storage by 4x
• Dimension reduction at the API is the simplest lever

Test Your Knowledge

Twenty-two lessons on embeddings and RAG: the essentials fit in five questions.

The thread of the course: meaning becomes measurable—search, ranking, recommendation are just applications. Production (scaling, caching) does the rest.