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
openaiPython 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 IMMEDIATEworks for single-node; for distributed systems, use RedisINCRBYFLOATor Postgres row locks. - Async: The
openai.AsyncOpenAIclient follows the same pattern. Wrapawait client.chat.completions.createand 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
printis 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
usagebreakdown.
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.