n4nAI

How to set per-model budgets for LLM API spend

Step-by-step tutorial: implement per-model LLM budget limits in Python using OpenAI-compatible usage metering, with fallback routing and SQLite persistence.

n4n Team4 min read824 words

Audio narration

Coming soon — every post will get a voice note here.

Setting per-model LLM budget limits is the only way to keep a multi-model app from blowing up your invoice when a retry loop goes rogue or a user floods the input. This tutorial builds a lightweight client wrapper that enforces those limits at the call site using the usage metadata returned by any OpenAI-compatible endpoint, then shows how to persist state and shift models when a limit is hit.

Prerequisites

  • Python 3.10+ with pip install openai (v1.x client)
  • An API key for an OpenAI-compatible gateway (OpenAI, n4n.ai, or self-hosted)
  • Familiarity with the openai Python client and basic JSON config

The code below points the client at a base_url of your choice. It does not depend on a specific vendor’s proprietary fields.

1. Define your budget policy

Start with a static config that declares per-model LLM budget limits in dollars. Treat this file like a firewall rule: commit it to version control, review changes in PRs, and keep dev limits strictly lower than prod.

{
  "budgets": {
    "gpt-4o": 50.0,
    "gpt-4o-mini": 10.0,
    "mistral-large": 25.0
  },
  "prices_per_million_tokens": {
    "gpt-4o": {"input": 2.50, "output": 10.00},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "mistral-large": {"input": 2.00, "output": 6.00}
  }
}

Prices are illustrative placeholders—replace them with your contract rates or public list prices. The wrapper reads this file at startup and never calls home for pricing.

2. Track spend from real usage, not guesses

The OpenAI-compatible response includes a usage object with prompt_tokens and completion_tokens. That is the source of truth. Estimating from string length or tiktoken locally will diverge from what the provider actually bills, especially with cached tokens or vendor-specific preprocessing.

import json

def load_config(path: str) -> dict:
    with open(path) as f:
        return json.load(f)

def cost_for(model: str, usage: dict, cfg: dict) -> float:
    prices = cfg["prices_per_million_tokens"].get(model)
    if not prices:
        raise ValueError(f"No price configured for {model}")
    in_cost = usage["prompt_tokens"] / 1_000_000 * prices["input"]
    out_cost = usage["completion_tokens"] / 1_000_000 * prices["output"]
    return in_cost + out_cost

If your gateway supports prompt caching, the usage object may contain prompt_tokens_details.cached_tokens. Those are often billed at a discount; extend the price table with a cached_input field and subtract accordingly.

3. Build the enforcing client

We wrap openai.OpenAI. Before each call, we run a conservative pre-check using max_tokens. After the call, we record the real cost. If a call would exceed the cap, we block it before sending a single token to the provider.

Implementation

from openai import OpenAI

class BudgetGuard:
    def __init__(self, cfg: dict, base_url: str, api_key: str, spent: dict | None = None):
        self.cfg = cfg
        self.client = OpenAI(base_url=base_url, api_key=api_key)
        self.spent = spent or {m: 0.0 for m in cfg["budgets"]}

    def _check(self, model: str, estimated: float):
        limit = self.cfg["budgets"].get(model)
        if limit is None:
            return  # untracked model, allow
        if self.spent[model] + estimated > limit:
            raise RuntimeError(
                f"Budget exceeded for {model}: "
                f"{self.spent[model] + estimated:.4f} > {limit:.4f}"
            )

    def chat(self, model: str, messages: list, **kwargs):
        est_tokens = kwargs.get("max_tokens", 1000)
        price = self.cfg["prices_per_million_tokens"][model]
        est = est_tokens / 1_000_000 * price["output"]
        self._check(model, est)

        resp = self.client.chat.completions.create(
            model=model, messages=messages, **kwargs
        )
        usage = resp.usage.model_dump()
        real_cost = cost_for(model, usage, self.cfg)
        self.spent[model] += real_cost
        return resp, real_cost

Enforcing per-model LLM budget limits at runtime means the guard must be instantiated once per process and shared across all call sites. Do not create a new BudgetGuard per request.

Checkpoint: first call under budget

cfg = load_config("budget.json")
guard = BudgetGuard(cfg, "https://api.openai.com/v1", "sk-...")

resp, cost = guard.chat(
    "gpt-4o-mini",
    [{"role": "user", "content": "Say hi in 5 words."}],
    max_tokens=20
)
print(f"Cost: ${cost:.4f}, total spent: ${guard.spent['gpt-4o-mini']:.4f}")

Expected output (exact digits vary by token count):

Cost: $0.0001, total spent: $0.0001

The ledger incremented exactly once.

Checkpoint: exceeding limit

Lower the gpt-4o-mini budget to 0.0002 and run the same call twice.

# second invocation raises:
# RuntimeError: Budget exceeded for gpt-4o-mini: 0.0003 > 0.0002

The second call never reaches the network. Fail closed—that is the behavior you want.

4. Add fallback routing when a limit is hit

Per-model LLM budget limits are not useful if they just hard-fail your product. Combine them with model tiers: when the primary model is capped, route to a cheaper one.

def chat_with_fallback(guard: BudgetGuard, model: str, messages: list, **kwargs):
    try:
        return guard.chat(model, messages, **kwargs)
    except RuntimeError as e:
        if "Budget exceeded" in str(e) and model != "gpt-4o-mini":
            print(f"Falling back from {model} to gpt-4o-mini")
            return guard.chat("gpt-4o-mini", messages, **kwargs)
        raise

If you route through n4n.ai, its automatic fallback when a provider is rate-limited or degraded complements this logic; your budget guard handles soft spend caps, the gateway handles hard provider outages.

5. Persist the ledger

In-memory spend resets on restart, which defeats the purpose. Use SQLite to survive process deaths and give you a single file to inspect.

import sqlite3

def init_db(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS spend (
            model TEXT PRIMARY KEY,
            dollars REAL NOT NULL
        )
    """)

def load_spent(conn) -> dict:
    rows = conn.execute("SELECT model, dollars FROM spend").fetchall()
    return {r[0]: r[1] for r in rows}

def save_spent(conn, spent: dict):
    for model, dollars in spent.items():
        conn.execute(
            "INSERT INTO spend(model, dollars) VALUES(?, ?) "
            "ON CONFLICT(model) DO UPDATE SET dollars=excluded.dollars",
            (model, dollars)
        )
    conn.commit()

Wire it into BudgetGuard by loading self.spent from the DB in __init__ and calling save_spent after each successful charge. Centralizing per-model LLM budget limits across restarts this way turns the guard into a real control plane.

6. Streaming and usage

Streaming completions do not return usage in the first chunk. Pass stream_options={"include_usage": True} and capture the final chunk:

stream = guard.client.chat.completions.create(
    model=model, messages=messages, stream=True,
    stream_options={"include_usage": True}
)
usage = None
for chunk in stream:
    if chunk.usage:
        usage = chunk.usage.model_dump()
real_cost = cost_for(model, usage, guard.cfg)
guard.spent[model] += real_cost

If you forget include_usage, you will silently undercount spend. Always assert usage is not None before charging.

7. Production considerations

  • Concurrency: Use a lock or atomic increments if multiple workers share a ledger. SQLite with BEGIN IMMEDIATE works for single-node; for distributed systems, use Redis INCRBYFLOAT or Postgres row locks.
  • Async: The openai.AsyncOpenAI client follows the same pattern. Wrap await client.chat.completions.create and await the stream iterator.
  • Granularity: Set per-model LLM budget limits per environment (dev/staging/prod) and per API key. A dev key should have a $5 cap, not $500.
  • Alerting: Before raising, emit a metric. A print is not enough; ship the exceeded event to Prometheus or your log pipeline so finance can see trends.
  • Cache-control: Some gateways forward provider cache-control hints. If your endpoint supports prompt caching, cached tokens are cheaper—adjust your price table accordingly and verify the usage breakdown.

8. Verifying the full flow

Spin up the guard with a tiny budget, run a loop that calls the model until it throws, then inspect the DB.

conn = sqlite3.connect("ledger.db")
init_db(conn)
cfg = load_config("budget.json")
guard = BudgetGuard(cfg, "https://api.openai.com/v1", "sk-...", load_spent(conn))

for i in range(5):
    try:
        resp, cost = chat_with_fallback(
            guard, "gpt-4o",
            [{"role": "user", "content": f"Count {i}"}],
            max_tokens=50
        )
        save_spent(conn, guard.spent)
    except RuntimeError as e:
        print("Stopped:", e)
        break

conn.close()

Expected terminal output shows successful gpt-4o calls until its limit is hit, then fallback to gpt-4o-mini, then eventually a hard stop if both are capped. The DB row for gpt-4o reflects the exact spent dollars, not an estimate.

9. Why not just use provider dashboards?

Provider dashboards show spend after the fact, often with hours of delay. They do not block a runaway loop at 2 a.m. Enforcing per-model LLM budget limits client-side gives you synchronous control and lets you swap models mid-request. The dashboard remains useful for reconciliation; the guard prevents the incident.

Closing

Implementing per-model LLM budget limits at the client level takes about 100 lines and saves you from the classic “I left a debug loop running overnight” incident. The usage object is reliable; trust it, persist it, and fail closed.

Tagsbudgetingcost-optimizationllmapi

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →

All cost optimization & model routing posts →