You're making hundreds of LLM calls a day across scripts, notebooks, and a few rogue agents. Your OpenAI bill arrives and you have no idea which call cost what. The dashboard is 24 hours behind. By the time you notice a runaway loop, you've burned $40 on a recursive summarizer you forgot existed.
The fix is one decorator that wraps any LLM call and logs tokens, cost, and latency to a JSONL file you can tail -f in real time. No refactor of existing code. No new infrastructure.
import functools, time, json
from datetime import datetime, timezone
from openai import OpenAI
client = OpenAI()
# Update quarterly — pricing changes more often than you'd think
PRICING = {
"gpt-4o": {"input": 0.0025, "output": 0.0100},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
"claude-sonnet-4-5": {"input": 0.003, "output": 0.015},
}
def track_tokens(log_path="llm_costs.jsonl"):
def decorator(fn):
@functools.wraps(fn)
def wrapper(*args, **kwargs):
start = time.perf_counter()
response = fn(*args, **kwargs)
elapsed = time.perf_counter() - start
usage = getattr(response, "usage", None)
if usage is None:
return response
model = getattr(response, "model", kwargs.get("model", "unknown"))
in_tok = usage.prompt_tokens
out_tok = usage.completion_tokens
rates = PRICING.get(model, {"input": 0, "output": 0})
cost = in_tok / 1000 * rates["input"] + out_tok / 1000 * rates["output"]
record = {
"ts": datetime.now(timezone.utc).isoformat(),
"fn": fn.__name__,
"model": model,
"in_tok": in_tok,
"out_tok": out_tok,
"cost": round(cost, 6),
"ms": int(elapsed * 1000),
}
with open(log_path, "a") as f:
f.write(json.dumps(record) + "\n")
return response
return wrapper
return decoratorJSONL, not JSON. Each call appends one line. No locks, no corruption if a script dies mid-write. JSON files need read-modify-write cycles that break under concurrency the moment two agents run in parallel.
The decorator returns the response unchanged. Drop-in, no refactor. If you can wrap a function, you can track it. The original call shape is preserved so type hints and downstream code keep working.
Pricing is a hardcoded dict, not an API call. Pricing APIs add latency and a new failure mode you don't want in a hot path. Hardcode the rates you actually use and update them quarterly. Six lines, zero dependencies.
Time goes in the log too. Tokens tell you cost; latency tells you whether to switch models. A 3-second gpt-4o call that could have been a 200ms gpt-4o-mini call is the waste most teams miss because token counts alone don't show it.
@track_tokens()
def summarize(text: str) -> str:
r = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"Summarize: {text}"}],
)
return r.choices[0].message.contentWorks with OpenAI out of the box. For Anthropic, map usage.input_tokens and usage.output_tokens to the same fields — two extra lines in the decorator. The same pattern works with LiteLLM if you want one decorator for every provider.
gpt-4o summarization step called 4,000 times a day with no business reason. Swapping to gpt-4o-mini cut $180/month to $11.Pair this with awk '{s+=$7} END{print "Daily: $"s}' llm_costs.jsonl for instant cost totals, or pipe the file into DuckDB for daily breakdowns by function and model. Ship it, then watch where the money actually goes.