6. Cost Control y Rate Limiting

Descripción

En la cápsula anterior implementaste error recovery y resilience: graceful degradation, fallbacks, circuit breakers, retry policies. Tu agente sobrevive fallos. Pero hay un fallo silencioso que no dispara excepciones ni rompe requests — el costo descontrolado. Un agente de IA no es un CRUD endpoint que cuesta fracciones de centavo por request. Un research query puede involucrar 5-15 llamadas al LLM, cada una consumiendo miles de tokens, más tool calls a APIs de pago, más embedding generation para memoria. Sin cost control, un agente en producción puede quemar el budget completo de un mes en horas.

Los números son reales: una query de investigación compleja al Research Agent puede consumir 15,000-30,000 tokens entre planning, tool calls, synthesis y reflection. Con GPT-4o a $2.50/M input y $10.00/M output, una sola query puede costar entre $0.10 y $0.50. Multiplica por 500 queries diarias y tienes un gasto de $50-$250 por día — $1,500-$7,500 al mes. Y eso es solo un agente. Si tienes un sistema multi-agente donde el supervisor delega a 3 sub-agentes, multiplica por 3-4x.

Conexión con el módulo: En la cápsula 04 configuraste monitoring y métricas de costo. En esta cápsula, pasas de medir costos a controlarlos activamente. Token budgets que abortan requests caras antes de que te arruinen. Rate limiting que protege contra abuso. Model routing que envía tareas baratas a modelos baratos. Caching que elimina llamadas redundantes. Cost awareness no es un nice-to-have — es responsabilidad del AI Engineer.

Anatomía del Costo de un Agente

Dónde se van los tokens

Para controlar costos, primero necesitas entender dónde se gasta cada centavo. Un agente no hace una sola llamada al LLM — hace muchas, y cada una tiene un costo diferente:

Anatomía de costo: Research Agent — una query típica
═══════════════════════════════════════════════════════════════

Fase               Input Tokens  Output Tokens  Modelo      Costo
──────────────────────────────────────────────────────────────────
Planning            2,200          800          gpt-4o      $0.014
Tool: web_search       —            —           Tavily      $0.001
Tool: web_search       —            —           Tavily      $0.001
Synthesis           8,500         1,200         gpt-4o      $0.033
Reflection          4,000          600          gpt-4o      $0.016
Respuesta final     3,800         1,500         gpt-4o      $0.025
Checkpoint write       —            —           PostgreSQL  $0.000
──────────────────────────────────────────────────────────────────
TOTAL              18,500         4,100                     $0.090

Con multi-agente (supervisor + 3 sub-agentes):
──────────────────────────────────────────────────────────────────
Supervisor routing   1,500          300          gpt-4o      $0.007
Sub-agente 1        12,000         2,800         gpt-4o      $0.058
Sub-agente 2         8,000         1,500         gpt-4o      $0.035
Sub-agente 3         6,500         1,200         gpt-4o      $0.029
Merge final          5,000         2,000         gpt-4o      $0.033
──────────────────────────────────────────────────────────────────
TOTAL              33,000         7,800                     $0.162

Los multiplicadores ocultos

Lo que ves en el prompt no es todo lo que pagas. Hay costos ocultos que se acumulan:

Calculando el costo real

from dataclasses import dataclass, field

MODEL_PRICING = {
    "gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000},
    "gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
    "gpt-4.1": {"input": 2.00 / 1_000_000, "output": 8.00 / 1_000_000},
    "gpt-4.1-mini": {"input": 0.40 / 1_000_000, "output": 1.60 / 1_000_000},
    "gpt-4.1-nano": {"input": 0.10 / 1_000_000, "output": 0.40 / 1_000_000},
}


@dataclass
class CostTracker:
    entries: list[dict] = field(default_factory=list)

    def record(self, model: str, input_tokens: int, output_tokens: int,
               component: str = "unknown") -> float:
        pricing = MODEL_PRICING.get(model, MODEL_PRICING["gpt-4o-mini"])
        cost = (input_tokens * pricing["input"]) + (output_tokens * pricing["output"])
        self.entries.append({
            "model": model, "input_tokens": input_tokens,
            "output_tokens": output_tokens, "cost_usd": cost,
            "component": component,
        })
        return cost

    def total_cost(self) -> float:
        return sum(e["cost_usd"] for e in self.entries)

    def cost_by_component(self) -> dict[str, float]:
        breakdown: dict[str, float] = {}
        for e in self.entries:
            breakdown[e["component"]] = breakdown.get(e["component"], 0) + e["cost_usd"]
        return breakdown

Si el 70% del costo está en reflection, sabes dónde optimizar.

Token Budgets por Request

El concepto

Un token budget es un límite duro: "esta request no puede consumir más de X tokens." Si el agente se acerca al límite, aborta la iteración actual y genera una respuesta con lo que tiene. Sin budgets, un agente con un loop de reflexión puede iterar indefinidamente, quemando tokens cada vez.

Implementación con LangGraph

Agrega token_budget y tokens_consumed al state del agente. Antes de cada LLM call, verifica si queda budget. Si no, genera una respuesta parcial con lo que tiene:

from typing import TypedDict
from langchain_core.messages import AIMessage


class AgentState(TypedDict):
    messages: list
    token_budget: int
    tokens_consumed: int
    budget_exceeded: bool


def check_budget(state: AgentState) -> AgentState:
    ratio = state["tokens_consumed"] / state["token_budget"] if state["token_budget"] > 0 else 0
    if ratio >= 0.95:
        state["budget_exceeded"] = True
    return state


def should_continue(state: AgentState) -> str:
    if state.get("budget_exceeded"):
        return "generate_partial_response"
    return "continue_processing"


def generate_partial_response(state: AgentState) -> AgentState:
    warning = (
        f"Budget de tokens alcanzado ({state['tokens_consumed']:,} / "
        f"{state['token_budget']:,}). Respondiendo con información parcial."
    )
    state["messages"].append(AIMessage(content=warning))
    return state

Callback de LangChain para tracking automático

En vez de trackear manualmente, usa un callback handler que intercepta cada llamada al LLM:

from langchain_core.callbacks import BaseCallbackHandler


class BudgetCallbackHandler(BaseCallbackHandler):
    def __init__(self, max_tokens: int = 50_000):
        self.max_tokens = max_tokens
        self.total_consumed = 0
        self.calls: list[dict] = []

    def on_llm_end(self, response, **kwargs):
        usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
        input_tokens = usage.get("prompt_tokens", 0)
        output_tokens = usage.get("completion_tokens", 0)
        total = input_tokens + output_tokens
        self.total_consumed += total
        self.calls.append({
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "total": total,
            "accumulated": self.total_consumed,
        })

        if self.total_consumed > self.max_tokens:
            raise TokenBudgetExceeded(
                f"Budget exceeded: {self.total_consumed:,} / {self.max_tokens:,}"
            )

    @property
    def remaining(self) -> int:
        return max(0, self.max_tokens - self.total_consumed)

    @property
    def utilization(self) -> float:
        return self.total_consumed / self.max_tokens if self.max_tokens > 0 else 0


class TokenBudgetExceeded(Exception):
    pass

Úsalo al invocar el agente:

budget_handler = BudgetCallbackHandler(max_tokens=50_000)

try:
    result = await graph_app.ainvoke(
        {"messages": [HumanMessage(content=query)]},
        config={"callbacks": [budget_handler]},
    )
except TokenBudgetExceeded:
    result = {"messages": [AIMessage(
        content="Se alcanzó el límite de tokens para esta consulta. "
        "La respuesta puede estar incompleta."
    )]}
finally:
    logger.info(f"Request used {budget_handler.total_consumed:,} tokens "
                f"({budget_handler.utilization:.0%} of budget)")

Rate Limiting por Usuario

Por qué necesitas rate limiting

Sin rate limiting, un solo usuario (o un script) puede agotar tu rate limit con OpenAI, dejar a los demás sin servicio, y generar una factura enorme. Rate limiting protege tu presupuesto, la experiencia de los demás, y tu relación con los LLM providers.

Implementación con SlowAPI

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware
from starlette.responses import JSONResponse

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)


@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=429,
        content={
            "error": "rate_limit_exceeded",
            "message": "Demasiadas solicitudes. Intenta de nuevo más tarde.",
            "retry_after_seconds": 60,
        },
        headers={"Retry-After": "60"},
    )


@app.post("/research")
@limiter.limit("10/minute;100/hour;500/day")
async def research(request: Request, query: str):
    result = await graph_app.ainvoke({"messages": [HumanMessage(content=query)]})
    return {"response": result["messages"][-1].content}

Rate limiting por plan de usuario

Un rate limit uniforme no es suficiente. Diferencia por plan:

from enum import Enum


class UserPlan(Enum):
    FREE = "free"
    PRO = "pro"
    ENTERPRISE = "enterprise"


PLAN_LIMITS = {
    UserPlan.FREE: {
        "requests_per_minute": 3, "requests_per_day": 50,
        "max_tokens_per_request": 20_000, "daily_token_budget": 500_000,
    },
    UserPlan.PRO: {
        "requests_per_minute": 15, "requests_per_day": 1_000,
        "max_tokens_per_request": 50_000, "daily_token_budget": 5_000_000,
    },
    UserPlan.ENTERPRISE: {
        "requests_per_minute": 60, "requests_per_day": 10_000,
        "max_tokens_per_request": 100_000, "daily_token_budget": 50_000_000,
    },
}

Token budget por usuario con Redis

Más allá de limitar requests, necesitas limitar tokens consumidos por usuario. Un usuario podría hacer 3 requests "legales" por minuto, pero cada una consumiendo 100,000 tokens:

import redis.asyncio as redis
from datetime import datetime, timezone

redis_client = redis.from_url("redis://localhost:6379")


async def check_user_token_budget(user_id: str, plan: UserPlan) -> dict:
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    key = f"token_budget:{user_id}:{today}"
    daily_limit = PLAN_LIMITS[plan]["daily_token_budget"]

    consumed = int(await redis_client.get(key) or 0)
    remaining = daily_limit - consumed

    return {
        "allowed": remaining > 0,
        "consumed_today": consumed,
        "daily_limit": daily_limit,
        "remaining": max(0, remaining),
    }


async def deduct_tokens(user_id: str, tokens: int):
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    key = f"token_budget:{user_id}:{today}"
    pipe = redis_client.pipeline()
    pipe.incrby(key, tokens)
    pipe.expire(key, 86400 * 2)
    await pipe.execute()

Middleware de cost control

Un middleware que verifica el budget antes de procesar cada request:

from starlette.middleware.base import BaseHTTPMiddleware


class CostControlMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path in ("/health", "/metrics", "/docs"):
            return await call_next(request)

        user_id = request.headers.get("X-User-ID")
        plan = UserPlan(request.headers.get("X-User-Plan", "free"))

        budget_status = await check_user_token_budget(user_id, plan)
        if not budget_status["allowed"]:
            return JSONResponse(status_code=429, content={
                "error": "daily_token_budget_exceeded",
                "consumed": budget_status["consumed_today"],
                "limit": budget_status["daily_limit"],
            })

        request.state.user_plan = plan
        request.state.max_tokens = PLAN_LIMITS[plan]["max_tokens_per_request"]
        return await call_next(request)

Model Routing por Costo

La idea central

No todas las tareas necesitan el modelo más capaz (y más caro). Clasificar una query no requiere GPT-4o — GPT-4o-mini lo hace igual de bien a 1/17 del costo. Planning, clasificación, extracción de parámetros: tareas que modelos baratos manejan perfectamente. Reserva el modelo premium para synthesis y razonamiento complejo.

Model Routing: Router (nano) → clasifica complejidad
├── SIMPLE  → nano  ($0.10/M) — FAQ, saludos, clasificación
├── MEDIUM  → mini  ($0.40/M) — planning, extraction, summaries
└── COMPLEX → full  ($2.50/M) — synthesis, research, razonamiento

Implementación del router

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

router_llm = ChatOpenAI(model="gpt-4.1-nano", temperature=0, max_tokens=20)
cheap_llm = ChatOpenAI(model="gpt-4.1-nano", temperature=0)
mid_llm = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
premium_llm = ChatOpenAI(model="gpt-4.1", temperature=0)

ROUTER_PROMPT = """Classify the task complexity. Respond with exactly one word:
- SIMPLE: greetings, FAQs, yes/no questions, format validation
- MEDIUM: summarization, data extraction, planning, translation
- COMPLEX: multi-step research, analysis, creative writing, reasoning

Task: {query}
Complexity:"""


async def route_by_cost(query: str) -> ChatOpenAI:
    """Selecciona el modelo más barato capaz de manejar la tarea."""
    response = await router_llm.ainvoke([
        HumanMessage(content=ROUTER_PROMPT.format(query=query))
    ])
    complexity = response.content.strip().upper()

    model_map = {
        "SIMPLE": cheap_llm,
        "MEDIUM": mid_llm,
        "COMPLEX": premium_llm,
    }
    return model_map.get(complexity, mid_llm)

Routing dentro de un agente LangGraph

El routing no solo aplica entre requests — aplica dentro de un mismo request. Los diferentes nodos del agente tienen diferentes requisitos de inteligencia:

NODE_MODEL_MAP = {
    "classify_query": "gpt-4.1-nano",
    "plan_research": "gpt-4.1-mini",
    "execute_search": "gpt-4.1-mini",
    "synthesize": "gpt-4.1",
    "reflect": "gpt-4.1-mini",
    "generate_response": "gpt-4.1",
}


def get_llm_for_node(node_name: str) -> ChatOpenAI:
    model = NODE_MODEL_MAP.get(node_name, "gpt-4.1-mini")
    return ChatOpenAI(model=model, temperature=0)

Impacto real del model routing

El ahorro no es marginal — es transformador:

Model routing reduce el costo a la mitad sin degradar la calidad de las respuestas finales. Las tareas simples se resuelven igual con un modelo barato — solo synthesis y razonamiento necesitan el premium.

Fallback por costo

Si el modelo premium no está disponible (rate limit, timeout), haces fallback descendente — del modelo asignado al siguiente más barato:

async def invoke_with_cost_fallback(query: str, node_name: str) -> str:
    models = ["gpt-4.1", "gpt-4.1-mini", "gpt-4.1-nano"]
    primary = NODE_MODEL_MAP.get(node_name, "gpt-4.1-mini")
    start_idx = models.index(primary) if primary in models else 0

    for model_name in models[start_idx:]:
        try:
            llm = ChatOpenAI(model=model_name, temperature=0, request_timeout=30)
            return (await llm.ainvoke([HumanMessage(content=query)])).content
        except Exception:
            continue
    raise RuntimeError("All models failed")

Cost Monitoring

Dashboard de costos en tiempo real

En la cápsula 04 implementaste métricas con Prometheus. Ahora extiéndelas para visibilidad completa de costos. Tu dashboard debe mostrar: costo de hoy vs. límite diario, costo mensual vs. budget, desglose por componente (planning, synthesis, reflection, tools), desglose por modelo, costo promedio por query (p50, p95, p99), y top usuarios por gasto.

Tracking con Prometheus y Redis

from prometheus_client import Counter, Gauge, Histogram
from datetime import datetime, timezone

cost_total = Counter("agent_cost_usd_total", "Total cost in USD", ["model", "component"])
cost_per_query = Histogram(
    "agent_cost_per_query_usd", "Cost per query",
    buckets=[0.01, 0.02, 0.05, 0.10, 0.20, 0.50, 1.00, 2.00],
)
daily_cost_gauge = Gauge("agent_daily_cost_usd", "Today's total cost")
monthly_budget_usage = Gauge("agent_monthly_budget_pct", "Monthly budget utilization %")

MONTHLY_BUDGET = 750.00


async def record_cost(
    model: str, input_tokens: int, output_tokens: int,
    component: str, user_id: str,
) -> float:
    pricing = MODEL_PRICING.get(model, MODEL_PRICING["gpt-4o-mini"])
    cost = (input_tokens * pricing["input"]) + (output_tokens * pricing["output"])
    cost_total.labels(model=model, component=component).inc(cost)

    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    month = datetime.now(timezone.utc).strftime("%Y-%m")
    pipe = redis_client.pipeline()
    pipe.incrbyfloat(f"cost:daily:{today}", cost)
    pipe.expire(f"cost:daily:{today}", 86400 * 7)
    pipe.incrbyfloat(f"cost:monthly:{month}", cost)
    pipe.expire(f"cost:monthly:{month}", 86400 * 45)
    pipe.incrbyfloat(f"cost:user:{user_id}:{today}", cost)
    pipe.expire(f"cost:user:{user_id}:{today}", 86400 * 7)
    await pipe.execute()

    daily = float(await redis_client.get(f"cost:daily:{today}") or 0)
    monthly = float(await redis_client.get(f"cost:monthly:{month}") or 0)
    daily_cost_gauge.set(daily)
    monthly_budget_usage.set((monthly / MONTHLY_BUDGET) * 100)
    return cost

Alertas de costo

Extiende el AlertManager de la cápsula 04 con reglas de costo:

async def check_cost_alerts(query_cost: float):
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    month = datetime.now(timezone.utc).strftime("%Y-%m")
    daily = float(await redis_client.get(f"cost:daily:{today}") or 0)
    monthly = float(await redis_client.get(f"cost:monthly:{month}") or 0)
    budget_pct = (monthly / MONTHLY_BUDGET) * 100

    if query_cost > 0.50:
        logger.warning(f"COST ALERT: Single query cost ${query_cost:.2f}")
    if daily > 100.0:
        logger.critical(f"COST ALERT: Daily cost ${daily:.2f} exceeds $100")
    elif daily > 50.0:
        logger.warning(f"COST ALERT: Daily cost ${daily:.2f} exceeds $50")
    if budget_pct > 90:
        logger.critical(f"COST ALERT: Monthly budget at {budget_pct:.0f}%")
    elif budget_pct > 75:
        logger.warning(f"COST ALERT: Monthly budget at {budget_pct:.0f}%")

Endpoint de visibilidad

@app.get("/costs")
async def get_costs():
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    month = datetime.now(timezone.utc).strftime("%Y-%m")
    daily = float(await redis_client.get(f"cost:daily:{today}") or 0)
    monthly = float(await redis_client.get(f"cost:monthly:{month}") or 0)

    return {
        "daily_cost_usd": round(daily, 2),
        "monthly_cost_usd": round(monthly, 2),
        "monthly_budget": MONTHLY_BUDGET,
        "budget_utilization_pct": round((monthly / MONTHLY_BUDGET) * 100, 1),
    }

Optimización de Costos

Prompt compression

Los prompts largos son caros. Cada token de input se paga en CADA llamada al LLM. Si tu system prompt tiene 3,000 tokens y haces 5 LLM calls por request, pagas 15,000 tokens solo en system prompt.

Comprime sin perder intención. Un system prompt de 120 tokens que dice lo mismo que uno de 35 tokens ahorra 71%. En 500 queries/día × 5 calls × 85 tokens ahorrados = 212,500 tokens/día (~$0.53/día con GPT-4o). Parece poco individualmente, pero es optimización gratuita — la calidad no cambia.

Caching agresivo

Cada cache hit es una LLM call que NO pagas. Cachea tool results con TTL apropiado (como viste en la cápsula 03), y para respuestas LLM exactas, usa un hash del prompt como key:

import hashlib

async def cached_llm_call(prompt: str, model: str = "gpt-4.1-mini", ttl: int = 300) -> str | None:
    cache_key = f"llm:{model}:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"
    cached = await redis_client.get(cache_key)
    return cached.decode() if cached else None

async def cache_llm_response(prompt: str, response: str, model: str = "gpt-4.1-mini", ttl: int = 300):
    cache_key = f"llm:{model}:{hashlib.sha256(prompt.encode()).hexdigest()[:16]}"
    await redis_client.setex(cache_key, ttl, response)

Reducir iteraciones del agente

Un agente ReAct puede iterar indefinidamente. Limita las iteraciones con contadores en el state: máximo 5 tool calls, máximo 2 rondas de reflection, máximo 10 LLM calls totales. Cuando se alcanza un límite, salta directamente a synthesis.

Optimizar el contexto window

Conversaciones largas acumulan historial. Usa trim_messages de LangChain para mantener solo los mensajes recientes dentro de un budget de tokens. Un window de 4,000 tokens con strategy="last" descarta los turnos más antiguos pero siempre preserva el system prompt.

Resumen de técnicas de optimización

Conexión con Proyecto

El Research Agent necesita cost control completo para ser production-ready:

1. Token budget por request — Máximo 50,000 tokens por research query. Si se excede, el agente genera una respuesta parcial con lo que tiene. El BudgetCallbackHandler intercepta cada LLM call y aborta si se pasa.

2. Rate limiting por plan — Free: 3 req/min, 50 req/día, 500K tokens/día. Pro: 15 req/min, 1000 req/día, 5M tokens/día. Implementado con SlowAPI + Redis para token budgets.

3. Model routing — El router (gpt-4.1-nano) clasifica cada query. Planning usa gpt-4.1-mini. Solo synthesis y respuesta final usan gpt-4.1. Resultado: costo promedio por query baja de $0.09 a $0.035.

4. Cost tracking en Redis — Cada LLM call registra modelo, tokens, componente, y usuario. Redis acumula costos diarios y mensuales. Endpoint /costs expone el estado en tiempo real.

5. Alertas de costo — Query individual > $0.50 → warning. Costo diario > $50 → warning. Costo diario > $100 → critical. Budget mensual > 75% → warning. Budget mensual > 90% → critical con notificación a Slack.

6. Caching — Tool results en Redis con TTL 10 min. Prompt compression. Window de 20 mensajes.

Resultado: un agente que cuesta ~$525/mes en vez de ~$1,350/mes, con la misma calidad, y que nunca se sale de presupuesto.

Troubleshooting

Problema 1: "El token budget se agota antes de generar respuesta"

Causa probable: El budget está demasiado bajo, o el system prompt y tool descriptions consumen demasiado. Si tu system prompt usa 2,000 tokens y tienes 10 tools (200 tokens c/u), ya consumiste 4,000 tokens antes de la primera LLM call real.

Solución: Audita cuánto consume cada componente fijo. Reduce el system prompt. Carga solo los tools relevantes (no los 10 siempre). Implementa budgets dinámicos basados en la complejidad clasificada por el router.

Problema 2: "El rate limiter bloquea usuarios legítimos"

Causa probable: Estás usando get_remote_address (IP) como key. En redes corporativas, todos los usuarios comparten la misma IP pública.

Solución: Usa el user_id autenticado como key: Limiter(key_func=lambda r: r.headers.get("X-User-ID", get_remote_address(r))). Nunca uses IP como único identificador en entornos con NAT/VPN.

Problema 3: "El model router elige el modelo incorrecto"

Causa probable: El prompt del router no es suficientemente claro, o el modelo nano no tiene la capacidad de clasificar bien.

Solución: Evalúa el router con 50-100 queries clasificadas manualmente. Si la accuracy es < 85%, sube a mini. Agrega few-shot examples. Considera reglas heurísticas como fallback (queries < 20 tokens → SIMPLE, queries con "analyze", "research" → COMPLEX).

Problema 4: "Los costos reportados no coinciden con la factura de OpenAI"

Causa probable: No estás contando todos los tokens. Los retries generan tokens que se cobran pero no siempre se registran en tu callback. Los embeddings para semantic cache y las llamadas de health check también suman.

Solución: Compara tus logs con el Usage dashboard de OpenAI. Registra tokens en retries. Incluye el costo de embeddings. Excluye health checks de las métricas de usuario pero inclúyelos en el costo total.

Problema 5: "El caching reduce calidad en queries time-sensitive"

Causa probable: El TTL del cache es demasiado largo para queries que requieren información en tiempo real.

Solución: Clasifica queries en "temporal" vs "atemporal" antes de consultar el cache. Queries con palabras como "hoy", "ahora", "último", "reciente" deben bypasear el cache o usar TTL muy corto (1-2 min). Agrega un header Cache-Control: no-cache para que el usuario fuerce una consulta fresca.

Ejercicios

Ejercicio 1: Calcular el costo real de un agente multi-agente

Tu sistema tiene un supervisor que delega a 3 sub-agentes. Cada sub-agente hace 2 LLM calls (planning + response) y 2 tool calls. El supervisor hace 2 LLM calls propias (routing + merge). Usa GPT-4o para todo. Calcula el costo de una query si cada LLM call consume en promedio 3,000 input tokens y 800 output tokens.

Ejercicio 2: Diseñar token budgets por tipo de query

Tu agente maneja 3 tipos de queries: FAQ (respuesta directa), Research (búsqueda + análisis), y Deep Analysis (multi-fuente + reflexión). Define token budgets apropiados y qué hacer cuando se excede cada uno.

QUERY_BUDGETS = {
    "faq": {"max_tokens": 10_000, "max_llm_calls": 2, "on_exceed": "answer_from_knowledge"},
    "research": {"max_tokens": 40_000, "max_llm_calls": 6, "on_exceed": "synthesize_partial"},
    "deep_analysis": {"max_tokens": 80_000, "max_llm_calls": 12, "on_exceed": "skip_reflection"},
}

Ejercicio 3: Implementar un cost dashboard endpoint

Implementa un endpoint /costs/dashboard que retorne: costo de hoy, costo de esta semana, costo de este mes, top 3 usuarios por costo hoy, y costo promedio por query.

from datetime import datetime, timezone, timedelta

@app.get("/costs/dashboard")
async def cost_dashboard():
    now = datetime.now(timezone.utc)
    today, month = now.strftime("%Y-%m-%d"), now.strftime("%Y-%m")

    daily = float(await redis_client.get(f"cost:daily:{today}") or 0)
    week_cost = sum(
        float(await redis_client.get(f"cost:daily:{(now - timedelta(days=i)).strftime('%Y-%m-%d')}") or 0)
        for i in range(7)
    )
    monthly = float(await redis_client.get(f"cost:monthly:{month}") or 0)

    user_keys = await redis_client.keys(f"cost:user:*:{today}")
    user_costs = [
        {"user_id": k.decode().split(":")[2], "cost_usd": round(float(await redis_client.get(k) or 0), 4)}
        for k in user_keys
    ]
    top_users = sorted(user_costs, key=lambda x: x["cost_usd"], reverse=True)[:3]

    query_count = max(int(await redis_client.get(f"query_count:{today}") or 1), 1)
    return {
        "today_usd": round(daily, 2), "this_week_usd": round(week_cost, 2),
        "this_month_usd": round(monthly, 2), "budget_remaining_usd": round(MONTHLY_BUDGET - monthly, 2),
        "avg_cost_per_query_usd": round(daily / query_count, 4), "top_users_today": top_users,
    }

Ejercicio 4: Rate limiting con token awareness

El rate limiting estándar cuenta requests, pero no todos los requests cuestan igual. Diseña un rate limiter que considere el costo acumulado del usuario, no solo el número de requests.

async def check_cost_rate_limit(user_id: str, plan: UserPlan) -> dict:
    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    user_cost = float(await redis_client.get(f"cost:user:{user_id}:{today}") or 0)
    max_cost = {UserPlan.FREE: 1.00, UserPlan.PRO: 10.00, UserPlan.ENTERPRISE: 100.00}[plan]

    user_tokens = int(await redis_client.get(f"tokens:user:{user_id}:{today}") or 0)
    max_tokens = PLAN_LIMITS[plan]["daily_token_budget"]

    allowed = user_cost < max_cost and user_tokens < max_tokens
    return {
        "allowed": allowed,
        "reason": None if allowed else ("daily_cost_exceeded" if user_cost >= max_cost else "daily_tokens_exceeded"),
        "cost_today": round(user_cost, 4), "cost_limit": max_cost,
    }

Ejercicio 5: Comparar estrategias de optimización

Tienes un agente que procesa 500 queries/día con un costo promedio de $0.12/query ($60/día). Tu objetivo es bajar a $30/día sin degradar calidad perceptiblemente. Propón 3 estrategias, estima el ahorro de cada una, y recomienda el orden de implementación.

Resumen

• Los agentes son caros por naturaleza. Múltiples LLM calls, tool calls, reflection loops, y multi-agente multiplican tokens. Una query puede costar $0.10-$0.50. Sin control, 500 queries/día se convierten en $1,500/mes.
• Token budgets son tu primera línea de defensa. Un límite duro por request previene queries desbocadas. El BudgetCallbackHandler intercepta cada LLM call y aborta si se excede.
• Rate limiting protege tu presupuesto y tus usuarios. Limita por request rate (SlowAPI) Y por token budget diario (Redis). Diferencia por plan: free, pro, enterprise.
• Model routing reduce costos 40-60% sin degradar calidad. Nano clasifica, mini planifica, full sintetiza. El ahorro es transformador.
• Cost monitoring con alertas previene sorpresas. Trackea por query, usuario, componente, modelo, día, mes. Alerta en umbrales. Endpoint /costs para visibilidad en tiempo real.
• Optimización de costos es un esfuerzo compuesto. Prompt compression + caching + limitar iteraciones + window de mensajes = 60-75% de reducción combinada.
• Cost awareness es responsabilidad del AI Engineer. Tú diseñas los prompts, configuras los modelos, decides las iteraciones. El costo es consecuencia directa de tus decisiones de diseño.

Próxima cápsula: Pydantic AI — Comparación — implementarás el mismo agente en Pydantic AI y compararás filosofías, trade-offs, y experiencia de desarrollo con el ecosistema LangChain/LangGraph que has usado durante toda la guía.

Recursos Adicionales

1. OpenAI Pricing — Precios actualizados por modelo, input/output tokens
2. Anthropic Pricing — Precios de Claude, comparación con OpenAI
3. SlowAPI Documentation — Rate limiting para FastAPI basado en limits
4. LangChain Callbacks — Callbacks para tracking de tokens y costos
5. Redis Rate Limiting Patterns — Token buckets y sliding window con Redis
6. Prompt Engineering for Cost Optimization — Técnicas de OpenAI para prompts eficientes