Tracking and cutting token costs in Python
Counting tokens before you send them, attributing every call to a feature, and the four changes that usually halve the bill.
The economics are the same in every language — what tokens cost and where the money goes is the model. This page is the Python implementation: how to count, how to attribute, and how to stop a loop from spending your month in an afternoon.
Count before you send#
The cheapest token is the one you notice before paying for it. Counting locally costs nothing and lets you refuse or trim a request before it leaves the process.
from functools import lru_cache
@lru_cache(maxsize=4)
def _encoder(model: str):
import tiktoken
try:
return tiktoken.encoding_for_model(model)
except KeyError:
return tiktoken.get_encoding("cl100k_base") # good enough for a budget check
def estimate_tokens(text: str, model: str = "gpt-4o") -> int:
return len(_encoder(model).encode(text))
def estimate_chars(text: str) -> int:
"""No dependency, no network: ~4 chars per token for English prose."""
return len(text) // 4For a hard guarantee on the way in, the provider's own token-counting endpoint is exact; it costs a round trip, so use it for the boundary case rather than every call.
Attribute every call#
The single most useful piece of infrastructure you can build here is small. Wrap the call, record the usage the API already gives you, and tag it.
from __future__ import annotations
import functools, json, os, time
from contextvars import ContextVar
from dataclasses import dataclass, asdict, field
from decimal import Decimal
# per-1M-token prices. Keep them in config, not in code — they change.
PRICES: dict[str, dict[str, Decimal]] = {
"small": {"in": Decimal("0.25"), "cached_in": Decimal("0.03"), "out": Decimal("1.25")},
"large": {"in": Decimal("3.00"), "cached_in": Decimal("0.30"), "out": Decimal("15.00")},
}
_ctx: ContextVar[dict] = ContextVar("llm_ctx", default={})
@dataclass(slots=True)
class Spend:
feature: str
model: str
input_tokens: int
cached_tokens: int
output_tokens: int
latency_ms: int
ok: bool
tenant: str | None = None
version: str = field(default_factory=lambda: os.getenv("GIT_SHA", "dev"))
@property
def usd(self) -> Decimal:
p = PRICES.get(self.model, PRICES["large"])
fresh = max(self.input_tokens - self.cached_tokens, 0)
return (
Decimal(fresh) * p["in"]
+ Decimal(self.cached_tokens) * p["cached_in"]
+ Decimal(self.output_tokens) * p["out"]
) / Decimal(1_000_000)
def record(spend: Spend) -> None:
line = asdict(spend) | {"usd": str(spend.usd), "ts": time.time()} | _ctx.get()
print(json.dumps(line), file=open(os.getenv("LLM_LOG", "llm-spend.jsonl"), "a"))def tracked(feature: str, model: str = "large"):
"""Wrap an LLM call so its usage is always recorded, success or failure."""
def deco(fn):
@functools.wraps(fn)
async def wrapper(*args, **kwargs):
start = time.perf_counter()
usage, ok = None, False
try:
result, usage = await fn(*args, **kwargs)
ok = True
return result
finally:
u = usage or {}
record(Spend(
feature=feature,
model=model,
input_tokens=u.get("input_tokens", 0),
cached_tokens=u.get("cache_read_input_tokens", 0),
output_tokens=u.get("output_tokens", 0),
latency_ms=int((time.perf_counter() - start) * 1000),
ok=ok,
))
return wrapper
return deco@tracked(feature="ticket_classification", model="small")
async def classify(text: str) -> tuple[Category, dict]:
resp = await client.messages.create(...)
return parse(resp), resp.usage.model_dump()Two details that matter more than they look:
finally, notelse. A failed call still consumed input tokens. If you only record successes, your cost-per-successful-result is wrong in the direction that flatters you.Decimal, notfloat. You are doing money arithmetic. See the failure-mode catalogue for why this is not pedantry.
ContextVar carries the tenant and request id down without threading them through every signature — set it once in your middleware.
Make the cache hit#
The largest single saving, and in Python it is usually a function-argument-ordering problem.
# BAD: the timestamp poisons everything after it
def build(question: str, docs: list[str]) -> list[dict]:
return [
{"role": "system", "content": f"Today is {date.today()}.\n{RULES}"},
{"role": "user", "content": "\n".join(docs) + question},
]
# GOOD: stable prefix first, volatile last
def build(question: str, docs: list[str]) -> list[dict]:
return [
{"role": "system", "content": RULES, # 4KB, never changes
"cache_control": {"type": "ephemeral"}},
{"role": "user", "content": "\n".join(sorted(docs)), # sorted: stable order
"cache_control": {"type": "ephemeral"}},
{"role": "user", "content": f"Today is {date.today()}.\n{question}"},
]Note sorted(docs). If your documents arrive from a set, a dict, or a database query with no ORDER BY, their order can vary between runs — and a different order means a different prefix means a cache miss on the whole block. This is a real and very Python-flavoured way to lose your cache: it fails silently and only shows up on the bill.
The same applies to tool definitions built from a dict. Sort them.
Use the batch API for anything that can wait#
Roughly half price, and evals, backfills and nightly jobs all qualify.
# instead of this, at full price, at 3am
results = await asyncio.gather(*(classify(r) for r in rows))
# submit a batch and collect later
batch = await client.messages.batches.create(requests=[
{"custom_id": str(r.id), "params": {...}} for r in rows
])The trade is latency — batches complete within hours rather than seconds. For an eval run or a one-off classification of a million rows, that is free money.
Enforce a budget before the call#
An alert tells you after the money is gone. A check stops it.
class BudgetExceeded(Exception): ...
@dataclass
class Budget:
limit_usd: Decimal
spent_usd: Decimal = Decimal("0")
max_turns: int = 12
turns: int = 0
def check(self, estimated_usd: Decimal) -> None:
if self.turns >= self.max_turns:
raise BudgetExceeded(f"turn limit {self.max_turns} reached")
if self.spent_usd + estimated_usd > self.limit_usd:
raise BudgetExceeded(
f"would exceed ${self.limit_usd} (spent ${self.spent_usd:.4f})"
)
def charge(self, actual_usd: Decimal) -> None:
self.spent_usd += actual_usd
self.turns += 1max_turns is not optional. An agent loop resends the whole conversation every turn, so cost grows roughly quadratically with turn count — ten turns over a 20k context is nearer 200k tokens than 20k. A turn cap is the cheapest protection against a runaway loop there is, and it is the one people add after the incident rather than before.
Retries multiply, so bound them#
from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception
def retryable(exc: BaseException) -> bool:
status = getattr(exc, "status_code", None)
return status == 429 or (status is not None and status >= 500)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential_jitter(initial=1, max=20),
retry=retry_if_exception(retryable), # never retry a 400 — it will fail identically
reraise=True,
)
async def call_model(**kw): ...Retrying a 400 is pure waste: the request is malformed and will be malformed again. Retrying without jitter turns a provider blip into a synchronised stampede from all your workers.
Read your own log#
The output of record() is JSONL, which pandas reads directly. Ten lines gets you the report that actually changes decisions.
import pandas as pd
df = pd.read_json("llm-spend.jsonl", lines=True)
df["usd"] = df["usd"].astype(float)
df["day"] = pd.to_datetime(df["ts"], unit="s").dt.date
print("\n— by feature —")
print(df.groupby("feature")
.agg(calls=("usd", "size"), usd=("usd", "sum"),
ok_rate=("ok", "mean"), p99_out=("output_tokens", lambda s: s.quantile(0.99)))
.sort_values("usd", ascending=False))
print("\n— cost per SUCCESSFUL result —")
by = df.groupby("feature").agg(total=("usd", "sum"), wins=("ok", "sum"))
print((by["total"] / by["wins"].clip(lower=1)).sort_values(ascending=False))
print("\n— cache hit rate —")
print((df["cached_tokens"].sum() / df["input_tokens"].clip(lower=1).sum()).round(3))
print("\n— top tenants —")
print(df.groupby("tenant")["usd"].sum().nlargest(10))The second block is the one worth staring at. Cost per call flatters you; cost per successful result includes everything you paid for output that failed to parse and had to be regenerated.
The four changes that usually do it
- Reorder prompts so the cache hits. Usually the largest single win, and it changes no behaviour.
- Cap turns in every agent loop. Protects against the worst case rather than the average one.
- Route classification and extraction to a small model. Most calls are not hard.
- Set
max_tokensdeliberately on every call. Output is the expensive side.
Common questions#
Is tiktoken accurate for non-OpenAI models?#
No — it is OpenAI's tokeniser. For other providers the count will be in the right order of magnitude, which is fine for a budget check, and wrong for accounting. Use each provider's own counting endpoint when you need exactness, and always reconcile against the usage the API returns.
Where should prices live?#
In configuration, loaded at startup, not in a Python literal — they change, and a code deploy to fix a price is a bad afternoon. Keep a dated table so historical spend stays reconcilable when a price changes mid-month.
Do I need a full observability stack?#
Not at first. A JSONL file and the pandas script above answer every question that matters until you are spending real money. When you outgrow it, the same records go to your existing metrics backend without changing what you record — which is why tagging from day one matters more than where it goes.
How do I stop async fan-out from blowing the budget?#
Bound the concurrency with a semaphore and check the budget inside the guarded section, not before it. asyncio.gather over a thousand rows with no limit will happily start a thousand requests, and by the time the first budget check fails you have already committed to all of them.
Get the Python agent pack
A battle-tested AGENTS.md, the review checklist, and the failure-mode cheat sheet for Python. One email, then occasional updates when the tooling shifts. No course pitch.
AGENTS.md now — no email needed.