Module 2: How Do Embeddings Work?

OpenAI API Advanced: Parameters and Production Patterns

Capsule overview

The OpenAI Embeddings API is simple on the surface, but it has parameters and optimizations that are critical for production: batch processing, reduced dimensions, encoding formats, robust error handling, rate limiting, and cost optimization.

In this capsule you'll learn the API's advanced parameters, how to process multiple texts efficiently with batch processing, implement retry logic with exponential backoff, handle rate limits, and optimize costs. You'll also see production-ready code you can use in your projects.

By the end, you'll be able to build robust embedding systems that handle volume at scale.


OpenAI Embeddings API: Full parameters

Basic API (seen in module 1):

from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

response = client.embeddings.create(
    model="text-embedding-3-small",
    input="Python is a language"
)

embedding = response.data[0].embedding

Advanced parameters:

response = client.embeddings.create(
    model="text-embedding-3-small",      # Model
    input="...",                         # Text or list of texts
    dimensions=512,                      # Reduce dimensionality (optional)
    encoding_format="float",             # "float" or "base64"
    user="user-123"                      # User identifier (optional)
)

Let's look at each parameter in detail.


Parameter: model

Available models (January 2026):

ModelDimensionsCost/1M tokensPerformanceUse
text-embedding-3-small1536$0.020GoodGeneral
text-embedding-3-large3072$0.130ExcellentHigh precision
text-embedding-ada-0021536$0.100Good (legacy)Legacy

Recommendation:

  • Prototype: text-embedding-3-small (cheap, fast)
  • Production (high quality): text-embedding-3-large (better performance)

Empirical comparison:

import numpy as np

texts = [
    "Python is a programming language",
    "Python is a snake"
]

# Small
embeddings_small = [
    client.embeddings.create(model="text-embedding-3-small", input=t).data[0].embedding
    for t in texts
]

# Large
embeddings_large = [
    client.embeddings.create(model="text-embedding-3-large", input=t).data[0].embedding
    for t in texts
]

# Compare similarity
def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

sim_small = cosine_similarity(embeddings_small[0], embeddings_small[1])
sim_large = cosine_similarity(embeddings_large[0], embeddings_large[1])

print(f"Similarity (small): {sim_small:.4f}")
print(f"Similarity (large): {sim_large:.4f}")

Typical output:

Similarity (small): 0.78
Similarity (large): 0.72  ← Better differentiation

large differentiates contexts better (better for RAG).


Parameter: dimensions

What it does:

Reduces the dimensionality of the embedding (without retraining the model).

By default:

  • text-embedding-3-small: 1536 dims
  • text-embedding-3-large: 3072 dims

With dimensions:

# Reduce from 1536 → 512
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="Python is popular",
    dimensions=512  # Reduce
)

embedding = response.data[0].embedding
print(f"Dimensions: {len(embedding)}")  # 512

Why reduce dimensions:

1. Smaller storage

# 1536 dims:
# 1M documents × 1536 × 4 bytes (float32) = 6 GB

# 512 dims:
# 1M documents × 512 × 4 bytes = 2 GB  ← 3x less storage

2. Faster search

# Fewer dimensions = faster dot product
# 512 dims → ~3x faster than 1536 dims

3. Trade-off: Loses some precision

# But the loss is minimal (~5% in benchmarks)

Precision vs dimensions benchmark:

from openai import OpenAI
import numpy as np

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# Test texts
query = "Python is a programming language"
docs = [
    "Python is an interpreted language",
    "JavaScript is a web language",
    "A cat sleeps on a couch"
]

# Test with different dimensions
dimensions_list = [512, 1024, 1536]

for dims in dimensions_list:
    # Embed query
    query_emb = client.embeddings.create(
        model="text-embedding-3-small",
        input=query,
        dimensions=dims
    ).data[0].embedding
    
    # Embed docs
    docs_embs = [
        client.embeddings.create(
            model="text-embedding-3-small",
            input=doc,
            dimensions=dims
        ).data[0].embedding
        for doc in docs
    ]
    
    # Calculate similarities
    sims = [np.dot(query_emb, doc_emb) for doc_emb in docs_embs]
    
    print(f"\nDimensions: {dims}")
    print(f"Similarities: {[f'{s:.4f}' for s in sims]}")

Expected output:

Dimensions: 512
Similarities: ['0.82', '0.65', '0.45']

Dimensions: 1024
Similarities: ['0.85', '0.68', '0.43']

Dimensions: 1536
Similarities: ['0.87', '0.70', '0.42']

Conclusion: 512 dims captures ~95% of the precision of 1536 dims.


Parameter: input (batch processing)

Single input:

# 1 text
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="Python is popular"
)

embedding = response.data[0].embedding

Batch input (multiple texts):

# Multiple texts in 1 call
texts = [
    "Python is popular",
    "JavaScript is fast",
    "Go is concurrent"
]

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts  # List of texts
)

# Extract embeddings
embeddings = [item.embedding for item in response.data]
print(f"Generated: {len(embeddings)} embeddings")

Why use batch:

1. More efficient (less HTTP overhead)

# Single: 100 calls × 50ms latency = 5 seconds
# Batch: 1 call × 100 texts = ~200ms

# Speedup: ~25x

2. Lower rate limiting

# OpenAI rate limits: requests per minute (RPM)
# 1 batch request = 1 RPM (no matter how many texts)

Batch limit:

# OpenAI allows up to ~2048 texts per batch
# (depends on total tokens)

MAX_BATCH_SIZE = 2048

def embed_batch(texts, batch_size=2048):
    """Embed multiple texts in batches"""
    embeddings = []
    
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=batch
        )
        
        batch_embeddings = [item.embedding for item in response.data]
        embeddings.extend(batch_embeddings)
    
    return embeddings

# Usage
texts = ["Text " + str(i) for i in range(5000)]
embeddings = embed_batch(texts, batch_size=2048)

Robust Error Handling

Common errors:

from openai import OpenAI, APIError, RateLimitError, APIConnectionError

try:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input="..."
    )
except RateLimitError:
    print("❌ Rate limit exceeded")
except APIConnectionError:
    print("❌ Network error")
except APIError as e:
    print(f"❌ API error: {e}")

Retry logic with exponential backoff:

import time
from openai import APIError, RateLimitError

def get_embedding_with_retry(text, max_retries=5):
    """
    Generate an embedding with automatic retry
    
    Args:
        text: The text to convert
        max_retries: Maximum number of retries
    
    Returns:
        The embedding, or None if it fails
    """
    for attempt in range(max_retries):
        try:
            response = client.embeddings.create(
                model="text-embedding-3-small",
                input=text
            )
            return response.data[0].embedding
        
        except RateLimitError:
            # Wait exponentially: 2^attempt seconds
            wait_time = 2 ** attempt
            print(f"Rate limit. Retry {attempt+1}/{max_retries} in {wait_time}s")
            time.sleep(wait_time)
        
        except APIError as e:
            print(f"API error: {e}")
            time.sleep(1)
    
    # If it fails after max_retries
    print(f"❌ Failed after {max_retries} attempts")
    return None

# Usage
embedding = get_embedding_with_retry("Python is popular")

Exponential backoff: 1s → 2s → 4s → 8s → 16s


Rate Limiting

OpenAI limits (tier-based):

TierRPM (requests/min)TPM (tokens/min)
Free3150,000
Tier 15002,000,000
Tier 25,00010,000,000
Tier 3+CustomCustom

If you exceed → RateLimitError


Manual rate limiter:

import time
from collections import deque

class RateLimiter:
    """Simple rate limiter (sliding window)"""
    
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()
    
    def acquire(self):
        """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
            if sleep_time > 0:
                time.sleep(sleep_time)
            self.requests.popleft()
        
        # Register the request
        self.requests.append(time.time())

# Usage (limit: 500 RPM)
limiter = RateLimiter(max_requests=500, window_seconds=60)

for text in large_dataset:
    limiter.acquire()  # Waits if necessary
    embedding = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    ).data[0].embedding

Cost Optimization

Strategy #1: Aggressive caching

import json
import hashlib
from pathlib import Path

class EmbeddingCache:
    """Disk-based embedding cache"""
    
    def __init__(self, cache_dir="./cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
    
    def get_cache_key(self, text):
        """Generate a unique key for the text"""
        return hashlib.md5(text.encode()).hexdigest()
    
    def get(self, text):
        """Get a cached embedding"""
        cache_file = self.cache_dir / f"{self.get_cache_key(text)}.json"
        
        if cache_file.exists():
            with open(cache_file, 'r') as f:
                return json.load(f)
        
        return None
    
    def set(self, text, embedding):
        """Save an embedding to the cache"""
        cache_file = self.cache_dir / f"{self.get_cache_key(text)}.json"
        
        with open(cache_file, 'w') as f:
            json.dump(embedding, f)
    
    def get_or_create(self, text):
        """Get from the cache or generate"""
        # Try the cache
        cached = self.get(text)
        if cached:
            return cached
        
        # Generate
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        embedding = response.data[0].embedding
        
        # Save to the cache
        self.set(text, embedding)
        
        return embedding

# Usage
cache = EmbeddingCache()

# First time: Calls the API
emb = cache.get_or_create("Python is popular")

# Second time: Uses the cache (doesn't call the API) ✅
emb = cache.get_or_create("Python is popular")

Strategy #2: Reduced dimensions

# Full dimensions (1536):
# 1M documents = 6 GB storage
# Cost: $0.020 / 1M tokens

# Reduced dimensions (512):
# 1M documents = 2 GB storage  ← 3x less
# Cost: SAME ($0.020 / 1M tokens)
# Precision: ~95% of 1536 dims

# ✅ Win-win: Less storage, same cost, minimal precision loss
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="...",
    dimensions=512  # Reduce
)

Strategy #3: Batch processing

# ❌ Single requests:
# 1000 texts × 50ms latency = 50 seconds

# ✅ Batch (100 texts per request):
# 10 requests × 200ms = 2 seconds
# Speedup: 25x

def embed_batch(texts, batch_size=100):
    """Embed with batching"""
    embeddings = []
    
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=batch
        )
        
        embeddings.extend([item.embedding for item in response.data])
    
    return embeddings

Exercises

Exercise 1: Batch processing

Implement a function that generates embeddings for 500 texts with batch processing:

texts = [f"Document {i}" for i in range(500)]

# Implement embed_batch()
See solution
def embed_batch(texts, batch_size=100):
    """Generate embeddings in batches"""
    embeddings = []
    
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=batch
        )
        
        batch_embeddings = [item.embedding for item in response.data]
        embeddings.extend(batch_embeddings)
        
        print(f"Batch {i//batch_size + 1}: {len(batch)} texts processed")
    
    return embeddings

# Usage
texts = [f"Document {i}" for i in range(500)]
embeddings = embed_batch(texts, batch_size=100)

print(f"\nTotal: {len(embeddings)} embeddings generated")

Output:

Batch 1: 100 texts processed
Batch 2: 100 texts processed
Batch 3: 100 texts processed
Batch 4: 100 texts processed
Batch 5: 100 texts processed

Total: 500 embeddings generated

Exercise 2: Retry logic

Implement retry with exponential backoff:

# Implement a function that retries up to 3 times
See solution
import time
from openai import APIError, RateLimitError

def get_embedding_with_retry(text, max_retries=3):
    """Generate an embedding with retry"""
    for attempt in range(max_retries):
        try:
            response = client.embeddings.create(
                model="text-embedding-3-small",
                input=text
            )
            return response.data[0].embedding
        
        except RateLimitError:
            wait_time = 2 ** attempt  # Exponential backoff
            print(f"Rate limit. Retry {attempt+1}/{max_retries} in {wait_time}s")
            time.sleep(wait_time)
        
        except APIError as e:
            print(f"API error: {e}")
            if attempt < max_retries - 1:
                time.sleep(1)
    
    print(f"❌ Failed after {max_retries} attempts")
    return None

# Test
embedding = get_embedding_with_retry("Python is popular")
if embedding:
    print(f"✅ Embedding generated: {len(embedding)} dims")

Summary

What you learned:

  • Advanced parameters: dimensions, encoding_format, batch input
  • Batch processing: Multiple texts in 1 request (25x speedup)
  • Error handling: Retry logic with exponential backoff
  • Rate limiting: Sliding window rate limiter
  • Cost optimization: Caching, reduced dimensions, batching

Key concepts:

  1. Batch processing → 25x speedup
  2. Reduced dimensions → 3x less storage
  3. Retry logic → Robustness against failures

Additional resources

  1. OpenAI Embeddings Docs - Official
  2. Rate Limits Guide - OpenAI
  3. Error Handling Best Practices - OpenAI
  4. Exponential Backoff - Wikipedia

In the next capsule

Capsule 08: Mini-Project - Robust API Client

You'll build:

  • A production-ready embeddings client
  • Caching, retry logic, rate limiting
  • Automatic batch processing
  • Logging and monitoring
  • A CLI for testing

From theory to a complete implementation.


Module 2 - Embeddings Deep Dive Guide API patterns for production: efficiency and robustness