Skip to main content

LLM Cost Optimization: Budget Enforcement & Hard Caps

Budget enforcement prevents cost surprises by setting hard spending caps at multiple levels (per-user, per-feature, per-project, per-month) and rejecting requests that would exceed them. A runaway loop (retry storm, infinite query generation) can burn $10,000 in minutes without enforcement; with hard caps, the system stops at $100 (your pre-set limit) and alerts an engineer. Enforcement requires three components: (1) a budget store (database) tracking spend per dimension, (2) a quota-check middleware that runs before each API call, and (3) alerting that pages an engineer when spend approaches limits. Mature production systems enforce budgets across four levels: per-user (e.g., free-tier users get $0.50/day), per-feature (e.g., code generation gets $500/month), per-project (e.g., all features in project X get $5,000/month), and per-organization (e.g., company-wide cap at $50,000/month). Enforcement is the cheapest insurance: investing one week to build a quota system saves tens of thousands in the first incident.

Designing a Multi-Level Budget Architecture​

A multi-level budget system tracks spend at user, feature, project, and organization levels, enforcing the tightest applicable limit. Here is a conceptual model:

LevelDimensionQuotaDurationUse Case
Useruser_id$0.50DailyRate-limit free users
Usercustomer_tier$10DailyLimit pro users per day
Featurefeature_name$500MonthlyCap code generation spend
Projectproject_id$5,000MonthlyDepartmental budget
Organizationorg_id$50,000MonthlyCompany-wide cap

When a request comes in (e.g., free-tier user requests code generation on project A), you check all applicable limits: user daily ($0.50 spent, can afford $0 more), feature monthly ($200 spent of $500), project monthly ($3,000 spent of $5,000), organization monthly ($40,000 spent of $50,000). The user hits their daily limit first, so the request is rejected with a message: "You've exhausted your free daily limit ($0.50). Upgrade to pro to get $10/day."

Implementing Quota Checks with a Quota Store​

A quota store is a database (PostgreSQL, Redis, Firestore) that tracks spend per dimension and budget tier. For Redis (low-latency in-memory option):

import anthropic
import redis
from datetime import datetime, timedelta
from typing import Optional

redis_client = redis.Redis(host='localhost', port=6379, db=0)

class BudgetConfig:
    """Budget limits per dimension."""
    def __init__(self):
        # Daily limits (reset every 24h)
        self.user_daily = {
            "free": 0.50,
            "pro": 10.0,
            "enterprise": 1000.0,
        }
        # Monthly limits (reset on 1st of month)
        self.feature_monthly = {
            "code_generation": 500.0,
            "data_analysis": 1000.0,
            "content_writing": 300.0,
        }
        self.project_monthly = 5000.0
        self.org_monthly = 50000.0

def get_budget_key(dimension: str, value: str, period: str) -> str:
    """Generate Redis key for budget tracking."""
    now = datetime.utcnow()
    if period == "daily":
        date_key = now.strftime("%Y-%m-%d")
    elif period == "monthly":
        date_key = now.strftime("%Y-%m")
    return f"budget:{dimension}:{value}:{date_key}"

def get_current_spend(dimension: str, value: str, period: str) -> float:
    """Query current spend from Redis."""
    key = get_budget_key(dimension, value, period)
    spend = redis_client.get(key)
    return float(spend) if spend else 0.0

def record_spend(dimension: str, value: str, period: str, amount: float) -> None:
    """Add to spend in Redis."""
    key = get_budget_key(dimension, value, period)
    ttl = 86400 if period == "daily" else 2592000  # 24h or 30d
    redis_client.incrby(key, int(amount * 1_000_000))  # Store as microdollars
    redis_client.expire(key, ttl)

def check_quota(
    user_id: str,
    user_tier: str,
    feature: str,
    project_id: str,
    org_id: str,
    estimated_cost: float,
    config: BudgetConfig,
) -> tuple[bool, Optional[str]]:
    """
    Check if request can proceed without exceeding any budget.
    Returns (allowed, reason_for_denial).
    """
    
    # Check user daily limit
    user_spend = get_current_spend("user", f"{user_id}:daily", "daily")
    user_limit = config.user_daily.get(user_tier, 0.0)
    if user_spend + estimated_cost > user_limit:
        return False, f"Daily limit reached: ${user_limit}. Spent: ${user_spend:.2f}"
    
    # Check feature monthly limit
    feature_spend = get_current_spend("feature", feature, "monthly")
    feature_limit = config.feature_monthly.get(feature, float('inf'))
    if feature_spend + estimated_cost > feature_limit:
        return False, f"Feature monthly limit: ${feature_limit}. Spent: ${feature_spend:.2f}"
    
    # Check project monthly limit
    project_spend = get_current_spend("project", project_id, "monthly")
    if project_spend + estimated_cost > config.project_monthly:
        return False, f"Project limit: ${config.project_monthly}. Spent: ${project_spend:.2f}"
    
    # Check organization monthly limit
    org_spend = get_current_spend("org", org_id, "monthly")
    if org_spend + estimated_cost > config.org_monthly:
        return False, f"Organization limit: ${config.org_monthly}. Spent: ${org_spend:.2f}"
    
    return True, None

def llm_request_with_quota(
    user_id: str,
    user_tier: str,
    feature: str,
    project_id: str,
    org_id: str,
    query: str,
    model: str = "claude-3-5-sonnet-20241022",
) -> dict:
    """
    Make LLM request only if quota allows.
    Returns {success, response, cost} or {success, error}.
    """
    config = BudgetConfig()
    client = anthropic.Anthropic()
    
    # Estimate cost before making request
    count_response = client.messages.count_tokens(
        model=model,
        messages=[{"role": "user", "content": query}],
    )
    estimated_input_tokens = count_response.input_tokens
    estimated_output_tokens = 200  # Assume average output
    
    # Pricing (June 2026)
    if "haiku" in model:
        input_price = 0.8
        output_price = 4.0
    else:  # Sonnet
        input_price = 3.0
        output_price = 15.0
    
    estimated_cost = (
        (estimated_input_tokens * input_price +
         estimated_output_tokens * output_price) / 1_000_000
    )
    
    # Check quota
    allowed, denial_reason = check_quota(
        user_id, user_tier, feature, project_id, org_id,
        estimated_cost, config
    )
    
    if not allowed:
        return {
            "success": False,
            "error": denial_reason,
        }
    
    # Quota OK, make request
    response = client.messages.create(
        model=model,
        max_tokens=300,
        messages=[{"role": "user", "content": query}],
    )
    
    actual_input_tokens = response.usage.input_tokens
    actual_output_tokens = response.usage.output_tokens
    actual_cost = (
        (actual_input_tokens * input_price +
         actual_output_tokens * output_price) / 1_000_000
    )
    
    # Record actual spend across all dimensions
    record_spend("user", f"{user_id}:daily", "daily", actual_cost)
    record_spend("feature", feature, "monthly", actual_cost)
    record_spend("project", project_id, "monthly", actual_cost)
    record_spend("org", org_id, "monthly", actual_cost)
    
    return {
        "success": True,
        "response": response.content[0].text,
        "cost": actual_cost,
        "tokens": {
            "input": actual_input_tokens,
            "output": actual_output_tokens,
        },
    }

This pattern enforces budgets in-memory (Redis) with microsecond latency. When any budget is exceeded, the request is denied before the (expensive) API call is made. The user sees a helpful error message: "Daily limit reached: $0.50. Spent: $0.50" rather than an unexpected bill.

Alerting on Budget Approach​

Enforce soft alerts when spend approaches (but has not exceeded) a budget—typically at 80% and 95% thresholds. This gives teams time to investigate and optimize before hitting hard caps. Here is an alerting pattern:

import logging

logger = logging.getLogger(__name__)

def check_budget_health(config: BudgetConfig) -> dict:
    """
    Audit all budgets and return health status.
    Alerts if any budget is >80% spent.
    """
    alerts = []
    
    # Check a sample of dimensions
    sample_users = ["user_1", "user_2", "user_3"]
    sample_projects = ["project_A", "project_B"]
    
    for user_id in sample_users:
        spend = get_current_spend("user", f"{user_id}:daily", "daily")
        tier = "pro"  # Assume pro for simplicity
        limit = config.user_daily[tier]
        pct = (spend / limit) * 100 if limit > 0 else 0
        
        if pct > 95:
            alerts.append({
                "severity": "critical",
                "message": f"User {user_id} at 95% of daily limit: ${spend:.2f}/${limit}",
                "dimension": "user_daily",
            })
        elif pct > 80:
            alerts.append({
                "severity": "warning",
                "message": f"User {user_id} at 80% of daily limit: ${spend:.2f}/${limit}",
                "dimension": "user_daily",
            })
    
    # Check feature budgets
    for feature, limit in config.feature_monthly.items():
        spend = get_current_spend("feature", feature, "monthly")
        pct = (spend / limit) * 100 if limit > 0 else 0
        
        if pct > 80:
            severity = "critical" if pct > 95 else "warning"
            alerts.append({
                "severity": severity,
                "message": f"Feature {feature} at {pct:.0f}% of monthly limit: ${spend:.2f}/${limit}",
                "dimension": "feature_monthly",
            })
    
    # Log and (optionally) page on-call engineer
    for alert in alerts:
        if alert["severity"] == "critical":
            logger.error(f"CRITICAL: {alert['message']}")
            # trigger_pagerduty_alert(alert)  # Uncomment in production
        else:
            logger.warning(f"WARNING: {alert['message']}")
    
    return {
        "total_alerts": len(alerts),
        "critical": len([a for a in alerts if a["severity"] == "critical"]),
        "warnings": len([a for a in alerts if a["severity"] == "warning"]),
        "alerts": alerts,
    }

# Run audit periodically
health = check_budget_health(BudgetConfig())
if health["critical"] > 0:
    print(f"ALERT: {health['critical']} critical budget issues!")

Run this audit every 1–2 hours in production. When it detects budget pressure, log to your alerting system (Datadog, PagerDuty, Slack) so teams can react. Alerts typically page the on-call engineer only if critical (>95% of budget spent), preventing noise while ensuring high-spend anomalies don't go unnoticed.

Gradual Rollout and Feedback Loop​

Deploy quota enforcement gradually: (1) week 1, log quota checks (warn on denial but allow request); (2) week 2, enforce on 10% of requests (real denials); (3) week 3, enforce on 50%; (4) week 4, enforce on 100%. Monitor error rates and user complaints; refine budget limits if needed. If users report "I'm being denied despite having budget," investigate: is your quota store out of sync? Are timestamps wrong (e.g., monthly budget resets at wrong time)? A gradual rollout surfaces these issues early.

Key Takeaways​

  • Enforce budgets at four levels: per-user (daily), per-feature (monthly), per-project (monthly), per-organization (monthly).
  • Use a fast quota store (Redis) to check spend before making expensive API calls.
  • Reject requests that would exceed budget with clear error messages guiding users.
  • Alert at 80% threshold, page on-call at 95% threshold to catch issues early.
  • Roll out enforcement gradually (logging, 10%, 50%, 100%) to catch integration issues.

Frequently Asked Questions​

What if a user genuinely needs more budget one month?​

Implement a budget override mechanism: a manager or finance team can increase limits for a specific user/feature/period. Log all overrides in an audit trail for billing reconciliation. Overrides should require approval (not self-serve) and should email compliance.

How do I handle sudden traffic spikes that approach budget?​

Set up auto-scaling budgets: if spend exceeds 50% of monthly budget before mid-month, flag for review. If trend continues, either (1) increase budget, or (2) reduce feature availability (disable code generation for free users). Auto-scaling prevents surprises while being responsive.

Should budget enforcement be strict (deny immediately) or gradual (warn, then deny)?​

Start with strict enforcement for new systems (fail fast). If existing systems have erratic patterns, use gradual enforcement: warn at 80%, hard-stop at 100%. Gradual enforcement buys time for ops teams to investigate spikes.

What if enforcement fails (quota store unreachable)?​

Design quota checks to fail open (allow request) with a warning if the quota store is down. This prevents cascading failures. Log all "fail open" requests prominently so you can audit and reprocess later. In practice, quota stores (Redis) are highly available; failures are rare.

Can I implement budget enforcement without a quota store?​

Minimally, yes: use API-level rate limiting and bill-level limits (your cloud provider's cap on API spend). But application-level quota gives you finer granularity (per-user, per-feature) and better alerting. Recommend building a quota store even if simple (single Redis instance).

Further Reading​