Módulo 7: Comparación técnica de proveedores

Proyecto: Decision Tool

Siguiente →

Llegaste al cierre del módulo de comparación técnica. Vas a construir una herramienta Python automatizada que materializa todo lo aprendido: toma como input un perfil de requisitos (volumen, latencia objetivo, presupuesto, restricciones, importancia de cada dimensión) y emite una recomendación con justificación cuantitativa.

Es el deliverable que te llevas a tu portfolio y al Módulo 8. Cualquier persona del equipo (o tú mismo en 3 meses) puede correrlo, pasarle un perfil distinto, y obtener la respuesta sin redoing el análisis desde cero.

Al terminar este proyecto vas a tener:

  • Una CLI Python que ranquea proveedores dado un perfil de uso
  • Configuración declarativa (YAML/JSON) para registrar nuevos proveedores sin tocar código
  • Reporte detallado con score, breakdown por dimensión, y razones cualitativas
  • Tests que validan la lógica de filtrado, normalización y scoring
  • Documentación de cómo extenderla para tu producto específico

Especificación

Entradas

Un archivo de perfil del producto, ejemplo perfil_producto.yaml:

nombre: "Chatbot soporte SaaS B2B"
volumen_mensual: 200000
tokens_input_promedio: 600
tokens_output_promedio: 250
warm_pool_horas_mes: 200

restricciones:
  data_residency_eu: false
  soc2_obligatorio: true
  presupuesto_max_usd: 1000
  hipaa_obligatorio: false
  latencia_p95_max_s: 5.0

pesos:
  latencia: 0.30
  costo: 0.20
  calidad: 0.50

Un catálogo de proveedores, providers.yaml:

proveedores:
  - nombre: "OpenAI GPT-4o-mini"
    tipo: "managed"
    latencia_p95_s: 2.3
    precio_input_per_1m: 0.15
    precio_output_per_1m: 0.60
    calidad_score: 86
    cumple_soc2: true
    cumple_hipaa_baa: true
    cumple_data_residency_eu: false
    notas: "Default seguro, lock-in medio"

  - nombre: "Anthropic Claude 3.5 Sonnet"
    tipo: "managed"
    latencia_p95_s: 2.8
    precio_input_per_1m: 3.00
    precio_output_per_1m: 15.00
    calidad_score: 92
    cumple_soc2: true
    cumple_hipaa_baa: true
    cumple_data_residency_eu: false
    notas: "Razonamiento superior; caro"

  - nombre: "OpenRouter Mistral 7B"
    tipo: "managed"
    latencia_p95_s: 3.0
    precio_input_per_1m: 0.07
    precio_output_per_1m: 0.07
    calidad_score: 69
    cumple_soc2: false
    cumple_hipaa_baa: false
    cumple_data_residency_eu: false
    notas: "Costo bajo; calidad limitada"

  - nombre: "Modal Mistral 7B (A10G + warm)"
    tipo: "serverless_gpu"
    latencia_p95_s: 2.7
    precio_gpu_per_second: 0.000306
    segundos_gpu_por_request: 2.0
    calidad_score: 69
    cumple_soc2: true
    cumple_hipaa_baa: false
    cumple_data_residency_eu: true
    notas: "Control de modelo; cold starts a mitigar"

  - nombre: "Self-hosted Ollama Mistral (EU)"
    tipo: "selfhosted"
    latencia_p95_s: 5.5
    precio_vm_per_hour: 1.10
    calidad_score: 69
    cumple_soc2: true       # depende de tu compliance
    cumple_hipaa_baa: true
    cumple_data_residency_eu: true
    notas: "Compliance estricto; requiere DevOps"

Salida

Un reporte CLI bien formateado con:

  1. Resumen del perfil de entrada
  2. Lista de proveedores descartados por restricciones (con razón)
  3. Ranking con scores normalizados
  4. Top 1 con composición del score
  5. Razones cualitativas adicionales
  6. Análisis de sensibilidad básico

Implementación

Crea decision_tool.py:

# decision_tool.py
"""
Decision Tool para selección de proveedor LLM.

Uso:
    python decision_tool.py --perfil perfil_producto.yaml --providers providers.yaml
"""
import argparse
import json
import sys
from dataclasses import dataclass, field
from pathlib import Path
import yaml


# ============================================================
# Modelos de datos
# ============================================================
@dataclass
class Proveedor:
    nombre: str
    tipo: str
    latencia_p95_s: float
    calidad_score: float
    cumple_soc2: bool = True
    cumple_hipaa_baa: bool = True
    cumple_data_residency_eu: bool = False
    notas: str = ""
    # Pricing fields - usados según tipo
    precio_input_per_1m: float | None = None
    precio_output_per_1m: float | None = None
    precio_gpu_per_second: float | None = None
    segundos_gpu_por_request: float | None = None
    precio_vm_per_hour: float | None = None

    @classmethod
    def from_dict(cls, d: dict) -> "Proveedor":
        return cls(**{k: v for k, v in d.items() if k in cls.__annotations__})

    def costo_mensual_estimado(
        self,
        requests_mes: int,
        tokens_input: int,
        tokens_output: int,
        warm_pool_horas: int,
    ) -> float:
        if self.tipo == "managed":
            return (
                requests_mes * tokens_input * self.precio_input_per_1m / 1_000_000
                + requests_mes * tokens_output * self.precio_output_per_1m / 1_000_000
            )
        elif self.tipo == "serverless_gpu":
            gpu_activa = requests_mes * self.segundos_gpu_por_request * self.precio_gpu_per_second
            warm = warm_pool_horas * 3600 * self.precio_gpu_per_second
            storage = 2.0
            return gpu_activa + warm + storage
        elif self.tipo == "selfhosted":
            return 720 * self.precio_vm_per_hour
        else:
            raise ValueError(f"Tipo desconocido: {self.tipo}")


@dataclass
class Perfil:
    nombre: str
    volumen_mensual: int
    tokens_input_promedio: int
    tokens_output_promedio: int
    warm_pool_horas_mes: int
    restricciones: dict
    pesos: dict


# ============================================================
# Carga
# ============================================================
def cargar_perfil(path: Path) -> Perfil:
    with open(path) as f:
        data = yaml.safe_load(f)
    return Perfil(
        nombre=data["nombre"],
        volumen_mensual=data["volumen_mensual"],
        tokens_input_promedio=data["tokens_input_promedio"],
        tokens_output_promedio=data["tokens_output_promedio"],
        warm_pool_horas_mes=data.get("warm_pool_horas_mes", 0),
        restricciones=data.get("restricciones", {}),
        pesos=data["pesos"],
    )


def cargar_proveedores(path: Path) -> list[Proveedor]:
    with open(path) as f:
        data = yaml.safe_load(f)
    return [Proveedor.from_dict(p) for p in data["proveedores"]]


# ============================================================
# Filtrado por restricciones
# ============================================================
def filtrar(proveedores: list[Proveedor], perfil: Perfil) -> tuple[list[Proveedor], list[tuple[str, list[str]]]]:
    r = perfil.restricciones
    validos = []
    descartados = []
    for p in proveedores:
        razones = []
        if r.get("data_residency_eu") and not p.cumple_data_residency_eu:
            razones.append("no cumple EU data residency")
        if r.get("soc2_obligatorio") and not p.cumple_soc2:
            razones.append("no cumple SOC2")
        if r.get("hipaa_obligatorio") and not p.cumple_hipaa_baa:
            razones.append("no cumple HIPAA BAA")
        if r.get("latencia_p95_max_s") and p.latencia_p95_s > r["latencia_p95_max_s"]:
            razones.append(f"latencia P95 {p.latencia_p95_s}s > max {r['latencia_p95_max_s']}s")

        costo = p.costo_mensual_estimado(
            perfil.volumen_mensual,
            perfil.tokens_input_promedio,
            perfil.tokens_output_promedio,
            perfil.warm_pool_horas_mes,
        )
        if r.get("presupuesto_max_usd") and costo > r["presupuesto_max_usd"]:
            razones.append(f"costo ${costo:.0f} > presupuesto ${r['presupuesto_max_usd']:.0f}")

        if razones:
            descartados.append((p.nombre, razones))
        else:
            validos.append(p)
    return validos, descartados


# ============================================================
# Normalización y scoring
# ============================================================
def normalizar(valores: list[float], menor_es_mejor: bool) -> list[float]:
    if not valores or max(valores) == min(valores):
        return [1.0] * len(valores)
    vmin, vmax = min(valores), max(valores)
    if menor_es_mejor:
        return [1 - (v - vmin) / (vmax - vmin) for v in valores]
    return [(v - vmin) / (vmax - vmin) for v in valores]


def scorear(proveedores: list[Proveedor], perfil: Perfil) -> list[dict]:
    if not proveedores:
        return []

    latencias = [p.latencia_p95_s for p in proveedores]
    costos = [
        p.costo_mensual_estimado(
            perfil.volumen_mensual,
            perfil.tokens_input_promedio,
            perfil.tokens_output_promedio,
            perfil.warm_pool_horas_mes,
        )
        for p in proveedores
    ]
    calidades = [p.calidad_score for p in proveedores]

    n_lat = normalizar(latencias, menor_es_mejor=True)
    n_cost = normalizar(costos, menor_es_mejor=True)
    n_qual = normalizar(calidades, menor_es_mejor=False)

    pesos = perfil.pesos
    resultados = []
    for p, lat, cost, qual, nl, nc, nq in zip(
        proveedores, latencias, costos, calidades, n_lat, n_cost, n_qual
    ):
        score = nl * pesos["latencia"] + nc * pesos["costo"] + nq * pesos["calidad"]
        resultados.append(
            {
                "proveedor": p,
                "latencia_s": lat,
                "costo_mensual_usd": cost,
                "calidad_score": qual,
                "n_latencia": nl,
                "n_costo": nc,
                "n_calidad": nq,
                "contribucion_latencia": nl * pesos["latencia"],
                "contribucion_costo": nc * pesos["costo"],
                "contribucion_calidad": nq * pesos["calidad"],
                "score_total": score,
            }
        )
    return sorted(resultados, key=lambda x: x["score_total"], reverse=True)


# ============================================================
# Análisis de sensibilidad
# ============================================================
def sensibilidad(proveedores: list[Proveedor], perfil: Perfil) -> dict:
    """¿El ganador cambia si movemos pesos ±0.1?"""
    pesos_base = perfil.pesos.copy()
    base = scorear(proveedores, perfil)
    if not base:
        return {"estable": False, "variaciones": []}

    ganador_base = base[0]["proveedor"].nombre
    variaciones = []
    for dim, delta in [("latencia", 0.1), ("latencia", -0.1), ("costo", 0.1), ("costo", -0.1), ("calidad", 0.1), ("calidad", -0.1)]:
        nuevos_pesos = pesos_base.copy()
        nuevos_pesos[dim] = max(0, min(1, nuevos_pesos[dim] + delta))
        # Renormalizar para que sumen 1
        total = sum(nuevos_pesos.values())
        nuevos_pesos = {k: v / total for k, v in nuevos_pesos.items()}
        perfil_temp = Perfil(
            **{**perfil.__dict__, "pesos": nuevos_pesos}
        )
        r = scorear(proveedores, perfil_temp)
        if r:
            variaciones.append(
                {"cambio": f"{dim} {'+' if delta > 0 else ''}{delta}", "ganador": r[0]["proveedor"].nombre}
            )

    distintos = {v["ganador"] for v in variaciones}
    return {"estable": len(distintos) == 1 and ganador_base in distintos, "variaciones": variaciones}


# ============================================================
# Reporte
# ============================================================
def reporte(perfil: Perfil, descartados, ranking, sens):
    sep = "=" * 80
    print(f"\n{sep}")
    print(f"  DECISION TOOL — {perfil.nombre}")
    print(f"{sep}\n")

    print(f"📋 Perfil de uso:")
    print(f"   Volumen: {perfil.volumen_mensual:,} req/mes")
    print(f"   Tokens: {perfil.tokens_input_promedio} input + {perfil.tokens_output_promedio} output")
    print(f"   Warm pool: {perfil.warm_pool_horas_mes} hrs/mes")
    print(f"\n⚖️  Pesos: lat={perfil.pesos['latencia']}, cost={perfil.pesos['costo']}, qual={perfil.pesos['calidad']}")

    if descartados:
        print(f"\n❌ Descartados ({len(descartados)}):")
        for nombre, razones in descartados:
            print(f"   {nombre}: {'; '.join(razones)}")

    if not ranking:
        print("\n⚠️  Ningún proveedor cumple las restricciones. Revisar perfil.")
        return

    print(f"\n📊 Ranking ({len(ranking)} candidatos):")
    print(f"   {'Proveedor':<40} {'P95':>6} {'$/mes':>10} {'Cal':>5} {'Score':>7}")
    print(f"   {'-' * 70}")
    for r in ranking:
        p = r["proveedor"]
        print(f"   {p.nombre:<40} {r['latencia_s']:>5.2f}s {r['costo_mensual_usd']:>9.0f} {r['calidad_score']:>4.0f} {r['score_total']:>7.3f}")

    top = ranking[0]
    p = top["proveedor"]
    print(f"\n🏆 Recomendación: {p.nombre}")
    print(f"   Score total: {top['score_total']:.3f}")
    print(f"   Composición:")
    print(f"     Latencia:  {top['contribucion_latencia']:.3f} (norm {top['n_latencia']:.2f})")
    print(f"     Costo:     {top['contribucion_costo']:.3f} (norm {top['n_costo']:.2f})")
    print(f"     Calidad:   {top['contribucion_calidad']:.3f} (norm {top['n_calidad']:.2f})")
    if p.notas:
        print(f"   Notas: {p.notas}")

    print(f"\n🔬 Análisis de sensibilidad (movimiento ±0.1 en pesos):")
    if sens["estable"]:
        print(f"   ✅ Decisión robusta — el ganador no cambia con variaciones de pesos")
    else:
        print(f"   ⚠️  Decisión sensible a pesos — el ganador cambia según ponderación:")
        for v in sens["variaciones"]:
            print(f"     {v['cambio']:<15} → {v['ganador']}")


# ============================================================
# CLI
# ============================================================
def main():
    parser = argparse.ArgumentParser(description="Decision tool para selección de proveedor LLM")
    parser.add_argument("--perfil", required=True, type=Path, help="YAML con perfil del producto")
    parser.add_argument("--providers", required=True, type=Path, help="YAML con catálogo de proveedores")
    parser.add_argument("--json", action="store_true", help="Output como JSON")
    args = parser.parse_args()

    perfil = cargar_perfil(args.perfil)
    proveedores = cargar_proveedores(args.providers)

    validos, descartados = filtrar(proveedores, perfil)
    ranking = scorear(validos, perfil)
    sens = sensibilidad(validos, perfil)

    if args.json:
        print(json.dumps(
            {
                "perfil": perfil.nombre,
                "ranking": [
                    {
                        "proveedor": r["proveedor"].nombre,
                        "score": r["score_total"],
                        "costo_mensual": r["costo_mensual_usd"],
                        "latencia_p95": r["latencia_s"],
                        "calidad": r["calidad_score"],
                    }
                    for r in ranking
                ],
                "descartados": [{"nombre": n, "razones": r} for n, r in descartados],
                "sensibilidad_estable": sens["estable"],
            },
            indent=2,
        ))
    else:
        reporte(perfil, descartados, ranking, sens)


if __name__ == "__main__":
    main()

Tests

Crea test_decision_tool.py:

# test_decision_tool.py
import pytest
from decision_tool import Proveedor, Perfil, filtrar, scorear, normalizar, sensibilidad


def perfil_basico():
    return Perfil(
        nombre="test",
        volumen_mensual=100_000,
        tokens_input_promedio=500,
        tokens_output_promedio=200,
        warm_pool_horas_mes=0,
        restricciones={"presupuesto_max_usd": 500},
        pesos={"latencia": 0.3, "costo": 0.4, "calidad": 0.3},
    )


def test_normalizar_menor_es_mejor():
    n = normalizar([1.0, 2.0, 4.0], menor_es_mejor=True)
    assert n[0] == 1.0   # más bajo = mejor = 1
    assert n[2] == 0.0   # más alto = peor = 0
    assert 0 < n[1] < 1  # intermedio


def test_normalizar_iguales_devuelve_unos():
    n = normalizar([5.0, 5.0], menor_es_mejor=True)
    assert n == [1.0, 1.0]


def test_filtrar_por_presupuesto():
    p_caro = Proveedor(
        nombre="Caro", tipo="managed", latencia_p95_s=2.0, calidad_score=90,
        precio_input_per_1m=10, precio_output_per_1m=30,
    )
    p_barato = Proveedor(
        nombre="Barato", tipo="managed", latencia_p95_s=3.0, calidad_score=70,
        precio_input_per_1m=0.1, precio_output_per_1m=0.5,
    )
    validos, descartados = filtrar([p_caro, p_barato], perfil_basico())
    assert len(validos) == 1
    assert validos[0].nombre == "Barato"
    assert any("presupuesto" in r for _, razones in descartados for r in razones)


def test_filtrar_por_data_residency():
    perfil = perfil_basico()
    perfil.restricciones = {"data_residency_eu": True}
    p_usa = Proveedor(nombre="USA", tipo="managed", latencia_p95_s=2, calidad_score=80,
                       precio_input_per_1m=0.1, precio_output_per_1m=0.1,
                       cumple_data_residency_eu=False)
    p_eu = Proveedor(nombre="EU", tipo="managed", latencia_p95_s=2, calidad_score=80,
                      precio_input_per_1m=0.1, precio_output_per_1m=0.1,
                      cumple_data_residency_eu=True)
    validos, _ = filtrar([p_usa, p_eu], perfil)
    assert {v.nombre for v in validos} == {"EU"}


def test_scoring_premia_mejor_en_todo():
    """Un proveedor estrictamente mejor en todo debe ganar."""
    p1 = Proveedor(nombre="Mejor", tipo="managed", latencia_p95_s=1.0, calidad_score=95,
                    precio_input_per_1m=0.05, precio_output_per_1m=0.05)
    p2 = Proveedor(nombre="Peor", tipo="managed", latencia_p95_s=5.0, calidad_score=60,
                    precio_input_per_1m=5.0, precio_output_per_1m=15.0)
    ranking = scorear([p1, p2], perfil_basico())
    assert ranking[0]["proveedor"].nombre == "Mejor"


def test_sensibilidad_detecta_inestabilidad():
    # Dos proveedores cerca en score → sensibilidad ALTA
    p1 = Proveedor(nombre="A", tipo="managed", latencia_p95_s=2.0, calidad_score=85,
                    precio_input_per_1m=0.3, precio_output_per_1m=0.6)
    p2 = Proveedor(nombre="B", tipo="managed", latencia_p95_s=2.1, calidad_score=84,
                    precio_input_per_1m=0.3, precio_output_per_1m=0.6)
    # Casi gemelos — algunos cambios de pesos deberían cambiar ganador o no
    sens = sensibilidad([p1, p2], perfil_basico())
    # No assert estricto — solo verificamos que la función devuelve estructura válida
    assert "estable" in sens
    assert "variaciones" in sens

Correr:

pip install pytest pyyaml
pytest test_decision_tool.py -v

Uso real

# Caso 1: B2B SaaS standard
python decision_tool.py --perfil perfil_producto.yaml --providers providers.yaml

# Caso 2: cliente con HIPAA
cat > perfil_hipaa.yaml <<EOF
nombre: "Asistente médico"
volumen_mensual: 50000
tokens_input_promedio: 800
tokens_output_promedio: 300
warm_pool_horas_mes: 100
restricciones:
  hipaa_obligatorio: true
  soc2_obligatorio: true
  presupuesto_max_usd: 2000
pesos:
  latencia: 0.2
  costo: 0.2
  calidad: 0.6
EOF

python decision_tool.py --perfil perfil_hipaa.yaml --providers providers.yaml

# Caso 3: salida JSON para integrar en pipeline
python decision_tool.py --perfil perfil_producto.yaml --providers providers.yaml --json > recomendacion.json

README sugerido

# LLM Provider Decision Tool

Recomienda el proveedor óptimo de LLM dado tu perfil de uso.

## Instalación

```bash
pip install pyyaml
```

## Uso

```bash
python decision_tool.py --perfil tu_perfil.yaml --providers providers.yaml
```

## Estructura de archivos

- `perfil_producto.yaml`: requisitos y prioridades de tu producto
- `providers.yaml`: catálogo de proveedores con sus métricas
- `decision_tool.py`: el script

## Extender

Para agregar un proveedor: editar `providers.yaml`. Tipos soportados:
- `managed`: pricing por tokens (OpenAI, Anthropic, OpenRouter)
- `serverless_gpu`: pricing por segundo de GPU (Modal, Replicate)
- `selfhosted`: pricing por hora de VM (AWS, GCP)

Para cambiar pesos: editar `perfil_producto.yaml`. Los pesos deben sumar 1.0.

## Tests

```bash
pytest test_decision_tool.py -v
```

## Importante

Los números de proveedores en `providers.yaml` son **referencias**. Reemplaza con tus benchmarks reales (cápsulas 02-04 del path) antes de usar en producción.

Auto-evaluación

Antes de cerrar el módulo:

  • El script corre sin errores con los archivos de ejemplo
  • Cambiar pesos en perfil_producto.yaml cambia el ranking
  • Activar restricciones (hipaa_obligatorio, data_residency_eu) descarta proveedores correctamente
  • Los tests pasan (pytest -v)
  • La salida JSON con --json es válida y parseable
  • El análisis de sensibilidad detecta cuando hay empate cercano

Conexión con el resto del path

Lo que construiste es el input cualitativo del Módulo 8 (Unified Client):

  • Tu decision tool dice "OpenAI gana en este perfil"
  • El Unified Client del M08 te deja cambiar de proveedor con un parámetro
  • Cuando tu perfil cambie (precios, restricciones), corres decision tool → obtienes nuevo ganador → cambias provider en Unified Client → migración casi cero

Tu trabajo aquí no termina con M07. Empieza a tomar valor real en M08.

Módulo 7 completado

Hiciste algo poco común: convertiste la elección de proveedor LLM en proceso reproducible y defendible. Vas a poder mostrar este decision tool en entrevistas o usar tu propio output en decisiones reales de producto.

Concretamente, ahora puedes:

  • ✅ Benchmarkear latencia con P50/P95/P99 reales
  • ✅ Calcular costos honestos con costos ocultos incluidos
  • ✅ Medir calidad con rubric + LLM-as-judge
  • ✅ Combinar dimensiones en matriz ponderada con sensibilidad
  • ✅ Migrar entre proveedores conociendo el costo de cada migración
  • ✅ Ranquear opciones con una herramienta CLI automatizada

Siguiente módulo

Módulo 8 — Unified AI Client es el cierre del path. Vas a construir el cliente que encapsula todos los proveedores que conociste detrás de una sola interfaz, con fallback automático, configuración declarativa, y testing multi-provider. Es el deliverable que llevas a tu portfolio como prueba de dominio del topic.

Recursos

  1. TOPSIS algorithm (Python implementation) — alternativa al método de scoring lineal.
  2. Click — Python CLI framework — para evolucionar tu CLI si crece.
  3. PyYAML docs — para entender más opciones de carga.
  4. Hypothesis — property-based testing — testing más profundo de la lógica de scoring.
  5. Decision making with multi-objective optimization — teoría aplicable si escalas el tool.
← AnteriorTabla comparativa completaSiguiente →Módulo 8: Unified AI Client — Proyecto integrador final