24 Oct 2026 · ia, custo, observabilidade · 4 min
Cost per consumer: how much each service spends on models
Without knowing who spends, model cost becomes a surprise at month end. Recording each call with consumer, model and day, then grouping, shows where the money goes. In practice most of the spend sits in a single consumer.
The problem
An agent, a bot and a summary script share the same API key. At month end the invoice arrives, but nobody knows who asked for what. The bill has no label, only a total.
The fix is to record each call with three fields beyond the model: who called, how many tokens, how much it cost. A script then groups by consumer, model and day, and the ranking appears.
The record format
A JSONL file, one line per call or per batch. Minimum fields:
{"ts": "2026-10-20T09:12:00Z", "consumer": "bot-telegram", "model": "gpt-4o-mini", "calls": 1, "prompt_tokens": 45000, "completion_tokens": 3000, "cost_usd": 0.0090}
ts accepts ISO 8601 or epoch. calls is optional and defaults to 1. cost_usd is optional; if missing, the script sums zero and you compute later from the provider price table.
The artifact
#!/usr/bin/env python3
"""Agrupa gasto de modelo por consumidor, modelo e dia.
Lê um arquivo JSONL com um registro por chamada (ou por lote) e imprime
o ranking de consumidores e o detalhe por dia e modelo.
Campos aceitos em cada linha:
ts ou date instante (ISO 8601 ou epoch)
consumer nome do serviço ou tarefa que fez a chamada
model nome do modelo
calls quantas chamadas (opcional, padrão 1)
prompt_tokens tokens de entrada
completion_tokens tokens de saída
cost_usd custo da chamada em dólares (opcional)
Uso:
python3 custo_por_consumidor.py uso.jsonl
python3 custo_por_consumidor.py # roda com exemplo embutido
"""
import json
import sys
from collections import defaultdict
from datetime import datetime, timezone
def para_data(valor):
if isinstance(valor, (int, float)):
dt = datetime.fromtimestamp(valor, tz=timezone.utc)
else:
s = str(valor).strip()
try:
dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
except ValueError:
dt = datetime.strptime(s[:10], "%Y-%m-%d").replace(tzinfo=timezone.utc)
return dt.date().isoformat()
def carregar(caminho):
with open(caminho, encoding="utf-8") as f:
for linha in f:
linha = linha.strip()
if not linha:
continue
yield json.loads(linha)
EXEMPLO = [
{"ts": "2026-10-20T09:12:00Z", "consumer": "bot-telegram", "model": "gpt-4o-mini",
"calls": 1, "prompt_tokens": 45000, "completion_tokens": 3000, "cost_usd": 0.0090},
{"ts": "2026-10-20T10:00:00Z", "consumer": "resumo-diario", "model": "gpt-4o-mini",
"calls": 1, "prompt_tokens": 12000, "completion_tokens": 800, "cost_usd": 0.0024},
{"ts": "2026-10-21T08:00:00Z", "consumer": "resumo-diario", "model": "gpt-4o-mini",
"calls": 1, "prompt_tokens": 11000, "completion_tokens": 700, "cost_usd": 0.0022},
{"ts": "2026-10-21T11:30:00Z", "consumer": "transcricao", "model": "whisper-1",
"calls": 1, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0120},
{"ts": "2026-10-22T14:00:00Z", "consumer": "bot-telegram", "model": "gpt-4o-mini",
"calls": 1, "prompt_tokens": 60000, "completion_tokens": 4000, "cost_usd": 0.0120},
]
def main():
if len(sys.argv) > 1:
registros = list(carregar(sys.argv[1]))
else:
registros = EXEMPLO
grupos = defaultdict(lambda: {"chamadas": 0, "prompt": 0,
"completion": 0, "custo": 0.0})
for r in registros:
chave = (r.get("consumer", "?"), r.get("model", "?"),
para_data(r.get("ts") or r.get("date")))
g = grupos[chave]
g["chamadas"] += int(r.get("calls", 1) or 1)
g["prompt"] += int(r.get("prompt_tokens", 0) or 0)
g["completion"] += int(r.get("completion_tokens", 0) or 0)
g["custo"] += float(r.get("cost_usd", 0) or 0)
por_consumidor = defaultdict(lambda: {"chamadas": 0, "prompt": 0,
"completion": 0, "custo": 0.0})
for (consumidor, modelo, dia), g in grupos.items():
c = por_consumidor[consumidor]
c["chamadas"] += g["chamadas"]
c["prompt"] += g["prompt"]
c["completion"] += g["completion"]
c["custo"] += g["custo"]
print("CONSUMO POR CONSUMIDOR (total da janela)")
print(f"{'consumidor':20} {'chamadas':>9} {'prompt':>10} "
f"{'completion':>11} {'custo_usd':>10}")
for consumidor, c in sorted(por_consumidor.items(),
key=lambda x: -x[1]["custo"]):
print(f"{consumidor:20} {c['chamadas']:9} {c['prompt']:10} "
f"{c['completion']:11} {c['custo']:10.4f}")
print("\nDETALHE POR DIA E MODELO")
for (consumidor, modelo, dia), g in sorted(grupos.items()):
print(f"{dia} {consumidor:20} {modelo:14} chamadas={g['chamadas']:4} "
f"prompt={g['prompt']:8} completion={g['completion']:7} "
f"custo={g['custo']:.4f}")
total = sum(c["custo"] for c in por_consumidor.values())
print(f"\nTOTAL: {len(registros)} registros, "
f"{len(por_consumidor)} consumidores, "
f"custo total = {total:.4f} USD")
if __name__ == "__main__":
main()
Run with your data:
python3 custo_por_consumidor.py uso.jsonl
Without a file, runs with a built-in example and prints the same format.
How to record the call
If you call the provider directly, wrap the call in a function that writes the line after the response. The response object carries usage.prompt_tokens and usage.completion_tokens.
import json, time
def call(consumer, model, fn, *args, **kwargs):
resp = fn(*args, **kwargs)
usage = getattr(resp, "usage", None)
record = {
"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"consumer": consumer,
"model": model,
"calls": 1,
"prompt_tokens": getattr(usage, "prompt_tokens", 0) if usage else 0,
"completion_tokens": getattr(usage, "completion_tokens", 0) if usage else 0,
"cost_usd": 0.0,
}
with open("uso.jsonl", "a", encoding="utf-8") as f:
f.write(json.dumps(record) + "\n")
return resp
cost_usd stays 0.0 and you fill it later from the provider price table, or compute it on the spot if you have the price per thousand tokens.
How to verify
Expected output has two parts. First the ranking:
CONSUMO POR CONSUMIDOR (total da janela)
consumidor chamadas prompt completion custo_usd
bot-telegram 2 105000 7000 0.0210
transcricao 1 0 0 0.0120
resumo-diario 2 23000 1500 0.0046
TOTAL: 5 registros, 3 consumidores, custo total = 0.0376 USD
Then the detail by day and model. If the numbers match the provider invoice, the recording is right.
If it goes wrong
KeyError: 'consumer'. Cause: some JSONL line lacks the field. Fix: standardize the producer to always write consumer, even if "unknown".
json.decoder.JSONDecodeError. Cause: broken line or stray comma. Fix: validate the file with python3 -m json.tool line by line, or use jq -c ..
Totals do not match the invoice. Cause: cached tokens not recorded, or cost computed with wrong price. Fix: add cache_read_tokens and cache_write_tokens if the provider charges for them, and check the price table on the call date.
Real example
In a 26-day window, the largest consumer was the conversation task, with 4194 calls and 3.58 USD. Second was background review, with 0.63 USD. Third was approval, with 0.18 USD. The window total was 4.55 USD.
FAQ
Do I need a gateway like LiteLLM? Not to start. The script reads what you already record. A gateway centralizes and adds a virtual key per consumer, which helps when the number of services grows. For a few consumers, JSONL is enough.
How do I record when using the SDK directly? Wrap the call in a function that writes the line after the response. The response object carries usage.prompt_tokens and usage.completion_tokens. You compute cost from the model price.
What if the provider does not return tokens? Estimate: 1 token is about 4 characters in English and 3 in Portuguese. Mark it as an estimate in the cost_status field.
Is it worth optimizing the largest consumer first? Yes. The Pareto rule applies here: the top of the ranking concentrates the spend. Cutting the largest consumer prompt by 20% saves more than trimming all the others.
What if I have many consumers? Group by service, not by request. The ranking stays useful and the file does not explode. For dozens of services, a centralized gateway helps.
How do I know if the token estimate is right? Compare with the provider tokenizer. For Portuguese, 1 token is about 3 characters. If your estimate is way off, adjust the factor.
Sources
LiteLLM documentation: docs.litellm.ai.
LiteLLM, open source gateway with spend tracking: https://github.com/BerriAI/litellm OpenAI production best practices: https://developers.openai.com/api/docs/guides/production-best-practices
Internal links: see ready-made MCP servers and cron the right way.