Módulo 12: LangSmith y Producción

Rate Limiting y Cost Control

Descripción de la cápsula

En la cápsula anterior aprendiste a trackear cuánto cuestan tus llamadas al modelo — tokens, dólares, desglose por operación. Pero trackear sin actuar es como tener un velocímetro sin frenos. Un solo usuario ejecutando 100 investigaciones en una hora puede consumir tu presupuesto diario completo. Sin rate limiting, tu sistema de AI es un grifo abierto de dinero.

Rate limiting en sistemas de AI no es lo mismo que rate limiting en APIs tradicionales. En una API REST, limitas requests por segundo para proteger el servidor. En un agente de AI, limitas requests para proteger tu presupuesto. Una sola ejecución del Research Assistant puede hacer 5-10 llamadas al modelo — si un usuario dispara 50 ejecuciones, son 250-500 llamadas al modelo en minutos. A $0.035 por ejecución con GPT-4.1, eso son $1.75 en unos minutos. Multiplica por 100 usuarios haciendo lo mismo y tienes una factura de $175 en una hora.

LangChain incluye InMemoryRateLimiter para controlar el ritmo de llamadas al modelo. Combinado con el cost tracking de la cápsula anterior y el model routing del Módulo 4, puedes construir un sistema de control de costos completo: limitar la velocidad de requests, establecer presupuestos por usuario y por proyecto, y degradar automáticamente a modelos más baratos cuando el presupuesto se agota.

InMemoryRateLimiter: control de velocidad

InMemoryRateLimiter es la implementación built-in de LangChain para limitar la tasa de requests al modelo. Funciona con un token bucket algorithm: tienes un "cubo" de tokens disponibles, cada request consume uno, y los tokens se reponen a una tasa fija.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter
import time

rate_limiter = InMemoryRateLimiter(
    requests_per_second=1,
    check_every_n_seconds=0.1,
    max_bucket_size=2,
)

model = init_chat_model("openai:gpt-4.1-mini")
limited_model = model.with_rate_limiter(rate_limiter)

prompts = [
    "Di 'uno'.",
    "Di 'dos'.",
    "Di 'tres'.",
    "Di 'cuatro'.",
    "Di 'cinco'.",
]

start = time.time()
for i, prompt in enumerate(prompts):
    t = time.time() - start
    response = limited_model.invoke(prompt)
    elapsed = time.time() - start
    print(f"  [{elapsed:.1f}s] Prompt {i+1}: {response.content.strip()}")

total = time.time() - start
print(f"\nTiempo total: {total:.1f}s (sin rate limit sería ~2-3s)")
# Output esperado:
#   [0.5s] Prompt 1: uno
#   [1.0s] Prompt 2: dos
#   [2.0s] Prompt 3: tres
#   [3.0s] Prompt 4: cuatro
#   [4.0s] Prompt 5: cinco
#
# Tiempo total: 4.5s (sin rate limit sería ~2-3s)

Los parámetros de InMemoryRateLimiter:

Parámetro	Significado	Ejemplo
`requests_per_second`	Tasa de reposición de tokens	1 = un request por segundo
`check_every_n_seconds`	Frecuencia de verificación	0.1 = verifica cada 100ms
`max_bucket_size`	Capacidad máxima del bucket	2 = permite burst de 2 requests

max_bucket_size: permitir bursts controlados

El max_bucket_size permite acumular tokens cuando no hay actividad. Si tu tasa es 1 request/segundo y no hay requests durante 5 segundos, el bucket acumula hasta max_bucket_size tokens, permitiendo un burst cuando vuelve la actividad.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter
import time

rate_limiter = InMemoryRateLimiter(
    requests_per_second=2,
    check_every_n_seconds=0.1,
    max_bucket_size=5,
)

model = init_chat_model("openai:gpt-4.1-mini")
limited_model = model.with_rate_limiter(rate_limiter)

print("Esperando 3 segundos para acumular tokens en el bucket...")
time.sleep(3)

start = time.time()
for i in range(8):
    response = limited_model.invoke(f"Di '{i+1}'.")
    elapsed = time.time() - start
    print(f"  [{elapsed:.1f}s] Request {i+1}: {response.content.strip()}")

total = time.time() - start
print(f"\nTiempo total: {total:.1f}s")
print("Los primeros 5 fueron rápidos (bucket lleno), luego 2/segundo")
# Output esperado:
# Esperando 3 segundos para acumular tokens en el bucket...
#   [0.3s] Request 1: 1
#   [0.6s] Request 2: 2
#   [0.9s] Request 3: 3
#   [1.2s] Request 4: 4
#   [1.5s] Request 5: 5
#   [2.0s] Request 6: 6
#   [2.5s] Request 7: 7
#   [3.0s] Request 8: 8
#
# Tiempo total: 3.0s
# Los primeros 5 fueron rápidos (bucket lleno), luego 2/segundo

Rate limiting en agentes

Cuando aplicas rate limiting a un agente, cada llamada al modelo (no cada invocación del agente) es limitada. Un agente que hace 3 llamadas al modelo por ejecución se ve afectado 3 veces por el rate limiter.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_core.rate_limiters import InMemoryRateLimiter
import time

rate_limiter = InMemoryRateLimiter(
    requests_per_second=2,
    check_every_n_seconds=0.1,
    max_bucket_size=3,
)

model = init_chat_model("openai:gpt-4.1-mini")
limited_model = model.with_rate_limiter(rate_limiter)

@tool
def search(query: str) -> str:
    """Busca información sobre un tema."""
    return f"Resultado: {query} es un concepto importante en AI engineering."

agent = create_agent(limited_model, [search])

start = time.time()
result = agent.invoke({"messages": [("user", "Busca qué es LangGraph y resúmelo.")]})
elapsed = time.time() - start

print(f"Respuesta: {result['messages'][-1].content[:100]}...")
print(f"Tiempo de ejecución: {elapsed:.1f}s")
print("(Incluye rate limiting entre cada llamada interna al modelo)")
# Output esperado:
# Respuesta: LangGraph es un framework de código abierto para construir aplicaciones con agentes de AI...
# Tiempo de ejecución: 2.5s
# (Incluye rate limiting entre cada llamada interna al modelo)

Per-user rate limiting: tiers de servicio

En producción, diferentes usuarios tienen diferentes límites. Un usuario free no debería consumir los mismos recursos que un usuario enterprise.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter
import time

TIER_LIMITS = {
    "free": {"requests_per_second": 0.5, "max_bucket_size": 2},
    "pro": {"requests_per_second": 2, "max_bucket_size": 5},
    "enterprise": {"requests_per_second": 10, "max_bucket_size": 20},
}

class UserRateLimitManager:
    def __init__(self):
        self.limiters: dict[str, InMemoryRateLimiter] = {}
        self.models: dict[str, object] = {}
        self.base_model = init_chat_model("openai:gpt-4.1-mini")

    def get_limited_model(self, user_id: str, tier: str):
        """Retorna un modelo con rate limiting apropiado para el tier del usuario."""
        key = f"{user_id}:{tier}"
        if key not in self.limiters:
            limits = TIER_LIMITS[tier]
            self.limiters[key] = InMemoryRateLimiter(
                requests_per_second=limits["requests_per_second"],
                check_every_n_seconds=0.1,
                max_bucket_size=limits["max_bucket_size"],
            )
            self.models[key] = self.base_model.with_rate_limiter(self.limiters[key])
        return self.models[key]


manager = UserRateLimitManager()

users = [
    ("alice", "free"),
    ("bob", "pro"),
    ("corp-acme", "enterprise"),
]

for user_id, tier in users:
    model = manager.get_limited_model(user_id, tier)
    limits = TIER_LIMITS[tier]

    start = time.time()
    for i in range(3):
        response = model.invoke(f"Di '{i+1}'.")
    elapsed = time.time() - start

    print(f"[{tier:>10}] {user_id}: 3 requests en {elapsed:.1f}s "
          f"(límite: {limits['requests_per_second']} req/s)")
# Output esperado:
# [      free] alice: 3 requests en 5.2s (límite: 0.5 req/s)
# [       pro] bob: 3 requests en 1.8s (límite: 2 req/s)
# [enterprise] corp-acme: 3 requests en 0.9s (límite: 10 req/s)

Cost budgets: límites de gasto

Rate limiting controla la velocidad. Cost budgets controlan el gasto total. Necesitas ambos: un usuario puede respetar el rate limit pero ejecutar miles de requests en un día.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from dataclasses import dataclass, field
from datetime import datetime

PRICING = {
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1": {"input": 2.00, "output": 8.00},
}


@dataclass
class CostBudget:
    user_id: str
    daily_limit: float
    monthly_limit: float
    spent_today: float = 0.0
    spent_month: float = 0.0
    requests_today: int = 0

    def check_budget(self, estimated_cost: float) -> tuple[bool, str]:
        """Verifica si hay presupuesto disponible."""
        if self.spent_today + estimated_cost > self.daily_limit:
            return False, f"daily_limit_exceeded (${self.spent_today:.4f}/${self.daily_limit:.2f})"
        if self.spent_month + estimated_cost > self.monthly_limit:
            return False, f"monthly_limit_exceeded (${self.spent_month:.2f}/${self.monthly_limit:.2f})"
        return True, "ok"

    def record(self, cost: float):
        """Registra un gasto."""
        self.spent_today += cost
        self.spent_month += cost
        self.requests_today += 1

    def alert_status(self) -> str:
        """Retorna el estado de alerta."""
        pct = (self.spent_today / self.daily_limit * 100) if self.daily_limit > 0 else 0
        if pct >= 100:
            return "BLOCKED"
        if pct >= 80:
            return "WARNING"
        if pct >= 50:
            return "NOTICE"
        return "OK"


class CostController:
    def __init__(self, model_name: str = "gpt-4.1-mini"):
        self.model_name = model_name
        self.budgets: dict[str, CostBudget] = {}

    def register_user(self, user_id: str, daily: float, monthly: float):
        self.budgets[user_id] = CostBudget(
            user_id=user_id,
            daily_limit=daily,
            monthly_limit=monthly,
        )

    def estimate_cost(self, input_tokens: int = 200, output_tokens: int = 300) -> float:
        """Estima el costo de una llamada típica."""
        pricing = PRICING[self.model_name]
        return (input_tokens / 1_000_000) * pricing["input"] + \
               (output_tokens / 1_000_000) * pricing["output"]

    def can_proceed(self, user_id: str) -> tuple[bool, str]:
        """Verifica si el usuario puede hacer una llamada."""
        if user_id not in self.budgets:
            return False, "user_not_registered"
        estimated = self.estimate_cost()
        return self.budgets[user_id].check_budget(estimated)

    def record_usage(self, user_id: str, usage_metadata: dict):
        """Registra el uso real de una llamada."""
        pricing = PRICING[self.model_name]
        cost = (usage_metadata["input_tokens"] / 1_000_000) * pricing["input"] + \
               (usage_metadata["output_tokens"] / 1_000_000) * pricing["output"]
        self.budgets[user_id].record(cost)
        return cost


controller = CostController("gpt-4.1-mini")
controller.register_user("free-user", daily=0.001, monthly=0.02)
controller.register_user("pro-user", daily=0.01, monthly=0.20)

model = init_chat_model("openai:gpt-4.1-mini")

for i in range(10):
    for user_id in ["free-user", "pro-user"]:
        can, reason = controller.can_proceed(user_id)

        if not can:
            budget = controller.budgets[user_id]
            print(f"  [{user_id:>10}] Request {i+1}: BLOCKED — {reason}")
            continue

        response = model.invoke(f"Concepto {i+1} de AI engineering en 1 oración.")
        cost = controller.record_usage(user_id, response.usage_metadata)
        budget = controller.budgets[user_id]
        alert = budget.alert_status()
        alert_str = f" [{alert}]" if alert != "OK" else ""
        print(f"  [{user_id:>10}] Request {i+1}: ${cost:.6f} | "
              f"Daily: ${budget.spent_today:.6f}/{budget.daily_limit:.4f}{alert_str}")

print(f"\nResumen final:")
for uid, budget in controller.budgets.items():
    pct = (budget.spent_today / budget.daily_limit * 100) if budget.daily_limit > 0 else 0
    print(f"  {uid}: {budget.requests_today} requests, "
          f"${budget.spent_today:.6f} / ${budget.daily_limit:.4f} ({pct:.0f}%)")
# Output esperado:
#   [ free-user] Request 1: $0.000120 | Daily: $0.000120/0.0010
#   [  pro-user] Request 1: $0.000118 | Daily: $0.000118/0.0100
#   [ free-user] Request 2: $0.000115 | Daily: $0.000235/0.0010
#   [  pro-user] Request 2: $0.000122 | Daily: $0.000240/0.0100
#   ...
#   [ free-user] Request 8: $0.000119 | Daily: $0.000930/0.0010 [WARNING]
#   [  pro-user] Request 8: $0.000118 | Daily: $0.000960/0.0100
#   [ free-user] Request 9: BLOCKED — daily_limit_exceeded ($0.000930/0.00)
#   [  pro-user] Request 9: $0.000115 | Daily: $0.001075/0.0100
#   [ free-user] Request 10: BLOCKED — daily_limit_exceeded ($0.000930/0.00)
#   [  pro-user] Request 10: $0.000120 | Daily: $0.001195/0.0100
#
# Resumen final:
#   free-user: 8 requests, $0.000930 / $0.0010 (93%)
#   pro-user: 10 requests, $0.001195 / $0.0100 (12%)

Circuit breaker: pausa automática ante costos anómalos

Un circuit breaker detecta cuando la tasa de gasto excede lo normal y pausa el sistema antes de que el daño sea mayor. Si tu gasto promedio es $0.05/hora y de repente sube a $0.50/hora, algo anda mal — un loop infinito, un prompt inyectado que causa respuestas enormes, o un ataque de un usuario malicioso.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from dataclasses import dataclass, field
from datetime import datetime
import time

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}


@dataclass
class CostCircuitBreaker:
    """Circuit breaker que se activa si el gasto por minuto excede el threshold."""
    cost_per_minute_threshold: float
    window_seconds: int = 60
    state: str = "closed"
    cost_history: list = field(default_factory=list)
    tripped_at: str = ""

    def record_cost(self, cost: float) -> str:
        """Registra un costo y verifica si el circuit breaker debe activarse."""
        now = time.time()
        self.cost_history.append((now, cost))

        cutoff = now - self.window_seconds
        self.cost_history = [(t, c) for t, c in self.cost_history if t > cutoff]

        window_cost = sum(c for _, c in self.cost_history)

        if window_cost > self.cost_per_minute_threshold:
            self.state = "open"
            self.tripped_at = datetime.now().isoformat()
            return "TRIPPED"

        self.state = "closed"
        return "OK"

    def can_proceed(self) -> tuple[bool, str]:
        if self.state == "open":
            return False, f"Circuit breaker OPEN (tripped at {self.tripped_at})"
        return True, "OK"

    def reset(self):
        self.state = "closed"
        self.cost_history = []
        self.tripped_at = ""


breaker = CostCircuitBreaker(cost_per_minute_threshold=0.001)
model = init_chat_model("openai:gpt-4.1-mini")

for i in range(12):
    can, reason = breaker.can_proceed()
    if not can:
        print(f"  Request {i+1}: BLOCKED — {reason}")
        continue

    response = model.invoke(f"Explica concepto #{i+1} de producción de AI en 3 oraciones detalladas.")
    usage = response.usage_metadata
    cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
           (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]

    status = breaker.record_cost(cost)
    window_cost = sum(c for _, c in breaker.cost_history)
    print(f"  Request {i+1}: ${cost:.6f} | Window: ${window_cost:.6f} | Status: {status}")
# Output esperado:
#   Request 1: $0.000180 | Window: $0.000180 | Status: OK
#   Request 2: $0.000175 | Window: $0.000355 | Status: OK
#   Request 3: $0.000190 | Window: $0.000545 | Status: OK
#   Request 4: $0.000185 | Window: $0.000730 | Status: OK
#   Request 5: $0.000178 | Window: $0.000908 | Status: OK
#   Request 6: $0.000182 | Window: $0.001090 | Status: TRIPPED
#   Request 7: BLOCKED — Circuit breaker OPEN (tripped at 2026-03-08T...)
#   Request 8: BLOCKED — Circuit breaker OPEN (tripped at 2026-03-08T...)
#   ...

Auto-degradación: modelo más barato cuando el presupuesto aprieta

En vez de bloquear al usuario cuando se acerca al límite, puedes degradar automáticamente a un modelo más barato. El usuario sigue recibiendo respuestas, pero con un modelo más económico (y potencialmente menos capaz).

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1": {"input": 2.00, "output": 8.00},
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

MODEL_TIERS = [
    ("gpt-4.1", "openai:gpt-4.1"),
    ("gpt-4.1-mini", "openai:gpt-4.1-mini"),
    ("gpt-4.1-nano", "openai:gpt-4.1-nano"),
]


class AutoDegradeController:
    def __init__(self, daily_budget: float):
        self.daily_budget = daily_budget
        self.spent = 0.0
        self.models = {name: init_chat_model(model_id) for name, model_id in MODEL_TIERS}
        self.degradation_log = []

    def select_model(self) -> tuple[str, object]:
        """Selecciona el modelo basado en el presupuesto restante."""
        remaining_pct = 1.0 - (self.spent / self.daily_budget) if self.daily_budget > 0 else 0

        if remaining_pct > 0.5:
            name = "gpt-4.1"
        elif remaining_pct > 0.2:
            name = "gpt-4.1-mini"
        else:
            name = "gpt-4.1-nano"

        return name, self.models[name]

    def invoke(self, prompt: str) -> tuple[str, str]:
        """Invoca el modelo apropiado y registra el costo."""
        model_name, model = self.select_model()
        response = model.invoke(prompt)
        usage = response.usage_metadata
        pricing = PRICING[model_name]
        cost = (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
               (usage["output_tokens"] / 1_000_000) * pricing["output"]

        self.spent += cost
        remaining_pct = (1 - self.spent / self.daily_budget) * 100
        self.degradation_log.append({
            "model": model_name, "cost": cost, "remaining_pct": remaining_pct,
        })
        return model_name, response.content


controller = AutoDegradeController(daily_budget=0.002)

prompts = [
    "Explica qué es observabilidad en AI en 3 oraciones.",
    "¿Qué es tracing en LangSmith? 2 oraciones.",
    "¿Qué es evaluation en AI? 2 oraciones.",
    "¿Qué es rate limiting? 1 oración.",
    "¿Qué es un circuit breaker? 1 oración.",
    "Define monitoring. 1 oración.",
    "Define deployment. 1 oración.",
    "Define scaling. 1 oración.",
    "Define resilience. 1 oración.",
    "Define caching. 1 oración.",
]

for i, prompt in enumerate(prompts):
    model_name, content = controller.invoke(prompt)
    remaining = (1 - controller.spent / controller.daily_budget) * 100
    print(f"  [{i+1:>2}] {model_name:<14} | Budget: {remaining:>5.1f}% | {content[:60]}...")

print(f"\nGasto total: ${controller.spent:.6f} / ${controller.daily_budget:.4f}")
print(f"Degradaciones: {len(set(d['model'] for d in controller.degradation_log))} modelos usados")
# Output esperado:
#   [ 1] gpt-4.1        | Budget:  75.0% | La observabilidad en AI se refiere a la capacidad de entende...
#   [ 2] gpt-4.1        | Budget:  55.0% | LangSmith es una plataforma de observabilidad que permite tr...
#   [ 3] gpt-4.1        | Budget:  40.0% | La evaluación en AI es el proceso sistemático de medir la ca...
#   [ 4] gpt-4.1-mini   | Budget:  36.0% | Rate limiting es una técnica que controla la cantidad de sol...
#   [ 5] gpt-4.1-mini   | Budget:  32.0% | Un circuit breaker es un patrón de diseño que detecta fallo...
#   [ 6] gpt-4.1-mini   | Budget:  28.0% | Monitoring es la práctica de observar continuamente el rendi...
#   [ 7] gpt-4.1-mini   | Budget:  24.0% | Deployment es el proceso de poner una aplicación en producc...
#   [ 8] gpt-4.1-nano   | Budget:  22.0% | Scaling es ajustar recursos según demanda....
#   [ 9] gpt-4.1-nano   | Budget:  20.0% | Resilience es la capacidad de recuperarse de fallos....
#   [10] gpt-4.1-nano   | Budget:  18.0% | Caching es almacenar datos para acceso rápido....
#
# Gasto total: $0.001640 / $0.0020
# Degradaciones: 3 modelos usados

El sistema empieza con GPT-4.1 (mejor calidad) y degrada a mini y luego a nano conforme el presupuesto se agota. El usuario no se bloquea — la calidad baja gradualmente.

Combinando rate limiting + cost control + model routing

En producción necesitas las tres capas juntas. Rate limiting controla la velocidad, cost control controla el gasto total, y model routing optimiza la relación costo-calidad.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter
from dataclasses import dataclass

PRICING = {
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

TIER_CONFIG = {
    "free": {
        "rate": {"requests_per_second": 0.5, "max_bucket_size": 2},
        "budget_daily": 0.001,
        "default_model": "gpt-4.1-nano",
    },
    "pro": {
        "rate": {"requests_per_second": 2, "max_bucket_size": 5},
        "budget_daily": 0.01,
        "default_model": "gpt-4.1-mini",
    },
}


@dataclass
class UserSession:
    user_id: str
    tier: str
    spent: float = 0.0
    requests: int = 0
    blocked_requests: int = 0


class ProductionController:
    def __init__(self):
        self.sessions: dict[str, UserSession] = {}
        self.rate_limiters: dict[str, InMemoryRateLimiter] = {}
        self.models = {
            "gpt-4.1-mini": init_chat_model("openai:gpt-4.1-mini"),
            "gpt-4.1-nano": init_chat_model("openai:gpt-4.1-nano"),
        }

    def register_user(self, user_id: str, tier: str):
        self.sessions[user_id] = UserSession(user_id=user_id, tier=tier)
        config = TIER_CONFIG[tier]
        self.rate_limiters[user_id] = InMemoryRateLimiter(
            requests_per_second=config["rate"]["requests_per_second"],
            check_every_n_seconds=0.1,
            max_bucket_size=config["rate"]["max_bucket_size"],
        )

    def get_model_for_user(self, user_id: str):
        """Selecciona modelo: degrada si el presupuesto se agota."""
        session = self.sessions[user_id]
        config = TIER_CONFIG[session.tier]
        budget = config["budget_daily"]
        remaining_pct = 1.0 - (session.spent / budget) if budget > 0 else 0

        if remaining_pct < 0:
            return None, None

        if remaining_pct < 0.2:
            model_name = "gpt-4.1-nano"
        else:
            model_name = config["default_model"]

        limiter = self.rate_limiters[user_id]
        model = self.models[model_name].with_rate_limiter(limiter)
        return model_name, model

    def invoke(self, user_id: str, prompt: str) -> tuple[bool, str, str]:
        """Invoca con todas las protecciones: rate limit + budget + model routing."""
        session = self.sessions[user_id]
        model_name, model = self.get_model_for_user(user_id)

        if model is None:
            session.blocked_requests += 1
            return False, "BUDGET_EXCEEDED", ""

        response = model.invoke(prompt)
        usage = response.usage_metadata
        pricing = PRICING[model_name]
        cost = (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
               (usage["output_tokens"] / 1_000_000) * pricing["output"]

        session.spent += cost
        session.requests += 1
        return True, model_name, response.content


ctrl = ProductionController()
ctrl.register_user("alice", "free")
ctrl.register_user("bob", "pro")

for i in range(8):
    for user_id in ["alice", "bob"]:
        ok, info, content = ctrl.invoke(user_id, f"Concepto {i+1} de AI. 1 oración.")
        session = ctrl.sessions[user_id]
        budget = TIER_CONFIG[session.tier]["budget_daily"]
        pct = (session.spent / budget * 100) if budget > 0 else 0

        if ok:
            print(f"  [{user_id:>6}] #{i+1}: {info:<14} ${session.spent:.6f}/{budget:.4f} ({pct:.0f}%)")
        else:
            print(f"  [{user_id:>6}] #{i+1}: BLOCKED — {info}")

print(f"\nResumen:")
for uid, session in ctrl.sessions.items():
    print(f"  {uid}: {session.requests} OK, {session.blocked_requests} blocked, ${session.spent:.6f} spent")
# Output esperado:
#   [ alice] #1: gpt-4.1-nano   $0.000015/0.0010 (2%)
#   [   bob] #1: gpt-4.1-mini   $0.000120/0.0100 (1%)
#   [ alice] #2: gpt-4.1-nano   $0.000028/0.0010 (3%)
#   [   bob] #2: gpt-4.1-mini   $0.000238/0.0100 (2%)
#   ...
#   [ alice] #7: gpt-4.1-nano   $0.000098/0.0010 (10%)
#   [   bob] #7: gpt-4.1-mini   $0.000840/0.0100 (8%)
#   [ alice] #8: gpt-4.1-nano   $0.000112/0.0010 (11%)
#   [   bob] #8: gpt-4.1-mini   $0.000960/0.0100 (10%)
#
# Resumen:
#   alice: 8 OK, 0 blocked, $0.000112 spent
#   bob: 8 OK, 0 blocked, $0.000960 spent

Rate limiting en multi-agent (conecta con M10)

En un sistema multi-agente, múltiples agentes comparten el mismo presupuesto. El researcher, analyst, y writer del Research Assistant hacen llamadas independientes, pero todas cuentan contra el mismo presupuesto del usuario.

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter

shared_limiter = InMemoryRateLimiter(
    requests_per_second=3,
    check_every_n_seconds=0.1,
    max_bucket_size=5,
)

base_model = init_chat_model("openai:gpt-4.1-mini")
shared_model = base_model.with_rate_limiter(shared_limiter)

shared_budget = {"spent": 0.0, "limit": 0.005}
PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}


def agent_call(agent_name: str, prompt: str) -> tuple[bool, str]:
    """Simula una llamada de un agente con presupuesto compartido."""
    if shared_budget["spent"] >= shared_budget["limit"]:
        return False, f"[{agent_name}] Budget exceeded"

    response = shared_model.invoke(prompt)
    usage = response.usage_metadata
    cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
           (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]
    shared_budget["spent"] += cost

    return True, f"[{agent_name}] ${cost:.6f} | Total: ${shared_budget['spent']:.6f}"


agents_flow = [
    ("researcher", "Busca información sobre AI en educación. 2 oraciones."),
    ("researcher", "Busca tendencias en edtech. 2 oraciones."),
    ("analyst", "Analiza cómo AI transforma la educación. 3 hallazgos en 3 oraciones."),
    ("writer", "Escribe un resumen de 3 oraciones sobre AI en educación."),
]

print("Multi-agent execution con presupuesto compartido:")
print(f"Budget: ${shared_budget['limit']:.4f}\n")

for agent_name, prompt in agents_flow:
    ok, msg = agent_call(agent_name, prompt)
    status = "OK" if ok else "BLOCKED"
    print(f"  [{status}] {msg}")

remaining = shared_budget["limit"] - shared_budget["spent"]
print(f"\nPresupuesto restante: ${remaining:.6f}")
# Output esperado:
# Multi-agent execution con presupuesto compartido:
# Budget: $0.0050
#
#   [OK] [researcher] $0.000120 | Total: $0.000120
#   [OK] [researcher] $0.000115 | Total: $0.000235
#   [OK] [analyst] $0.000220 | Total: $0.000455
#   [OK] [writer] $0.000160 | Total: $0.000615
#
# Presupuesto restante: $0.004385

El rate limiter compartido asegura que ningún agente monopolice las llamadas al modelo. El presupuesto compartido asegura que la ejecución completa no exceda el límite del usuario.

Cost alerts: notificaciones de gasto

En producción, las alertas tempranas previenen sorpresas en la factura. Configura alertas progresivas: noticia al 50%, advertencia al 80%, bloqueo al 100%.

from dotenv import load_dotenv
load_dotenv()

from dataclasses import dataclass
from datetime import datetime


@dataclass
class CostAlert:
    level: str
    message: str
    timestamp: str
    user_id: str
    spent: float
    limit: float


class AlertSystem:
    def __init__(self):
        self.alerts: list[CostAlert] = []
        self.notified: dict[str, set] = {}

    def check(self, user_id: str, spent: float, limit: float) -> list[CostAlert]:
        """Verifica y genera alertas si aplica."""
        if user_id not in self.notified:
            self.notified[user_id] = set()

        pct = (spent / limit * 100) if limit > 0 else 0
        new_alerts = []

        thresholds = [
            (50, "NOTICE", "Budget at 50% — consider optimizing prompts or switching models"),
            (80, "WARNING", "Budget at 80% — auto-degradation will activate at 80%"),
            (95, "CRITICAL", "Budget at 95% — service will be limited shortly"),
            (100, "BLOCKED", "Budget exceeded — requests blocked until next period"),
        ]

        for threshold, level, msg_template in thresholds:
            if pct >= threshold and threshold not in self.notified[user_id]:
                alert = CostAlert(
                    level=level,
                    message=f"[{user_id}] {msg_template} ({pct:.0f}%)",
                    timestamp=datetime.now().isoformat(),
                    user_id=user_id,
                    spent=spent,
                    limit=limit,
                )
                self.alerts.append(alert)
                new_alerts.append(alert)
                self.notified[user_id].add(threshold)

        return new_alerts


alerts = AlertSystem()

simulated_spending = [
    ("user-001", 0.003, 0.01),
    ("user-001", 0.005, 0.01),
    ("user-001", 0.008, 0.01),
    ("user-001", 0.0095, 0.01),
    ("user-001", 0.011, 0.01),
    ("user-002", 0.04, 0.10),
    ("user-002", 0.085, 0.10),
]

for user_id, spent, limit in simulated_spending:
    new_alerts = alerts.check(user_id, spent, limit)
    for alert in new_alerts:
        icon = {"NOTICE": "ℹ️", "WARNING": "⚠️", "CRITICAL": "🔴", "BLOCKED": "🚫"}.get(alert.level, "")
        print(f"  {icon} [{alert.level:>8}] {alert.message}")

print(f"\nTotal alerts generated: {len(alerts.alerts)}")
# Output esperado:
#   ℹ️ [  NOTICE] [user-001] Budget at 50% — consider optimizing prompts or switching models (50%)
#   ⚠️ [ WARNING] [user-001] Budget at 80% — auto-degradation will activate at 80% (80%)
#   🔴 [CRITICAL] [user-001] Budget at 95% — service will be limited shortly (95%)
#   🚫 [ BLOCKED] [user-001] Budget exceeded — requests blocked until next period (110%)
#   ℹ️ [  NOTICE] [user-002] Budget at 50% — consider optimizing prompts or switching models (85%)
#   ⚠️ [ WARNING] [user-002] Budget at 80% — auto-degradation will activate at 80% (85%)
#
# Total alerts generated: 6

Troubleshooting

Problema 1: El rate limiter bloquea demasiado tiempo

Causa: requests_per_second es muy bajo o max_bucket_size es 1, lo que significa que no hay capacidad de burst.

Solución: Ajusta los parámetros según tu caso de uso:

# Para API interna con uso moderado
limiter = InMemoryRateLimiter(
    requests_per_second=5,
    check_every_n_seconds=0.1,
    max_bucket_size=10,
)

# Para usuario free que debe ir lento
limiter = InMemoryRateLimiter(
    requests_per_second=0.5,
    check_every_n_seconds=0.5,
    max_bucket_size=2,
)

Problema 2: El presupuesto se agota sin generar alertas

Causa: Las alertas se verifican después de cada gasto. Si un solo gasto es muy grande (ej: un prompt con 50K tokens), puede saltar del 40% al 120% sin pasar por los umbrales intermedios.

Solución: Verifica todos los umbrales en orden en cada check, no solo el siguiente:

for threshold, level, msg in thresholds:
    if pct >= threshold and threshold not in notified:
        # Genera alerta para CADA umbral alcanzado
        notified.add(threshold)

Problema 3: El rate limiter no se comparte entre agentes

Causa: Cada agente crea su propio rate limiter en vez de compartir uno.

Solución: Crea el rate limiter una vez y pásalo a todos los agentes:

shared_limiter = InMemoryRateLimiter(requests_per_second=5, ...)
researcher_model = base_model.with_rate_limiter(shared_limiter)
analyst_model = base_model.with_rate_limiter(shared_limiter)
writer_model = base_model.with_rate_limiter(shared_limiter)

Problema 4: `InMemoryRateLimiter` no persiste entre reinicios

Causa: Es in-memory — se reinicia cuando el proceso termina.

Solución: Para producción, usa un rate limiter basado en Redis u otro store persistente. InMemoryRateLimiter es suficiente para un solo proceso, pero no para sistemas distribuidos.

Problema 5: La auto-degradación cambia de modelo en medio de una conversación

Causa: El presupuesto cruza un umbral durante la ejecución del agente, haciendo que las primeras llamadas usen GPT-4.1 y las últimas GPT-4.1-nano.

Solución: Fija el modelo al inicio de cada ejecución completa, no entre llamadas:

model_name = select_model_for_budget(user_budget)
# Usa este modelo para TODA la ejecución del agente

Ejercicios

Ejercicio 1: Rate limiter básico con medición de throughput (Fácil)

Crea un rate limiter de 2 requests/segundo y mide cuánto tardan 6 requests secuenciales. Compara con el tiempo teórico.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter
import time

limiter = InMemoryRateLimiter(
    requests_per_second=2,
    check_every_n_seconds=0.1,
    max_bucket_size=2,
)

model = init_chat_model("openai:gpt-4.1-mini")
limited = model.with_rate_limiter(limiter)

start = time.time()
for i in range(6):
    response = limited.invoke(f"Di '{i+1}'.")
    elapsed = time.time() - start
    print(f"  [{elapsed:.1f}s] Request {i+1}: {response.content.strip()}")

total = time.time() - start
theoretical = (6 - 2) / 2
print(f"\nTiempo real: {total:.1f}s")
print(f"Tiempo teórico mínimo: ~{theoretical:.1f}s + latencia del modelo")
# Output esperado:
#   [0.3s] Request 1: 1
#   [0.6s] Request 2: 2
#   [1.1s] Request 3: 3
#   [1.6s] Request 4: 4
#   [2.1s] Request 5: 5
#   [2.6s] Request 6: 6
#
# Tiempo real: 2.6s
# Tiempo teórico mínimo: ~2.0s + latencia del modelo

Explicación: Con max_bucket_size=2, los primeros 2 requests salen inmediatos. Los siguientes 4 esperan 0.5s cada uno (2 req/s). El tiempo real es teórico + latencia de la API.

Ejercicio 2: Cost controller con bloqueo al 100% (Fácil)

Crea un cost controller con presupuesto de $0.001. Ejecuta requests hasta que se bloquee y reporta cuántos requests pasaron.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {"gpt-4.1-mini": {"input": 0.40, "output": 1.60}}

class SimpleCostController:
    def __init__(self, budget: float):
        self.budget = budget
        self.spent = 0.0
        self.passed = 0
        self.blocked = 0

    def try_charge(self, usage: dict) -> bool:
        cost = (usage["input_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["input"] + \
               (usage["output_tokens"] / 1_000_000) * PRICING["gpt-4.1-mini"]["output"]
        if self.spent + cost > self.budget:
            self.blocked += 1
            return False
        self.spent += cost
        self.passed += 1
        return True

ctrl = SimpleCostController(budget=0.001)
model = init_chat_model("openai:gpt-4.1-mini")

for i in range(20):
    response = model.invoke(f"Explica concepto #{i+1} de AI en 2 oraciones.")
    if ctrl.try_charge(response.usage_metadata):
        pct = ctrl.spent / ctrl.budget * 100
        print(f"  #{i+1}: OK | Spent: ${ctrl.spent:.6f} ({pct:.0f}%)")
    else:
        print(f"  #{i+1}: BLOCKED | Spent: ${ctrl.spent:.6f} / ${ctrl.budget:.4f}")
        break

print(f"\nResultado: {ctrl.passed} pasaron, {ctrl.blocked} bloqueados")
# Output esperado:
#   #1: OK | Spent: $0.000125 (13%)
#   #2: OK | Spent: $0.000248 (25%)
#   ...
#   #7: OK | Spent: $0.000880 (88%)
#   #8: BLOCKED | Spent: $0.000880 / $0.0010
#
# Resultado: 7 pasaron, 1 bloqueados

Explicación: El controller acumula costos y bloquea cuando el siguiente gasto excedería el presupuesto. Simple y efectivo para proteger el presupuesto.

Ejercicio 3: Auto-degradación por presupuesto (Medio)

Implementa un sistema que use GPT-4.1-mini cuando hay >50% de presupuesto y GPT-4.1-nano cuando hay <50%. Muestra el cambio de modelo durante la ejecución.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model

PRICING = {
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

models = {
    "gpt-4.1-mini": init_chat_model("openai:gpt-4.1-mini"),
    "gpt-4.1-nano": init_chat_model("openai:gpt-4.1-nano"),
}

budget = 0.001
spent = 0.0

for i in range(12):
    remaining_pct = (1 - spent / budget) * 100 if budget > 0 else 0
    model_name = "gpt-4.1-mini" if remaining_pct > 50 else "gpt-4.1-nano"

    if spent >= budget:
        print(f"  #{i+1}: STOPPED — budget exhausted")
        break

    response = models[model_name].invoke(f"Concepto {i+1} de AI. 1 oración corta.")
    usage = response.usage_metadata
    pricing = PRICING[model_name]
    cost = (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
           (usage["output_tokens"] / 1_000_000) * pricing["output"]

    spent += cost
    remaining_pct = (1 - spent / budget) * 100
    print(f"  #{i+1}: {model_name:<14} | ${cost:.6f} | Budget: {remaining_pct:.0f}%")

print(f"\nTotal: ${spent:.6f} / ${budget:.4f}")
# Output esperado:
#   #1: gpt-4.1-mini   | $0.000120 | Budget: 88%
#   #2: gpt-4.1-mini   | $0.000115 | Budget: 76%
#   #3: gpt-4.1-mini   | $0.000118 | Budget: 64%
#   #4: gpt-4.1-mini   | $0.000122 | Budget: 52%
#   #5: gpt-4.1-nano   | $0.000025 | Budget: 49%
#   #6: gpt-4.1-nano   | $0.000022 | Budget: 47%
#   ...

Explicación: Cuando el presupuesto cruza el 50%, el sistema degrada a nano automáticamente. Nota cómo el costo por request baja dramáticamente, extendiendo la vida del presupuesto.

Ejercicio 4: Sistema de alertas con 3 niveles (Medio)

Crea un sistema de alertas que notifique al 50%, 80%, y 100% del presupuesto. Simula gasto progresivo y muestra las alertas generadas.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from datetime import datetime

class BudgetAlerts:
    def __init__(self, budget: float):
        self.budget = budget
        self.spent = 0.0
        self.triggered = set()
        self.log = []

    def spend(self, amount: float) -> list[str]:
        self.spent += amount
        pct = (self.spent / self.budget * 100) if self.budget > 0 else 0
        alerts = []

        levels = [(50, "NOTICE"), (80, "WARNING"), (100, "CRITICAL")]
        for threshold, level in levels:
            if pct >= threshold and threshold not in self.triggered:
                self.triggered.add(threshold)
                msg = f"[{level}] Budget at {pct:.0f}% — ${self.spent:.6f}/${self.budget:.4f}"
                alerts.append(msg)
                self.log.append({"level": level, "pct": pct, "time": datetime.now().isoformat()})

        return alerts


system = BudgetAlerts(budget=0.001)

increments = [0.0001, 0.0001, 0.0001, 0.0001, 0.0001,
              0.0001, 0.0001, 0.0001, 0.0001, 0.0002]

for i, amount in enumerate(increments):
    alerts = system.spend(amount)
    pct = system.spent / system.budget * 100
    print(f"  Spend #{i+1}: +${amount:.4f} | Total: {pct:.0f}%")
    for alert in alerts:
        print(f"    → {alert}")

print(f"\nAlerts triggered: {len(system.log)}")
# Output esperado:
#   Spend #1: +$0.0001 | Total: 10%
#   Spend #2: +$0.0001 | Total: 20%
#   Spend #3: +$0.0001 | Total: 30%
#   Spend #4: +$0.0001 | Total: 40%
#   Spend #5: +$0.0001 | Total: 50%
#     → [NOTICE] Budget at 50% — $0.000500/$0.0010
#   Spend #6: +$0.0001 | Total: 60%
#   Spend #7: +$0.0001 | Total: 70%
#   Spend #8: +$0.0001 | Total: 80%
#     → [WARNING] Budget at 80% — $0.000800/$0.0010
#   Spend #9: +$0.0001 | Total: 90%
#   Spend #10: +$0.0002 | Total: 110%
#     → [CRITICAL] Budget at 110% — $0.001100/$0.0010
#
# Alerts triggered: 3

Explicación: Las alertas se disparan una sola vez por umbral (el triggered set previene duplicados). En producción, cada nivel enviaría la notificación por el canal apropiado (log, email, Slack, PagerDuty).

Ejercicio 5: Circuit breaker con auto-reset (Difícil)

Implementa un circuit breaker que se activa cuando el gasto en 30 segundos excede un threshold, pero se auto-resetea después de 10 segundos de pausa.

Ver solución

from dotenv import load_dotenv
load_dotenv()

import time

class AutoResetCircuitBreaker:
    def __init__(self, cost_threshold: float, window_s: int = 30, cooldown_s: int = 10):
        self.cost_threshold = cost_threshold
        self.window_s = window_s
        self.cooldown_s = cooldown_s
        self.history: list[tuple[float, float]] = []
        self.state = "closed"
        self.opened_at: float = 0

    def record(self, cost: float) -> str:
        now = time.time()

        if self.state == "open":
            if now - self.opened_at > self.cooldown_s:
                self.state = "closed"
                self.history = []
                print(f"    [CIRCUIT] Auto-reset after {self.cooldown_s}s cooldown")
            else:
                return "OPEN"

        self.history.append((now, cost))
        cutoff = now - self.window_s
        self.history = [(t, c) for t, c in self.history if t > cutoff]

        window_cost = sum(c for _, c in self.history)
        if window_cost > self.cost_threshold:
            self.state = "open"
            self.opened_at = now
            return "TRIPPED"

        return "OK"

    def can_proceed(self) -> bool:
        if self.state == "open":
            if time.time() - self.opened_at > self.cooldown_s:
                self.state = "closed"
                self.history = []
                return True
            return False
        return True


breaker = AutoResetCircuitBreaker(cost_threshold=0.0005, window_s=30, cooldown_s=3)

costs = [0.0001, 0.0001, 0.00015, 0.00012, 0.0001, 0.0001, 0.0001, 0.0001]

for i, cost in enumerate(costs):
    if not breaker.can_proceed():
        print(f"  #{i+1}: BLOCKED (circuit open, waiting for cooldown...)")
        time.sleep(1)
        if breaker.can_proceed():
            print(f"  #{i+1}: Circuit reset! Proceeding...")
            status = breaker.record(cost)
            print(f"  #{i+1}: ${cost:.5f} | Status: {status}")
        else:
            continue
    else:
        status = breaker.record(cost)
        print(f"  #{i+1}: ${cost:.5f} | Status: {status}")

    if status == "TRIPPED":
        print(f"    Circuit breaker TRIPPED! Pausing for {breaker.cooldown_s}s...")
        time.sleep(breaker.cooldown_s + 0.5)
# Output esperado:
#   #1: $0.00010 | Status: OK
#   #2: $0.00010 | Status: OK
#   #3: $0.00015 | Status: OK
#   #4: $0.00012 | Status: OK
#   #5: $0.00010 | Status: TRIPPED
#     Circuit breaker TRIPPED! Pausing for 3s...
#     [CIRCUIT] Auto-reset after 3s cooldown
#   #6: $0.00010 | Status: OK
#   #7: $0.00010 | Status: OK
#   #8: $0.00010 | Status: OK

Explicación: El circuit breaker se abre cuando el gasto acumulado en la ventana excede el threshold. Después del cooldown, se auto-resetea con historial limpio. Esto previene bloqueos permanentes mientras protege contra picos de gasto.

Ejercicio 6: Controller completo con rate limit + budget + degradación (Difícil)

Construye un controller que combine rate limiting (2 req/s), presupuesto ($0.002), y auto-degradación (GPT-4.1-mini → nano al 60%). Ejecuta 15 prompts y muestra la transición completa.

Ver solución

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.rate_limiters import InMemoryRateLimiter
import time

PRICING = {
    "gpt-4.1-mini": {"input": 0.40, "output": 1.60},
    "gpt-4.1-nano": {"input": 0.10, "output": 0.40},
}

limiter = InMemoryRateLimiter(
    requests_per_second=2,
    check_every_n_seconds=0.1,
    max_bucket_size=3,
)

models = {
    "gpt-4.1-mini": init_chat_model("openai:gpt-4.1-mini").with_rate_limiter(limiter),
    "gpt-4.1-nano": init_chat_model("openai:gpt-4.1-nano").with_rate_limiter(limiter),
}

budget = 0.002
spent = 0.0
degradation_threshold = 0.60

start = time.time()
for i in range(15):
    if spent >= budget:
        print(f"  #{i+1}: STOPPED — budget exhausted (${spent:.6f})")
        break

    remaining_pct = 1 - (spent / budget) if budget > 0 else 0
    model_name = "gpt-4.1-mini" if remaining_pct > (1 - degradation_threshold) else "gpt-4.1-nano"

    response = models[model_name].invoke(f"Concepto {i+1} de producción AI. 1 oración.")
    usage = response.usage_metadata
    pricing = PRICING[model_name]
    cost = (usage["input_tokens"] / 1_000_000) * pricing["input"] + \
           (usage["output_tokens"] / 1_000_000) * pricing["output"]

    spent += cost
    elapsed = time.time() - start
    remaining_pct_display = (1 - spent / budget) * 100
    print(f"  #{i+1:>2} [{elapsed:>5.1f}s] {model_name:<14} ${cost:.6f} | "
          f"Budget: {remaining_pct_display:>5.1f}% | Total: ${spent:.6f}")

print(f"\nFinal: ${spent:.6f} / ${budget:.4f} en {time.time() - start:.1f}s")
# Output esperado:
#   # 1 [  0.5s] gpt-4.1-mini   $0.000120 | Budget:  94.0% | Total: $0.000120
#   # 2 [  1.0s] gpt-4.1-mini   $0.000118 | Budget:  88.1% | Total: $0.000238
#   # 3 [  1.5s] gpt-4.1-mini   $0.000115 | Budget:  82.4% | Total: $0.000353
#   # 4 [  2.0s] gpt-4.1-mini   $0.000122 | Budget:  76.3% | Total: $0.000475
#   # 5 [  2.5s] gpt-4.1-mini   $0.000119 | Budget:  70.4% | Total: $0.000594
#   # 6 [  3.0s] gpt-4.1-mini   $0.000125 | Budget:  64.2% | Total: $0.000719
#   # 7 [  3.5s] gpt-4.1-nano   $0.000025 | Budget:  62.9% | Total: $0.000744
#   # 8 [  4.0s] gpt-4.1-nano   $0.000022 | Budget:  61.8% | Total: $0.000766
#   ...
#
# Final: $0.000900 / $0.0020 en 7.5s

Explicación: Las tres capas trabajan juntas: el rate limiter espacía los requests (visible en los timestamps), el presupuesto bloquea al agotar, y la auto-degradación cambia de mini a nano al cruzar el 60%. Nota cómo el costo por request baja dramáticamente después de la degradación.

Resumen

En esta cápsula aprendiste:

InMemoryRateLimiter controla la velocidad de requests al modelo usando un token bucket algorithm con requests_per_second, check_every_n_seconds, y max_bucket_size
.with_rate_limiter() aplica el limiter a cualquier modelo de forma transparente — el caller no necesita saber que hay rate limiting
Per-user rate limiting asigna diferentes velocidades según el tier del usuario (free/pro/enterprise)
Cost budgets establecen límites de gasto diario y mensual por usuario, con verificación antes de cada llamada
Circuit breaker detecta tasas de gasto anómalas y pausa el sistema automáticamente para prevenir descontrol
Auto-degradación cambia a modelos más baratos conforme el presupuesto se agota, en vez de bloquear al usuario
Rate limit + cost control + model routing combinados crean un sistema de control de costos completo para producción
Presupuesto compartido en multi-agent asegura que todos los agentes (researcher, analyst, writer) cuentan contra el mismo límite del usuario
Cost alerts con notificaciones progresivas (50%, 80%, 100%) previenen sorpresas en la factura

Próxima cápsula: Production Checklist y Deployment — la checklist completa para poner tu agente de AI en producción.

Recursos adicionales

LangChain Rate Limiting — Guía oficial de rate limiting en modelos
InMemoryRateLimiter API — Referencia de la API
Token Bucket Algorithm — Teoría detrás del rate limiting
OpenAI Rate Limits — Límites de la API de OpenAI
Circuit Breaker Pattern — Martin Fowler sobre circuit breakers
LangSmith Usage Dashboard — Monitoreo de uso y costos en LangSmith

Módulo 12 — LangChain & LangGraph: From Chains to Agents

← AnteriorToken Usage Tracking y Prompt Caching Siguiente →Production Checklist y Deployment

Rate Limiting y Cost Control

Descripción de la cápsula

InMemoryRateLimiter: control de velocidad

max_bucket_size: permitir bursts controlados

Rate limiting en agentes

Per-user rate limiting: tiers de servicio

Cost budgets: límites de gasto

Circuit breaker: pausa automática ante costos anómalos

Auto-degradación: modelo más barato cuando el presupuesto aprieta

Combinando rate limiting + cost control + model routing

Rate limiting en multi-agent (conecta con M10)

Cost alerts: notificaciones de gasto

Troubleshooting

Problema 1: El rate limiter bloquea demasiado tiempo

Problema 2: El presupuesto se agota sin generar alertas

Problema 3: El rate limiter no se comparte entre agentes

Problema 4: InMemoryRateLimiter no persiste entre reinicios

Problema 5: La auto-degradación cambia de modelo en medio de una conversación

Ejercicios

Ejercicio 1: Rate limiter básico con medición de throughput (Fácil)

Ejercicio 2: Cost controller con bloqueo al 100% (Fácil)

Ejercicio 3: Auto-degradación por presupuesto (Medio)

Ejercicio 4: Sistema de alertas con 3 niveles (Medio)

Ejercicio 5: Circuit breaker con auto-reset (Difícil)

Ejercicio 6: Controller completo con rate limit + budget + degradación (Difícil)

Resumen

Recursos adicionales

Problema 4: `InMemoryRateLimiter` no persiste entre reinicios