# Tracking and cutting token costs in Python

> Source: https://learn-python.com/ai/tokenomics/
> Part of Learn Python, free to read.

The economics are the same in every language — [what tokens cost and where the money goes](https://codelearningdojo.com/token-economics/) 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.

```python src/app/tokens.py
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) // 4
```

:::warn A local count is an estimate, not the invoice
Tokenisers differ between providers and model families, and none of them account for the tokens a provider adds for system scaffolding, tool schemas or images. Use the local count to *decide* — is this request too big, should I trim, is this within budget — and use the `usage` field the API returns to *account*. Never reconcile a bill against a local count.
:::

For 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.

```python src/app/costs.py
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"))
```

```python
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
```

```python
@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`, not `else`.** 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`, not `float`.** You are doing money arithmetic. See [the failure-mode catalogue](/review/failure-modes/) 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.

```python
# 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.

:::tip Assert on the cache, in a test
```python
async def test_prefix_is_stable():
    a = build("q1", ["doc-b", "doc-a"])
    b = build("q2", ["doc-a", "doc-b"])
    assert a[0] == b[0] and a[1] == b[1]      # only the last message differs
```
That test catches the day someone adds `f"Session {uuid4()}"` to the system prompt.
:::

## Use the batch API for anything that can wait

Roughly half price, and evals, backfills and nightly jobs all qualify.

```python
# 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](/ai/evals/) 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.

```python src/app/budget.py
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 += 1
```

`max_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

```python
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.

```python scripts/spend_report.py
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.

:::verdict The four changes that usually do it
1. **Reorder prompts so the cache hits.** Usually the largest single win, and it changes no behaviour.
2. **Cap turns in every agent loop.** Protects against the worst case rather than the average one.
3. **Route classification and extraction to a small model.** Most calls are not hard.
4. **Set `max_tokens` deliberately 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.