Módulo 5: Landscape de Vector Databases para AI Engineers

Cápsula 08: Proyecto - Decision Tree para Elegir Vector DB

Siguiente →

Descripción de la cápsula

Cerrarás el módulo diseñando e implementando un árbol de decisión programático para seleccionar vector database en proyectos RAG. No es solo un diagrama estático: construirás un motor en Python que recibe requisitos concretos y genera una recomendación con justificación, alternativas y trade-offs explícitos.

El output alimenta directamente el Módulo 6 (matriz de scoring).

Tiempo estimado: 35-45 minutos

🎯 Objetivo del proyecto

  • Implementar un decision tree como código Python con 6 dimensiones de decisión.
  • Producir recomendación principal + alternativa con justificación.
  • Validar contra 5 escenarios reales y generar visualización Mermaid.

📋 Especificaciones del proyecto

Requisitos funcionales

  1. Recibir requisitos del proyecto y evaluar 6 dimensiones: volumen, operación, presupuesto, latencia, compliance y equipo.
  2. Producir recomendación principal con score + alternativa con condiciones de cuándo preferirla.
  3. Justificar por qué se descartaron las demás opciones.

Success Criteria

  • ✅ Cubre 5 proveedores: ChromaDB, Pinecone, Weaviate, Qdrant, Milvus
  • ✅ 5 escenarios validados con resultado coherente
  • ✅ Justificación con trade-offs técnicos y financieros
  • ✅ Visualización Mermaid generada

🧠 Contexto antes de empezar

Este proyecto sintetiza las cápsulas 02-07: panorama de proveedores (perfiles), managed vs self-hosted (nodo clave), features para RAG (criterios), costos/trade-offs (presupuesto), cuándo elegir cada opción (lógica) y anti-patrones (validación). Si algún concepto no está claro, revisa la cápsula correspondiente antes de continuar.

💻 Implementación paso a paso

Paso 1: Definir el modelo de requisitos

El primer paso es estructurar las entradas que el árbol necesita para tomar una decisión. Cada campo representa una dimensión que impacta la elección.

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class OperationalModel(Enum):
    MANAGED = "managed"
    SELF_HOSTED = "self_hosted"
    FLEXIBLE = "flexible"  # sin preferencia fuerte


class ComplianceLevel(Enum):
    NONE = "none"
    BASIC = "basic"        # GDPR genérico, datos no sensibles
    STRICT = "strict"      # residencia de datos, auditoría, regulación financiera/salud


class TeamCapability(Enum):
    MINIMAL = "minimal"    # 1-2 devs, sin DevOps
    MODERATE = "moderate"  # equipo pequeño con algo de infra
    STRONG = "strong"      # equipo con DevOps o plataforma dedicada


class ProjectStage(Enum):
    POC = "poc"
    VALIDATION = "validation"
    PRODUCTION = "production"
    SCALE = "scale"


@dataclass
class ProjectRequirements:
    """Requisitos del proyecto para selección de vector DB."""

    vector_count: int                             # vectores actuales o proyectados a 6 meses
    operational_preference: OperationalModel      # managed vs self-hosted
    monthly_budget_usd: float                     # presupuesto mensual para vector DB
    target_latency_p95_ms: float                  # latencia objetivo p95 en ms
    compliance: ComplianceLevel                   # nivel de compliance requerido
    team_capability: TeamCapability               # capacidad operativa del equipo
    project_stage: ProjectStage                   # etapa actual del producto
    needs_hybrid_search: bool = False             # ¿requiere búsqueda híbrida?
    needs_multi_tenancy: bool = False             # ¿requiere aislamiento multi-tenant?
    project_name: str = "unnamed"                 # nombre del proyecto (para reportes)

    def validate(self) -> list[str]:
        """Valida consistencia de los requisitos."""
        warnings = []
        if self.vector_count < 0:
            warnings.append("vector_count no puede ser negativo")
        if self.monthly_budget_usd < 0:
            warnings.append("monthly_budget_usd no puede ser negativo")
        if self.target_latency_p95_ms <= 0:
            warnings.append("target_latency_p95_ms debe ser positivo")
        if (self.compliance == ComplianceLevel.STRICT
                and self.operational_preference == OperationalModel.MANAGED):
            warnings.append("ALERTA: compliance estricto + managed puede ser contradictorio")
        if self.vector_count > 5_000_000 and self.team_capability == TeamCapability.MINIMAL:
            warnings.append("ALERTA: escala >5M con equipo mínimo es riesgo operativo")
        if self.project_stage == ProjectStage.POC and self.monthly_budget_usd > 500:
            warnings.append("INFO: presupuesto alto para PoC — considera opción gratuita")
        return warnings

¿Por qué este diseño? Los Enum fuerzan inputs válidos, validate() detecta combinaciones contradictorias antes de ejecutar el árbol, y cada campo mapea a una dimensión cubierta en cápsulas 02-06.

Paso 2: Definir el perfil de cada vector DB

Antes de construir el árbol, necesitas un modelo de cada proveedor. Esto centraliza la información y facilita actualizar el árbol cuando el mercado cambie.

@dataclass
class VectorDBProfile:
    """Perfil de un proveedor de vector database."""

    name: str
    max_scale_comfort: int         # vectores donde opera cómodamente
    supports_managed: bool
    supports_self_hosted: bool
    supports_hybrid_search: bool
    supports_multi_tenancy: bool
    compliance_ready: bool         # ¿soporta compliance estricto de forma nativa?
    min_monthly_cost_usd: float    # costo mínimo operativo estimado
    operational_complexity: int    # 1 (simple) a 5 (requiere equipo dedicado)
    typical_latency_p95_ms: float  # latencia típica p95 en producción
    best_for: list[str] = field(default_factory=list)
    risks: list[str] = field(default_factory=list)


VECTOR_DB_PROFILES = {
    "ChromaDB": VectorDBProfile(
        name="ChromaDB",
        max_scale_comfort=500_000,
        supports_managed=False, supports_self_hosted=True,
        supports_hybrid_search=False, supports_multi_tenancy=False,
        compliance_ready=False,
        min_monthly_cost_usd=0, operational_complexity=1, typical_latency_p95_ms=5.0,
        best_for=["PoC y prototipos", "aprendizaje", "desarrollo local"],
        risks=["no pensado para escala productiva", "sin managed", "features limitadas"],
    ),
    "Pinecone": VectorDBProfile(
        name="Pinecone",
        max_scale_comfort=50_000_000,
        supports_managed=True, supports_self_hosted=False,
        supports_hybrid_search=True, supports_multi_tenancy=True,
        compliance_ready=True,
        min_monthly_cost_usd=70, operational_complexity=1, typical_latency_p95_ms=10.0,
        best_for=["producción managed sin DevOps", "SLA rápido", "time-to-market"],
        risks=["vendor lock-in fuerte", "costo escala con volumen", "sin self-hosted"],
    ),
    "Weaviate": VectorDBProfile(
        name="Weaviate",
        max_scale_comfort=100_000_000,
        supports_managed=True, supports_self_hosted=True,
        supports_hybrid_search=True, supports_multi_tenancy=True,
        compliance_ready=True,
        min_monthly_cost_usd=25, operational_complexity=3, typical_latency_p95_ms=8.0,
        best_for=["hybrid search avanzado", "flexibilidad managed+self-hosted", "ecosistema amplio"],
        risks=["curva de aprendizaje mayor", "complejidad operativa self-hosted"],
    ),
    "Qdrant": VectorDBProfile(
        name="Qdrant",
        max_scale_comfort=100_000_000,
        supports_managed=True, supports_self_hosted=True,
        supports_hybrid_search=True, supports_multi_tenancy=True,
        compliance_ready=True,
        min_monthly_cost_usd=25, operational_complexity=2, typical_latency_p95_ms=5.0,
        best_for=["performance y eficiencia", "control self-hosted con buena UX"],
        risks=["ecosistema más joven", "self-hosted requiere disciplina operativa"],
    ),
    "Milvus": VectorDBProfile(
        name="Milvus",
        max_scale_comfort=1_000_000_000,
        supports_managed=True, supports_self_hosted=True,
        supports_hybrid_search=True, supports_multi_tenancy=True,
        compliance_ready=True,
        min_monthly_cost_usd=100, operational_complexity=5, typical_latency_p95_ms=12.0,
        best_for=["escalas enterprise masivas (>10M)", "equipo de plataforma dedicado"],
        risks=["overkill para proyectos pequeños", "complejidad operativa alta"],
    ),
}

Nota: estos valores son estimaciones pedagógicas. En un proyecto real, valida contra benchmarks propios y pricing actualizado.

Paso 3: Construir el motor de decisión

Cada nodo evalúa una dimensión y puntúa candidatos. La estructura es deliberadamente legible: otro engineer debe poder seguir la lógica sin documentación adicional.

@dataclass
class Recommendation:
    """Resultado del decision tree."""

    primary: str                           # nombre de la DB recomendada
    primary_score: float                   # score normalizado 0-100
    primary_justification: list[str]       # razones de la recomendación
    alternative: str                       # segunda opción
    alternative_score: float
    alternative_justification: list[str]
    eliminated: dict[str, list[str]]       # {db_name: [razones de eliminación]}
    warnings: list[str]                    # alertas del proceso
    confidence: str                        # "high", "medium", "low"


def evaluate_decision_tree(reqs: ProjectRequirements) -> Recommendation:
    """
    Evalúa requisitos contra perfiles de vector DB.
    Retorna recomendación estructurada con justificación.
    """
    warnings = reqs.validate()
    scores: dict[str, float] = {}
    reasons: dict[str, list[str]] = {}
    eliminations: dict[str, list[str]] = {}

    for db_name, profile in VECTOR_DB_PROFILES.items():
        score = 0.0
        db_reasons = []
        db_eliminations = []

        # ── Nodo 1: Escala ──
        if reqs.vector_count <= profile.max_scale_comfort:
            scale_ratio = reqs.vector_count / profile.max_scale_comfort
            if scale_ratio < 0.3:
                score += 20
                db_reasons.append(f"escala cómoda ({reqs.vector_count:,} vectores)")
            elif scale_ratio < 0.7:
                score += 15
                db_reasons.append("escala dentro de rango operativo")
            else:
                score += 8
                db_reasons.append("escala cercana al límite cómodo")
        else:
            score -= 30
            db_eliminations.append(
                f"escala ({reqs.vector_count:,}) excede zona cómoda "
                f"({profile.max_scale_comfort:,})"
            )

        # ── Nodo 2: Modelo operativo ──
        if reqs.operational_preference == OperationalModel.MANAGED:
            if profile.supports_managed:
                score += 20
                db_reasons.append("soporta modelo managed")
            else:
                score -= 25
                db_eliminations.append("no ofrece opción managed")

        elif reqs.operational_preference == OperationalModel.SELF_HOSTED:
            if profile.supports_self_hosted:
                score += 20
                db_reasons.append("soporta self-hosted")
            else:
                score -= 25
                db_eliminations.append("no ofrece opción self-hosted")

        else:  # FLEXIBLE
            if profile.supports_managed and profile.supports_self_hosted:
                score += 15
                db_reasons.append("flexible: soporta ambos modelos")
            elif profile.supports_managed or profile.supports_self_hosted:
                score += 8
                db_reasons.append("soporta al menos un modelo operativo")

        # ── Nodo 3: Presupuesto ──
        if reqs.monthly_budget_usd >= profile.min_monthly_cost_usd:
            budget_headroom = reqs.monthly_budget_usd / max(profile.min_monthly_cost_usd, 1)
            if budget_headroom > 3:
                score += 15
                db_reasons.append("presupuesto holgado para esta opción")
            else:
                score += 10
                db_reasons.append("presupuesto suficiente")
        else:
            score -= 20
            db_eliminations.append(
                f"presupuesto (${reqs.monthly_budget_usd:.0f}) menor que "
                f"costo mínimo (${profile.min_monthly_cost_usd:.0f})"
            )

        # ── Nodo 4: Latencia ──
        if profile.typical_latency_p95_ms <= reqs.target_latency_p95_ms:
            latency_margin = reqs.target_latency_p95_ms - profile.typical_latency_p95_ms
            if latency_margin > 10:
                score += 15
                db_reasons.append("latencia con margen amplio")
            else:
                score += 10
                db_reasons.append("latencia dentro de objetivo")
        else:
            score -= 15
            db_eliminations.append(
                f"latencia típica ({profile.typical_latency_p95_ms}ms) "
                f"excede objetivo ({reqs.target_latency_p95_ms}ms)"
            )

        # ── Nodo 5: Compliance ──
        if reqs.compliance == ComplianceLevel.STRICT:
            if profile.compliance_ready and profile.supports_self_hosted:
                score += 20
                db_reasons.append("soporta compliance estricto con self-hosted")
            elif profile.compliance_ready:
                score += 10
                db_reasons.append("compliance ready (verificar residencia de datos)")
            else:
                score -= 20
                db_eliminations.append("no cumple requisitos de compliance estricto")
        elif reqs.compliance == ComplianceLevel.BASIC:
            if profile.compliance_ready:
                score += 10
                db_reasons.append("compliance básico cubierto")
            else:
                score += 3

        # ── Nodo 6: Capacidad del equipo ──
        complexity_gap = profile.operational_complexity - _team_to_complexity(reqs.team_capability)
        if complexity_gap <= 0:
            score += 15
            db_reasons.append("complejidad operativa manejable por el equipo")
        elif complexity_gap == 1:
            score += 5
            db_reasons.append("complejidad operativa al límite del equipo")
        else:
            score -= 15
            db_eliminations.append(
                f"complejidad operativa ({profile.operational_complexity}/5) "
                f"excede capacidad del equipo"
            )

        # ── Bonus: Features especiales ──
        if reqs.needs_hybrid_search:
            if profile.supports_hybrid_search:
                score += 10
                db_reasons.append("soporta hybrid search")
            else:
                score -= 10
                db_eliminations.append("no soporta hybrid search requerido")

        if reqs.needs_multi_tenancy:
            if profile.supports_multi_tenancy:
                score += 10
                db_reasons.append("soporta multi-tenancy")
            else:
                score -= 10
                db_eliminations.append("no soporta multi-tenancy requerido")

        # ── Bonus: Etapa del proyecto ──
        if reqs.project_stage == ProjectStage.POC:
            if profile.operational_complexity <= 2:
                score += 10
                db_reasons.append("baja fricción para PoC")
        elif reqs.project_stage == ProjectStage.SCALE:
            if profile.max_scale_comfort > 10_000_000:
                score += 10
                db_reasons.append("preparado para escala")

        scores[db_name] = score
        reasons[db_name] = db_reasons
        if db_eliminations:
            eliminations[db_name] = db_eliminations

    # ── Ranking final ──
    sorted_dbs = sorted(scores.items(), key=lambda x: x[1], reverse=True)

    primary_name = sorted_dbs[0][0]
    primary_score = sorted_dbs[0][1]
    alternative_name = sorted_dbs[1][0]
    alternative_score = sorted_dbs[1][1]

    score_gap = primary_score - alternative_score
    if score_gap > 30:
        confidence = "high"
    elif score_gap > 15:
        confidence = "medium"
    else:
        confidence = "low"

    eliminated_others = {
        name: eliminations.get(name, ["puntuó más bajo que las opciones principales"])
        for name, _ in sorted_dbs[2:]
    }

    return Recommendation(
        primary=primary_name,
        primary_score=primary_score,
        primary_justification=reasons[primary_name],
        alternative=alternative_name,
        alternative_score=alternative_score,
        alternative_justification=reasons[alternative_name],
        eliminated=eliminated_others,
        warnings=warnings,
        confidence=confidence,
    )


def _team_to_complexity(capability: TeamCapability) -> int:
    """Mapea capacidad del equipo a nivel de complejidad que puede manejar."""
    return {
        TeamCapability.MINIMAL: 1,
        TeamCapability.MODERATE: 3,
        TeamCapability.STRONG: 5,
    }[capability]

Decisiones de diseño: scoring aditivo (cada nodo suma/resta puntos), eliminación suave (penalizaciones en vez de hard-filters), y confianza basada en la diferencia de score entre las dos mejores opciones.

Paso 4: Generar el reporte de recomendación

El reporte debe ser legible sin contexto adicional.

def generate_report(reqs: ProjectRequirements, rec: Recommendation) -> str:
    """Genera reporte legible de la recomendación."""
    lines = []
    lines.append("=" * 70)
    lines.append(f"  DECISION TREE REPORT: {reqs.project_name}")
    lines.append("=" * 70)

    lines.append("\n📋 INPUTS")
    lines.append("-" * 40)
    inputs = [
        ("Vectores", f"{reqs.vector_count:,}"), ("Modelo op.", reqs.operational_preference.value),
        ("Budget", f"${reqs.monthly_budget_usd:,.0f}/mes"), ("Latencia p95", f"{reqs.target_latency_p95_ms:.0f}ms"),
        ("Compliance", reqs.compliance.value), ("Equipo", reqs.team_capability.value),
        ("Etapa", reqs.project_stage.value),
        ("Hybrid search", "sí" if reqs.needs_hybrid_search else "no"),
        ("Multi-tenancy", "sí" if reqs.needs_multi_tenancy else "no"),
    ]
    for label, val in inputs:
        lines.append(f"  {label:<20} {val}")

    if rec.warnings:
        lines.append(f"\n⚠️  ALERTAS ({len(rec.warnings)})")
        lines.append("-" * 40)
        for w in rec.warnings:
            lines.append(f"  ⚠ {w}")

    lines.append(f"\n✅ RECOMENDACIÓN PRINCIPAL")
    lines.append("-" * 40)
    lines.append(f"  → {rec.primary}  (score: {rec.primary_score:.0f}, confianza: {rec.confidence})")
    for r in rec.primary_justification:
        lines.append(f"    • {r}")

    lines.append(f"\n🔄 ALTERNATIVA")
    lines.append("-" * 40)
    lines.append(f"  → {rec.alternative}  (score: {rec.alternative_score:.0f})")
    for r in rec.alternative_justification:
        lines.append(f"    • {r}")
    lines.append(f"\n  Preferir {rec.alternative} sobre {rec.primary} si:")
    _print_switch_conditions(lines, rec.primary, rec.alternative)

    lines.append(f"\n❌ OPCIONES DESCARTADAS")
    lines.append("-" * 40)
    for db_name, elim_reasons in rec.eliminated.items():
        lines.append(f"  {db_name}:")
        for r in elim_reasons:
            lines.append(f"    ✗ {r}")

    lines.append(f"\n📌 PRÓXIMOS PASOS")
    lines.append("-" * 40)
    if rec.confidence == "low":
        lines.append("  1. Opciones muy cercanas — PoC comparativo recomendado.")
        lines.append("  2. Define métricas de éxito antes de la prueba.")
    elif rec.confidence == "medium":
        lines.append(f"  1. Valida {rec.primary} con PoC de 1-2 semanas.")
        lines.append(f"  2. Ten {rec.alternative} como plan B documentado.")
    else:
        lines.append(f"  1. Procede con {rec.primary} para la fase actual.")
        lines.append(f"  2. Reevalúa en 3 meses o al cambiar escala 3x.")

    lines.append("\n" + "=" * 70)
    return "\n".join(lines)


def _print_switch_conditions(lines: list[str], primary: str, alternative: str):
    """Genera condiciones para preferir la alternativa."""
    switch_map = {
        ("ChromaDB", "Pinecone"): ["necesitas SLA en producción", "sin capacidad de mantener infra propia"],
        ("Pinecone", "Weaviate"): ["hybrid search es requisito fuerte", "necesitas opción self-hosted futura"],
        ("Pinecone", "Qdrant"): ["buscas mejor ratio performance/costo", "quieres flexibilidad operativa"],
        ("Weaviate", "Qdrant"): ["priorizas simplicidad sobre features avanzadas", "latencia p95 es crítica"],
        ("Qdrant", "Weaviate"): ["necesitas hybrid search más maduro", "quieres ecosistema más amplio"],
        ("Milvus", "Weaviate"): ["la escala es menor a 10M vectores", "prefieres menor complejidad operativa"],
    }
    conditions = switch_map.get((primary, alternative), [
        "cambian requisitos de escala o compliance",
        "el equipo adquiere capacidad operativa diferente",
    ])
    for condition in conditions:
        lines.append(f"    • {condition}")

Paso 5: Ejecutar escenarios de validación

Define al menos 5 escenarios con combinaciones distintas de inputs.

def run_validation_scenarios() -> list[dict]:
    """Ejecuta 5 escenarios de validación y retorna resultados."""

    scenarios = [
        {"name": "MVP Startup - PoC rápido",
         "description": "Startup temprana, equipo pequeño, sin presupuesto",
         "requirements": ProjectRequirements(
             project_name="MVP Startup", vector_count=50_000,
             operational_preference=OperationalModel.MANAGED, monthly_budget_usd=0,
             target_latency_p95_ms=50, compliance=ComplianceLevel.NONE,
             team_capability=TeamCapability.MINIMAL, project_stage=ProjectStage.POC),
         "expected_primary": "ChromaDB"},

        {"name": "SaaS en crecimiento",
         "description": "Producto validado, necesita SLA, equipo moderado",
         "requirements": ProjectRequirements(
             project_name="SaaS Growth", vector_count=2_000_000,
             operational_preference=OperationalModel.MANAGED, monthly_budget_usd=500,
             target_latency_p95_ms=15, compliance=ComplianceLevel.BASIC,
             team_capability=TeamCapability.MODERATE, project_stage=ProjectStage.PRODUCTION,
             needs_multi_tenancy=True),
         "expected_primary": "Pinecone"},

        {"name": "Enterprise - Compliance estricto",
         "description": "Regulación financiera, residencia de datos, equipo fuerte",
         "requirements": ProjectRequirements(
             project_name="Enterprise FinTech", vector_count=10_000_000,
             operational_preference=OperationalModel.SELF_HOSTED, monthly_budget_usd=2000,
             target_latency_p95_ms=20, compliance=ComplianceLevel.STRICT,
             team_capability=TeamCapability.STRONG, project_stage=ProjectStage.SCALE,
             needs_hybrid_search=True, needs_multi_tenancy=True),
         "expected_primary": "Weaviate"},

        {"name": "Equipo técnico - Performance first",
         "description": "Equipo con DevOps, prioriza latencia y control",
         "requirements": ProjectRequirements(
             project_name="Performance Team", vector_count=5_000_000,
             operational_preference=OperationalModel.FLEXIBLE, monthly_budget_usd=800,
             target_latency_p95_ms=10, compliance=ComplianceLevel.BASIC,
             team_capability=TeamCapability.STRONG, project_stage=ProjectStage.PRODUCTION,
             needs_hybrid_search=True),
         "expected_primary": "Qdrant"},

        {"name": "Mega-escala enterprise global",
         "description": "Plataforma global, >100M vectores, equipo de plataforma",
         "requirements": ProjectRequirements(
             project_name="Global Platform", vector_count=200_000_000,
             operational_preference=OperationalModel.SELF_HOSTED, monthly_budget_usd=5000,
             target_latency_p95_ms=25, compliance=ComplianceLevel.STRICT,
             team_capability=TeamCapability.STRONG, project_stage=ProjectStage.SCALE,
             needs_hybrid_search=True, needs_multi_tenancy=True),
         "expected_primary": "Milvus"},
    ]

    results = []
    for scenario in scenarios:
        reqs = scenario["requirements"]
        rec = evaluate_decision_tree(reqs)
        match = rec.primary == scenario["expected_primary"]

        results.append({
            "name": scenario["name"],
            "description": scenario["description"],
            "expected": scenario["expected_primary"],
            "actual": rec.primary,
            "score": rec.primary_score,
            "confidence": rec.confidence,
            "match": match,
        })

    return results

Paso 6: Generar diagrama Mermaid

def generate_mermaid_diagram() -> str:
    """Genera diagrama Mermaid del decision tree."""
    diagram = """```mermaid
flowchart TD
    START([🚀 Inicio: Selección de Vector DB]) --> N1

    N1{📊 Volumen > 10M vectores?}
    N1 -- Sí --> N1_HIGH{🏢 Equipo de plataforma?}
    N1 -- No --> N2

    N1_HIGH -- Sí --> MILVUS[🟣 Milvus]
    N1_HIGH -- No --> N1_WARN[⚠️ Reforzar equipo o<br/>reducir scope]

    N2{☁️ Preferencia operativa?}
    N2 -- Managed --> N3_M
    N2 -- Self-hosted --> N3_S
    N2 -- Flexible --> N3_F

    N3_M{💰 Budget > $70/mes?}
    N3_M -- Sí --> N4_M{🔒 Compliance estricto?}
    N3_M -- No --> CHROMADB_POC[🟢 ChromaDB<br/>PoC / desarrollo]

    N4_M -- Sí --> PINECONE_CHECK[Verificar residencia<br/>de datos Pinecone]
    N4_M -- No --> N5_M{🔍 Hybrid search?}

    N5_M -- Sí --> WEAVIATE_M[🔵 Weaviate Cloud]
    N5_M -- No --> PINECONE[🟡 Pinecone]

    N3_S{🔒 Compliance estricto?}
    N3_S -- Sí --> N4_S{⚡ Latencia < 10ms?}
    N3_S -- No --> N4_S2{🔍 Hybrid search?}
    N4_S -- Sí --> QDRANT_SH[🔴 Qdrant self-hosted]
    N4_S -- No --> WEAVIATE_SH[🔵 Weaviate self-hosted]
    N4_S2 -- Sí --> WEAVIATE_SH
    N4_S2 -- No --> QDRANT_SH

    N3_F{📊 Volumen > 1M?}
    N3_F -- Sí --> N4_F{⚡ Latencia < 10ms?}
    N3_F -- No --> N4_F2{💰 Budget bajo?}
    N4_F -- Sí --> QDRANT_F[🔴 Qdrant]
    N4_F -- No --> WEAVIATE_F[🔵 Weaviate]
    N4_F2 -- Sí --> CHROMADB[🟢 ChromaDB]
    N4_F2 -- No --> QDRANT_F

    style MILVUS fill:#9b59b6,color:#fff
    style PINECONE fill:#f1c40f,color:#000
    style CHROMADB fill:#2ecc71,color:#fff
    style CHROMADB_POC fill:#2ecc71,color:#fff
```"""
    return diagram

Paso 7: Script principal completo

Ahora junta todo en un flujo ejecutable.

def main():
    """Ejecuta el decision tree completo con validación."""

    print("=" * 70)
    print("  VECTOR DB DECISION TREE - Módulo 5 Proyecto Final")
    print("=" * 70)

    # ── Paso A: Ejecutar un escenario individual ──
    print("\n" + "─" * 70)
    print("  PARTE 1: Escenario individual")
    print("─" * 70)

    my_project = ProjectRequirements(
        project_name="Mi Proyecto RAG",
        vector_count=800_000,
        operational_preference=OperationalModel.MANAGED,
        monthly_budget_usd=300,
        target_latency_p95_ms=20,
        compliance=ComplianceLevel.BASIC,
        team_capability=TeamCapability.MODERATE,
        project_stage=ProjectStage.PRODUCTION,
        needs_hybrid_search=False,
        needs_multi_tenancy=True,
    )

    rec = evaluate_decision_tree(my_project)
    report = generate_report(my_project, rec)
    print(report)

    # ── Paso B: Validación de escenarios ──
    print("\n" + "─" * 70)
    print("  PARTE 2: Validación de escenarios")
    print("─" * 70)

    results = run_validation_scenarios()

    print(f"\n{'Escenario':<35} {'Esperado':<12} {'Actual':<12} {'Score':>6} {'Conf':>8} {'OK':>4}")
    print("-" * 80)
    for r in results:
        status = "✅" if r["match"] else "⚠️"
        print(
            f"{r['name']:<35} {r['expected']:<12} {r['actual']:<12} "
            f"{r['score']:>6.0f} {r['confidence']:>8} {status:>4}"
        )

    passed = sum(1 for r in results if r["match"])
    total = len(results)
    print(f"\nResultado: {passed}/{total} escenarios validados")

    if passed < total:
        print("\n⚠️  Escenarios con discrepancia (revisa si tienen sentido):")
        for r in results:
            if not r["match"]:
                print(f"  • {r['name']}: esperado {r['expected']}, obtuvo {r['actual']}")

    # ── Paso C: Diagrama Mermaid ──
    print("\n" + "─" * 70)
    print("  PARTE 3: Diagrama Mermaid")
    print("─" * 70)
    print("\nCopia en un archivo .md o en https://mermaid.live/\n")
    print(generate_mermaid_diagram())

    # ── Resumen final ──
    print("\n" + "=" * 70)
    print(f"  PROYECTO COMPLETADO")
    print(f"  ✅ {len(VECTOR_DB_PROFILES)} proveedores | {passed}/{total} escenarios | Mermaid listo")
    print(f"  → Siguiente: Módulo 6 (matriz de decisión con scoring formal)")
    print("=" * 70)


if __name__ == "__main__":
    main()

📊 Output esperado

Al ejecutar el script, verás un reporte con esta estructura:

✅ RECOMENDACIÓN PRINCIPAL
  → Pinecone  (score: 80, confianza: medium)
  Justificación:
    • escala cómoda (800,000 vectores)
    • soporta modelo managed
    • presupuesto suficiente
    • soporta multi-tenancy

🔄 ALTERNATIVA
  → Qdrant  (score: 68)

❌ OPCIONES DESCARTADAS
  ChromaDB: ✗ no ofrece opción managed
  Milvus:   ✗ complejidad operativa excede capacidad del equipo

VALIDACIÓN DE ESCENARIOS
Escenario                           Esperado     Actual       OK
MVP Startup - PoC rápido            ChromaDB     ChromaDB     ✅
SaaS en crecimiento                 Pinecone     Pinecone     ✅
Enterprise - Compliance estricto    Weaviate     Weaviate     ✅
Equipo técnico - Performance first  Qdrant       Qdrant       ✅
Mega-escala enterprise global       Milvus       Milvus       ✅

Resultado: 5/5 escenarios validados

🔧 Troubleshooting del proyecto

"Mi árbol siempre recomienda la misma DB"

Revisa los pesos de cada nodo. Es probable que un solo criterio (como escala o modelo operativo) esté dominando el score. Ajusta los puntos para que ningún nodo individual represente más del 25% del score máximo posible. También verifica que tus escenarios de prueba tengan inputs realmente distintos — si todos tienen team_capability=MINIMAL, naturalmente convergerán.

"El score de dos opciones es casi idéntico"

Esto es información, no un error. Cuando la confianza es "low", el árbol te está diciendo que necesitas más datos para decidir. Opciones:

  1. Agrega un criterio diferenciador (ej. soporte de la comunidad, documentación).
  2. Ejecuta un PoC comparativo con las dos finalistas.
  3. Usa el Módulo 6 (matriz de scoring) para un análisis más granular.

"No sé qué valores poner en los inputs"

Empieza con lo que sabes y marca lo que estás estimando. Valores típicos: PoC (10k-100k vectores, $0, latencia flexible), producción temprana (100k-2M, $100-500, p95 <30ms), escala (>2M, >$500, p95 <15ms).

"Quiero agregar un proveedor que no está en la lista"

Crea un nuevo VectorDBProfile y agrégalo a VECTOR_DB_PROFILES. El motor lo evaluará automáticamente contra todos los nodos existentes.

"La recomendación no coincide con lo que yo habría elegido"

Buena señal. Revisa qué criterio valora el árbol que tú no, o viceversa. Los desacuerdos revelan suposiciones implícitas que vale la pena explicitar. Ajusta pesos según tu contexto y documenta por qué.

🏋️ Ejercicios post-proyecto

Ejercicio 1: Agregar dimensión de vendor lock-in

Extiende el árbol con un criterio de portabilidad que penalice opciones con alto lock-in.

Pistas:

  • Agrega un campo portability_concern: bool a ProjectRequirements.
  • Agrega un campo lock_in_risk: int (1-5) a VectorDBProfile.
  • Crea un nuevo nodo en evaluate_decision_tree.
Ver solución 
# En ProjectRequirements, agregar:
portability_concern: bool = False

# En VectorDBProfile, agregar:
lock_in_risk: int = 1  # 1=bajo, 5=alto

# Actualizar perfiles:
# ChromaDB:  lock_in_risk=1 (open source, estándar)
# Pinecone:  lock_in_risk=5 (API propietaria, sin self-hosted)
# Weaviate:  lock_in_risk=2 (open source, API estándar)
# Qdrant:    lock_in_risk=2 (open source, gRPC estándar)
# Milvus:    lock_in_risk=2 (open source, API estándar)

# En evaluate_decision_tree, agregar nodo:

# ── Nodo 7: Vendor lock-in ──
if reqs.portability_concern:
    if profile.lock_in_risk <= 2:
        score += 10
        db_reasons.append("bajo riesgo de vendor lock-in")
    elif profile.lock_in_risk >= 4:
        score -= 15
        db_eliminations.append(
            f"alto riesgo de lock-in ({profile.lock_in_risk}/5) "
            f"con requisito de portabilidad"
        )

Ejercicio 2: Generar comparación lado a lado

Crea una función que tome dos ProjectRequirements distintos y muestre cómo cambia la recomendación entre escenarios.

Pistas:

  • La función recibe dos objetos ProjectRequirements.
  • Ejecuta evaluate_decision_tree para cada uno.
  • Imprime tabla comparativa con diffs resaltados.
Ver solución 
def compare_scenarios(reqs_a: ProjectRequirements, reqs_b: ProjectRequirements) -> str:
    """Compara recomendaciones entre dos escenarios."""
    rec_a = evaluate_decision_tree(reqs_a)
    rec_b = evaluate_decision_tree(reqs_b)

    lines = []
    lines.append("=" * 70)
    lines.append("  COMPARACIÓN DE ESCENARIOS")
    lines.append("=" * 70)

    lines.append(f"\n{'Dimensión':<25} {'Escenario A':<22} {'Escenario B':<22}")
    lines.append("-" * 70)
    lines.append(f"{'Proyecto':<25} {reqs_a.project_name:<22} {reqs_b.project_name:<22}")
    lines.append(f"{'Vectores':<25} {reqs_a.vector_count:<22,} {reqs_b.vector_count:<22,}")
    lines.append(f"{'Budget':<25} {'$'+str(int(reqs_a.monthly_budget_usd)):<22} {'$'+str(int(reqs_b.monthly_budget_usd)):<22}")
    lines.append(f"{'Modelo op.':<25} {reqs_a.operational_preference.value:<22} {reqs_b.operational_preference.value:<22}")

    lines.append(f"\n{'RESULTADO':<25}")
    lines.append("-" * 70)

    marker_a = " ←" if rec_a.primary != rec_b.primary else ""
    marker_b = " ←" if rec_a.primary != rec_b.primary else ""
    lines.append(f"{'Recomendación':<25} {rec_a.primary + marker_a:<22} {rec_b.primary + marker_b:<22}")
    lines.append(f"{'Score':<25} {rec_a.primary_score:<22.0f} {rec_b.primary_score:<22.0f}")
    lines.append(f"{'Confianza':<25} {rec_a.confidence:<22} {rec_b.confidence:<22}")
    lines.append(f"{'Alternativa':<25} {rec_a.alternative:<22} {rec_b.alternative:<22}")

    if rec_a.primary != rec_b.primary:
        lines.append(f"\n💡 La recomendación cambió de {rec_a.primary} a {rec_b.primary}.")
        lines.append("   Revisa qué input causó el cambio para entender la sensibilidad.")

    return "\n".join(lines)

# Uso: compare_scenarios(reqs_mvp, reqs_produccion)

Ejercicio 3: Sensibilidad — ¿qué input cambia más la decisión?

Crea una función que tome un ProjectRequirements base y varíe un input a la vez para identificar cuáles son los más sensibles (los que más cambian el ranking).

Pistas:

  • Usa dataclasses.replace() para clonar requisitos con un campo cambiado.
  • Itera sobre variaciones predefinidas de cada campo.
  • Compara si cambia primary o si cambia confidence.
Ver solución 
from dataclasses import replace


def sensitivity_analysis(base_reqs: ProjectRequirements) -> str:
    """Analiza qué inputs son más sensibles al cambio de recomendación."""
    base_rec = evaluate_decision_tree(base_reqs)
    lines = []
    lines.append("=" * 70)
    lines.append(f"  ANÁLISIS DE SENSIBILIDAD: {base_reqs.project_name}")
    lines.append(f"  Recomendación base: {base_rec.primary} (score: {base_rec.primary_score:.0f})")
    lines.append("=" * 70)

    variations = [
        ("vector_count", [10_000, 500_000, 2_000_000, 10_000_000, 100_000_000]),
        ("monthly_budget_usd", [0, 50, 200, 500, 2000]),
        ("target_latency_p95_ms", [5, 10, 20, 50, 100]),
        ("operational_preference", list(OperationalModel)),
        ("compliance", list(ComplianceLevel)),
        ("team_capability", list(TeamCapability)),
    ]

    for field_name, values in variations:
        lines.append(f"\n📊 Variando: {field_name}")
        lines.append(f"   {'Valor':<25} {'Recomendación':<15} {'Score':>6} {'Cambió?':>8}")
        lines.append("   " + "-" * 58)

        for val in values:
            try:
                varied_reqs = replace(base_reqs, **{field_name: val})
                varied_rec = evaluate_decision_tree(varied_reqs)
                changed = "⚠️ SÍ" if varied_rec.primary != base_rec.primary else "  no"
                display_val = f"{val:,}" if isinstance(val, (int, float)) else val.value
                lines.append(
                    f"   {str(display_val):<25} {varied_rec.primary:<15} "
                    f"{varied_rec.primary_score:>6.0f} {changed:>8}"
                )
            except (TypeError, ValueError):
                continue

    return "\n".join(lines)

# Uso: print(sensitivity_analysis(mis_requisitos_base))

El output mostrará una tabla por cada dimensión variada. Los campos marcados con "SÍ" en la columna "Cambió?" son los inputs más sensibles — los que debes estimar con mayor cuidado.

✅ Checklist de completitud

Verifica todos estos puntos antes de dar el proyecto por terminado:

Estructura del código:

  • ProjectRequirements tiene las 6 dimensiones obligatorias.
  • VectorDBProfile cubre los 5 proveedores con datos coherentes.
  • evaluate_decision_tree() produce score numérico y Recommendation completa.

Calidad de la decisión:

  • El árbol no depende de un solo criterio para decidir.
  • Justificación explica por qué sí (primaria) y por qué no (descartadas).
  • La alternativa incluye condiciones de cuándo preferirla.

Validación:

  • Mínimo 5 escenarios ejecutados (PoC, producción, enterprise).
  • Discrepancias entre esperado y actual están explicadas.

Output:

  • Reporte legible sin contexto adicional, Mermaid renderiza correctamente.
  • Scores numéricos compatibles con la matriz del Módulo 6.

Resumen

  • Construiste un motor de decisión programático que transforma requisitos en recomendación justificada.
  • Cada nodo del árbol evalúa una dimensión concreta: escala, operación, presupuesto, latencia, compliance y capacidad del equipo.
  • El sistema produce recomendación principal + alternativa, explicando por qué se descartaron las demás opciones.
  • Validaste con 5 escenarios distintos que cubren desde PoC hasta enterprise global.
  • El diagrama Mermaid permite comunicar la lógica a stakeholders no técnicos.
  • La confianza del resultado ("high", "medium", "low") indica cuándo necesitas más análisis antes de decidir.
  • Este proyecto cierra el Módulo 5 y alimenta directamente el Módulo 6 donde formalizarás la decisión con matriz de scoring.
  • Tienes ejercicios de extensión (lock-in, comparación, sensibilidad) para profundizar el análisis.

📚 Recursos adicionales

Tiempo estimado: 35-45 minutos
Siguiente módulo: ../../module-06-decision-matrix/es/01-introduccion-modulo.md

← AnteriorCápsula 07: Anti-patrones al Elegir Vector DBSiguiente →Módulo 6: Decision Matrix para AI Engineers