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):
| Model | Dimensions | Cost/1M tokens | Performance | Use |
|---|---|---|---|---|
text-embedding-3-small | 1536 | $0.020 | Good | General |
text-embedding-3-large | 3072 | $0.130 | Excellent | High precision |
text-embedding-ada-002 | 1536 | $0.100 | Good (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 dimstext-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):
| Tier | RPM (requests/min) | TPM (tokens/min) |
|---|---|---|
| Free | 3 | 150,000 |
| Tier 1 | 500 | 2,000,000 |
| Tier 2 | 5,000 | 10,000,000 |
| Tier 3+ | Custom | Custom |
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, batchinput - ✅ 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:
- Batch processing → 25x speedup
- Reduced dimensions → 3x less storage
- Retry logic → Robustness against failures
Additional resources
- OpenAI Embeddings Docs - Official
- Rate Limits Guide - OpenAI
- Error Handling Best Practices - OpenAI
- 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