pipeline-llm-consciente-de-costos

Pipeline LLM Consciente de Costos

Patrones para controlar los costos de las API de LLM mientras se mantiene la calidad. Combina enrutamiento de modelos, seguimiento de presupuesto, lógica de reintentos y almacenamiento en caché de indicaciones en un pipeline componible.

Cuándo Activarlo

• Al construir aplicaciones que llaman a API de LLM (Claude, GPT, etc.)
• Al procesar lotes de elementos con complejidad variable
• Cuando se necesita mantenerse dentro de un presupuesto de gasto de API
• Al optimizar costos sin sacrificar calidad en tareas complejas

Conceptos Clave

1. Enrutamiento de Modelos por Complejidad de la Tarea

Selecciona automáticamente modelos más baratos para tareas simples, reservando modelos costosos 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 modelo basado en 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 tracker (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 tracker 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 Específica

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

from anthropic import (
    APIConnectionError,
    InternalServerError,
    RateLimitError,
)

_RETRYABLE_ERRORS = (APIConnectionError, RateLimitError, InternalServerError)
_MAX_RETRIES = 3

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

4. Almacenamiento en Caché de Indicaciones

Almacenar en caché indicaciones del sistema largas para evitar reenviarlas 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 + almacenamiento en 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 (heredado)	$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 (heredado)	$15.00	$75.00	15x

Mejores Prácticas

• Comience con el modelo más barato y solo enrute a modelos costosos cuando se cumplan los umbrales de complejidad
• Establezca límites de presupuesto explícitos antes de procesar lotes: falle temprano en lugar de gastar de más
• Registre las decisiones de selección de modelos para poder ajustar los umbrales basándose en datos reales
• Use el almacenamiento en caché de indicaciones para indicaciones del sistema de más de 1024 tokens: ahorra tanto costo como latencia
• Nunca reintente errores de autenticación o validación — solo fallos transitorios (red, límite de tasa, error del servidor)

Antipatrones a Evitar

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

Cuándo Usarlo

• Cualquier aplicación que llame a Claude, OpenAI o API de LLM similares
• Pipelines de procesamiento por lotes donde los costos se acumulan rápidamente
• Arquitecturas multimodelo que necesitan enrutamiento inteligente
• Sistemas de producción que necesitan barreras de seguridad de presupuesto