Cápsula 03: Setup de Pinecone serverless — el primer índice production-ready en 30 minutos

Descripción de la cápsula

Decidiste migrar (cápsula 02). Ahora viene el primer paso concreto: crear tu índice Pinecone. Suena simple — y lo es — pero hay tres decisiones que se toman al crear el índice y que NO se pueden cambiar después sin recrear todo: dimensiones, métrica de distancia, y región/cloud. Esta cápsula te enseña a tomar esas decisiones bien la primera vez.

Vas a configurar un índice serverless (sin pensar en infraestructura), validar que las dimensiones coinciden con tu modelo de embeddings, y dejar todo listo con scripts idempotentes que cualquier dev del equipo pueda re-ejecutar sin romper estado.

Al finalizar esta cápsula serás capaz de:

• ✅ Crear un índice Pinecone serverless con dimensión y métrica correctas
• ✅ Configurar API keys y environment variables apropiadamente
• ✅ Implementar setup idempotente que se puede correr en cualquier ambiente (dev, staging, prod)
• ✅ Validar que la dimensión del índice matchea el modelo de embeddings antes de indexar
• ✅ Decidir entre serverless y pod-based según tu caso
• ✅ Anticipar el error #1 de migración: dimensiones incompatibles entre el índice y los embeddings

Tiempo estimado: 30-35 minutos

Las tres decisiones inmovibles al crear un índice

Cuando creas un índice Pinecone, tres parámetros se fijan para siempre. Cambiarlos requiere recrear el índice y re-indexar todo:

1. Dimensión

pc.create_index(
    name="my-rag",
    dimension=1536,  # ← se fija acá
    ...
)

Cómo elegir: debe matchear EXACTAMENTE la dimensión de tu modelo de embeddings.

Trampa común: crear índice con dim=1536 (asumiendo OpenAI), pero después decidir migrar a text-embedding-3-large (dim=3072). El índice no acepta los nuevos embeddings — hay que recrear.

2. Métrica de distancia

pc.create_index(
    name="my-rag",
    metric="cosine",  # ← se fija acá
    ...
)

Opciones:

• cosine — default para text embeddings de OpenAI, Cohere, Sentence Transformers
• euclidean (L2) — algunos casos específicos (image embeddings, recomendaciones)
• dotproduct — cuando los embeddings ya están normalizados

Recomendación: para 95% de casos de RAG con text embeddings, cosine. Si tu modelo de embeddings recomienda otra métrica, usar esa.

3. Región y cloud

pc.create_index(
    name="my-rag",
    spec=ServerlessSpec(
        cloud="aws",         # aws | gcp | azure
        region="us-east-1",  # ← elegir según latencia a tus usuarios
    ),
)

Cómo elegir región:

• Si tu app está en AWS us-east-1, índice también en us-east-1.
• Latencia de red: ~5-15ms en misma región, 50-100ms entre regiones.
• Pinecone serverless no soporta multi-región replicación automática (Enterprise sí).

Setup paso a paso

Paso 1: instalación y configuración

pip install "pinecone-client>=4.0"

# .env
PINECONE_API_KEY=...  # de la console de Pinecone
PINECONE_INDEX_NAME=production-rag
PINECONE_ENVIRONMENT=production

Paso 2: cliente y verificación

# pinecone_setup.py
import os
from pinecone import Pinecone, ServerlessSpec
from dotenv import load_dotenv


load_dotenv()


def get_pinecone_client() -> Pinecone:
    """Inicializa cliente Pinecone con API key."""
    api_key = os.getenv("PINECONE_API_KEY")
    if not api_key:
        raise EnvironmentError("PINECONE_API_KEY no configurada")
    return Pinecone(api_key=api_key)


# Quick smoke test
pc = get_pinecone_client()
print(f"Indexes existentes: {pc.list_indexes().names()}")

Paso 3: creación idempotente del índice

def ensure_index(
    pc: Pinecone,
    name: str,
    dimension: int = 1536,
    metric: str = "cosine",
    cloud: str = "aws",
    region: str = "us-east-1",
):
    """
    Crea índice si no existe. Si existe, valida config y devuelve referencia.
    Idempotente: se puede correr múltiples veces sin romper.
    """
    existing_indexes = [idx.name for idx in pc.list_indexes()]

    if name in existing_indexes:
        # Validar que la config existente matchea
        index_info = pc.describe_index(name)
        if index_info.dimension != dimension:
            raise ValueError(
                f"Index '{name}' existe con dimension {index_info.dimension}, "
                f"esperado {dimension}. Recrear o usar otro nombre."
            )
        if index_info.metric != metric:
            raise ValueError(
                f"Index '{name}' existe con metric {index_info.metric}, "
                f"esperado {metric}."
            )
        print(f"Index '{name}' ya existe, config OK")
    else:
        print(f"Creando index '{name}'...")
        pc.create_index(
            name=name,
            dimension=dimension,
            metric=metric,
            spec=ServerlessSpec(cloud=cloud, region=region),
        )
        # Esperar a que el index esté listo (puede tomar 30-60 segundos)
        import time
        while True:
            status = pc.describe_index(name).status["ready"]
            if status:
                break
            print("Esperando que index esté ready...")
            time.sleep(2)
        print(f"Index '{name}' creado")

    return pc.Index(name)


# Uso
pc = get_pinecone_client()
index = ensure_index(
    pc,
    name="production-rag",
    dimension=1536,    # text-embedding-3-small
    metric="cosine",
    cloud="aws",
    region="us-east-1",
)

Paso 4: smoke test

def smoke_test(index):
    """Verifica que el índice responde correctamente."""
    # Stats del índice
    stats = index.describe_index_stats()
    print(f"Total vectors: {stats['total_vector_count']}")
    print(f"Dimension: {stats['dimension']}")
    print(f"Index fullness: {stats.get('index_fullness', 'N/A')}")

    # Insert + delete de prueba
    test_vector = [0.1] * 1536  # vector trivial
    index.upsert([("smoke_test_vector", test_vector, {"test": True})])
    print("Insert OK")

    # Query
    results = index.query(vector=test_vector, top_k=1, include_metadata=True)
    print(f"Query OK, results: {len(results['matches'])}")

    # Cleanup
    index.delete(ids=["smoke_test_vector"])
    print("Delete OK")


smoke_test(index)

Output esperado:

Total vectors: 0
Dimension: 1536
Index fullness: 0
Insert OK
Query OK, results: 1
Delete OK

Validación: dimensión matchea modelo de embeddings

def validate_embedding_compatibility(index, embedding_function):
    """
    Genera un embedding de prueba y verifica que coincide con la dimensión del índice.
    Crítico antes de hacer ingest masivo.
    """
    test_text = "test"
    test_embedding = embedding_function(test_text)

    if isinstance(test_embedding, list):
        actual_dim = len(test_embedding)
    elif hasattr(test_embedding, "shape"):
        actual_dim = test_embedding.shape[0]
    else:
        raise TypeError(f"Embedding type unexpected: {type(test_embedding)}")

    index_dim = index.describe_index_stats()["dimension"]

    if actual_dim != index_dim:
        raise ValueError(
            f"Dimension mismatch! Embedding produce dim={actual_dim}, "
            f"índice espera dim={index_dim}. Recrear índice o cambiar modelo."
        )

    print(f"OK: embedding ({actual_dim}d) compatible con index ({index_dim}d)")


# Uso antes de ingest
from openai import OpenAI
openai_client = OpenAI()


def get_openai_embedding(text):
    response = openai_client.embeddings.create(
        input=text,
        model="text-embedding-3-small",
    )
    return response.data[0].embedding


validate_embedding_compatibility(index, get_openai_embedding)

Serverless vs pod-based

Pinecone ofrece dos tipos de índice:

Recomendación inicial: serverless. Sin compromisos, paga por lo que usas. Si después la latencia o el costo crecen mucho, evalúa pod-based.

# Serverless (recomendado para empezar)
pc.create_index(
    name="my-rag",
    dimension=1536,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1"),
)


# Pod-based (avanzado, latencia crítica)
from pinecone import PodSpec

pc.create_index(
    name="my-rag-pods",
    dimension=1536,
    metric="cosine",
    spec=PodSpec(
        environment="us-east-1-aws",
        pod_type="p1.x1",  # tipo y tamaño
        pods=2,             # número de pods
    ),
)

Configuración por ambiente

# config.py
import os
from typing import NamedTuple


class PineconeConfig(NamedTuple):
    api_key: str
    index_name: str
    dimension: int
    metric: str
    cloud: str
    region: str


def get_config(env: str = None) -> PineconeConfig:
    """Configuración separada por ambiente: dev, staging, prod."""
    env = env or os.getenv("APP_ENV", "dev")

    base_index = os.getenv("PINECONE_INDEX_BASENAME", "rag")

    return PineconeConfig(
        api_key=os.getenv("PINECONE_API_KEY"),
        index_name=f"{base_index}-{env}",  # rag-dev, rag-staging, rag-prod
        dimension=int(os.getenv("EMBEDDING_DIMENSION", "1536")),
        metric=os.getenv("PINECONE_METRIC", "cosine"),
        cloud=os.getenv("PINECONE_CLOUD", "aws"),
        region=os.getenv("PINECONE_REGION", "us-east-1"),
    )


# Uso
config = get_config()  # auto-detecta APP_ENV
print(f"Usando index: {config.index_name}")  # ej: rag-prod

Beneficios:

• Dev/staging/prod completamente separados (no contaminas datos).
• Cambio de ambiente con un env var, sin tocar código.
• Configuración versionada (git) sin secrets (estos en .env).

Trampas y errores comunes

Trampa 1: dimensión incorrecta sin validación

El error: crear índice con dimension=1536 (asumido) y después usar embeddings de 384 dim. Inserts fallan.

Síntoma: PineconeApiException: Vector dimension 384 does not match the dimension of the index 1536.

Cómo prevenir: validate_embedding_compatibility antes de ingestar masivamente.

Trampa 2: métrica incorrecta

El error: índice con metric="euclidean" cuando los embeddings esperan cosine.

Síntoma: sistema funciona pero retrieval da rankings extraños. Cosine y euclidean dan rankings similares pero NO idénticos.

Cómo prevenir: usar la métrica que recomienda el modelo de embeddings (cosine para OpenAI, Cohere, ST).

Trampa 3: API key sin permisos suficientes

El error: API key de "read-only" para crear índices.

Síntoma: 403 Forbidden al crear.

Cómo prevenir: crear API key específica con permiso "Manage Indexes" para el script de setup. Después puedes usar key con menos permisos para queries.

Trampa 4: olvidar time.sleep después de crear

El error:

pc.create_index(...)
index = pc.Index(name)
index.upsert(...)  # fail! índice todavía no está ready

Síntoma: error 503 o timeout.

Cómo prevenir: poll del status hasta ready=True antes de usar (visto en ensure_index).

Trampa 5: index name con caracteres inválidos

El error: pc.create_index(name="my_rag_index") (underscore).

Síntoma: Pinecone solo permite lowercase + hyphens. my_rag_index falla.

Cómo prevenir: usar siempre lowercase + -. my-rag-index ✓

Trampa 6: misma cuenta para dev y prod

El error: dev y prod usan la misma API key, mismo índice.

Síntoma: un dev hace cleanup en dev y borra datos de producción.

Cómo prevenir: API keys separadas, índices con nombres distintos (rag-dev vs rag-prod).

Ejercicio aplicado

Escenario: estás migrando un sistema RAG existente a Pinecone. Detalles:

• Modelo de embeddings actual: OpenAI text-embedding-3-small (1536 dim)
• Equipo evalúa subir a text-embedding-3-large (3072 dim) en próximos 6 meses
• App deployada en GCP us-central1
• 3 ambientes: dev, staging, prod
• Multi-tenant con 50 clientes

Tu trabajo:

1. Configura los índices apropiados (¿uno o tres? ¿qué dimensión?).
2. Diseña script idempotente de setup.
3. Plan de migración para el cambio de modelo en 6 meses.

# config.py
INDICES_CONFIG = {
    "dev": {
        "name": "rag-dev",
        "dimension": 1536,
        "metric": "cosine",
        "cloud": "gcp",
        "region": "us-central1",
    },
    "staging": {
        "name": "rag-staging",
        "dimension": 1536,
        "metric": "cosine",
        "cloud": "gcp",
        "region": "us-central1",
    },
    "prod": {
        "name": "rag-prod",
        "dimension": 1536,
        "metric": "cosine",
        "cloud": "gcp",
        "region": "us-central1",
    },
}

# scripts/setup_pinecone.py
"""
Setup idempotente de índices Pinecone.
Uso: APP_ENV=prod python scripts/setup_pinecone.py
"""
import os
import sys
import time
from pinecone import Pinecone, ServerlessSpec


def main():
    env = os.getenv("APP_ENV")
    if env not in ["dev", "staging", "prod"]:
        print("APP_ENV debe ser dev | staging | prod")
        sys.exit(1)

    # Validación extra: prod requiere confirmación
    if env == "prod":
        confirm = input("⚠️  Modificando índice de PRODUCCIÓN. Continuar? (yes/no): ")
        if confirm != "yes":
            print("Abortado.")
            sys.exit(0)

    config = INDICES_CONFIG[env]
    pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))

    existing = [idx.name for idx in pc.list_indexes()]

    if config["name"] in existing:
        info = pc.describe_index(config["name"])
        # Validaciones
        assert info.dimension == config["dimension"], (
            f"Dim mismatch! Existe={info.dimension}, "
            f"config={config['dimension']}"
        )
        assert info.metric == config["metric"]
        print(f"OK: {config['name']} existe con config correcta")
    else:
        print(f"Creando {config['name']}...")
        pc.create_index(
            name=config["name"],
            dimension=config["dimension"],
            metric=config["metric"],
            spec=ServerlessSpec(cloud=config["cloud"], region=config["region"]),
        )

        # Wait until ready
        while not pc.describe_index(config["name"]).status["ready"]:
            print("Esperando ready...")
            time.sleep(3)
        print(f"✅ {config['name']} creado")


if __name__ == "__main__":
    main()

pc.create_index(
    name="rag-prod-v2",  # ← v2 indica nueva dimensión
    dimension=3072,
    metric="cosine",
    spec=ServerlessSpec(cloud="gcp", region="us-central1"),
)

Resumen y siguiente paso

Lo que aprendiste:

• Tres decisiones inmovibles al crear un índice: dimensión, métrica, región/cloud.
• Dimensión debe matchear EXACTAMENTE el modelo de embeddings.
• Cosine para 95% de RAG con text embeddings; cambiar solo si el modelo lo recomienda.
• Setup idempotente: ensure_index que valida config existente o crea nueva.
• Validación pre-ingest: confirmar dimensión del modelo coincide con índice.
• Serverless es default razonable; pod-based para latencia crítica o tráfico constante.
• Configuración por ambiente con índices separados (rag-dev, rag-staging, rag-prod).
• Trampas: API key con permisos insuficientes, no esperar a que índice esté ready, name con underscore, mismo índice para dev y prod.

Checkpoint: antes de avanzar, deberías poder:

• Crear índice Pinecone con configuración correcta.
• Validar dimensión matchea modelo de embeddings.
• Implementar setup idempotente que se puede correr en cualquier ambiente.

Siguiente cápsula: 04 — Migración ChromaDB → Pinecone.

Tienes el índice. Ahora viene la migración real: trasladar los datos desde ChromaDB sin perder vectores ni metadata, sin downtime para producción, con rollback plan si algo sale mal.

Recursos

1. Pinecone — Quickstart — Setup oficial
2. Pinecone — Create Index — Parámetros completos
3. Pinecone — Serverless vs Pods — Decisión
4. OpenAI Embeddings — Dimensiones por modelo
5. Pinecone — API Reference — SDK detallado
6. Pinecone Examples GitHub — Código real

Tiempo estimado: 30-35 minutos Siguiente: 04-migrating-from-chromadb-to-pinecone.md