AI Request Monitoring

Ma Mahalakshmi V Updated 19 Sep 2026
8 min read ·Lesson 174 of 224

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.

0 Comments

Reviewed before they appear

No comments yet.

OpenAI SDK
Introduction to the OpenAI SDK Setting Up Python Creating an API Key Your First Call — client.responses.create() and response.output_text Understanding Billing, Credits, and What a Request Costs Why Responses Replaced Chat Completions Anatomy of a Request: model, input, and instructions Anatomy of a Response: The Typed output Array, Not Just Text Roles: User, Assistant, and Developer/System Choosing a Model, and Reading the Models Page Instead of Memorizing Names Instructions vs. Input Writing Prompts That Get Consistent Results Few-Shot Examples Reasoning Models and the reasoning Parameter Debugging a Prompt That Misbehaves Why Streaming Matters for User Experience stream=True and Iterating Over Events Handling the Event Types You Actually Care About Background Mode for Long-Running Jobs Project — Add Live Streaming to Your Chatbot The Problem With Parsing Free Text JSON Schema and Strict Mode Pydantic Models With the SDK's Parse Helpers Handling Refusals and Validation Failures Project — A Resume-to-JSON Extractor Working With input_image input_file, PDFs, and the Files API Image Generation Speech-to-Text and Text-to-Speech Project: A PDF Question-Answering Script What Function Calling Is Defining a Tool Schema The Full Loop Multiple Tools Errors, Timeouts, and Untrusted Arguments Project: A Weather Assistant Web Search File Search and Vector Stores Code Interpreter Remote MCP Servers and Connectors Project: A Research Assistant What an Embedding Is, Without the Maths Generating and Storing Embeddings Similarity Search From Scratch Hosted Vector Stores vs. Rolling Your Own A Small RAG App Over a Folder of Notes Agents vs. a Single API Call — When You Need One pip install openai Giving Agents Tools Handoffs and Multi-Agent Triage Guardrails and Approvals Tracing and Observing What Your Agent Did A Multi-Agent Support Desk Error Codes and What Each One Means Retries, Timeouts, and Backoff Rate Limits and Spend Limits Prompt Caching and Cost Optimisation The Batch API for Bulk Work Async Clients and Concurrency Moderation and Safety Best Practices Designing the App Backend With FastAPI Streaming to a Simple Frontend Deploying and a Cost/Safety Checklist Why Web Search Is Useful for Current Information Using the Web Search Tool with the Responses API Configuring Search Behavior for Application Use Cases Understanding Citations and Source Attribution Where to Go Next Building a Research Assistant with Web Search Combining Web Search with Structured Outputs Handling Conflicting or Low-Quality Web Sources Reducing Unsupported Claims with Grounded Generation Testing Freshness-Sensitive AI Answers Production Considerations for Web-Grounded Applications Understanding File Search and Retrieval-Augmented Generation Creating and Organizing Vector Stores Uploading Documents for Retrieval Connecting Vector Stores to Responses API Requests Designing Document Metadata and Filtering Strategies Building a PDF Question-Answering Application Improving Retrieval Quality With Better Document Preparation Handling Missing Evidence and Retrieval Failures Combining File Search With Web Search Building a Production Knowledge-Base Assistant What the Code Interpreter Tool Is Designed For Running Python-Based Analysis Through the OpenAI SDK Uploading Datasets for Analysis Analyzing CSV and Spreadsheet Data Generating Charts and Data Summaries Handling Generated Files and Downloadable Artifacts Building a Data-Analysis Assistant Combining Code Execution with Structured Outputs Validating Generated Calculations and Results Security and Sandbox Considerations for Code Execution Understanding Multimodal Input with the OpenAI SDK Sending Images to a Model Image Analysis from URLs and Uploaded Files Extracting Text and Information from Screenshots Building an Image-Question-Answering Application Combining Image Input with Structured Output Analyzing Multiple Images in One Request Handling Image Quality and Input Limitations Designing Multimodal Prompts for Reliable Results Building a Practical Vision-Powered Python Application Understanding Speech-to-Text and Text-to-Speech Workflows Transcribing Audio with the OpenAI SDK Working with Uploaded Audio Files Handling Timestamps and Transcription Metadata Building a Meeting Transcription Workflow Generating Spoken Responses from Text Handling Long Audio and Processing Failures Combining Audio with Text and Tool Calling Building an End-to-End Python Voice Application What Embeddings Are and When to Use Them Generating Embeddings With the OpenAI API Preparing Text for Embedding Comparing Vectors With Cosine Similarity Building a Simple Semantic Search Engine in Python Storing Embeddings in a Database Metadata Filtering for Semantic Search Chunking Strategies for Better Retrieval Evaluating Semantic Search Quality Building a Document Similarity Application When Batch Processing Makes Sense Designing Large-Volume AI Processing Pipelines Using Asynchronous Python with the OpenAI SDK Running Concurrent Requests Safely Controlling Concurrency and Avoiding Rate Limits Tracking Batch Job Progress Handling Partial Failures in Bulk Workloads Retrying Failed Items Without Duplicating Successful Work Designing Resumable AI Processing Jobs Building a Production Batch-Processing Pipeline Batch Processing Makes Sense Large-Scale AI Processing Pipelines Async Python with OpenAI SDK Safe Concurrent Requests Concurrency & Rate Limits Batch Progress Tracking Partial Failure Handling Safe Retry Handling Resumable AI Jobs Production Batch Pipeline System–User Data Separation Reusable App Instructions Prompt Templates & Variables Extraction & Classification Prompts Summarization & Transformation Prompts Explicit Output Requirements Prompt Version Management Prompt Testing & Evaluation Reusable Python Prompt Library API Key Security Secure API Key Storage Secure Secret Management Prompt Injection Prevention Trusted vs. Untrusted Content Tool Argument Validation Sensitive Data Handling Secure Logging AI Action Authorization Production AI Security Checklist Why AI Applications Need Evaluation Beyond Unit Tests Unit Testing OpenAI SDK Integration Code Mocking API Responses in Python Tests Testing Structured Outputs Against Schemas Testing Tool-Calling Workflows Building a Small Evaluation Dataset Measuring Accuracy, Consistency, and Failure Rates Regression Testing Prompts and Model Changes Human Evaluation Versus Automated Evaluation Creating a Repeatable Evaluation Pipeline AI Request Monitoring Token Cost Management Usage Metrics Design Reducing Model Calls Prompt & Context Optimization Model Selection & Optimization AI Caching Strategies Interactive Latency Optimization Usage Dashboards & Budget Alerts Performance & Cost Checklist Every API Call Starts Fresh Fixing API Statelessness Server-Side Conversation Memory Limits of Response Chaining What We're Building Conversation Memory Challenges Preparing an OpenAI SDK Application for Deployment Environment-Specific Configuration for Development and Production Deploying a Python AI Service with Docker Container Health Checks and Startup Configuration Managing Secrets in Cloud Deployments Background Workers for Long-Running AI Tasks Queues and Asynchronous Job Architectures Scaling AI Workloads Horizontally Monitoring Production Incidents and Failures Production Deployment Checklist for OpenAI SDK Applications Reusable OpenAI Service Classes AI Client Dependency Injection Typed AI Responses Python Configuration Management AI Request Decorators Centralized AI Error Handling Clean SDK Abstractions Reusable OpenAI Utilities Internal AI Python Libraries SDK Integration Maintenance Production AI Chatbot Document Q&A System Web Research Assistant Customer Support Agent AI Data Analysis Assistant Image Analysis App Meeting Transcription & Summary Semantic Document Search Multi-Tool AI Agent Production OpenAI SDK App Why "It Looked Fine When I Tested It" Isn't Enough Timing Note Status Note Pre-Decision Status Note Current Availability Note
Ask about this post
AI Ask about this post

Ask questions about AI Request Monitoring and get answers drawn from it.

Signed-in readers only.