Module 2: How Do Embeddings Work?

Mini-Project: Robust API Client for Production

Next →

Project overview

You'll build a production-ready embeddings client that implements all the patterns seen in this module: batch processing, caching, retry logic with exponential backoff, rate limiting, normalization, and logging. This client is reusable in real projects.

By completing this project, you'll have a solid Python library for generating embeddings at scale with all the production optimizations.

Project objectives

Features:

  1. ✅ Generate embeddings (single and batch)
  2. ✅ Disk caching (avoid regenerating)
  3. ✅ Retry logic (exponential backoff)
  4. ✅ Rate limiting (respect RPM limits)
  5. ✅ Automatic batch processing
  6. ✅ Optional normalization
  7. ✅ Detailed logging
  8. ✅ CLI for testing

Project structure

embeddings-client/
├── src/
│   ├── __init__.py
│   ├── client.py           # Main client
│   ├── cache.py            # Cache system
│   ├── rate_limiter.py     # Rate limiter
│   └── utils.py            # Utilities (normalize, etc.)
├── tests/
│   └── test_client.py
├── cache/                  # Embedding cache (auto-generated)
├── .env
├── requirements.txt
└── README.md

Initial setup

1. requirements.txt

openai==1.54.0
python-dotenv==1.0.0
numpy==1.26.4

2. .env

OPENAI_API_KEY=your-api-key-here

3. Install dependencies

pip install -r requirements.txt

Implementation

Step 1: utils.py (Utilities)

"""
Utilities for embeddings
"""
import numpy as np
from typing import List

def normalize_embedding(embedding: List[float]) -> List[float]:
    """
    Normalize an embedding to magnitude 1.0
    
    Args:
        embedding: The vector to normalize
    
    Returns:
        The normalized embedding
    """
    embedding = np.array(embedding)
    norm = np.linalg.norm(embedding)
    
    if norm == 0:
        return embedding.tolist()
    
    return (embedding / norm).tolist()

def cosine_similarity(emb_a: List[float], emb_b: List[float]) -> float:
    """
    Calculate cosine similarity between two embeddings
    
    Args:
        emb_a: First embedding
        emb_b: Second embedding
    
    Returns:
        Similarity score [0, 1]
    """
    a = np.array(emb_a)
    b = np.array(emb_b)
    
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

Step 2: cache.py (Cache system)

"""
Cache system for embeddings
"""
import json
import hashlib
from pathlib import Path
from typing import Optional, List

class EmbeddingCache:
    """Disk-based embedding cache"""
    
    def __init__(self, cache_dir: str = "./cache"):
        """
        Initialize the cache
        
        Args:
            cache_dir: Directory to store the cache
        """
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
    
    def _get_cache_key(self, text: str, model: str, dimensions: Optional[int]) -> str:
        """
        Generate a unique key for text + config
        
        Args:
            text: The text
            model: The model used
            dimensions: The dimensions (None = default)
        
        Returns:
            An MD5 hash as the key
        """
        # Combine text + model + dimensions
        cache_string = f"{text}:{model}:{dimensions}"
        return hashlib.md5(cache_string.encode()).hexdigest()
    
    def get(self, text: str, model: str, dimensions: Optional[int] = None) -> Optional[List[float]]:
        """
        Get an embedding from the cache
        
        Returns:
            The embedding, or None if it doesn't exist
        """
        cache_file = self.cache_dir / f"{self._get_cache_key(text, model, dimensions)}.json"
        
        if cache_file.exists():
            with open(cache_file, 'r') as f:
                data = json.load(f)
                return data['embedding']
        
        return None
    
    def set(self, text: str, model: str, dimensions: Optional[int], embedding: List[float]) -> None:
        """
        Save an embedding to the cache
        """
        cache_file = self.cache_dir / f"{self._get_cache_key(text, model, dimensions)}.json"
        
        with open(cache_file, 'w') as f:
            json.dump({
                'text': text[:100],  # Save a snippet
                'model': model,
                'dimensions': dimensions,
                'embedding': embedding
            }, f)
    
    def clear(self) -> int:
        """
        Clear the entire cache
        
        Returns:
            The number of files removed
        """
        count = 0
        for cache_file in self.cache_dir.glob("*.json"):
            cache_file.unlink()
            count += 1
        return count

Step 3: rate_limiter.py (Rate limiting)

"""
Rate limiter to respect API limits
"""
import time
from collections import deque
from typing import Optional

class RateLimiter:
    """Rate limiter with a sliding window"""
    
    def __init__(self, max_requests: int, window_seconds: int):
        """
        Initialize the rate limiter
        
        Args:
            max_requests: Maximum requests in the window
            window_seconds: Window size in seconds
        """
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()
    
    def acquire(self) -> None:
        """
        Wait if necessary to avoid exceeding the rate limit
        """
        now = time.time()
        
        # Remove requests outside the window
        while self.requests and self.requests[0] < now - self.window_seconds:
            self.requests.popleft()
        
        # If we're at the limit, wait
        if len(self.requests) >= self.max_requests:
            sleep_time = self.requests[0] + self.window_seconds - now + 0.1
            if sleep_time > 0:
                time.sleep(sleep_time)
            self.requests.popleft()
        
        # Register the request
        self.requests.append(time.time())
    
    def reset(self) -> None:
        """Reset the rate limiter"""
        self.requests.clear()

Step 4: client.py (Main client)

"""
Robust embeddings client for production
"""
import logging
import time
from typing import List, Optional, Union
from openai import OpenAI, APIError, RateLimitError, APIConnectionError
from dotenv import load_dotenv
import os

from .cache import EmbeddingCache
from .rate_limiter import RateLimiter
from .utils import normalize_embedding

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class EmbeddingsClient:
    """Production-ready embeddings client"""
    
    def __init__(
        self,
        model: str = "text-embedding-3-small",
        dimensions: Optional[int] = None,
        use_cache: bool = True,
        cache_dir: str = "./cache",
        max_retries: int = 3,
        max_rpm: int = 500,
        normalize: bool = False
    ):
        """
        Initialize the client
        
        Args:
            model: The embedding model
            dimensions: Embedding dimensions (None = default)
            use_cache: Use the disk cache
            cache_dir: The cache directory
            max_retries: Maximum number of retries
            max_rpm: Maximum requests per minute
            normalize: Normalize embeddings to magnitude 1.0
        """
        # Setup the OpenAI client
        load_dotenv()
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        
        # Config
        self.model = model
        self.dimensions = dimensions
        self.use_cache = use_cache
        self.max_retries = max_retries
        self.normalize = normalize
        
        # Cache
        if use_cache:
            self.cache = EmbeddingCache(cache_dir)
        
        # Rate limiter
        self.rate_limiter = RateLimiter(max_requests=max_rpm, window_seconds=60)
        
        # Stats
        self.stats = {
            'total_requests': 0,
            'cache_hits': 0,
            'api_calls': 0,
            'errors': 0
        }
        
        logger.info(f"EmbeddingsClient initialized (model={model}, cache={use_cache})")
    
    def embed(self, text: str) -> Optional[List[float]]:
        """
        Generate an embedding for a text
        
        Args:
            text: The text to convert
        
        Returns:
            The embedding, or None if it fails
        """
        self.stats['total_requests'] += 1
        
        # Try the cache
        if self.use_cache:
            cached = self.cache.get(text, self.model, self.dimensions)
            if cached:
                self.stats['cache_hits'] += 1
                logger.debug(f"Cache hit: {text[:50]}...")
                return cached
        
        # Generate the embedding
        embedding = self._generate_embedding_with_retry(text)
        
        if embedding is None:
            return None
        
        # Normalize if required
        if self.normalize:
            embedding = normalize_embedding(embedding)
        
        # Save to the cache
        if self.use_cache and embedding:
            self.cache.set(text, self.model, self.dimensions, embedding)
        
        return embedding
    
    def embed_batch(self, texts: List[str], batch_size: int = 100) -> List[Optional[List[float]]]:
        """
        Generate embeddings for multiple texts
        
        Args:
            texts: A list of texts
            batch_size: The batch size
        
        Returns:
            A list of embeddings
        """
        embeddings = []
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i+batch_size]
            
            # Process the batch
            batch_embeddings = self._process_batch(batch)
            embeddings.extend(batch_embeddings)
            
            logger.info(f"Batch {i//batch_size + 1}: {len(batch)} texts processed")
        
        return embeddings
    
    def _process_batch(self, texts: List[str]) -> List[Optional[List[float]]]:
        """
        Process a batch of texts (with cache)
        
        Args:
            texts: A list of texts
        
        Returns:
            A list of embeddings
        """
        embeddings = []
        texts_to_generate = []
        indices_to_generate = []
        
        # Check the cache
        for idx, text in enumerate(texts):
            if self.use_cache:
                cached = self.cache.get(text, self.model, self.dimensions)
                if cached:
                    embeddings.append(cached)
                    self.stats['cache_hits'] += 1
                    continue
            
            # Needs generation
            embeddings.append(None)  # Placeholder
            texts_to_generate.append(text)
            indices_to_generate.append(idx)
        
        # Generate the missing embeddings
        if texts_to_generate:
            generated = self._generate_embeddings_batch_with_retry(texts_to_generate)
            
            # Insert the generated embeddings
            for idx, emb in zip(indices_to_generate, generated):
                if emb:
                    # Normalize if required
                    if self.normalize:
                        emb = normalize_embedding(emb)
                    
                    embeddings[idx] = emb
                    
                    # Cache
                    if self.use_cache:
                        self.cache.set(texts[idx], self.model, self.dimensions, emb)
        
        return embeddings
    
    def _generate_embedding_with_retry(self, text: str) -> Optional[List[float]]:
        """
        Generate an embedding with retry logic
        
        Args:
            text: The text
        
        Returns:
            The embedding, or None if it fails
        """
        for attempt in range(self.max_retries):
            try:
                # Rate limiting
                self.rate_limiter.acquire()
                
                # API call
                response = self.client.embeddings.create(
                    model=self.model,
                    input=text,
                    dimensions=self.dimensions
                )
                
                self.stats['api_calls'] += 1
                return response.data[0].embedding
            
            except RateLimitError:
                wait_time = 2 ** attempt
                logger.warning(f"Rate limit. Retry {attempt+1}/{self.max_retries} in {wait_time}s")
                time.sleep(wait_time)
            
            except (APIError, APIConnectionError) as e:
                logger.error(f"API error: {e}")
                if attempt < self.max_retries - 1:
                    time.sleep(1)
        
        # Failed
        self.stats['errors'] += 1
        logger.error(f"Failed to generate embedding for: {text[:50]}...")
        return None
    
    def _generate_embeddings_batch_with_retry(self, texts: List[str]) -> List[Optional[List[float]]]:
        """
        Generate a batch of embeddings with retry
        
        Args:
            texts: A list of texts
        
        Returns:
            A list of embeddings
        """
        for attempt in range(self.max_retries):
            try:
                # Rate limiting
                self.rate_limiter.acquire()
                
                # API call (batch)
                response = self.client.embeddings.create(
                    model=self.model,
                    input=texts,
                    dimensions=self.dimensions
                )
                
                self.stats['api_calls'] += 1
                return [item.embedding for item in response.data]
            
            except RateLimitError:
                wait_time = 2 ** attempt
                logger.warning(f"Rate limit. Retry {attempt+1}/{self.max_retries} in {wait_time}s")
                time.sleep(wait_time)
            
            except (APIError, APIConnectionError) as e:
                logger.error(f"API error: {e}")
                if attempt < self.max_retries - 1:
                    time.sleep(1)
        
        # Failed
        self.stats['errors'] += len(texts)
        logger.error(f"Failed to generate batch of {len(texts)} embeddings")
        return [None] * len(texts)
    
    def get_stats(self) -> dict:
        """
        Get the client's statistics
        
        Returns:
            A dict with stats
        """
        cache_hit_rate = 0
        if self.stats['total_requests'] > 0:
            cache_hit_rate = self.stats['cache_hits'] / self.stats['total_requests']
        
        return {
            **self.stats,
            'cache_hit_rate': f"{cache_hit_rate:.2%}"
        }
    
    def clear_cache(self) -> int:
        """
        Clear the cache
        
        Returns:
            The number of files removed
        """
        if self.use_cache:
            return self.cache.clear()
        return 0

Testing

Create test_client.py

"""
Tests for the embeddings client
"""
from src.client import EmbeddingsClient
from src.utils import cosine_similarity

def test_single_embedding():
    """Test: Generate 1 embedding"""
    client = EmbeddingsClient(use_cache=False)
    
    embedding = client.embed("Python is a language")
    
    assert embedding is not None
    assert len(embedding) == 1536  # Default dimensions
    print("✅ Test single embedding passed")

def test_batch_embedding():
    """Test: Generate a batch of embeddings"""
    client = EmbeddingsClient(use_cache=False)
    
    texts = ["Python", "JavaScript", "Go"]
    embeddings = client.embed_batch(texts)
    
    assert len(embeddings) == 3
    assert all(emb is not None for emb in embeddings)
    print("✅ Test batch embedding passed")

def test_cache():
    """Test: The cache works"""
    client = EmbeddingsClient(use_cache=True)
    
    text = "Test caching"
    
    # First call (API)
    emb1 = client.embed(text)
    api_calls_1 = client.stats['api_calls']
    
    # Second call (cache)
    emb2 = client.embed(text)
    api_calls_2 = client.stats['api_calls']
    
    assert emb1 == emb2
    assert api_calls_2 == api_calls_1  # No new API call
    assert client.stats['cache_hits'] == 1
    
    print("✅ Test cache passed")
    
    # Cleanup
    client.clear_cache()

def test_normalization():
    """Test: Normalization works"""
    import numpy as np
    
    client = EmbeddingsClient(normalize=True, use_cache=False)
    
    embedding = client.embed("Python is popular")
    magnitude = np.linalg.norm(embedding)
    
    assert 0.99 <= magnitude <= 1.01  # Magnitude ~1.0
    print(f"✅ Test normalization passed (magnitude={magnitude:.4f})")

def test_reduced_dimensions():
    """Test: Reduced dimensions"""
    client = EmbeddingsClient(dimensions=512, use_cache=False)
    
    embedding = client.embed("Python")
    
    assert len(embedding) == 512
    print("✅ Test reduced dimensions passed")

if __name__ == "__main__":
    print("Running tests...\n")
    
    test_single_embedding()
    test_batch_embedding()
    test_cache()
    test_normalization()
    test_reduced_dimensions()
    
    print("\n✅ All tests passed!")

Testing CLI

Create main.py

"""
CLI for testing the client
"""
from src.client import EmbeddingsClient
from src.utils import cosine_similarity

def main():
    """Interactive demo"""
    print("=== Embeddings Client Demo ===\n")
    
    # Initialize the client
    client = EmbeddingsClient(
        model="text-embedding-3-small",
        dimensions=512,  # Reduced
        use_cache=True,
        normalize=True,
        max_rpm=100
    )
    
    # Test 1: Single embedding
    print("Test 1: Single embedding")
    embedding = client.embed("Python is a programming language")
    print(f"✅ Embedding generated: {len(embedding)} dims\n")
    
    # Test 2: Similarity
    print("Test 2: Semantic similarity")
    emb_a = client.embed("Python is popular")
    emb_b = client.embed("Python is widely used")
    emb_c = client.embed("A cat sleeps on a couch")
    
    sim_ab = cosine_similarity(emb_a, emb_b)
    sim_ac = cosine_similarity(emb_a, emb_c)
    
    print(f"'Python is popular' vs 'Python is widely used': {sim_ab:.4f}")
    print(f"'Python is popular' vs 'A cat sleeps on a couch': {sim_ac:.4f}\n")
    
    # Test 3: Batch
    print("Test 3: Batch processing")
    texts = [f"Document {i}" for i in range(10)]
    embeddings = client.embed_batch(texts, batch_size=5)
    print(f"✅ {len(embeddings)} embeddings generated\n")
    
    # Stats
    print("Stats:")
    stats = client.get_stats()
    for key, value in stats.items():
        print(f"  {key}: {value}")

if __name__ == "__main__":
    main()

Run:

python main.py

Expected output:

=== Embeddings Client Demo ===

Test 1: Single embedding
✅ Embedding generated: 512 dims

Test 2: Semantic similarity
'Python is popular' vs 'Python is widely used': 0.8721
'Python is popular' vs 'A cat sleeps on a couch': 0.4532

Test 3: Batch processing
✅ 10 embeddings generated

Stats:
  total_requests: 13
  cache_hits: 0
  api_calls: 3
  errors: 0
  cache_hit_rate: 0.00%

Project validation

Checklist:

  • The client generates single and batch embeddings ✅
  • The cache works (the second call doesn't hit the API) ✅
  • Retry logic handles rate limits ✅
  • The rate limiter respects RPM ✅
  • Normalization works (magnitude ~1.0) ✅
  • Reduced dimensions work (512 dims) ✅
  • Logging reports progress ✅
  • Tests pass without errors ✅

Optional extensions

1. Add async support (for high volume):

import asyncio
from openai import AsyncOpenAI

class AsyncEmbeddingsClient:
    """Async version of the client"""
    
    async def embed(self, text: str):
        # Implement the async version
        pass

2. Add a database cache (Redis/PostgreSQL):

# Instead of JSON files, use Redis:
import redis

class RedisCache:
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379)
    
    def get(self, key):
        # Implement
        pass

Project summary

What you implemented:

  • ✅ Robust client: Production-ready embeddings client
  • ✅ Caching: Disk-based cache (JSON)
  • ✅ Retry logic: Automatic exponential backoff
  • ✅ Rate limiting: Sliding window rate limiter
  • ✅ Batch processing: Automatic with cache integration
  • ✅ Normalization: Optional (magnitude 1.0)
  • ✅ Logging: Detailed for monitoring
  • ✅ Tests: A complete validation suite

Patterns applied:

  1. Exponential backoff (handling rate limits)
  2. Cache-aside pattern (check cache first)
  3. Batch processing (reduce API calls)
  4. Dependency injection (flexible config)

Module 2 conclusion

What you learned in the module:

Architecture:

  • ✅ Transformer encoders (self-attention, multi-head attention)
  • ✅ Tokenization (BPE, WordPiece, tiktoken)
  • ✅ Contextualization (dynamic embeddings)
  • ✅ Pooling strategies (mean, CLS, max)
  • ✅ Normalization (L2 norm, unit magnitude)

Production API:

  • ✅ Advanced OpenAI API (batch, dimensions, retry)
  • ✅ Production patterns (caching, rate limiting)
  • ✅ Robust client (complete project)

Lines of code generated: ~1,200 (production-ready)

Next module

Module 3: Embedding Models Comparison

You'll learn:

  • OpenAI vs Sentence-BERT vs BGE
  • Open-source embeddings (local)
  • Benchmarking (MTEB)
  • Domain-specific embeddings
  • Multi-language embeddings
  • When to use which model

From architecture to model comparison.

Module 2 completed ✅ The complete pipeline: Tokens → Contextualization → Pooling → Normalization → Production

← PreviousOpenAI API Advanced: Parameters and Production PatternsNext →Introduction to Module 3: Embedding Models Compared