Cápsula 07: Migración sin Downtime (ChromaDB → Pinecone)

Descripción de la cápsula

Migrar una vector database en producción exige proteger al usuario final. Esta cápsula te lleva paso a paso por un patrón gradual para mover datos y tráfico de ChromaDB a Pinecone sin interrupciones notorias. Cubrimos mapeo de IDs, esquemas de metadata, estrategia de dual-write, routing canary, y validación post-migración con scripts Python completos.

¿Por qué migrar de ChromaDB a Pinecone?

ChromaDB es excelente para desarrollo y prototipos. Pero cuando tu RAG crece — más de ~500K vectores, necesidad de SLA, multi-región, o simplemente no quieres mantener infra — Pinecone ofrece zero-ops, escalado automático y latencias bajas. La migración no tiene que ser un "big bang": con el patrón de esta cápsula lo haces sin downtime.

Diferencias críticas: ChromaDB vs Pinecone

Antes de migrar, entiende las diferencias que impactan el diseño de la migración:

Estas diferencias determinan el mapeo de IDs, la transformación de metadata, y si re-embedes o exportas vectores existentes.

Estrategia recomendada: las 5 fases

Vista general

Fase 1: Preparación
   └─ Mapeo ID, metadata, decisión re-embed vs export
Fase 2: Carga inicial en destino
   └─ Script de migración batch
Fase 3: Dual-write temporal
   └─ Escribir en ambos durante N días
Fase 4: Dual-read y validación
   └─ Comparar resultados, accuracy, latencia
Fase 5: Canary + cutover
   └─ Redirigir % tráfico, monitorear, cortar origen

Fase 1: Preparación

1.1 Mapeo de IDs (ChromaDB UUIDs → Pinecone strings)

ChromaDB usa UUIDs internos; Pinecone acepta strings arbitrarios. Puedes:

Opción A — Conservar UUID como string (recomendado):

# ChromaDB devuelve IDs como str "550e8400-e29b-41d4-a716-446655440000"
# Pinecone acepta hasta 512 bytes — UUID encaja perfectamente
pinecone_id = chroma_id  # sin cambios

Opción B — Prefix para evitar colisiones:

Si en algún momento mezclaste IDs de distintas colecciones o sistemas:

def chroma_to_pinecone_id(chroma_id: str, prefix: str = "doc_") -> str:
    """Mapea ID ChromaDB a ID Pinecone con prefix."""
    return f"{prefix}{chroma_id.replace('-', '')}" if prefix else chroma_id

Opción C — IDs semánticos (si los tienes):

Si tus documentos tienen un ID de negocio (ej. doc_12345), úsalo directamente en ambos sistemas para facilitar trazabilidad.

# Si guardaste metadata con "doc_id" en ChromaDB:
metadata = chroma_result["metadatas"][0]
pinecone_id = metadata.get("doc_id") or chroma_result["ids"][0]

1.2 Mapeo de metadata

Pinecone solo acepta tipos escalares. ChromaDB es más permisivo. Necesitas una función de transformación:

from typing import Any

def chroma_metadata_to_pinecone(metadata: dict[str, Any]) -> dict[str, str | int | float | bool]:
    """Convierte metadata ChromaDB a schema compatible con Pinecone."""
    allowed = (str, int, float, bool)
    result = {}
    for k, v in metadata.items():
        if v is None:
            continue
        if isinstance(v, allowed):
            result[k] = v
        elif isinstance(v, list) and all(isinstance(x, str) for x in v):
            result[k] = v  # list[str] permitido en Pinecone
        else:
            result[k] = str(v)  # dict, list[dict], datetime, etc. → serializar
    return result

Reglas típicas:

• None → omitir
• datetime → string ISO
• list[dict] → string JSON
• Nombres de keys: evita caracteres especiales; Pinecone usa . para nested (ej. user.name)

1.3 Embeddings existentes: ¿re-embed o exportar?

Export (recomendado si el model no cambia):

# ChromaDB almacena vectores; puedes leerlos sin re-embeder
collection = client.get_collection("my_collection")
results = collection.get(include=["embeddings", "metadatas", "documents"])
# results["embeddings"] es list[list[float]]

Re-embed: Úsalo si migras a otro model (ej. text-embedding-3-large) o si ChromaDB no te da acceso fácil a vectores en batch.

Fase 2: Script de migración batch completo

Script completo para migrar una colección ChromaDB a un index Pinecone:

#!/usr/bin/env python3
"""
Migración ChromaDB → Pinecone (batch inicial).
Uso: python migrate_chroma_to_pinecone.py --chroma-path ./chroma_db --collection docs
"""
import argparse
import chromadb
from pinecone import Pinecone
from chroma_metadata_to_pinecone import chroma_metadata_to_pinecone  # función anterior


def migrate_collection(
    chroma_path: str,
    collection_name: str,
    pinecone_api_key: str,
    pinecone_index: str,
    pinecone_namespace: str = "default",
    batch_size: int = 100,
) -> dict:
    """
    Migra una colección ChromaDB a Pinecone.
    Retorna stats: {uploaded, failed, errors}.
    """
    client = chromadb.PersistentClient(path=chroma_path)
    collection = client.get_collection(collection_name)
    pc = Pinecone(api_key=pinecone_api_key)
    index = pc.Index(pinecone_index)

    # Obtener todos los datos (embeddings incluidos)
    results = collection.get(
        include=["embeddings", "metadatas", "documents"]
    )
    ids = results["ids"]
    embeddings = results["embeddings"]
    metadatas = results.get("metadatas") or [{}] * len(ids)
    documents = results.get("documents") or [None] * len(ids)

    uploaded = 0
    failed = 0
    errors = []

    for i in range(0, len(ids), batch_size):
        batch_ids = ids[i : i + batch_size]
        batch_embeddings = embeddings[i : i + batch_size]
        batch_metadatas = metadatas[i : i + batch_size]
        batch_docs = documents[i : i + batch_size]

        vectors = []
        for j, (cid, emb, meta, doc) in enumerate(
            zip(batch_ids, batch_embeddings, batch_metadatas, batch_docs)
        ):
            meta_clean = chroma_metadata_to_pinecone(meta or {})
            if doc is not None:
                meta_clean["text"] = doc[:40_000]  # límite Pinecone metadata ~40KB
            vectors.append({
                "id": cid,
                "values": emb,
                "metadata": meta_clean,
            })

        try:
            index.upsert(
                vectors=vectors,
                namespace=pinecone_namespace,
            )
            uploaded += len(vectors)
        except Exception as e:
            failed += len(vectors)
            errors.append({"batch": i // batch_size, "error": str(e)})

    return {"uploaded": uploaded, "failed": failed, "errors": errors}


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--chroma-path", required=True)
    parser.add_argument("--collection", required=True)
    parser.add_argument("--pinecone-index", required=True)
    parser.add_argument("--namespace", default="default")
    parser.add_argument("--batch-size", type=int, default=100)
    args = parser.parse_args()

    import os
    api_key = os.environ.get("PINECONE_API_KEY")
    if not api_key:
        raise SystemExit("PINECONE_API_KEY no definida")

    stats = migrate_collection(
        chroma_path=args.chroma_path,
        collection_name=args.collection,
        pinecone_api_key=api_key,
        pinecone_index=args.pinecone_index,
        pinecone_namespace=args.namespace,
        batch_size=args.batch_size,
    )
    print(f"Migración: {stats['uploaded']} subidos, {stats['failed']} fallidos")
    if stats["errors"]:
        for e in stats["errors"][:5]:
            print(f"  Error: {e}")

Fase 3: Dual-write durante la transición

Durante la ventana de transición (típicamente 7–14 días), escribes en ambos sistemas. Así cualquier documento nuevo o actualizado queda en ChromaDB y en Pinecone.

Implementación con abstracción

from abc import ABC, abstractmethod
from typing import Optional
import chromadb
from pinecone import Pinecone


class VectorStore(ABC):
    @abstractmethod
    def upsert(self, ids: list[str], embeddings: list[list[float]], metadatas: list[dict]):
        pass

    @abstractmethod
    def query(self, embedding: list[float], top_k: int = 5, filter_: Optional[dict] = None):
        pass


class ChromaStore(VectorStore):
    def __init__(self, path: str, collection: str):
        self.client = chromadb.PersistentClient(path=path)
        self.collection = self.client.get_collection(collection)

    def upsert(self, ids, embeddings, metadatas):
        self.collection.upsert(ids=ids, embeddings=embeddings, metadatas=metadatas)

    def query(self, embedding, top_k=5, filter_=None):
        return self.collection.query(
            query_embeddings=[embedding],
            n_results=top_k,
            where=filter_,
            include=["metadatas", "distances"],
        )


class PineconeStore(VectorStore):
    def __init__(self, api_key: str, index: str, namespace: str = "default"):
        pc = Pinecone(api_key=api_key)
        self.index = pc.Index(index)
        self.namespace = namespace

    def upsert(self, ids, embeddings, metadatas):
        vectors = [
            {"id": iid, "values": emb, "metadata": chroma_metadata_to_pinecone(m or {})}
            for iid, emb, m in zip(ids, embeddings, metadatas)
        ]
        self.index.upsert(vectors=vectors, namespace=self.namespace)

    def query(self, embedding, top_k=5, filter_=None):
        r = self.index.query(
            vector=embedding,
            top_k=top_k,
            filter=filter_,
            namespace=self.namespace,
            include_metadata=True,
        )
        return {"ids": [m.id for m in r.matches], "metadatas": [m.metadata for m in r.matches]}


class DualWriteStore(VectorStore):
    """Escribe en ambos, lee del origen (ChromaDB) durante dual-write."""

    def __init__(self, chroma: ChromaStore, pinecone: PineconeStore):
        self.chroma = chroma
        self.pinecone = pinecone

    def upsert(self, ids, embeddings, metadatas):
        self.chroma.upsert(ids, embeddings, metadatas)
        try:
            self.pinecone.upsert(ids, embeddings, metadatas)
        except Exception as e:
            # Log pero no fallar — ChromaDB es source of truth
            import logging
            logging.warning(f"Dual-write Pinecone failed: {e}")

    def query(self, embedding, top_k=5, filter_=None):
        return self.chroma.query(embedding, top_k, filter_)

Durante dual-write, el lector sigue siendo ChromaDB. Solo cuando valides equivalencia pasas a dual-read y luego a Pinecone puro.

Fase 4: Dual-read y validación de equivalencia

4.1 Comparación de resultados

Necesitas un set de queries reales (ej. 50–200 preguntas de producción o sintéticas). Para cada query:

1. Generas embedding.
2. Consultas ChromaDB y Pinecone.
3. Comparas los top-k devueltos (IDs, scores, orden).

def compare_results(chroma_res: dict, pinecone_res: dict, top_k: int = 5) -> dict:
    """
    Compara resultados ChromaDB vs Pinecone.
    Retorna: overlap (Jaccard), order_correlation, score_diff.
    """
    chroma_ids = set(chroma_res.get("ids", [[]])[0][:top_k])
    pinecone_ids = set(pinecone_res.get("ids", [])[:top_k])

    overlap = len(chroma_ids & pinecone_ids) / top_k if top_k else 0
    jaccard = len(chroma_ids & pinecone_ids) / len(chroma_ids | pinecone_ids) if chroma_ids or pinecone_ids else 1.0

    # Diferencia de scores (si ambos usan cosine)
    chroma_dists = chroma_res.get("distances", [[]])[0][:top_k]
    pinecone_scores = [m.get("score") for m in (pinecone_res.get("metadatas") or [])[:top_k]]
    score_diff = 0
    if chroma_dists and pinecone_scores:
        score_diff = sum(abs(c - p) for c, p in zip(chroma_dists, pinecone_scores)) / min(len(chroma_dists), len(pinecone_scores))

    return {
        "overlap": overlap,
        "jaccard": jaccard,
        "score_diff": score_diff,
    }

4.2 Script de validación completa

def validate_migration(
    chroma: ChromaStore,
    pinecone: PineconeStore,
    test_queries: list[str],
    embedding_fn,
    top_k: int = 5,
) -> dict:
    """
    Valida que ChromaDB y Pinecone devuelven resultados equivalentes.
    Retorna métricas agregadas.
    """
    overlaps = []
    score_diffs = []

    for q in test_queries:
        emb = embedding_fn(q)
        chroma_res = chroma.query(emb, top_k=top_k)
        pinecone_res = pinecone.query(emb, top_k=top_k)
        cmp = compare_results(chroma_res, pinecone_res, top_k)
        overlaps.append(cmp["overlap"])
        score_diffs.append(cmp["score_diff"])

    return {
        "mean_overlap": sum(overlaps) / len(overlaps) if overlaps else 0,
        "min_overlap": min(overlaps) if overlaps else 0,
        "mean_score_diff": sum(score_diffs) / len(score_diffs) if score_diffs else 0,
        "n_queries": len(test_queries),
    }

Criterios típicos para avanzar:

• mean_overlap >= 0.9 (90% de los IDs en top-k coinciden)
• mean_score_diff < 0.05 (scores muy parecidos)
• Si no se cumplen, investiga: filtros distintos, normalización, orden de inserción.

Fase 5: Canary y cutover

5.1 Canary traffic routing

En lugar de cortar todo el tráfico de golpe, rediriges un porcentaje (5% → 25% → 50% → 100%) a Pinecone durante ventanas de tiempo (ej. 24–48 h por escalón).

import random

def get_vector_store(use_pinecone_prob: float = 0) -> VectorStore:
    """Devuelve Chroma o Pinecone según probabilidad (canary)."""
    chroma = ChromaStore(path="./chroma_db", collection="docs")
    pinecone = PineconeStore(api_key="...", index="docs", namespace="default")
    if random.random() < use_pinecone_prob:
        return pinecone
    return chroma

# En tu API: store = get_vector_store(use_pinecone_prob=0.05)  # 5% canary

5.2 Métricas para decidir avance o rollback

Regla: Si dos de tres métricas empeoran, detén el avance y mantén dual-write. No aumentes el % de canary hasta resolver.

5.3 Checklist previo al cutover

• Dataset destino sincronizado y validado
• Dual-write activo por ventana mínima (ej. 7 días)
• Dual-read validado (mean_overlap ≥ 0.9)
• Canary plan definido (porcentaje y duración por escalón)
• Rollback probado en staging
• Criterios de rollback documentados y conocidos por el equipo

Criterios de rollback

Activa rollback (vuelve a ChromaDB al 100%) si:

1. Accuracy cae por debajo del umbral (ej. mean_overlap < 0.85)
2. Error rate supera el límite (ej. > 0.5%)
3. Latencia empeora de forma sostenida (p95 > 1.5× origen)
4. Incidentes críticos relacionados con Pinecone (outages, timeouts)

Troubleshooting

1. "Dual-write genera inconsistencias entre Chroma y Pinecone"

Causas típicas: Orden de escritura distinto, fallos silenciosos en uno de los dos, metadata que no se transforma bien.

Solución:

• Valida que el orden de escritura sea idempotente: Chroma primero, Pinecone segundo.
• Si Pinecone falla, registra el error y reintenta en cola asíncrona; no dejes documentos solo en Chroma.
• Revisa chroma_metadata_to_pinecone: tipos no permitidos (dict, list de objetos) causan errores en Pinecone.

2. "Canary muestra latencia interpleentente o picos"

Causas: Cold starts (serverless Pinecone), red, diferencias de carga entre momentos.

Solución:

• Aumenta la ventana de observación (ej. 48 h mínimo por escalón).
• Compara p50, p95, p99; si p50 es estable y p99 es alto, puede ser cold start.
• Considera warm-up: haz queries periódicas de bajo volumen para mantener el index "caliente".

3. "IDs de ChromaDB no coinciden con IDs esperados en Pinecone"

Causas: UUID con formato distinto, encoding, o IDs generados por ChromaDB automáticamente que no guardaste.

Solución:

• Usa include=["embeddings","metadatas","documents"] en get() y reconstruye un mapeo doc_id → chroma_uuid si guardaste doc_id en metadata.
• Si no, el UUID de ChromaDB es válido en Pinecone; pásalo tal cual. Verifica que no estés añadiendo/quitando guiones ni prefijos por error.

4. "Metadata con tipos complejos falla al insertar en Pinecone"

Causas: Pinecone solo acepta str, int, float, bool, list[str].

Solución:

• Serializa a string: datetime → ISO, dict/list[dict] → json.dumps().
• Evita keys con . si no quieres nested; normaliza nombres (snake_case recomendado).
• Valida con un script de dry-run que intente upsert de una muestra antes de la migración completa.

5. "El equipo presiona por cortar rápido"

Riesgo: Sin criterios de rollback claros y sin validación suficiente, aumentas la probabilidad de incidentes.

Solución:

• Documenta criterios de rollback (accuracy, error rate, latencia) y compártelos con el equipo.
• Define un "mínimo viable" de dual-write (ej. 7 días) y de canary (ej. 5% → 25% → 100% en 3 escalones).
• Si hay prisa, reduce la ventana pero no elimines la validación ni los criterios de rollback.

Ejercicios

Ejercicio 1: Mapeo de metadata

Tienes metadata en ChromaDB con {"created_at": datetime(2025,3,1), "tags": ["a","b"], "nested": {"x": 1}}. Escribe la versión compatible con Pinecone.

from datetime import datetime
import json

metadata = {
    "created_at": datetime(2025, 3, 1),
    "tags": ["a", "b"],
    "nested": {"x": 1},
}

def to_pinecone(m):
    out = {}
    for k, v in m.items():
        if v is None:
            continue
        if isinstance(v, (str, int, float, bool)):
            out[k] = v
        elif isinstance(v, list) and all(isinstance(x, str) for x in v):
            out[k] = v
        elif isinstance(v, datetime):
            out[k] = v.isoformat()
        elif isinstance(v, (dict, list)):
            out[k] = json.dumps(v)
    return out

print(to_pinecone(metadata))
# {"created_at": "2025-03-01T00:00:00", "tags": ["a","b"], "nested": "{\"x\": 1}"}

Ejercicio 2: Script de comparación de resultados

Implementa una función que, dados dos listas de IDs ordenados (Chroma vs Pinecone), devuelva el overlap en posición (cuántos IDs coinciden en la misma posición) y el overlap en conjunto (cuántos IDs están en ambos top-k).

def overlap_metrics(chroma_ids: list[str], pinecone_ids: list[str], k: int = 5) -> dict:
    chroma_ids = chroma_ids[:k]
    pinecone_ids = pinecone_ids[:k]
    set_c = set(chroma_ids)
    set_p = set(pinecone_ids)
    position_overlap = sum(1 for i in range(min(len(chroma_ids), len(pinecone_ids)))
                          if chroma_ids[i] == pinecone_ids[i])
    set_overlap = len(set_c & set_p) / k if k else 0
    jaccard = len(set_c & set_p) / len(set_c | set_p) if set_c or set_p else 1.0
    return {
        "position_overlap": position_overlap,
        "position_overlap_ratio": position_overlap / k if k else 0,
        "set_overlap": set_overlap,
        "jaccard": jaccard,
    }

Ejercicio 3: Dual-write con retry

Extiende DualWriteStore para que, si el upsert a Pinecone falla, reintente hasta 3 veces con backoff exponencial (1s, 2s, 4s) antes de registrar el fallo.

import time

def upsert_with_retry(pinecone_store: PineconeStore, ids, embeddings, metadatas, max_retries=3):
    for attempt in range(max_retries):
        try:
            pinecone_store.upsert(ids, embeddings, metadatas)
            return
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

# En DualWriteStore.upsert:
try:
    upsert_with_retry(self.pinecone, ids, embeddings, metadatas)
except Exception as e:
    logging.warning(f"Dual-write Pinecone failed after retries: {e}")

Ejercicio 4: Canary por header

En vez de probabilidad aleatoria, implementa canary por header HTTP: si X-Use-Pinecone: true, usa Pinecone; si no, ChromaDB. ¿Qué ventaja tiene sobre el enfoque aleatorio?

def get_store_from_header(header_value: str | None) -> VectorStore:
    use_pinecone = (header_value or "").lower() in ("1", "true", "yes")
    return pinecone if use_pinecone else chroma

# Ventaja: Permite testing manual (curl -H "X-Use-Pinecone: true" ...)
# y A/B testing controlado (solo ciertos usuarios/tenant via header).
# Aleatorio es mejor para canary "ciego" de porcentaje de tráfico.

Ejercicio 5: Validación de count

Antes de comparar resultados por query, valida que el número total de vectores en ChromaDB coincida con el de Pinecone. Escribe un script que devuelva ambos conteos.

def count_chroma(client, collection_name: str) -> int:
    col = client.get_collection(collection_name)
    return col.count()

def count_pinecone(index, namespace: str = "default") -> int:
    # Pinecone: describe_index_stats() devuelve conteo por namespace
    stats = index.describe_index_stats()
    ns = stats.namespaces.get(namespace)
    return ns.vector_count if ns else 0

# Uso:
chroma_count = count_chroma(client, "docs")
pinecone_count = count_pinecone(pc_index, "default")
assert chroma_count == pinecone_count, f"Mismatch: {chroma_count} vs {pinecone_count}"

Ejercicio 6: Decisión re-embed vs export

Tienes 100K documentos en ChromaDB con embeddings de text-embedding-ada-002. Quieres migrar a Pinecone. Opciones: (A) exportar vectores, (B) re-embed con el mismo model, (C) re-embed con text-embedding-3-small. Elabora una tabla de pros/contras y recomienda una opción según: (i) tiempo limitado, (ii) quieres mejorar calidad de retrieval.

Resumen

• Migración sin downtime requiere fases: preparación (mapeo ID/metadata, decisión re-embed), carga batch, dual-write, validación dual-read, canary y cutover.
• ChromaDB vs Pinecone difieren en IDs (UUID vs string), metadata (tipos permitidos) y filtros; el mapeo explícito evita errores.
• Dual-write mantiene ambos sistemas sincronizados; el lector sigue siendo ChromaDB hasta validar equivalencia.
• Validación se basa en overlap de resultados (mean_overlap ≥ 0.9) y comparación de latencias/error rate.
• Canary reduce riesgo redirigiendo un % de tráfico a Pinecone; si dos de tres métricas empeoran, detén el avance.
• Rollback debe estar definido (accuracy, error rate, latencia) y probado en staging antes del cutover.
• Con este patrón estás listo para consolidar el checklist final de production readiness.

Recursos adicionales

1. Martin Fowler — Canary Release — Patrón canary aplicado a releases
2. Pinecone — Migration guide — Guía oficial de migración
3. ChromaDB — Exporting data — Cómo exportar colecciones
4. Pinecone — Metadata filtering — Sintaxis de filtros
5. Strangler Fig Pattern — Patrón para reemplazar sistemas gradualmente
6. Zero-downtime migrations (blog) — Estrategias de despliegue sin caídas
7. Pinecone — Python SDK — Cliente oficial
8. ChromaDB — Python Client API — API reference

Tiempo estimado: 45–60 minutos
Siguiente: 08-proyecto-production-readiness-checklist.md