Cost optimization

Fallback corrige fallos. Pero en producción real, también quieres optimizar costo activamente: routing al provider más barato cuando la tarea lo permite, y al más caro/mejor cuando la calidad importa.

En esta cápsula vas a agregar estrategias de routing al UnifiedClient. La interfaz pública casi no cambia — agregamos un parámetro priority o configuración por defecto, y el cliente decide internamente.

Al terminar vas a poder:

• Configurar múltiples políticas de routing: cost-first, quality-first, balanced
• Cambiar política por request sin re-instanciar el cliente
• Calcular ahorros reales aplicando routing inteligente
• Combinar routing con fallback (compatible, no excluyente)

Modelo mental: routing vs fallback

Son dos features distintas que se confunden:

Son complementarios. Routing decide quién es "primary" para esta request; si ese primary falla, fallback sigue funcionando como en la cápsula 04.

Política simple: tag por provider

Anotamos cada provider en su config con tags que describen sus características:

providers:
  openai-mini:
    name: openai-mini
    type: openai
    model: gpt-4o-mini
    api_key_env: OPENAI_API_KEY
    tags: ["quality", "balanced"]
    cost_tier: "medium"

  openrouter-mistral:
    name: openrouter-mistral
    type: openai_compatible
    model: mistralai/mistral-7b-instruct
    api_key_env: OPENROUTER_API_KEY
    base_url: https://openrouter.ai/api/v1
    tags: ["cheap", "balanced"]
    cost_tier: "low"

  ollama-mistral:
    name: ollama-mistral
    type: openai_compatible
    model: mistral
    base_url: http://localhost:11434/v1
    api_key_env: OLLAMA_DUMMY_KEY
    tags: ["cheap", "privacy", "offline"]
    cost_tier: "free"

  openai-gpt4o:
    name: openai-gpt4o
    type: openai
    model: gpt-4o
    api_key_env: OPENAI_API_KEY
    tags: ["quality"]
    cost_tier: "high"

Estrategias predefinidas

Definimos 3 estrategias de uso común:

Puedes agregar más (e.g., latency-first, privacy-first con tag privacy). El patrón es el mismo.

Implementación: ampliando UnifiedClient

Agrega esto al modelo ProviderConfig en models.py:

class ProviderConfig(BaseModel):
    name: str
    type: Literal["openai", "openai_compatible", "modal_custom"]
    model: str | None = None
    base_url: str | None = None
    api_key_env: str | None = None
    base_url_env: str | None = None
    api_token_env: str | None = None
    price_input_per_1m: float | None = None
    price_output_per_1m: float | None = None
    # Nuevos campos:
    tags: list[str] = Field(default_factory=list)
    cost_tier: Literal["free", "low", "medium", "high"] | None = None

Crea unified_ai_client/routing.py:

# unified_ai_client/routing.py
from typing import Literal
from .adapters.base import BaseAdapter
from .models import ProviderConfig

Priority = Literal["cost-first", "quality-first", "balanced"]

_TIER_RANK = {"free": 0, "low": 1, "medium": 2, "high": 3, None: 999}


def ordenar_por_prioridad(
    adapters: list[BaseAdapter],
    prioridad: Priority,
) -> list[BaseAdapter]:
    """
    Devuelve la lista de adapters reordenada según la política de prioridad.
    No filtra; siempre devuelve los mismos adapters, en orden distinto.
    """
    if prioridad == "cost-first":
        return sorted(adapters, key=lambda a: _TIER_RANK[a.config.cost_tier])
    if prioridad == "quality-first":
        return sorted(adapters, key=lambda a: -_TIER_RANK[a.config.cost_tier])
    if prioridad == "balanced":
        # Mantiene orden, pero filtra a los marcados como "balanced"
        ordenados = [a for a in adapters if "balanced" in a.config.tags]
        if not ordenados:
            return adapters  # fallback: no había ninguno marcado balanced
        return ordenados
    raise ValueError(f"Prioridad desconocida: {prioridad}")

Actualiza client.py:

# unified_ai_client/client.py — solo cambios
from .routing import ordenar_por_prioridad, Priority


class UnifiedClient:
    def __init__(
        self,
        config: ClientConfig,
        umbral_circuito: int = 3,
        duracion_circuito_s: int = 60,
        prioridad_default: Priority | None = None,
    ):
        # ... resto igual
        self.prioridad_default = prioridad_default
        # ...

    def chat(
        self,
        prompt: str,
        *,
        system: str | None = None,
        max_tokens: int = 256,
        temperature: float = 0.7,
        use_fallback: bool = True,
        priority: Priority | None = None,
    ) -> ChatResponse:
        messages = []
        if system:
            messages.append(Message(role="system", content=system))
        messages.append(Message(role="user", content=prompt))
        return self.chat_with_messages(
            messages,
            max_tokens=max_tokens,
            temperature=temperature,
            use_fallback=use_fallback,
            priority=priority,
        )

    def chat_with_messages(
        self,
        messages: list[Message],
        max_tokens: int = 256,
        temperature: float = 0.7,
        use_fallback: bool = True,
        priority: Priority | None = None,
    ) -> ChatResponse:
        # Aplicar política de routing
        prio = priority or self.prioridad_default
        if prio:
            adapters_ordenados = ordenar_por_prioridad(self.adapters, prio)
        else:
            adapters_ordenados = self.adapters

        # Si fallback desactivado, solo el primero
        if not use_fallback:
            adapters_ordenados = adapters_ordenados[:1]

        errores: dict[str, Exception] = {}
        for adapter in adapters_ordenados:
            if self._circuito_abierto(adapter.name):
                errores[adapter.name] = ProviderError(adapter.name, "Circuit abierto")
                continue
            try:
                return self._intentar_con_retry(adapter, messages, max_tokens, temperature)
            except (AuthError, RateLimitError, TimeoutError, ProviderError) as e:
                errores[adapter.name] = e
                self._registrar_fallo(adapter.name)
                continue

        raise AllProvidersFailedError(errores)

Verificación

Crea examples/test_routing.py:

# examples/test_routing.py
from unified_ai_client import UnifiedClient

client = UnifiedClient.from_yaml("examples/clients_fallback.yaml")

# Cost-first: empieza por Ollama (gratis)
r_cost = client.chat("Di hola", max_tokens=20, priority="cost-first")
print(f"\ncost-first → primero intentó: {r_cost.provider}")

# Quality-first: empieza por GPT-4o si está en tu config, si no gpt-4o-mini
r_qual = client.chat("Di hola", max_tokens=20, priority="quality-first")
print(f"quality-first → primero intentó: {r_qual.provider}")

# Balanced: providers con tag "balanced"
r_bal = client.chat("Di hola", max_tokens=20, priority="balanced")
print(f"balanced → primero intentó: {r_bal.provider}")

Output esperado (asumiendo todos los providers funcionando):

cost-first → primero intentó: ollama-mistral
quality-first → primero intentó: openai-gpt4o (si está en config) o openai-mini
balanced → primero intentó: openai-mini (o el primer "balanced" en config)

Política configurable: priority default a nivel cliente

Si tu app siempre quiere cost-first, configúralo a nivel cliente:

client = UnifiedClient.from_yaml(
    "examples/clients_fallback.yaml",
    prioridad_default="cost-first",
)

# Todas las requests usan cost-first sin tener que pasarlo
client.chat("Pregunta 1")
client.chat("Pregunta 2")

# Una request específica puede override
client.chat("Pregunta crítica", priority="quality-first")

Patrones avanzados

Pattern 1 — Routing por contexto del request

Para casos donde la calidad necesaria depende del request (chat casual → barato; análisis complejo → caro), tu app determina el priority y lo pasa:

def chat_inteligente(prompt: str) -> str:
    if "explica" in prompt or "analiza" in prompt:
        return client.chat(prompt, priority="quality-first").text
    return client.chat(prompt, priority="cost-first").text

Pattern 2 — Routing por usuario / tier

Tu producto tiene usuarios free y paid. Free usa el modelo barato; paid el caro:

def chat_para_usuario(prompt: str, tier: str) -> str:
    priority = "quality-first" if tier == "paid" else "cost-first"
    return client.chat(prompt, priority=priority).text

Pattern 3 — Routing por hora del día

En horario laboral usas managed (rápido); de noche usas tu Modal cheaper:

from datetime import datetime

def chat_segun_hora(prompt: str) -> str:
    hora = datetime.now().hour
    priority = "balanced" if 9 <= hora < 18 else "cost-first"
    return client.chat(prompt, priority=priority).text

Cuánto ahorra el routing en práctica

Asume distribución típica de tu producto:

• 70% de requests son "casual" (priority="cost-first" → Mistral barato)
• 30% de requests son "complejos" (priority="quality-first" → GPT-4o-mini)

Comparación con "siempre OpenAI" para 1M req/mes (500 tokens promedio):

• Sin routing (siempre GPT-4o-mini): 1M × 700 × $0.50 / 1M = ~$350/mes
• Con routing (70% Mistral OR + 30% GPT-4o-mini):
  • 700k × 700 × $0.07 / 1M = $34
  • 300k × 700 × $0.50 / 1M = $105
  • Total: $139/mes
• Ahorro: ~60% sin perder calidad en los casos donde importa.

Esto es plata real. Justifica completamente la complejidad agregada de routing.

Trampas comunes

Trampa 1 — "Routing por hora rompe la UX de un usuario activo de noche." Si tu usuario obtiene calidad distinta según la hora, eso es bug, no feature. Routing por hora aplica a workloads batch (jobs internos), no a usuarios finales.

Trampa 2 — "cost-tier 'free' siempre gana en cost-first." Ollama local "gratis" no escala si tu servidor no aguanta tráfico. Si lo tienes en cost-tier free pero solo soporta 10 req/min, cuando tengas 100 req/min vas a tener cola. Define límites o sube su cost-tier.

Trampa 3 — "Olvidé que routing y fallback interactúan." Con priority="cost-first", tu cliente intenta Ollama primero. Si Ollama está caído, fallback va a OpenRouter, después OpenAI. El orden de fallback es el orden post-priority, no el orden de declaración. Verifica que esto sea lo que quieres.

Trampa 4 — "Routing oculta que mi provider barato es malo." Si Mistral 7B responde incorrectamente al 30% de tus requests "casuales", estás dando producto malo al 30% de tus usuarios. Routing por costo asume que todos los providers cumplen tu SLA de calidad — verifica antes con la cápsula 04 del Módulo 7.

Trampa 5 — "Cost-tier sin actualizar." Si OpenAI baja el precio de gpt-4o-mini un 50% mañana, tu cost-tier: medium está desactualizado y rankings de routing son incorrectos. Revisa los cost-tier trimestralmente.

Ejercicio

Agrega una nueva estrategia: priority="privacy-first" que filtra providers a solo los que tengan tag privacy (en tu config: Ollama local). Si ningún provider califica, lanza ConfigError. Diferencia importante: no es "ordenar" — es "filtrar".

Priority = Literal["cost-first", "quality-first", "balanced", "privacy-first"]


def ordenar_por_prioridad(
    adapters: list[BaseAdapter],
    prioridad: Priority,
) -> list[BaseAdapter]:
    if prioridad == "cost-first":
        return sorted(adapters, key=lambda a: _TIER_RANK[a.config.cost_tier])
    if prioridad == "quality-first":
        return sorted(adapters, key=lambda a: -_TIER_RANK[a.config.cost_tier])
    if prioridad == "balanced":
        ordenados = [a for a in adapters if "balanced" in a.config.tags]
        return ordenados if ordenados else adapters
    if prioridad == "privacy-first":
        privacy = [a for a in adapters if "privacy" in a.config.tags]
        if not privacy:
            from .exceptions import ConfigError
            raise ConfigError(
                "priority='privacy-first' requiere al menos un provider con tag 'privacy'"
            )
        return privacy
    raise ValueError(f"Prioridad desconocida: {prioridad}")

r = client.chat("Procesar datos médicos sensibles", priority="privacy-first")
print(f"Resuelto por (solo on-prem): {r.provider}")

Resumen

Aprendiste:

• ✅ Routing es ordenar/filtrar providers según política; fallback es reaccionar a fallos
• ✅ Tags y cost_tier en config permiten políticas declarativas
• ✅ Tres estrategias predefinidas: cost-first, quality-first, balanced
• ✅ Configurar default a nivel cliente, override por request
• ✅ Patterns combinables: routing por contexto del prompt, por tier del usuario, por hora
• ✅ Routing y fallback son compatibles y se combinan naturalmente

Checkpoint: si puedes hacer client.chat("...", priority="cost-first") y ver en logs que intentó primero el provider más barato, estás listo.

Siguiente cápsula

06 — Monitoring y métricas. Tu cliente decide bien pero no sabes qué decisiones está tomando. Agregamos tracking que te diga: cuántas requests por provider, latencia agregada, costo acumulado, errores por tipo. Es la pieza que falta para que tu cliente sea verdaderamente production-ready.

Recursos

1. LiteLLM Router strategies — referencia de cómo otros lo hacen.
2. Langfuse — LLM cost tracking — herramienta para cost tracking en producción.
3. OpenRouter price comparison API — precios de todos los modelos en un endpoint.
4. Anyscale routing patterns — perspectivas de equipos que routean a escala.