Logging Every Model Call

Logging a model call is easy. Logging it so that six weeks later you can answer “why did the bill treble on Tuesday” in one query is a design decision, and it is made when you choose the fields.

Design the log around five questions

These are the questions that actually get asked. Every field in the record below exists because one of them needs it; nothing is captured because it was available.

• Where is the money going? Needs model, token counts and a caller label per call.
• Why is it slow, and for whom? Needs duration, and enough calls to compute p95 rather than a mean.
• What did the model actually see? Needs the resolved prompt, not the template. Reproducing a bad answer from a template plus variables is guesswork.
• Is the failure rate moving? Needs an outcome field with a status code, and retries recorded rather than hidden.
• Which request did this belong to? Needs a correlation id shared with your application logs.

The record

One row per attempt, not per logical call. If a call succeeds on the third try, three rows exist and the third has attempt = 3. Collapsing retries into one row is how a retry storm stays invisible while the bill doubles.

# record.py
from dataclasses import dataclass, asdict, field
from typing import Any


@dataclass
class CallRecord:
    request_id: str            # correlates with your application logs
    call_id: str               # unique per attempt
    attempt: int
    model: str
    caller: str                # "summarise_ticket", "rerank" — the code path
    started_at: float          # unix seconds
    duration_ms: float
    ttft_ms: float | None      # streaming only; None otherwise
    status: str                # "ok" | "http_error" | "timeout" | "parse_error"
    http_status: int | None
    prompt_tokens: int | None
    completion_tokens: int | None
    cost_micros: int | None    # integer millionths, never a float
    cached: bool
    error: str | None
    prompt_sha: str            # sha256 of the resolved messages
    prompt: Any | None         # the messages themselves, if retention allows
    response_text: str | None

    def to_json(self) -> dict:
        return asdict(self)

cost_micros as an integer is not fussiness. Floating-point money accumulated over a million rows drifts, and a per-call cost of 0.000023 is exactly the magnitude where float summation error becomes visible in a monthly total — money in integers is the longer argument.

prompt_sha earns its place separately from prompt. It lets you group identical prompts, measure a cache hit rate and find duplicated work, and it survives a retention policy that deletes the prompt text itself.

The decorator

# instrument.py
import functools
import hashlib
import json
import time
import uuid
from contextvars import ContextVar

import httpx

from record import CallRecord

current_request_id: ContextVar[str] = ContextVar("request_id", default="-")


def prompt_hash(messages: list[dict]) -> str:
    blob = json.dumps(messages, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:32]


def instrumented(caller: str, sink):
    """Wrap f(payload, **kw) -> response body. sink() takes a CallRecord."""

    def decorate(func):
        @functools.wraps(func)
        def wrapper(payload: dict, *args, attempt: int = 1, **kwargs):
            started = time.time()
            clock = time.perf_counter()
            base = dict(
                request_id=current_request_id.get(),
                call_id=uuid.uuid4().hex,
                attempt=attempt,
                model=payload.get("model", "?"),
                caller=caller,
                started_at=started,
                ttft_ms=None,
                cached=False,
                prompt_sha=prompt_hash(payload.get("messages", [])),
                prompt=payload.get("messages"),
            )
            try:
                body = func(payload, *args, **kwargs)
            except httpx.HTTPStatusError as exc:
                sink(CallRecord(
                    **base,
                    duration_ms=(time.perf_counter() - clock) * 1000,
                    status="http_error",
                    http_status=exc.response.status_code,
                    prompt_tokens=None, completion_tokens=None, cost_micros=None,
                    error=exc.response.text[:500], response_text=None,
                ))
                raise
            except httpx.TimeoutException as exc:
                sink(CallRecord(
                    **base,
                    duration_ms=(time.perf_counter() - clock) * 1000,
                    status="timeout", http_status=None,
                    prompt_tokens=None, completion_tokens=None, cost_micros=None,
                    error=f"{type(exc).__name__}: {exc}", response_text=None,
                ))
                raise

            usage = body.get("usage") or {}
            text = body["choices"][0]["message"].get("content")
            sink(CallRecord(
                **base,
                duration_ms=(time.perf_counter() - clock) * 1000,
                status="ok", http_status=200,
                prompt_tokens=usage.get("prompt_tokens"),
                completion_tokens=usage.get("completion_tokens"),
                cost_micros=cost_micros(
                    base["model"], usage.get("prompt_tokens"),
                    usage.get("completion_tokens")),
                error=None,
                response_text=text[:4000] if text else None,
            ))
            return body

        return wrapper

    return decorate

time.perf_counter() for the duration and time.time() for the timestamp, deliberately. perf_counter is monotonic, so an NTP correction or a daylight-saving change cannot produce a negative duration; time.time() is what a human needs to correlate with anything else. Using one for both is a bug that appears twice a year.

The sink is a parameter so that tests can pass a list and production can pass a writer. In development, appending JSON lines to a file is entirely adequate:

import json, threading

_lock = threading.Lock()

def jsonl_sink(path: str):
    def write(record) -> None:
        line = json.dumps(record.to_json(), ensure_ascii=False, default=str)
        with _lock:
            with open(path, "a", encoding="utf-8") as fh:
                fh.write(line + "\n")
    return write


call = instrumented("summarise_ticket", jsonl_sink("llm_calls.jsonl"))(raw_call)

Computing cost without lying about it

Cost is tokens times a price you hold, so the price table is data and it has to be dated. A hard-coded constant that was right in March is a log that is quietly wrong from April onwards, and nobody re-derives it.

# pricing.py
from dataclasses import dataclass


@dataclass(frozen=True)
class Price:
    """Micro-units (millionths of a currency unit) per one million tokens."""
    input_per_mtok: int
    output_per_mtok: int
    effective_from: str        # ISO date; kept so a stale table is visible


# Fill these in from your provider's own pricing page and record the date.
PRICES: dict[str, Price] = {
    # "vendor/model-name": Price(input_per_mtok=..., output_per_mtok=...,
    #                            effective_from="2026-08-04"),
}


def cost_micros(model: str, prompt_tokens: int | None,
                completion_tokens: int | None) -> int | None:
    price = PRICES.get(model)
    if price is None or prompt_tokens is None or completion_tokens is None:
        return None            # unknown, and recorded as unknown
    return round(
        prompt_tokens * price.input_per_mtok / 1_000_000
        + completion_tokens * price.output_per_mtok / 1_000_000
    )

Returning None for an unpriced model is the important behaviour. The alternative — defaulting to zero — makes a newly added model look free, which is precisely the model whose cost you most need to see. A weekly query for cost_micros IS NULL tells you which models have appeared without a price.

Making one record findable

A ContextVar carries the request id through the call stack without threading it through every function signature, and unlike a global it is correct under both threads and asyncio tasks — each task gets its own copy.

# middleware.py (FastAPI)
import uuid

from starlette.middleware.base import BaseHTTPMiddleware

from instrument import current_request_id


class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        request_id = request.headers.get("x-request-id") or uuid.uuid4().hex
        token = current_request_id.set(request_id)
        try:
            response = await call_next(request)
        finally:
            current_request_id.reset(token)
        response.headers["x-request-id"] = request_id
        return response

Returning the id in a response header is what closes the loop: a user reports a bad answer, pastes the id from the UI, and one grep finds every model call behind it. Without that, support tickets are matched by timestamp and guesswork.

The queries

Load the JSONL into DuckDB or SQLite and these five answer the five questions. Having them written down in the repository is what turns a log into a tool.

-- 1. Where the money goes, by code path and model
SELECT caller, model,
       count(*)              AS calls,
       sum(cost_micros)/1e6  AS cost_units,
       sum(completion_tokens) AS out_tokens
FROM calls WHERE status = 'ok'
GROUP BY 1, 2 ORDER BY 3 DESC;

-- 2. Latency, as percentiles rather than a mean
SELECT caller,
       count(*) AS n,
       quantile_cont(duration_ms, 0.50) AS p50,
       quantile_cont(duration_ms, 0.95) AS p95,
       max(duration_ms)                 AS worst
FROM calls WHERE status = 'ok' GROUP BY 1 ORDER BY p95 DESC;

-- 3. Retry rate, which a success-rate dashboard hides completely
SELECT date_trunc('hour', to_timestamp(started_at)) AS hour,
       count(*)                                     AS attempts,
       count(*) FILTER (WHERE attempt > 1)          AS retries,
       100.0 * count(*) FILTER (WHERE attempt > 1) / count(*) AS pct
FROM calls GROUP BY 1 ORDER BY 1 DESC LIMIT 48;

-- 4. Failures, grouped so one bug does not look like forty
SELECT status, http_status, substr(error, 1, 60) AS detail, count(*)
FROM calls WHERE status <> 'ok' GROUP BY 1, 2, 3 ORDER BY 4 DESC;

-- 5. Duplicated work: identical prompts you paid for more than once
SELECT prompt_sha, count(*) AS times, sum(cost_micros)/1e6 AS spent
FROM calls WHERE status = 'ok'
GROUP BY 1 HAVING count(*) > 1 ORDER BY spent DESC LIMIT 20;

Query 5 is the one that pays for the whole exercise on its first run. It is a direct list of calls a cache would have removed, priced — caching model responses is the fix, and this query is how you decide whether it is worth building. Query 2 uses percentiles because a mean latency hides the tail that users actually complain about.

From a file to something you can query

A JSONL file is the right start and stops being enough at a predictable point. DuckDB reads the file directly, with no import step, which covers a surprising amount of ground:

pip install duckdb

duckdb -c "
  CREATE VIEW calls AS SELECT * FROM read_json_auto('llm_calls*.jsonl');
  SELECT caller, count(*), sum(cost_micros)/1e6 AS cost
  FROM calls WHERE status='ok' GROUP BY 1 ORDER BY 3 DESC;
"

Three things eventually force a move to a real table, and it is worth knowing which one you have hit rather than migrating on principle.

• Several processes write at once. The lock in jsonl_sink is per process, so four uvicorn workers appending to one file will interleave partial lines under load. One file per process — include the PID in the filename — solves it without any new infrastructure, and the glob above still reads them all.
• The write is on the request path. Appending to a local file is fast; posting a record to a remote collector synchronously is not, and it adds its own failure mode to every model call. Push records onto a queue.Queue and drain them from a background thread, and drop rather than block when the queue is full — a full telemetry buffer must never take down the thing it is observing.
• Somebody other than you needs the numbers. That is the point at which the table wants to live in your existing database with a retention policy and an index on started_at, or in an observability tool. OpenTelemetry for LLM calls covers emitting the same fields as spans, and observability tools covers the products that consume them.

Whatever the destination, rotate. An unrotated JSONL file with full prompts grows faster than people expect — a few kilobytes per call at a few calls per second is gigabytes a week — and the day it fills the disk is the day the service stops, because the sink raises inside the call path.

What not to put in the log

• The API key. Never log the headers dict. It is one line to forget and it puts a credential in every log aggregator you ship to.
• Raw personal data, without a decision. The prompt is the most useful field and often the most sensitive. Redact before writing, keep prompt_sha unconditionally so grouping still works, and set a retention period — PII in LLM logs and redaction cover the mechanics.
• Whole responses, at full volume. The truncation to 4,000 characters above is deliberate. Log every prompt at low traffic; sample at high traffic, keeping all errors and a percentage of successes — trace sampling is the pattern.
• Anything you would not accept losing. A log is not the record of a business fact. If a classification result matters, it belongs in a table with a foreign key, not in a line somebody may rotate away.