OpenAI API Advanced: Parámetros y Patterns de Producción

Descripción de la cápsula

La OpenAI Embeddings API es simple en superficie, pero tiene parámetros y optimizaciones críticas para producción: batch processing, dimensiones reducidas, encoding formats, error handling robusto, rate limiting, y cost optimization.

En esta cápsula aprenderás los parámetros avanzados de la API, cómo procesar múltiples textos eficientemente con batch processing, implementar retry logic con exponential backoff, manejar rate limits, y optimizar costos. También verás código production-ready que puedes usar en tus proyectos.

Al final, podrás construir sistemas de embeddings robustos que manejan volumen a escala.

OpenAI Embeddings API: Parámetros completos

API básica (visto en módulo 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 es un lenguaje"
)

embedding = response.data[0].embedding

Parámetros avanzados:

response = client.embeddings.create(
    model="text-embedding-3-small",      # Modelo
    input="...",                         # Texto o lista de textos
    dimensions=512,                      # Reducir dimensionalidad (opcional)
    encoding_format="float",             # "float" o "base64"
    user="user-123"                      # Identificador de usuario (opcional)
)

Veamos cada parámetro en detalle.

Parámetro: model

Modelos disponibles (Enero 2026):

Recomendación:

• Prototipo: text-embedding-3-small (barato, rápido)
• Producción (alta calidad): text-embedding-3-large (mejor performance)

Comparación empírica:

import numpy as np

texts = [
    "Python es un lenguaje de programación",
    "Python es una serpiente"
]

# 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
]

# Comparar similaridad
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}")

Output típico:

Similarity (small): 0.78
Similarity (large): 0.72  ← Mejor diferenciación

large diferencia mejor contextos (mejor para RAG).

Parámetro: dimensions

Qué hace:

Reduce dimensionalidad del embedding (sin re-entrenar modelo).

Por defecto:

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

Con dimensions:

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

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

Por qué reducir dimensiones:

✅ 1. Storage más pequeño

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

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

✅ 2. Búsqueda más rápida

# Menos dimensiones = dot product más rápido
# 512 dims → ~3x más rápido que 1536 dims

❌ 3. Trade-off: Pierde algo de precisión

# Pero pérdida es mínima (~5% en benchmarks)

Benchmark de precision vs dimensiones:

from openai import OpenAI
import numpy as np

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

# Textos de prueba
query = "Python es un lenguaje de programación"
docs = [
    "Python es un lenguaje interpretado",
    "JavaScript es un lenguaje web",
    "Gato duerme en sofá"
]

# Test con diferentes dimensiones
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
    ]
    
    # Calcular 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]}")

Output esperado:

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']

Conclusión: 512 dims captura ~95% de la precisión de 1536 dims.

Parámetro: input (batch processing)

Single input:

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

embedding = response.data[0].embedding

Batch input (múltiples textos):

# Múltiples textos en 1 llamada
texts = [
    "Python es popular",
    "JavaScript es rápido",
    "Go es concurrente"
]

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts  # Lista de textos
)

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

Por qué usar batch:

✅ 1. Más eficiente (menos overhead HTTP)

# Single: 100 llamadas × 50ms latencia = 5 segundos
# Batch: 1 llamada × 100 textos = ~200ms

# Speedup: ~25x

✅ 2. Rate limiting más bajo

# OpenAI rate limits: requests per minute (RPM)
# 1 batch request = 1 RPM (no importa cuántos textos)

Límite de batch:

# OpenAI permite hasta ~2048 textos por batch
# (depende de total de tokens)

MAX_BATCH_SIZE = 2048

def embed_batch(texts, batch_size=2048):
    """Embed múltiples textos en 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

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

Error Handling Robusto

Errores comunes:

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 con exponential backoff:

import time
from openai import APIError, RateLimitError

def get_embedding_with_retry(text, max_retries=5):
    """
    Generar embedding con retry automático
    
    Args:
        text: Texto a convertir
        max_retries: Máximo de reintentos
    
    Returns:
        Embedding o None si falla
    """
    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:
            # Esperar exponencialmente: 2^attempt segundos
            wait_time = 2 ** attempt
            print(f"Rate limit. Retry {attempt+1}/{max_retries} en {wait_time}s")
            time.sleep(wait_time)
        
        except APIError as e:
            print(f"API error: {e}")
            time.sleep(1)
    
    # Si falla después de max_retries
    print(f"❌ Failed después de {max_retries} intentos")
    return None

# Uso
embedding = get_embedding_with_retry("Python es popular")

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

Rate Limiting

Límites de OpenAI (tier-based):

Si excedes → RateLimitError

Rate limiter manual:

import time
from collections import deque

class RateLimiter:
    """Rate limiter simple (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):
        """Esperar si necesario para no exceder rate limit"""
        now = time.time()
        
        # Remover requests fuera de ventana
        while self.requests and self.requests[0] < now - self.window_seconds:
            self.requests.popleft()
        
        # Si estamos en límite, esperar
        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()
        
        # Registrar request
        self.requests.append(time.time())

# Uso (límite: 500 RPM)
limiter = RateLimiter(max_requests=500, window_seconds=60)

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

Cost Optimization

Estrategia #1: Caching agresivo

import json
import hashlib
from pathlib import Path

class EmbeddingCache:
    """Cache de embeddings en disco"""
    
    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):
        """Generar key único para texto"""
        return hashlib.md5(text.encode()).hexdigest()
    
    def get(self, text):
        """Obtener embedding cacheado"""
        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):
        """Guardar embedding en 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):
        """Obtener de cache o generar"""
        # Intentar cache
        cached = self.get(text)
        if cached:
            return cached
        
        # Generar
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        embedding = response.data[0].embedding
        
        # Guardar en cache
        self.set(text, embedding)
        
        return embedding

# Uso
cache = EmbeddingCache()

# Primera vez: Llama API
emb = cache.get_or_create("Python es popular")

# Segunda vez: Usa cache (no llama API) ✅
emb = cache.get_or_create("Python es popular")

Estrategia #2: Dimensiones reducidas

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

# Reduced dimensions (512):
# 1M documentos = 2 GB storage  ← 3x menos
# Costo: MISMO ($0.020 / 1M tokens)
# Precision: ~95% de 1536 dims

# ✅ Win-win: Menos storage, mismo costo, mínima pérdida de precisión
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="...",
    dimensions=512  # Reducir
)

Estrategia #3: Batch processing

# ❌ Single requests:
# 1000 textos × 50ms latency = 50 segundos

# ✅ Batch (100 textos por request):
# 10 requests × 200ms = 2 segundos
# Speedup: 25x

def embed_batch(texts, batch_size=100):
    """Embed con 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

Ejercicios

Ejercicio 1: Batch processing

Implementa función que genera embeddings para 500 textos con batch processing:

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

# Implementa embed_batch()

def embed_batch(texts, batch_size=100):
    """Generar embeddings en 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)} textos procesados")
    
    return embeddings

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

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

Batch 1: 100 textos procesados
Batch 2: 100 textos procesados
Batch 3: 100 textos procesados
Batch 4: 100 textos procesados
Batch 5: 100 textos procesados

Total: 500 embeddings generados

Ejercicio 2: Retry logic

Implementa retry con exponential backoff:

# Implementa función que reintenta hasta 3 veces

import time
from openai import APIError, RateLimitError

def get_embedding_with_retry(text, max_retries=3):
    """Generar embedding con 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} en {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 después de {max_retries} intentos")
    return None

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

Resumen

Qué aprendiste:

• ✅ Parámetros avanzados: dimensions, encoding_format, batch input
• ✅ Batch processing: Múltiples textos en 1 request (25x speedup)
• ✅ Error handling: Retry logic con exponential backoff
• ✅ Rate limiting: Sliding window rate limiter
• ✅ Cost optimization: Caching, dimensions reducidas, batching

Conceptos clave:

1. Batch processing → 25x speedup
2. Dimensiones reducidas → 3x menos storage
3. Retry logic → Robustez ante failures

Recursos adicionales

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

En la siguiente cápsula

Cápsula 08: Mini-Proyecto - API Client Robusto

Construirás:

• Cliente de embeddings production-ready
• Caching, retry logic, rate limiting
• Batch processing automático
• Logging y monitoring
• CLI para testing

De teoría a implementación completa.

Módulo 2 - Embeddings Deep Dive Guide API patterns para producción: eficiencia y robustez