AI Request Monitoring
Tracking requests, latency, errors, and token usage
Unit 12, Lesson 4 introduced prompt caching and a few cost levers. This unit builds a full observability and cost-management practice around an application, starting with the foundation everything else depends on: knowing exactly what your application is doing every time it calls a model.
Why Observability Comes Before Optimization
You cannot reduce cost, improve latency, or debug failures in a system you cannot see into. Every optimization technique covered later in this unit — caching, model selection, prompt trimming — requires a baseline measurement to know whether the change helped. Without structured tracking, teams end up guessing: "it feels slower today" or "the bill went up but we don't know why." A logging layer around every model call turns those guesses into answerable questions.
The four signals that matter most for an LLM-backed application are:
- Requests — how many calls are made, to which model, for which feature.
- Latency — how long each call takes, end to end and broken into phases.
- Errors — what fails, how often, and why.
- Token usage — how many input and output tokens each call consumes, which drives cost directly.
Tracking all four together, per request, is what makes later analysis possible. If you only log token counts, you can compute cost but not diagnose why a particular feature is slow. If you only log latency, you can see slowness but not correlate it with a token spike caused by an oversized prompt.
Structuring a Request Log Record
Rather than scattering print statements around your code, define a single structured record that is populated once per model call and written to a log sink. This gives every call a consistent shape that downstream tools (dashboards, alerting, cost reports) can rely on.
import time
import uuid
from dataclasses import dataclass, field, asdict
from typing import Optional
@dataclass
class RequestLog:
request_id: str
feature: str
model: str
started_at: float
finished_at: Optional[float] = None
input_tokens: Optional[int] = None
output_tokens: Optional[int] = None
status: str = "in_progress"
error_type: Optional[str] = None
error_message: Optional[str] = None
@property
def latency_ms(self) -> Optional[float]:
if self.finished_at is None:
return None
return (self.finished_at - self.started_at) * 1000
def to_dict(self) -> dict:
d = asdict(self)
d["latency_ms"] = self.latency_ms
return d
def new_request_log(feature: str, model: str) -> RequestLog:
return RequestLog(
request_id=str(uuid.uuid4()),
feature=feature,
model=model,
started_at=time.time(),
)
This example does three things worth calling out. First, request_id is generated with uuid.uuid4() rather than left implicit, because every downstream system — logs, traces, support tickets — needs a stable identifier to correlate a single user-facing action with the model call(s) it triggered. Second, feature records which part of the application made the call (for example "summarize_ticket" or "generate_reply"), which is essential later for per-feature cost attribution — a topic covered in Lesson 3. Third, latency_ms is a computed property rather than a stored field, so it is always consistent with started_at and finished_at and cannot drift out of sync if one field is updated without the other.
Wrapping the Model Call
The log record is only useful if it is populated consistently. The cleanest way to guarantee that is to wrap every model call in a helper function that always fills in the record, whether the call succeeds or fails.
from openai import OpenAI, APIError
client = OpenAI()
def call_model_with_logging(feature: str, messages: list[dict], model: str = "gpt-5.6-terra") -> tuple[str, RequestLog]:
log = new_request_log(feature=feature, model=model)
try:
response = client.chat.completions.create(
model=model,
messages=messages,
)
log.input_tokens = response.usage.prompt_tokens
log.output_tokens = response.usage.completion_tokens
log.status = "success"
return response.choices[0].message.content, log
except APIError as exc:
log.status = "error"
log.error_type = type(exc).__name__
log.error_message = str(exc)
raise
finally:
log.finished_at = time.time()
emit_log(log)
def emit_log(log: RequestLog) -> None:
# In production this would write to a log aggregator (e.g. structured
# JSON to stdout for collection by a log pipeline). For now, print.
print(log.to_dict())
Note: The exact attribute names on
response.usage(prompt_tokens,completion_tokens,total_tokens) reflect the OpenAI SDK's usage object at the time of writing. Confirm these field names against the SDK version you have installed, since usage object shapes have changed across SDK versions and may again.
The try/except/finally structure is deliberate, not incidental. The except block captures failures so that even an error produces a complete, queryable log record — this is what lets you later compute an error rate per feature or per model. The finally block guarantees finished_at is always set and the log is always emitted, regardless of whether the call succeeded, failed, or raised an unexpected exception. If you instead set finished_at only in the success path, every failed call would have latency_ms == None, silently corrupting your latency dashboards by dropping exactly the requests you most need to see (slow calls are more likely to time out and fail).
Note also that the function re-raises the exception after logging it. Logging should never swallow errors — the caller still needs to know the call failed so it can retry, fall back, or surface an error to the user. Observability code should be a transparent layer around your logic, not a replacement for proper error handling.
Capturing Latency Phases, Not Just Totals
A single end-to-end latency number tells you that something is slow but not where. In a real application, the time between "user submits a request" and "response is displayed" is made up of several phases: building the prompt, waiting on the network, waiting on the model to generate tokens, and post-processing the response. Splitting these apart is what lets you tell whether a slowdown is your code or the model.
def call_model_with_phase_timing(feature: str, messages: list[dict], model: str = "gpt-5.6-terra") -> dict:
t0 = time.time()
# Phase 1: prompt construction (placeholder — replace with real work)
prompt_built_at = time.time()
response = client.chat.completions.create(model=model, messages=messages)
response_received_at = time.time()
# Phase 3: post-processing (placeholder — replace with real work)
processed_at = time.time()
return {
"prompt_build_ms": (prompt_built_at - t0) * 1000,
"model_call_ms": (response_received_at - prompt_built_at) * 1000,
"post_process_ms": (processed_at - response_received_at) * 1000,
"total_ms": (processed_at - t0) * 1000,
}
In practice, model_call_ms usually dominates, but when prompt_build_ms is unexpectedly large (for example, because it involves a slow database query to assemble context), phase timing is the only way to notice that your own code, not the model, is the bottleneck. This distinction directly informs which optimization technique applies: a slow model call is addressed with the model-selection and caching techniques in later lessons, while a slow prompt-build phase is addressed with ordinary application performance work — indexing a database, caching a lookup, parallelizing independent fetches.
Testing the Logging Layer Without Calling the API
Because call_model_with_logging depends on an external client, it should be tested with a fake client rather than a real API call. This keeps tests fast, deterministic, and free.
class FakeUsage:
def __init__(self, prompt_tokens: int, completion_tokens: int):
self.prompt_tokens = prompt_tokens
self.completion_tokens = completion_tokens
class FakeMessage:
def __init__(self, content: str):
self.content = content
class FakeChoice:
def __init__(self, content: str):
self.message = FakeMessage(content)
class FakeResponse:
def __init__(self, content: str, prompt_tokens: int, completion_tokens: int):
self.choices = [FakeChoice(content)]
self.usage = FakeUsage(prompt_tokens, completion_tokens)
class FakeCompletions:
def __init__(self, response: FakeResponse):
self._response = response
def create(self, model: str, messages: list[dict]):
return self._response
def test_log_records_token_usage():
fake_response = FakeResponse("hello there", prompt_tokens=12, completion_tokens=4)
global client
original_client = client
class FakeClient:
def __init__(self):
self.chat = type("Chat", (), {"completions": FakeCompletions(fake_response)})()
client = FakeClient()
try:
content, log = call_model_with_logging("greeting", [{"role": "user", "content": "hi"}])
assert content == "hello there"
assert log.input_tokens == 12
assert log.output_tokens == 4
assert log.status == "success"
assert log.latency_ms is not None
print("PASS: log records token usage and success status")
finally:
client = original_client
test_log_records_token_usage()
The fake objects (FakeUsage, FakeMessage, FakeChoice, FakeResponse, FakeCompletions) mirror only the shape of the real SDK objects that call_model_with_logging actually touches — nothing more. This is the dependency-injection pattern: instead of mocking library internals, you substitute an object that satisfies the same interface your code depends on. The test then asserts on the observable outcome (the returned content and the populated log fields) rather than on implementation details, which means the test keeps working even if the internals of call_model_with_logging change, as long as its contract does not.
Common Mistakes
Logging only on success. If error paths don't produce a log record, your error rate looks artificially low and your latency numbers are biased toward fast, successful calls — exactly the opposite of what you need when debugging a slowdown or an outage.
Storing raw prompt and response text in logs by default. Full request and response bodies are useful for debugging but often contain user data. Log token counts, timing, and metadata by default, and only capture full content behind an explicit, access-controlled debug flag.
Using wall-clock print statements instead of structured records. Unstructured text logs cannot be aggregated, filtered by feature, or queried for percentiles. A structured record (a dict, or a well-defined class) that maps directly to JSON is what allows any log aggregation tool to compute rates, sums, and percentiles later.
Best Practices
Give every request a unique ID and propagate it. If a user-facing action triggers multiple model calls (for example, a retrieval step followed by a generation step), tag all of them with the same request_id or a shared trace_id so you can reconstruct the full chain later.
Always record which model served the request. Model identifiers change over time as you experiment or as providers deprecate versions. Without this field, a cost or latency shift after a model change is invisible in your data.
Emit logs asynchronously where possible. Writing a structured log to stdout or a lightweight queue should not add meaningful latency to the user-facing request; avoid synchronous writes to slow external logging services on the critical path.