Per-user LLM spend tracking is the control plane that keeps a multi-tenant product from bleeding money on a single noisy neighbor. You cannot allocate costs or enforce quotas if you only see an aggregate provider bill at the end of the month. The fix is to treat every completion as a metered event tied to a user ID from the first line of code, not a spreadsheet task after the invoice arrives.
Step 1: Decide your attribution primitive
Pick one stable identifier per end user: an internal user_id, not an email or session token. If you resell to teams, compose it with tenant_id so you keep both grains. The OpenAI-compatible request shape accepts a user field exactly for this purpose, and most gateways pass it through to upstream providers.
from openai import OpenAI
client = OpenAI(
base_url="https://api.n4n.ai/v1", # OpenAI-compatible, 240+ models
api_key="sk-...",
)
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Summarize this ticket"}],
user="tenant_42:user_881", # stable composite key, set server-side
)The user string lands in the gateway’s logs and, critically, comes back in the response metadata if your gateway meters per token. n4n.ai forwards this field and returns usage in the standard usage object, so you never have to scrape provider CSV exports to answer “how much did user X cost us?”
Never accept the user value from a browser client. Inject it from your auth context after verifying the session. A client that claims user="tenant_1:admin" will happily inflate someone else’s bill.
Step 2: Capture usage at the call site
Do not aggregate token counts in a separate middleware layer that guesses at what happened. Emit a usage event the moment you receive the response. The response shape is deterministic for non-streaming calls:
usage = resp.usage
event = {
"user_key": resp.user if hasattr(resp, "user") else "tenant_42:user_881",
"model": resp.model,
"prompt_tokens": usage.prompt_tokens,
"completion_tokens": usage.completion_tokens,
"total_tokens": usage.total_tokens,
"request_id": resp.id,
"ts": resp.created,
}Streaming changes the mechanics but not the principle. The final chunk carries usage. Collect it after the loop, not per chunk.
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Draft a plan"}],
stream=True,
user="tenant_42:user_881",
)
collected = ""
usage = None
for chunk in stream:
if chunk.usage:
usage = chunk.usage
if chunk.choices[0].delta.content:
collected += chunk.choices[0].delta.content
# only now emit event with `usage`
assert usage is not None, "stream ended without usage"If a stream fails mid-way and you retry, the first attempt’s partial tokens are not returned in a finalized usage object, so they cost you nothing in the ledger. That is correct: you only pay for completed tokens.
Step 3: Write to an append-only ledger
A relational table with a unique constraint on a request identifier is the simplest correct design. Avoid updating rows in place; you want an audit trail.
CREATE TABLE token_ledger (
id BIGSERIAL PRIMARY KEY,
request_id TEXT UNIQUE NOT NULL,
user_key TEXT NOT NULL,
model TEXT NOT NULL,
prompt_tokens INT NOT NULL,
completion_tokens INT NOT NULL,
total_tokens INT NOT NULL,
created_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX ON token_ledger (user_key, created_at);Insert with conflict ignore so retries at the network layer cannot double-count:
import psycopg2
def record_event(conn, event):
with conn.cursor() as cur:
cur.execute(
"""INSERT INTO token_ledger
(request_id, user_key, model, prompt_tokens, completion_tokens, total_tokens, created_at)
VALUES (%s,%s,%s,%s,%s,%s, to_timestamp(%s))
ON CONFLICT (request_id) DO NOTHING""",
(event["request_id"], event["user_key"], event["model"],
event["prompt_tokens"], event["completion_tokens"], event["total_tokens"], event["ts"]),
)
conn.commit()This ledger is the source of truth for per-user LLM spend tracking. Every token your app consumes is an immutable row.
Step 4: Handle fallback and retries without double counting
Automatic fallback when a provider is rate-limited or degraded is mandatory for production. If your gateway retries upstream, you must not emit two ledger rows for one logical call. Two rules keep this safe:
- Generate your own idempotency key and send it as a header your gateway echoes back.
- Use that key—not the provider’s response ID—as
request_idin the ledger.
import uuid
req_id = str(uuid.uuid4())
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "Translate this"}],
user="tenant_42:user_881",
extra_headers={"x-request-id": req_id}, # n4n.ai honors client routing directives
)
event["request_id"] = req_id # not resp.id
record_event(conn, event)If the gateway fails over from Anthropic to OpenAI mid-request, your req_id stays constant. The ledger sees one row. Without this, a retry storm during a provider outage can 10x a user’s apparent spend.
Step 5: Aggregate for billing and quotas
Raw rows are not a dashboard. Roll them up with a materialized view refreshed every minute or five.
CREATE MATERIALIZED VIEW daily_user_tokens AS
SELECT
user_key,
model,
date_trunc('day', created_at) AS day,
sum(prompt_tokens) AS prompt_tokens,
sum(completion_tokens) AS completion_tokens,
sum(total_tokens) AS total_tokens
FROM token_ledger
GROUP BY user_key, model, date_trunc('day', created_at);Convert tokens to currency in application code, never in SQL. Provider prices change; keep a versioned price table.
PRICES = { # per 1K tokens, illustrative only — replace with your contracted rates
"gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006},
}
def compute_cost(model, prompt_tokens, completion_tokens):
p = PRICES.get(model, {"prompt": 0, "completion": 0})
return (prompt_tokens / 1000) * p["prompt"] + (completion_tokens / 1000) * p["completion"]Per-user LLM spend tracking then reduces to a single aggregate query:
SELECT user_key, sum(total_tokens) AS tokens, sum(total_tokens) FILTER (...)
FROM token_ledger
WHERE created_at > now() - interval '30 days'
GROUP BY user_key ORDER BY tokens DESC;Step 6: Enforce limits and alert
Quotas are just a compare against the ledger before you call the model. Check month-to-date usage:
def within_quota(conn, user_key, limit_tokens):
with conn.cursor() as cur:
cur.execute("""SELECT coalesce(sum(total_tokens),0) FROM token_ledger
WHERE user_key=%s AND created_at > date_trunc('month', now())""",
(user_key,))
used = cur.fetchone()[0]
return used < limit_tokensReturn HTTP 429 if false. For alerting, run the same query on a cron and page if a single user_key exceeds a sanity threshold (e.g., 5x median). This catches both bugs and abuse.
Step 7: Verify success
You need three signals in CI and in production:
- Ledger row count equals successful request count (minus intentional skips).
- Sum of
total_tokensper user matches the gateway’s own usage metering filtered by thatuser. - Retried requests appear exactly once.
A pytest snippet that proves dedup:
def test_ledger_dedup():
for i in range(10):
e = make_event(f"req_{i}", user_key="test:user")
record_event(conn, e)
record_event(conn, e) # simulate at-least-once delivery
with conn.cursor() as cur:
cur.execute("SELECT count(*) FROM token_ledger WHERE user_key='test:user'")
assert cur.fetchone()[0] == 10Run it against a throwaway Postgres. If it passes, your per-user LLM spend tracking is correct at the storage layer.
Edge cases that will bite you
Cache hits: providers return prompt_tokens_details.cached_tokens. If your gateway forwards cache-control hints, those tokens are often billed at a discount. Store them in a separate column or you will over-bill.
cached = 0
if usage.prompt_tokens_details and usage.prompt_tokens_details.cached_tokens:
cached = usage.prompt_tokens_details.cached_tokensModel aliases: the same backend can appear as gpt-4o or openai/gpt-4o. Normalize to a canonical string before insert or your aggregates fragment.
Batch endpoints: some providers accept arrays of requests in one call and return an array of usage objects. Loop and emit one ledger row per sub-request with a derived request_id like batch_123:0.
Cross-tenant leakage: always set user from server-side trust. A single misplaced passthrough turns your cost dashboard into a liar.
Per-user LLM spend tracking is not a reporting afterthought you bolt on at Series B. Build the ledger at commit one, and every later feature—billing, quotas, fraud detection—is a SELECT, not a project.