Module 12: LangSmith and Production

Rate Limiting and Cost Control

Next →

Capsule overview

In the previous capsule you learned to track what your model calls cost — tokens, dollars, breakdown by operation. But tracking without acting is like having a speedometer with no brakes. A single user running 100 research queries in an hour can burn through your entire daily budget. Without rate limiting, your AI system is an open faucet of money.

Rate limiting in AI systems isn't the same as rate limiting in traditional APIs. In a REST API, you limit requests per second to protect the server. In an AI agent, you limit requests to protect your budget. A single Research Assistant run can make 5-10 model calls — if a user fires off 50 runs, that's 250-500 model calls in minutes. At $0.035 per run with GPT-4.1, that's $1.75 in a few minutes. Multiply by 100 users doing the same and you've got a $175 bill in an hour.

LangChain ships InMemoryRateLimiter to control the pace of model calls. Combined with the cost tracking from the previous capsule and the model routing from Module 4, you can build a complete cost control system: throttle request speed, set budgets per user and per project, and automatically degrade to cheaper models when the budget runs low.

InMemoryRateLimiter: speed control

InMemoryRateLimiter is LangChain's built-in implementation for limiting the rate of model requests. It works with a token bucket algorithm: you have a "bucket" of available tokens, each request consumes one, and tokens refill at a fixed rate.

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 = [
    "Say 'one'.",
    "Say 'two'.",
    "Say 'three'.",
    "Say 'four'.",
    "Say 'five'.",
]

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"\nTotal time: {total:.1f}s (without rate limiting it'd be ~2-3s)")
# Expected output:
#   [0.5s] Prompt 1: one
#   [1.0s] Prompt 2: two
#   [2.0s] Prompt 3: three
#   [3.0s] Prompt 4: four
#   [4.0s] Prompt 5: five
#
# Total time: 4.5s (without rate limiting it'd be ~2-3s)

The InMemoryRateLimiter parameters:

ParameterMeaningExample
requests_per_secondToken refill rate1 = one request per second
check_every_n_secondsCheck frequency0.1 = checks every 100ms
max_bucket_sizeMaximum bucket capacity2 = allows a burst of 2 requests

max_bucket_size: allowing controlled bursts

max_bucket_size lets you accumulate tokens while there's no activity. If your rate is 1 request/second and there are no requests for 5 seconds, the bucket fills up to max_bucket_size tokens, allowing a burst when activity resumes.

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("Waiting 3 seconds to build up tokens in the bucket...")
time.sleep(3)

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

total = time.time() - start
print(f"\nTotal time: {total:.1f}s")
print("The first 5 were fast (full bucket), then 2/second")
# Expected output:
# Waiting 3 seconds to build up tokens in the 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
#
# Total time: 3.0s
# The first 5 were fast (full bucket), then 2/second

Rate limiting in agents

When you apply rate limiting to an agent, each model call (not each agent invocation) is limited. An agent that makes 3 model calls per run gets hit by the rate limiter 3 times.

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:
    """Search for information about a topic."""
    return f"Result: {query} is an important concept in AI engineering."

agent = create_agent(limited_model, [search])

start = time.time()
result = agent.invoke({"messages": [("user", "Search for what LangGraph is and summarize it.")]})
elapsed = time.time() - start

print(f"Response: {result['messages'][-1].content[:100]}...")
print(f"Execution time: {elapsed:.1f}s")
print("(Includes rate limiting between each internal model call)")
# Expected output:
# Response: LangGraph is an open-source framework for building applications with AI agents...
# Execution time: 2.5s
# (Includes rate limiting between each internal model call)

Per-user rate limiting: service tiers

In production, different users have different limits. A free user shouldn't consume the same resources as an enterprise user.

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):
        """Return a model with the rate limiting appropriate for the user's tier."""
        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"Say '{i+1}'.")
    elapsed = time.time() - start

    print(f"[{tier:>10}] {user_id}: 3 requests in {elapsed:.1f}s "
          f"(limit: {limits['requests_per_second']} req/s)")
# Expected output:
# [      free] alice: 3 requests in 5.2s (limit: 0.5 req/s)
# [       pro] bob: 3 requests in 1.8s (limit: 2 req/s)
# [enterprise] corp-acme: 3 requests in 0.9s (limit: 10 req/s)

Cost budgets: spending limits

Rate limiting controls speed. Cost budgets control total spend. You need both: a user can respect the rate limit and still run thousands of requests in a day.

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]:
        """Check whether there's budget available."""
        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):
        """Record a charge."""
        self.spent_today += cost
        self.spent_month += cost
        self.requests_today += 1

    def alert_status(self) -> str:
        """Return the alert status."""
        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:
        """Estimate the cost of a typical call."""
        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]:
        """Check whether the user can make a call."""
        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):
        """Record the real usage of a call."""
        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"AI engineering concept {i+1} in 1 sentence.")
        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"\nFinal summary:")
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}%)")
# Expected output:
#   [ 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
#
# Final summary:
#   free-user: 8 requests, $0.000930 / $0.0010 (93%)
#   pro-user: 10 requests, $0.001195 / $0.0100 (12%)

Circuit breaker: automatic pause on anomalous costs

A circuit breaker detects when the spend rate exceeds normal levels and pauses the system before the damage grows. If your average spend is $0.05/hour and it suddenly jumps to $0.50/hour, something is wrong — an infinite loop, an injected prompt causing enormous responses, or an attack by a malicious user.

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 that trips if spend per minute exceeds the 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:
        """Record a cost and check whether the circuit breaker should trip."""
        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"Explain AI production concept #{i+1} in 3 detailed sentences.")
    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}")
# Expected output:
#   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-degradation: a cheaper model when the budget gets tight

Instead of blocking the user when they approach the limit, you can automatically degrade to a cheaper model. The user keeps getting answers, just from a more economical (and potentially less capable) model.

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]:
        """Select the model based on remaining budget."""
        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]:
        """Invoke the appropriate model and record the cost."""
        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 = [
    "Explain what observability in AI is in 3 sentences.",
    "What is tracing in LangSmith? 2 sentences.",
    "What is evaluation in AI? 2 sentences.",
    "What is rate limiting? 1 sentence.",
    "What is a circuit breaker? 1 sentence.",
    "Define monitoring. 1 sentence.",
    "Define deployment. 1 sentence.",
    "Define scaling. 1 sentence.",
    "Define resilience. 1 sentence.",
    "Define caching. 1 sentence.",
]

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"\nTotal spend: ${controller.spent:.6f} / ${controller.daily_budget:.4f}")
print(f"Degradations: {len(set(d['model'] for d in controller.degradation_log))} models used")
# Expected output:
#   [ 1] gpt-4.1        | Budget:  75.0% | Observability in AI refers to the ability to understand w...
#   [ 2] gpt-4.1        | Budget:  55.0% | LangSmith is an observability platform that lets you trace...
#   [ 3] gpt-4.1        | Budget:  40.0% | Evaluation in AI is the systematic process of measuring th...
#   [ 4] gpt-4.1-mini   | Budget:  36.0% | Rate limiting is a technique that controls the number of r...
#   [ 5] gpt-4.1-mini   | Budget:  32.0% | A circuit breaker is a design pattern that detects failure...
#   [ 6] gpt-4.1-mini   | Budget:  28.0% | Monitoring is the practice of continuously observing the p...
#   [ 7] gpt-4.1-mini   | Budget:  24.0% | Deployment is the process of putting an application into p...
#   [ 8] gpt-4.1-nano   | Budget:  22.0% | Scaling is adjusting resources according to demand....
#   [ 9] gpt-4.1-nano   | Budget:  20.0% | Resilience is the ability to recover from failures....
#   [10] gpt-4.1-nano   | Budget:  18.0% | Caching is storing data for fast access....
#
# Total spend: $0.001640 / $0.0020
# Degradations: 3 models used

The system starts with GPT-4.1 (best quality) and degrades to mini and then nano as the budget drains. The user never gets blocked — quality drops gradually instead.

Combining rate limiting + cost control + model routing

In production you need all three layers together. Rate limiting controls speed, cost control caps total spend, and model routing optimizes the cost-quality tradeoff.

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):
        """Select a model: degrade if the budget runs low."""
        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]:
        """Invoke with every protection on: 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"AI concept {i+1}. 1 sentence.")
        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"\nSummary:")
for uid, session in ctrl.sessions.items():
    print(f"  {uid}: {session.requests} OK, {session.blocked_requests} blocked, ${session.spent:.6f} spent")
# Expected output:
#   [ 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%)
#
# Summary:
#   alice: 8 OK, 0 blocked, $0.000112 spent
#   bob: 8 OK, 0 blocked, $0.000960 spent

Rate limiting in multi-agent systems (connects with M10)

In a multi-agent system, several agents share the same budget. The Research Assistant's researcher, analyst, and writer make independent calls, but all of them count against the same user budget.

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]:
    """Simulate an agent call against a shared budget."""
    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", "Search for information about AI in education. 2 sentences."),
    ("researcher", "Search for trends in edtech. 2 sentences."),
    ("analyst", "Analyze how AI transforms education. 3 findings in 3 sentences."),
    ("writer", "Write a 3-sentence summary about AI in education."),
]

print("Multi-agent execution with a shared budget:")
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"\nRemaining budget: ${remaining:.6f}")
# Expected output:
# Multi-agent execution with a shared budget:
# 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
#
# Remaining budget: $0.004385

The shared rate limiter makes sure no single agent monopolizes the model calls. The shared budget makes sure the full run never exceeds the user's limit.

Cost alerts: spend notifications

In production, early alerts prevent invoice surprises. Set up progressive alerts: a notice at 50%, a warning at 80%, a block at 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]:
        """Check spend and generate alerts if applicable."""
        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)}")
# Expected output:
#   ℹ️ [  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

Problem 1: The rate limiter blocks for too long

Cause: requests_per_second is too low, or max_bucket_size is 1, which means there's no burst capacity.

Fix: Tune the parameters to your use case:

# For an internal API with moderate usage
limiter = InMemoryRateLimiter(
    requests_per_second=5,
    check_every_n_seconds=0.1,
    max_bucket_size=10,
)

# For a free user who should go slow
limiter = InMemoryRateLimiter(
    requests_per_second=0.5,
    check_every_n_seconds=0.5,
    max_bucket_size=2,
)

Problem 2: The budget drains without generating alerts

Cause: Alerts are checked after each charge. If a single charge is very large (e.g. a prompt with 50K tokens), it can jump from 40% to 120% without passing through the intermediate thresholds.

Fix: Check every threshold in order on each check, not just the next one:

for threshold, level, msg in thresholds:
    if pct >= threshold and threshold not in notified:
        # Generate an alert for EVERY threshold crossed
        notified.add(threshold)

Problem 3: The rate limiter isn't shared across agents

Cause: Each agent creates its own rate limiter instead of sharing one.

Fix: Create the rate limiter once and pass it to every agent:

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)

Problem 4: InMemoryRateLimiter doesn't survive restarts

Cause: It's in-memory — it resets when the process ends.

Fix: For production, use a rate limiter backed by Redis or another persistent store. InMemoryRateLimiter is enough for a single process, but not for distributed systems.

Problem 5: Auto-degradation switches models mid-conversation

Cause: The budget crosses a threshold during the agent run, so the first calls use GPT-4.1 and the last ones use GPT-4.1-nano.

Fix: Pin the model at the start of each full run, not between calls:

model_name = select_model_for_budget(user_budget)
# Use this model for the ENTIRE agent run

Exercises

Exercise 1: Basic rate limiter with throughput measurement (Easy)

Create a rate limiter of 2 requests/second and measure how long 6 sequential requests take. Compare it against the theoretical time.

See solution 
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"Say '{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"\nActual time: {total:.1f}s")
print(f"Theoretical minimum: ~{theoretical:.1f}s + model latency")
# Expected output:
#   [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
#
# Actual time: 2.6s
# Theoretical minimum: ~2.0s + model latency

Explanation: With max_bucket_size=2, the first 2 requests go out immediately. The next 4 wait 0.5s each (2 req/s). Actual time is theoretical plus API latency.

Exercise 2: Cost controller that blocks at 100% (Easy)

Build a cost controller with a $0.001 budget. Run requests until it blocks and report how many got through.

See solution 
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"Explain AI concept #{i+1} in 2 sentences.")
    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"\nResult: {ctrl.passed} passed, {ctrl.blocked} blocked")
# Expected output:
#   #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
#
# Result: 7 passed, 1 blocked

Explanation: The controller accumulates costs and blocks when the next charge would exceed the budget. Simple and effective for protecting the budget.

Exercise 3: Auto-degradation by budget (Medium)

Implement a system that uses GPT-4.1-mini when more than 50% of the budget remains and GPT-4.1-nano when less than 50% remains. Show the model switch during the run.

See solution 
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"AI concept {i+1}. 1 short sentence.")
    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}")
# Expected output:
#   #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%
#   ...

Explanation: When the budget crosses 50%, the system degrades to nano automatically. Notice how the cost per request drops dramatically, extending the life of the budget.

Exercise 4: A 3-level alert system (Medium)

Build an alert system that notifies at 50%, 80%, and 100% of the budget. Simulate progressive spend and show the alerts it generates.

See solution 
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)}")
# Expected output:
#   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

Explanation: Alerts fire only once per threshold (the triggered set prevents duplicates). In production, each level would send its notification through the appropriate channel (log, email, Slack, PagerDuty).

Exercise 5: Circuit breaker with auto-reset (Hard)

Implement a circuit breaker that trips when spend within 30 seconds exceeds a threshold, but auto-resets after a 10-second pause.

See solution 
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)
# Expected output:
#   #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

Explanation: The circuit breaker opens when accumulated spend in the window exceeds the threshold. After the cooldown, it auto-resets with a clean history. This prevents permanent blocks while still protecting against spending spikes.

Exercise 6: Complete controller with rate limit + budget + degradation (Hard)

Build a controller that combines rate limiting (2 req/s), a budget ($0.002), and auto-degradation (GPT-4.1-mini → nano at 60%). Run 15 prompts and show the full transition.

See solution 
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"AI production concept {i+1}. 1 sentence.")
    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} in {time.time() - start:.1f}s")
# Expected output:
#   # 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 in 7.5s

Explanation: The three layers work together: the rate limiter spaces out the requests (visible in the timestamps), the budget blocks once it's exhausted, and auto-degradation switches from mini to nano when it crosses 60%. Notice how the cost per request drops dramatically after the degradation.

Summary

In this capsule you learned:

  • InMemoryRateLimiter controls the speed of model requests using a token bucket algorithm with requests_per_second, check_every_n_seconds, and max_bucket_size
  • .with_rate_limiter() applies the limiter to any model transparently — the caller doesn't need to know rate limiting is there
  • Per-user rate limiting assigns different speeds based on the user's tier (free/pro/enterprise)
  • Cost budgets set daily and monthly spending limits per user, with a check before every call
  • Circuit breakers detect anomalous spend rates and pause the system automatically to prevent it from spiraling
  • Auto-degradation switches to cheaper models as the budget drains, instead of blocking the user
  • Rate limit + cost control + model routing combined create a complete cost control system for production
  • A shared budget in multi-agent systems ensures every agent (researcher, analyst, writer) counts against the same user limit
  • Cost alerts with progressive notifications (50%, 80%, 100%) prevent invoice surprises

Next capsule: Production Checklist and Deployment — the complete checklist for putting your AI agent into production.

Additional resources

  1. LangChain Rate Limiting — Official guide to rate limiting in models
  2. InMemoryRateLimiter API — API reference
  3. Token Bucket Algorithm — The theory behind rate limiting
  4. OpenAI Rate Limits — OpenAI API limits
  5. Circuit Breaker Pattern — Martin Fowler on circuit breakers
  6. LangSmith Usage Dashboard — Usage and cost monitoring in LangSmith

Module 12 — LangChain & LangGraph: From Chains to Agents

← PreviousToken Usage Tracking and Prompt CachingNext →Production Checklist and Deployment