pipeline-llm-consciente-de-costos

Pipeline LLM Consciente de Costos

Patrones para controlar los costos de las APIs LLM manteniendo la calidad. Combina enrutamiento de modelos, seguimiento de presupuesto, lógica de reintentos y almacenamiento en caché de prompts en un pipeline componible.

Cuándo Activar

• Construir aplicaciones que llaman a APIs LLM (Claude, GPT, etc.)
• Procesar lotes de elementos con complejidad variable
• Necesidad de mantenerse dentro de un presupuesto para el gasto en API
• Optimizar costos sin sacrificar calidad en tareas complejas

Conceptos Clave

1. Enrutamiento de Modelos por Complejidad de Tarea

Seleccionar automáticamente modelos más baratos para tareas simples, reservando modelos caros para las complejas.

MODEL_SONNET = "claude-sonnet-5"
MODEL_HAIKU = "claude-haiku-4-5-20251001"

_SONNET_TEXT_THRESHOLD = 10_000  # caracteres
_SONNET_ITEM_THRESHOLD = 30     # elementos

def select_model(
    text_length: int,
    item_count: int,
    force_model: str | None = None,
) -> str:
    """Selecciona el modelo según la complejidad de la tarea."""
    if force_model is not None:
        return force_model
    if text_length >= _SONNET_TEXT_THRESHOLD or item_count >= _SONNET_ITEM_THRESHOLD:
        return MODEL_SONNET  # Tarea compleja
    return MODEL_HAIKU  # Tarea simple (3-4x más barato)

2. Seguimiento de Costos Inmutable

Rastrea el gasto acumulado con dataclasses congeladas. Cada llamada a la API devuelve un nuevo rastreador — nunca muta el estado.

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CostRecord:
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float

@dataclass(frozen=True, slots=True)
class CostTracker:
    budget_limit: float = 1.00
    records: tuple[CostRecord, ...] = ()

    def add(self, record: CostRecord) -> "CostTracker":
        """Devuelve un nuevo rastreador con el registro añadido (nunca muta self)."""
        return CostTracker(
            budget_limit=self.budget_limit,
            records=(*self.records, record),
        )

    @property
    def total_cost(self) -> float:
        return sum(r.cost_usd for r in self.records)

    @property
    def over_budget(self) -> bool:
        return self.total_cost > self.budget_limit

3. Lógica de Reintentos Estrecha

Reintentar solo en errores transitorios. Fallar rápido en errores de autenticación o de solicitud incorrecta.

from anthropic import (
    APIConnectionError,
    InternalServerError,
    RateLimitError,
)

_ERRORES_REINTENTABLES = (APIConnectionError, RateLimitError, InternalServerError)
_MAX_REINTENTOS = 3

def call_with_retry(func, *, max_retries: int = _MAX_REINTENTOS):
    """Reintenta solo en errores transitorios, falla rápido en otros."""
    for intento in range(max_retries):
        try:
            return func()
        except _ERRORES_REINTENTABLES:
            if intento == max_retries - 1:
                raise
            time.sleep(2 ** intento)  # Retroceso exponencial
    # AuthenticationError, BadRequestError etc. → se lanza inmediatamente

4. Almacenamiento en Caché de Prompts

Almacenar en caché prompts largos del sistema para evitar reenviarlos en cada solicitud.

messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": system_prompt,
                "cache_control": {"type": "ephemeral"},  # Almacenar en caché esto
            },
            {
                "type": "text",
                "text": user_input,  # Parte variable
            },
        ],
    }
]

Composición

Combina las cuatro técnicas en una única función de pipeline:

def process(text: str, config: Config, tracker: CostTracker) -> tuple[Result, CostTracker]:
    # 1. Enrutar modelo
    model = select_model(len(text), estimated_items, config.force_model)

    # 2. Verificar presupuesto
    if tracker.over_budget:
        raise BudgetExceededError(tracker.total_cost, tracker.budget_limit)

    # 3. Llamar con reintentos + caché
    response = call_with_retry(lambda: client.messages.create(
        model=model,
        messages=build_cached_messages(system_prompt, text),
    ))

    # 4. Rastrear costo (inmutable)
    record = CostRecord(model=model, input_tokens=..., output_tokens=..., cost_usd=...)
    tracker = tracker.add(record)

    return parse_result(response), tracker

Referencia de Precios (2026)

Modelo	Entrada ($/1M tokens)	Salida ($/1M tokens)	Costo Relativo
Haiku 3.5 (legado)	$0.80	$4.00	0.8x
Haiku 4.5	$1.00	$5.00	1x
Sonnet 5	$2.00	$10.00	2x
Sonnet 4.6	$3.00	$15.00	3x
Opus 4.8	$5.00	$25.00	5x
Fable 5 / Mythos 5	$10.00	$50.00	10x
Opus 4.0 / 4.1 (legado)	$15.00	$75.00	15x

Mejores Prácticas

• Comienza con el modelo más barato y solo enruta a modelos caros cuando se cumplan los umbrales de complejidad
• Establece límites de presupuesto explícitos antes de procesar lotes — falla temprano en lugar de gastar de más
• Registra las decisiones de selección de modelo para poder ajustar los umbrales basándote en datos reales
• Usa almacenamiento en caché de prompts para prompts del sistema de más de 1024 tokens — ahorra tanto costo como latencia
• Nunca reintentes en errores de autenticación o validación — solo fallos transitorios (red, límite de velocidad, error del servidor)

Anti-patrones a Evitar

• Usar el modelo más caro para todas las solicitudes independientemente de la complejidad
• Reintentar en todos los errores (desperdicia presupuesto en fallos permanentes)
• Mutar el estado del rastreador de costos (dificulta la depuración y auditoría)
• Hardcodear nombres de modelo en todo el código base (usa constantes o configuración)
• Ignorar el almacenamiento en caché de prompts para prompts del sistema repetitivos

Cuándo Usar

• Cualquier aplicación que llame a Claude, OpenAI o APIs LLM similares
• Pipelines de procesamiento por lotes donde los costos se acumulan rápidamente
• Arquitecturas multi-modelo que necesitan enrutamiento inteligente
• Sistemas de producción que necesitan límites de presupuesto y protección contra gastos excesivos