Cápsula 04: Migración ChromaDB → Pinecone — sin perder datos ni romper queries

Descripción de la cápsula

La migración real es donde la mayoría de proyectos se complica. No es "exportar de ChromaDB y meter en Pinecone" — eso falla 50% de las veces. La diferencia entre una migración suave y un incidente está en los detalles: validación de integridad, manejo de errores por batch, dual-write durante transición, rollback plan, y A/B testing antes del switchover total.

Esta cápsula te enseña la migración correcta: estrategia incremental con downtime mínimo, scripts robustos con retry exponencial, validación de conteos e integridad después de cada paso, y plan de rollback si algo sale mal en producción.

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

• ✅ Implementar script de migración por batches con retry y logging
• ✅ Validar integridad: conteos, metadata preservada, embeddings consistentes
• ✅ Diseñar estrategia incremental (no big-bang) con dual-write durante transición
• ✅ Implementar rollback plan con feature flag
• ✅ Validar calidad de retrieval post-migración con eval set
• ✅ Anticipar los cuatro errores más comunes: payload size, metadata types, namespace incorrecto, dimensión mismatch

Tiempo estimado: 35-40 minutos

Estrategia: incremental, no big-bang

                            BIG-BANG (peligroso)

   Día 1: Stop app
   Día 1-3: Migrar todo a Pinecone
   Día 3: Switch app a Pinecone
   Día 3+: Esperar que funcione

   Si algo falla: rollback toma horas, downtime extendido


                          INCREMENTAL (recomendado)

   Semana 1: Setup Pinecone + script de migración + tests
   Semana 2: Dual-write (escribir en ChromaDB Y Pinecone para cada doc nuevo)
   Semana 3: Backfill (migrar histórico de ChromaDB a Pinecone)
   Semana 4: Dual-read con A/B test (query a ambos, comparar)
   Semana 5: Switch primario a Pinecone, ChromaDB como fallback
   Semana 6: Decommission ChromaDB

   Si algo falla: rollback en minutos via feature flag

La migración incremental es la regla. Big-bang solo se justifica para corpus chico (<100K docs) o sistemas con downtime planificado.

Script base: export desde ChromaDB

# migration_script.py
from typing import Iterable
import chromadb
from chromadb.utils import embedding_functions
import os


def export_from_chromadb(collection_name: str, batch_size: int = 1000) -> Iterable[dict]:
    """
    Exporta documentos en batches para no cargar todo en RAM.
    Yield batches de dicts con id, embedding, metadata.
    """
    chroma_client = chromadb.PersistentClient(path="./chroma_db")
    collection = chroma_client.get_collection(collection_name)

    total = collection.count()
    print(f"Exportando {total} docs de '{collection_name}'")

    offset = 0
    while offset < total:
        batch = collection.get(
            limit=batch_size,
            offset=offset,
            include=["embeddings", "metadatas", "documents"],
        )

        # Yield batch como lista de dicts
        items = []
        for i in range(len(batch["ids"])):
            items.append({
                "id": batch["ids"][i],
                "embedding": batch["embeddings"][i],
                "metadata": batch["metadatas"][i] or {},
                "document": batch["documents"][i],
            })

        yield items
        offset += batch_size
        print(f"  Exportados {min(offset, total)}/{total}")

Por qué batches: colecciones grandes (5M+ docs) no caben en RAM. Batch de 1000 mantiene RAM ~500MB.

Upsert a Pinecone con retry

import time
from pinecone import Pinecone


def upsert_to_pinecone(
    index,
    items: list,
    namespace: str = "",
    max_retries: int = 3,
) -> int:
    """
    Upsert batch a Pinecone con retry exponencial.
    Retorna count de items insertados exitosamente.
    """
    # Construir vectors en formato Pinecone
    vectors = []
    for item in items:
        # Pinecone metadata debe ser serializable y plano
        metadata = clean_metadata_for_pinecone(item["metadata"])

        # Agregar contenido del documento como metadata para retrieval posterior
        metadata["content"] = item["document"][:40000]  # Pinecone limit

        vectors.append({
            "id": item["id"],
            "values": item["embedding"],
            "metadata": metadata,
        })

    # Upsert con retry exponencial
    for attempt in range(max_retries):
        try:
            response = index.upsert(vectors=vectors, namespace=namespace)
            return response.upserted_count
        except Exception as e:
            if attempt == max_retries - 1:
                # Último intento, re-raise
                raise
            wait_time = 2 ** attempt  # 1, 2, 4 segundos
            print(f"  Error en attempt {attempt + 1}: {e}. Retry en {wait_time}s...")
            time.sleep(wait_time)


def clean_metadata_for_pinecone(metadata: dict) -> dict:
    """
    Pinecone restricts metadata to: strings, numbers, booleans, lists of strings.
    No nested dicts or other types.
    """
    cleaned = {}
    for key, value in metadata.items():
        if isinstance(value, (str, int, float, bool)):
            cleaned[key] = value
        elif isinstance(value, list):
            # Convertir todo a strings
            cleaned[key] = [str(v) for v in value]
        elif value is None:
            # Skip None values
            continue
        elif isinstance(value, dict):
            # Flatten dict: {a: {b: 1}} → {a_b: 1}
            for sub_key, sub_value in value.items():
                if isinstance(sub_value, (str, int, float, bool)):
                    cleaned[f"{key}_{sub_key}"] = sub_value
        else:
            # Convert otros tipos a string
            cleaned[key] = str(value)
    return cleaned

Detalles importantes:

• Pinecone tiene límite de 40,000 caracteres por metadata field. Truncar content.
• Metadata debe ser plana (sin dicts anidados). Flatten si es necesario.
• Solo tipos primitivos en metadata: str, int, float, bool, list[str].

Validación de integridad

Después de migrar, validar que NO se perdió nada:

def validate_migration(
    chroma_collection,
    pinecone_index,
    namespace: str = "",
) -> dict:
    """
    Validaciones post-migración:
    1. Conteos coinciden
    2. Sample de IDs aleatorios existen en ambos
    3. Metadata preservada
    """
    # 1. Conteos
    chroma_count = chroma_collection.count()
    pinecone_stats = pinecone_index.describe_index_stats()
    pinecone_count = pinecone_stats["namespaces"].get(namespace, {}).get("vector_count", 0)

    print(f"Counts: ChromaDB={chroma_count}, Pinecone={pinecone_count}")
    if chroma_count != pinecone_count:
        return {
            "valid": False,
            "error": f"Count mismatch: chroma={chroma_count} != pinecone={pinecone_count}",
        }

    # 2. Sample de IDs aleatorios
    import random
    chroma_data = chroma_collection.get(limit=20)  # primeros 20 como sample
    sample_ids = random.sample(chroma_data["ids"], min(20, len(chroma_data["ids"])))

    pinecone_fetched = pinecone_index.fetch(ids=sample_ids, namespace=namespace)
    missing_ids = [
        sid for sid in sample_ids
        if sid not in pinecone_fetched.get("vectors", {})
    ]

    if missing_ids:
        return {
            "valid": False,
            "error": f"Missing IDs in Pinecone: {missing_ids[:5]}",
        }

    # 3. Validar metadata preservada (sample)
    for sid in sample_ids[:5]:
        chroma_meta = chroma_collection.get(ids=[sid], include=["metadatas"])["metadatas"][0]
        pinecone_meta = pinecone_fetched["vectors"][sid]["metadata"]

        # Comparar campos críticos
        for critical_field in ["workspace_id", "doc_id", "type"]:
            if chroma_meta.get(critical_field) != pinecone_meta.get(critical_field):
                return {
                    "valid": False,
                    "error": f"Metadata mismatch for {sid} field {critical_field}",
                }

    return {
        "valid": True,
        "chroma_count": chroma_count,
        "pinecone_count": pinecone_count,
        "sample_validated": len(sample_ids),
    }


# Uso
result = validate_migration(chroma_collection, pinecone_index, namespace="acme_corp")
if result["valid"]:
    print("✅ Migración validada")
else:
    print(f"❌ Migración FAILED: {result['error']}")

Validación de calidad: eval set comparativo

Conteos OK no es suficiente. Validar que el retrieval da resultados similares:

def validate_quality_post_migration(
    chroma_collection,
    pinecone_index,
    eval_set: list,
    threshold_recall: float = 0.85,
):
    """
    Compara recall del retrieval entre ChromaDB y Pinecone.
    Si Pinecone tiene recall significativamente menor, la migración tiene problema.
    """
    chroma_recalls = []
    pinecone_recalls = []

    for item in eval_set:
        # ChromaDB query
        chroma_results = chroma_collection.query(
            query_texts=[item["query"]],
            n_results=5,
            where={"workspace_id": item["workspace_id"]},
        )
        chroma_top_5 = set(chroma_results["ids"][0])

        # Pinecone query
        from openai import OpenAI
        client = OpenAI()
        query_emb = client.embeddings.create(
            input=item["query"],
            model="text-embedding-3-small",
        ).data[0].embedding

        pinecone_results = pinecone_index.query(
            vector=query_emb,
            top_k=5,
            namespace=item["workspace_id"],
            include_metadata=True,
        )
        pinecone_top_5 = set(m["id"] for m in pinecone_results["matches"])

        # Recall vs ground truth
        relevant = set(item["expected_doc_ids"])
        chroma_recall = len(chroma_top_5 & relevant) / max(len(relevant), 1)
        pinecone_recall = len(pinecone_top_5 & relevant) / max(len(relevant), 1)

        chroma_recalls.append(chroma_recall)
        pinecone_recalls.append(pinecone_recall)

    avg_chroma = sum(chroma_recalls) / len(chroma_recalls)
    avg_pinecone = sum(pinecone_recalls) / len(pinecone_recalls)

    delta = avg_pinecone - avg_chroma
    print(f"Recall ChromaDB: {avg_chroma:.2%}")
    print(f"Recall Pinecone: {avg_pinecone:.2%}")
    print(f"Delta: {delta:+.2%}")

    if avg_pinecone < threshold_recall:
        return {
            "valid": False,
            "error": f"Pinecone recall {avg_pinecone:.2%} < threshold {threshold_recall:.2%}",
        }

    if delta < -0.03:  # más de 3% peor
        return {
            "valid": False,
            "error": f"Pinecone recall {delta:+.2%} peor que ChromaDB",
        }

    return {"valid": True, "chroma_recall": avg_chroma, "pinecone_recall": avg_pinecone}

Estrategia incremental con feature flag

Durante la transición, queremos poder switch instantáneo entre ChromaDB y Pinecone:

# retrieval.py
import os


VECTOR_DB_BACKEND = os.getenv("VECTOR_DB_BACKEND", "chromadb")  # chromadb | pinecone | dual


def retrieve(query: str, tenant: TenantContext, n_results: int = 5):
    """Retrieval que respeta el feature flag VECTOR_DB_BACKEND."""

    if VECTOR_DB_BACKEND == "chromadb":
        return retrieve_chromadb(query, tenant, n_results)

    elif VECTOR_DB_BACKEND == "pinecone":
        return retrieve_pinecone(query, tenant, n_results)

    elif VECTOR_DB_BACKEND == "dual":
        # Para A/B testing: ejecutar ambos, devolver chromadb pero loggear ambos
        chroma_results = retrieve_chromadb(query, tenant, n_results)
        try:
            pinecone_results = retrieve_pinecone(query, tenant, n_results)
            log_comparison(query, chroma_results, pinecone_results)
        except Exception as e:
            print(f"Pinecone failed (logging only): {e}")
        return chroma_results

    else:
        raise ValueError(f"Unknown backend: {VECTOR_DB_BACKEND}")

Plan de rollout:

1. Deploy con VECTOR_DB_BACKEND=chromadb (sin cambio).
2. Configurar VECTOR_DB_BACKEND=dual en producción → genera logs comparativos.
3. Si después de 1 semana los logs muestran resultados consistentes, cambiar a pinecone.
4. Si algo está mal, rollback a chromadb con cambio de env var (5 minutos, sin redeploy).

Trampas y errores comunes

Trampa 1: payload size demasiado grande

El error: batch de 5000 vectors con embeddings de 3072 dim cada uno.

Síntoma: RequestError: Request body too large. Pinecone tiene limit ~2MB por request.

Cómo prevenir: batch_size de 100-200 con embeddings 1536d. Si dim es mayor, batch_size más chico.

Trampa 2: metadata con tipos no soportados

El error:

metadata = {
    "tags": ["a", "b"],
    "config": {"nested": "dict"},  # ← Pinecone no soporta dict anidado
    "score": np.float64(0.5),       # ← Tipo numpy
}

Síntoma: ValidationError: metadata field "config" type not supported.

Cómo prevenir: clean_metadata_for_pinecone (visto arriba) flatten dicts y convierte tipos.

Trampa 3: dimensión mismatch silencioso

El error: ChromaDB tiene embeddings de 1536d. Índice Pinecone con dim=384 (de un test).

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

Cómo prevenir: validar dimensión del primer batch antes de procesar todo.

def validate_dimensions(items, expected_dim):
    if items and len(items[0]["embedding"]) != expected_dim:
        raise ValueError(
            f"Dimension mismatch: items={len(items[0]['embedding'])}, expected={expected_dim}"
        )

Trampa 4: namespace incorrecto

El error: olvidar el namespace en el upsert. Todo termina en namespace default "".

Síntoma: queries que filtran por namespace devuelven vacío.

Cómo prevenir: namespace siempre obligatorio en wrappers:

def safe_upsert(index, vectors, workspace_id):
    if not workspace_id:
        raise ValueError("workspace_id obligatorio")
    namespace = f"ws_{workspace_id}"
    index.upsert(vectors=vectors, namespace=namespace)

Trampa 5: no rate-limit el script de migración

El error: script paraleliza upsert con 50 workers concurrentes.

Síntoma: Pinecone rate limits, 429 errors, migración falla a mitad.

Cómo prevenir: Pinecone Standard permite ~1000 upserts/segundo. Con batch de 100, son 10 batches/segundo. Más de eso requiere plan superior.

# Throttling
time.sleep(0.05)  # 50ms entre batches → 20 batches/segundo

Trampa 6: corte abrupto de migración a mitad

El error: la VM corre fuera de memoria a mitad. Tienes 3M de 5M docs migrados.

Síntoma: estado mixto, no puedes saber qué falta.

Cómo prevenir: trackear progreso por offset en archivo. Si script falla, retomar desde último offset.

def migrate_with_checkpoint(checkpoint_file: str = "./migration_progress.txt"):
    last_offset = 0
    if os.path.exists(checkpoint_file):
        with open(checkpoint_file) as f:
            last_offset = int(f.read().strip())
        print(f"Resuming from offset {last_offset}")

    # ... migrate from last_offset onwards ...

    # Save progress periodically
    with open(checkpoint_file, "w") as f:
        f.write(str(current_offset))

Ejercicio aplicado

Escenario: eres AI Engineer, tienes que migrar a Pinecone. Datos:

• ChromaDB con 3M chunks de 50 tenants
• Modelo: OpenAI text-embedding-3-small (1536 dim)
• Sistema en producción 24/7, no hay ventana de mantenimiento
• Equipo: 3 personas
• Plazo: 2 semanas

Tu trabajo:

1. Diseña plan de migración incremental.
2. Implementa pseudo-código del script principal.
3. Define rollback plan.

Día 1-2: Setup + tests
  - Crear índice Pinecone (cápsula 03)
  - Implementar export_from_chromadb + upsert_to_pinecone
  - Tests con corpus chico (10K docs en staging)
  - Validar dimension/metric/calidad

Día 3-5: Backfill histórico (3M docs)
  - Script de migración con checkpoint (retomable)
  - Throttling para no romper rate limits
  - Logging de cada batch + retry exponencial
  - Tiempo estimado: 30-45 minutos puros, ~3-4 horas con throttling

Día 6-7: Validación post-backfill
  - Conteos coinciden (ChromaDB vs Pinecone)
  - Sample 100 IDs aleatorios, validar metadata preservada
  - Eval set comparativo (recall debe estar dentro de 2% del baseline)

Día 8-9: Dual-write
  - App escribe nuevos docs en AMBOS sistemas
  - Logs detectan inconsistencias
  - VECTOR_DB_BACKEND=chromadb (queries siguen ahí)

Día 10-12: Dual-read con A/B
  - VECTOR_DB_BACKEND=dual
  - Queries van a chromadb (primario) y pinecone (sombra)
  - Logs comparan resultados
  - Si delta de calidad <2%, OK para switch

Día 13: Switch primario
  - VECTOR_DB_BACKEND=pinecone
  - ChromaDB sigue como fallback en caso de error
  - Monitoring extra durante 24-48 horas

Día 14: Decommission
  - Si Pinecone estable por 48 horas, deprecate ChromaDB
  - Backup final de ChromaDB para retain
  - Cleanup de código legacy

# scripts/migrate_to_pinecone.py
import os
import time
from pathlib import Path

CHECKPOINT_FILE = Path("./migration_checkpoint.txt")
BATCH_SIZE = 100
THROTTLE_MS = 50


def migrate():
    chroma_collection = get_chroma_collection()
    pinecone_index = get_pinecone_index()

    # Resume desde checkpoint si existe
    last_offset = 0
    if CHECKPOINT_FILE.exists():
        last_offset = int(CHECKPOINT_FILE.read_text().strip())
        print(f"Resuming from offset {last_offset}")

    total = chroma_collection.count()
    failed_batches = []

    for batch_start in range(last_offset, total, BATCH_SIZE):
        # Export
        batch = chroma_collection.get(
            limit=BATCH_SIZE,
            offset=batch_start,
            include=["embeddings", "metadatas", "documents"],
        )

        # Group by namespace (workspace_id)
        by_namespace = {}
        for i in range(len(batch["ids"])):
            ws_id = batch["metadatas"][i].get("workspace_id", "default")
            namespace = f"ws_{ws_id}"
            by_namespace.setdefault(namespace, []).append({
                "id": batch["ids"][i],
                "values": batch["embeddings"][i],
                "metadata": clean_metadata({**batch["metadatas"][i], "content": batch["documents"][i][:40000]}),
            })

        # Upsert por namespace
        for namespace, vectors in by_namespace.items():
            try:
                upsert_with_retry(pinecone_index, vectors, namespace)
            except Exception as e:
                print(f"FAILED batch {batch_start} namespace {namespace}: {e}")
                failed_batches.append((batch_start, namespace))

        # Save checkpoint
        CHECKPOINT_FILE.write_text(str(batch_start + BATCH_SIZE))

        # Progress log
        progress = (batch_start + BATCH_SIZE) / total * 100
        print(f"  [{progress:.1f}%] Migrados hasta offset {batch_start + BATCH_SIZE}")

        # Throttle
        time.sleep(THROTTLE_MS / 1000)

    # Final report
    print(f"\nMigración completada.")
    print(f"Failed batches: {failed_batches}")
    if failed_batches:
        # Save list para retry manual
        Path("./failed_batches.json").write_text(json.dumps(failed_batches))


if __name__ == "__main__":
    migrate()

# Cambio de env var en producción
export VECTOR_DB_BACKEND=chromadb

# Restart de la app (rolling, sin downtime)
kubectl rollout restart deployment/rag-api

# 1. Mantener ChromaDB como primario (no cambiar)
# 2. Borrar índice Pinecone (datos migrados son inválidos)
pinecone delete-index --name production-rag

# 3. Análisis de root cause
# 4. Re-planificar migración con fix del problema

# Identificar qué docs fallaron
failed = json.loads(Path("./failed_batches.json").read_text())

# Migrar solo los failed con script ajustado
for batch_start, namespace in failed:
    # ... retry ...

Resumen y siguiente paso

Lo que aprendiste:

• Migración incremental >> big-bang. Dual-write y dual-read permiten rollback en 5 minutos.
• Export por batches (no cargar todo en RAM). Checkpoint en archivo permite retomar.
• Pinecone metadata: solo str, int, float, bool, list[str]. Dicts anidados deben aplanarse.
• Batch size de 100-200 para upserts. Throttling para no romper rate limits.
• Validación obligatoria post-migración: conteos + sample de IDs + calidad con eval set.
• Feature flag con VECTOR_DB_BACKEND permite switch instantáneo en producción.
• Rollback plan en tres niveles: inmediato, datos migrados, re-do.

Checkpoint: antes de avanzar, deberías poder:

• Implementar script de migración con retry, throttling, checkpoint.
• Validar integridad post-migración con conteos + sample + eval set.
• Diseñar rollback plan apropiado a tu sistema.

Siguiente cápsula: 05 — Namespaces como aislamiento multi-tenant nativo.

En ChromaDB usabas workspace_id como metadata para aislar tenants. Pinecone tiene un mecanismo más fuerte: namespaces. La cápsula 05 cubre cómo usarlos para aislamiento garantizado a nivel de motor.

Recursos

1. Pinecone — Upsert Data — API oficial
2. Pinecone — Migration Guide — Patrones de migración
3. Exponential Backoff (AWS Best Practices) — Para retry
4. Feature Flag Best Practices (LaunchDarkly) — Para rollout
5. ChromaDB — Get Operation — Para export
6. Pinecone — Rate Limits — Para throttling

Tiempo estimado: 35-40 minutos Siguiente: 05-namespaces-and-multi-tenancy.md