The mental model for LLM call logging is a structured audit trail, not application-level debug prints. Every call to an LLM is a billable, potentially user-visible event with inputs you chose and an output you may not have anticipated. The log record is your evidence that the call happened, what it contained, and what it cost. That record needs to exist whether the call succeeded, timed out, returned a refusal, or was rate-limited.
The minimum viable record has six fields. First, a unique trace ID you generate before the call, not after. This ID ties together multi-step chains, retrieval steps, and tool calls into a single logical transaction. Second, the input payload verbatim: system prompt, all conversation turns, any injected context. Reconstructing prompts from application logic later is unreliable because template logic changes. Third, the raw output including finish_reason ("stop", "length", "content_filter") because a truncated response and a refused response look identical to the user but need different remediation. Fourth, the model version string returned by the API, not the alias you requested. You might request "gpt-4o" but the API will tell you the pinned version actually served, like "gpt-4o-2024-08-06". That distinction matters when OpenAI silently promotes an alias to a new snapshot. Fifth, token counts from the usage object: prompt tokens, completion tokens, and cached tokens if your provider exposes them. Sixth, wall-clock latency measured around the API call, not the full request handler, so you can separate model latency from your own processing overhead.
Cost is a derived field, not a raw one. You compute it: (prompt_tokens * input_price_per_token) + (completion_tokens * output_price_per_token). Store the component values and the price schedule version so you can recalculate if pricing changes. Never hardcode prices in your logger because they change and you will not remember to update them. A reasonable pattern is a small YAML config file versioned in git that maps model strings to their price schedule. On startup, your logger reads it. When OpenAI changes prices, you update the file, re-deploy, and your historical records are still accurate for the prices that were active at that time.
In a real production scenario, consider a customer support bot serving 500 concurrent users. A prompt regression ships on a Friday afternoon. By Monday, CSAT scores are down. Without logs you spend two hours reconstructing what the prompts looked like last week. With logs you write a single SQL query: SELECT input, output, model, timestamp FROM llm_calls WHERE timestamp BETWEEN '2024-11-01' AND '2024-11-04' AND output LIKE '%unable to help%' LIMIT 50. You have the exact prompts that produced bad responses in under a minute. The fix is identifying the system prompt change in the diff, reverting it, and re-deploying. The whole incident takes 90 minutes instead of two days.
The main architectural tradeoff is synchronous versus asynchronous logging. Writing the log record to a database or HTTP endpoint inside the request handler adds latency to every LLM call. At low volume this is fine. At scale, you want to enqueue the record to a background queue (a local asyncio queue, Celery, or a message broker like SQS) and let a separate worker flush it. This decouples your application's response time from the reliability of your logging backend. The failure mode to watch: if your queue fills up and you drop records, you have silent blind spots. Always monitor queue depth and dead-letter any failed records rather than discarding them.
As you go from 10 users to 10k users to 10M users, the logging architecture changes more than the logging schema. At 10 users, writing JSON lines to a file or a Postgres table is fine. At 10k users, you want a time-series-friendly store like ClickHouse, BigQuery, or Loki + Parquet files on S3, because your LLM call table will have 50 columns and you will query it with filters on timestamp, model, user_id, and cost simultaneously. At 10M users, you are generating gigabytes of log data per day and you need partitioning, a retention policy, and probably a separate cold-storage tier for records older than 30 days. The schema you define on day one should anticipate this: store user_id, feature_name, experiment_variant, and environment from the start even if you do not query them yet. Adding columns to a 500GB table at 3am is a bad time.
Key Takeaways
- Log the full input, full output, model version string, token counts, and derived cost on every call.
- Write logs as structured JSON so they are queryable without custom parsers later.
- Never reconstruct prompts from application code after the fact — capture them at call time.
- Assign a unique trace ID per request so you can correlate multi-step chains into one record.
Pro tips
- Always log response.model, not the model string you passed in. Aliases like 'gpt-4o' get silently promoted to new snapshots and your alias logs tell you nothing about which snapshot actually served the response.
- Log finish_reason on every call. A finish_reason of 'length' means the model hit the token limit and stopped mid-sentence. If you are not logging this, you are serving truncated outputs to users and have no way to know how often it happens.
- Store the raw messages array, not a reconstructed summary. Prompt templates change, system prompt injection logic changes, and retrieved context changes. The only version you can trust is what was actually sent over the wire.
- Separate the log write from the return path using an asyncio task or background queue. This prevents a flaky logging backend (a slow Postgres write, a full disk, a network timeout to your observability vendor) from adding latency or errors to the critical path your user sees.
Common pitfalls
- Mistake: Logging the requested model alias instead of the served model version. Fix: Read
response.modelfrom the API response, not the variable you passed to the API call. - Mistake: Omitting system prompts from logs to 'save space'. Fix: Log the full messages array. System prompts change often and are the first thing you need when debugging a regression.
- Mistake: Computing cost from hardcoded per-token prices in application code. Fix: Keep prices in a versioned config file and record both the price schedule version and the token counts so you can recalculate accurately after a price change.
- Mistake: Not logging
finish_reason, so truncated and refused responses look identical. Fix: Always includeresponse.choices[0].finish_reasonin the log record and alert when its value is 'length' or 'content_filter'.
When to use a managed observability platform vs. building your own logging pipeline
| Option | Use when | Avoid when |
|---|---|---|
| Managed platform (Langfuse, Braintrust, Helicone) | Team is small, you want dashboards and evals out of the box, and you can tolerate sending call data to a third party. | Data residency requirements prohibit sending prompts/outputs off-prem, or you need custom schema fields that the platform does not expose. |
| Self-hosted structured logging to ClickHouse or BigQuery | You already have a data warehouse, need full schema control, and have volume above roughly 1M calls/month where vendor per-event pricing gets expensive. | Your team has no data engineering capacity to maintain the pipeline, schema migrations, and retention policies. |
| File-based JSON lines logging | Local development, CI pipelines, or early prototypes where you just need to inspect individual calls. | Production traffic at any meaningful scale. JSON lines on disk do not support concurrent writes, are not queryable, and are lost if the instance restarts. |
| OpenTelemetry spans (OTLP to Grafana, Datadog, etc.) | Your org already has OTel infrastructure and you want LLM calls to appear in the same trace as database queries and HTTP calls. | You need to store and search the full prompt and output text. OTLP span attributes have size limits that make storing multi-kilobyte prompts impractical. |
Code Example
# openai>=1.0.0
import openai, time, json, uuid
client = openai.OpenAI() # reads OPENAI_API_KEY from env
def call_and_log(system: str, user: str, model: str = "gpt-4o-mini") -> dict:
trace_id = str(uuid.uuid4())
t0 = time.perf_counter()
response = client.chat.completions.create(
model=model,
messages=[{"role": "system", "content": system},
{"role": "user", "content": user}],
)
latency_ms = (time.perf_counter() - t0) * 1000
usage = response.usage
record = {
"trace_id": trace_id,
"model": response.model, # exact version from API
"input_tokens": usage.prompt_tokens,
"output_tokens": usage.completion_tokens,
"latency_ms": round(latency_ms, 1),
"input": {"system": system, "user": user},
"output": response.choices[0].message.content,
}
print(json.dumps(record)) # replace with your log sink
return recordHow this code works
This code defines a helper function, call_and_log, designed to interact with Large Language Models (LLMs) and, crucially, to record detailed information about each interaction. Its primary job within the lesson is to standardize the logging of essential observability metrics such as input prompts, LLM responses, the specific model version used, token counts (related to cost), and the time taken for each API call. This systematic logging is fundamental for monitoring LLM usage, analyzing performance, and managing costs over time.
The call_and_log function orchestrates this process. It first generates a unique trace_id using uuid.uuid4() for easy tracking of individual requests. Before making the LLM call, it captures the start time with time.perf_counter(). The core interaction happens via client.chat.completions.create(), sending the system and user prompts to the specified model. A subtle but important detail is the model parameter defaults to "gpt-4o-mini"; if no model is provided, this one is used, which can be crucial for cost and capability planning. After receiving the response, the function calculates latency_ms and extracts usage details, including prompt_tokens and completion_tokens. All these data points, along with the exact response.model (which ensures the precise model version is logged), are then gathered into a record dictionary. Finally, print(json.dumps(record)) outputs this structured log, indicating where a real monitoring system would ingest this data.
Production-grade example
Adds retries with backoff, exact model version logging, per-call cost from versioned price config, and structured logs.
# openai>=1.0.0, structlog>=24.0.0
import asyncio, os, time, uuid, structlog
from decimal import Decimal
from typing import Any
import openai
from openai import AsyncOpenAI, APITimeoutError, RateLimitError, APIStatusError
log = structlog.get_logger()
# Price schedule — update when providers change rates, version in git
PRICE_PER_TOKEN: dict[str, dict[str, Decimal]] = {
"gpt-4o-2024-08-06": {"input": Decimal("0.0000025"), "output": Decimal("0.00001")},
"gpt-4o-mini-2024-07-18": {"input": Decimal("0.00000015"), "output": Decimal("0.0000006")},
}
client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"], timeout=30.0)
async def llm_call(
messages: list[dict],
model: str = "gpt-4o-mini",
feature: str = "unknown",
user_id: str | None = None,
max_retries: int = 3,
) -> str:
trace_id = str(uuid.uuid4())
bound = log.bind(trace_id=trace_id, feature=feature, user_id=user_id, model=model)
for attempt in range(max_retries):
t0 = time.perf_counter()
try:
response = await client.chat.completions.create(
model=model, messages=messages
)
except RateLimitError as exc:
wait = 2 ** attempt
bound.warning("rate_limit", attempt=attempt, wait_s=wait, error=str(exc))
await asyncio.sleep(wait)
continue
except APITimeoutError as exc:
bound.error("timeout", attempt=attempt, error=str(exc))
if attempt == max_retries - 1:
raise
await asyncio.sleep(2 ** attempt)
continue
except APIStatusError as exc:
bound.error("api_error", status=exc.status_code, error=str(exc))
raise
latency_ms = round((time.perf_counter() - t0) * 1000, 1)
usage = response.usage
served_model = response.model # pinned version, not alias
prices = PRICE_PER_TOKEN.get(served_model, {})
cost_usd = (
Decimal(usage.prompt_tokens) * prices.get("input", Decimal(0))
+ Decimal(usage.completion_tokens) * prices.get("output", Decimal(0))
) if prices else None
bound.info(
"llm_call",
served_model=served_model,
input_tokens=usage.prompt_tokens,
output_tokens=usage.completion_tokens,
cached_tokens=getattr(usage, "prompt_tokens_details", {}).get("cached_tokens", 0),
cost_usd=str(cost_usd) if cost_usd is not None else "unknown",
latency_ms=latency_ms,
finish_reason=response.choices[0].finish_reason,
attempt=attempt,
)
return response.choices[0].message.content
raise RuntimeError(f"LLM call failed after {max_retries} attempts")How this code works
This code defines a robust function, llm_call, to interact with large language models (LLMs) like OpenAI's GPT, specifically designed for AI Observability. Its main job is to make an LLM call and then meticulously log crucial details such as the input and output, the exact model version used, and the estimated cost, all while handling common network issues.
The llm_call function uses structlog to bind contextual information like a unique trace_id, feature, and user_id to every log entry, making it easy to trace individual LLM requests. It wraps the actual client.chat.completions.create call in a try...except block, implementing automatic retries with exponential backoff for transient errors like RateLimitError or APITimeoutError. After a successful call, it calculates latency_ms and extracts usage details (like prompt_tokens and completion_tokens). A subtle but important detail is retrieving served_model = response.model from the LLM's response itself, rather than just using the requested model parameter. This ensures the cost calculation, using the PRICE_PER_TOKEN dictionary, is based on the exact model version that processed the request, even if the initial request used a generic alias. Finally, bound.info logs all these collected metrics, including the cost_usd, for comprehensive monitoring.
Practice & master
Try the exercise, check your understanding, then mark this lesson mastered to track your path to pro.
Exercise
Build a logging decorator that wraps any function calling the OpenAI chat completions API. It should capture the messages array, the served model version, token counts, computed cost (use a hardcoded price for gpt-4o-mini: $0.00000015/input token, $0.0000006/output token), latency in milliseconds, and a trace ID. Write each record as a JSON line to a file called llm_calls.jsonl.
# openai>=1.0.0
import json, time, uuid, functools
from decimal import Decimal
from pathlib import Path
from openai import OpenAI
client = OpenAI() # OPENAI_API_KEY from env
LOG_FILE = Path("llm_calls.jsonl")
PRICES = {
# TODO: fill in gpt-4o-mini-2024-07-18 input and output prices
}
def log_llm_call(func):
"""Decorator: wraps a function that returns an OpenAI ChatCompletion."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
trace_id = str(uuid.uuid4())
t0 = time.perf_counter()
response = func(*args, **kwargs)
latency_ms = round((time.perf_counter() - t0) * 1000, 1)
# TODO: extract served_model, prompt_tokens, completion_tokens
# TODO: compute cost_usd using PRICES dict
# TODO: extract the messages array from kwargs or args
# TODO: build the record dict
# TODO: append record as a JSON line to LOG_FILE
return response
return wrapper
@log_llm_call
def chat(messages, model="gpt-4o-mini"):
return client.chat.completions.create(model=model, messages=messages)
if __name__ == "__main__":
chat([{"role": "user", "content": "What is 2 + 2?"}])Quick check
You log the model as 'gpt-4o' in every record. OpenAI silently upgrades the alias to a new snapshot. What is the impact on your logs?
Why should cost be stored as token counts plus a price schedule version rather than just a pre-computed dollar amount?
A response with finish_reason 'length' looks normal to the user but your system is not alerting on it. What is the most likely consequence?