Log and Monitor AI API Calls in Production

By the end of this guide your deployed script will write one machine-readable line for every call it makes to a language model, and you will have a short pandas script that turns those lines into four numbers: how often calls fail, how slow the slow ones are, how many tokens a typical request burns, and what yesterday cost. Budget about an hour. Nothing here needs a paid service, a sidecar agent, or a rewrite of your application.

The reason to bother is blunt: you cannot debug what you did not record. When a customer says "it was broken yesterday around lunch", the provider dashboard will show you call volume and nothing about which of your users hit which prompt with which model. Your own log line closes that gap.

This guide sits in Deploying Python AI Apps, and it assumes the app is already running somewhere — as a web service, as a scheduled job, or as a worker pulling from a queue.

Prerequisites

You need Python 3.10 or newer, a working API key, and a virtual environment. If any of those is missing, start with Create a Python Virtual Environment for AI and come back.

Beyond whatever your app already uses, this guide needs pandas for the analysis step and httpx for the alert:

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "openai>=1.40" "pandas>=2.2" "httpx>=0.27" "python-dotenv>=1.0"
pip freeze > requirements.txt

Two settings belong in your environment rather than your code, because you will want to change them without a deploy:

OPENAI_API_KEY=sk-your-key-here
LOG_PROMPT_TEXT=false
ALERT_WEBHOOK_URL=https://hooks.example.com/your-endpoint
PRICE_PER_1K_INPUT=0
PRICE_PER_1K_OUTPUT=0

Add .env to .gitignore before you commit anything:

echo ".env" >> .gitignore

Fill the two price values with the current per-thousand-token figures from your provider's pricing page. They change, so read them rather than trusting a number you saw in an article. For the wider question of where secrets should live once you leave your laptop, see Manage API Keys Safely in Production.

Step 1 — Emit one structured JSON line per call

A log line is useful in proportion to how easily a machine can read it. print(f"call took {seconds}s") is readable by you and by nobody else; a JSON object on a single line can be filtered, counted and loaded into a dataframe without writing a parser. That format has a name — JSON Lines, one complete JSON object per line — and every log tool on earth understands it.

Python's standard library already does the plumbing. You only need a custom formatter, which is a small class that decides how a log record turns into text. Save this as ai_logging.py:

import json
import logging
import sys
from datetime import datetime, timezone


class JsonLineFormatter(logging.Formatter):
    """Render each log record as one JSON object on a single line."""

    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.fromtimestamp(record.created, timezone.utc).isoformat(),
            "level": record.levelname,
            "event": record.getMessage(),
        }
        # Anything the caller passed as extra={"fields": {...}} is merged in.
        payload.update(getattr(record, "fields", {}))
        return json.dumps(payload, ensure_ascii=False, default=str)


handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonLineFormatter())

logger = logging.getLogger("ai_calls")
logger.setLevel(logging.INFO)
logger.addHandler(handler)
logger.propagate = False   # stops a second, unformatted copy on the root logger


if __name__ == "__main__":
    logger.info("ai_call", extra={"fields": {"model": "gpt-4o-mini", "outcome": "ok"}})

Run python ai_logging.py and you get exactly one line of JSON. Two details are doing real work. default=str means a stray date or decimal will be turned into text instead of raising an exception halfway through a request. propagate = False stops the record travelling up to Python's root logger, which is the usual cause of every line appearing twice.

Writing to standard output rather than to a file is deliberate. Container platforms, systemd and hosted runtimes all collect standard output for you; if you write to a file inside a container, the file disappears with the container. When you do want a local file for the analysis in Step 4, redirect at the command line: python app.py >> logs/ai_calls.jsonl.

Here is the shape of the whole arrangement, from the call your code makes to the alert that eventually fires.

Step 2 — Wrap the client so logging cannot be forgotten

Logging that depends on discipline decays. Six weeks from now somebody adds a second call site in a hurry, forgets the log statement, and that path becomes invisible exactly when it starts failing. The fix is to make the logged path the only path: one function that calls the model and records the outcome, and a rule that nothing else touches the client directly.

import os
import time
import uuid

from dotenv import load_dotenv
from openai import OpenAI, APIError

from ai_logging import logger

load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


def logged_chat(messages, model="gpt-4o-mini", user_id="anonymous", **kwargs):
    """Call the chat API and record exactly one JSON line, success or failure."""
    request_id = uuid.uuid4().hex[:12]
    started = time.perf_counter()
    fields = {"request_id": request_id, "model": model, "user_id": user_id}

    try:
        response = client.chat.completions.create(model=model, messages=messages, **kwargs)
    except APIError as exc:
        fields.update(
            outcome="error",
            error_type=type(exc).__name__,
            status_code=getattr(exc, "status_code", None),
            latency_ms=round((time.perf_counter() - started) * 1000),
        )
        logger.error("ai_call", extra={"fields": fields})
        raise

    usage = response.usage
    fields.update(
        outcome="ok",
        latency_ms=round((time.perf_counter() - started) * 1000),
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
        finish_reason=response.choices[0].finish_reason,
    )
    logger.info("ai_call", extra={"fields": fields})
    return response


if __name__ == "__main__":
    reply = logged_chat([{"role": "user", "content": "Say hello in five words."}])
    print(reply.choices[0].message.content)

Four choices in there are worth calling out.

The request_id is generated by you, before the call. That means a failed call still has an identifier, which a provider-assigned id cannot give you. Pass the same id into whatever your web framework logs and you can follow one user's click through your own logs and out to the model.

time.perf_counter() is used rather than time.time() because it is monotonic — it cannot jump backwards when the machine syncs its clock, so a latency figure can never come out negative.

The timer starts before the call and is read in both the success and the failure branch. Failed calls are usually the slow ones, and a latency chart that quietly drops them will look healthier than reality.

finish_reason tells you why generation stopped. A rising count of "length" means replies are being truncated by your max_tokens cap, which users experience as sentences that stop mid-word.

For error handling that goes further than recording the failure — backoff, retry budgets, and what to do when the provider rate-limits you — pair this with Fix the 429 Rate-Limit Error in Python and Fix Connection and Timeout Errors with AI APIs.

Step 3 — Redact prompt content by default

Here is the trap. Prompt text is the single most useful thing to have when debugging a bad answer, and the single most dangerous thing to keep. A support-bot prompt contains the customer's message. A document tool's prompt contains the document. Once that lands in a log aggregator, it has been copied to a search index, a backup and probably a laptop, and your retention policy now applies to text you never intended to store.

Default to metrics, not content. Log the shape of the prompt — how long it was, and a hash so you can tell whether two calls used the same input — and keep the text itself behind a switch.

import hashlib
import os

LOG_PROMPT_TEXT = os.getenv("LOG_PROMPT_TEXT", "false").lower() == "true"
SNIPPET_CHARS = 120


def prompt_fields(messages) -> dict:
    """Describe a prompt for the log without leaking what it says."""
    text = "\n".join(
        m["content"] for m in messages if isinstance(m.get("content"), str)
    )
    fields = {
        "prompt_chars": len(text),
        "prompt_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest()[:16],
    }
    if LOG_PROMPT_TEXT:
        fields["prompt_snippet"] = text[:SNIPPET_CHARS]
    return fields

A hash is a fixed-length fingerprint of some text: identical inputs always produce the same fingerprint, and you cannot work backwards from the fingerprint to the text. Sixteen characters of it is plenty to spot "this exact prompt ran four hundred times today" — which is how you discover a retry loop, or a caching opportunity worth taking up in Cache AI Responses in Python to Cut Costs.

Merge it into the wrapper by adding fields.update(prompt_fields(messages)) immediately after fields is created, so both the success and the error line carry it.

When you do need real text, make three commitments in writing before you flip LOG_PROMPT_TEXT to true: which accounts it applies to, how many days you keep it, and who can read it. Then make the deletion automatic, because a retention window nobody enforces is a retention window of forever.

from datetime import datetime, timedelta, timezone
from pathlib import Path

CUTOFF = datetime.now(timezone.utc) - timedelta(days=7)


def prune(path: Path) -> None:
    """Drop log lines older than the retention window, in place."""
    kept = []
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        ts = line.split('"ts": "', 1)[-1].split('"', 1)[0]
        if datetime.fromisoformat(ts) >= CUTOFF:
            kept.append(line)
    path.write_text("\n".join(kept) + "\n", encoding="utf-8")


if __name__ == "__main__":
    prune(Path("logs/ai_calls.jsonl"))

Run that nightly. Schedule Python AI Jobs with GitHub Actions shows one way to put it on a timer without a server.

The decision below is the one you will make for every new field somebody wants to add to the log.

Step 4 — Build the four numbers with pandas

Raw lines are evidence, not insight. Four aggregates cover the overwhelming majority of what goes wrong with a deployed AI feature, and all four come out of the fields you are already writing. Save this one as rollup.py; Step 5 imports from it.

import os

import pandas as pd
from dotenv import load_dotenv

load_dotenv()
PRICE_IN = float(os.getenv("PRICE_PER_1K_INPUT", "0"))
PRICE_OUT = float(os.getenv("PRICE_PER_1K_OUTPUT", "0"))

df = pd.read_json("logs/ai_calls.jsonl", lines=True)
df["ts"] = pd.to_datetime(df["ts"], utc=True)
df["day"] = df["ts"].dt.date

for col in ("input_tokens", "output_tokens", "latency_ms"):
    df[col] = pd.to_numeric(df.get(col), errors="coerce")

df["total_tokens"] = df["input_tokens"].fillna(0) + df["output_tokens"].fillna(0)
df["cost"] = (
    df["input_tokens"].fillna(0) / 1000 * PRICE_IN
    + df["output_tokens"].fillna(0) / 1000 * PRICE_OUT
)

daily = df.groupby("day").agg(
    calls=("request_id", "count"),
    error_rate=("outcome", lambda s: (s == "error").mean()),
    p95_latency_ms=("latency_ms", lambda s: s.quantile(0.95)),
    tokens_per_call=("total_tokens", "mean"),
    spend=("cost", "sum"),
)

print(daily.tail(7).round(3))
print("\nErrors by type:")
print(df[df["outcome"] == "error"]["error_type"].value_counts())
print("\nSlowest models:")
print(df.groupby("model")["latency_ms"].quantile(0.95).sort_values(ascending=False))

p95_latency_ms deserves a sentence of its own. It is the value that 95 percent of calls came in under. Averages flatter you: nineteen fast calls and one forty-second call average out to something respectable while one user in twenty stares at a spinner. The p95 shows that user. Watch it instead of the mean, and watch it per model — the last line of the script does that — because a single slow model can drag a whole product down while every other code path is fine.

The spend column is only as honest as the two price values in your .env, and per-token prices change, so re-read them when you notice the arithmetic drifting from your invoice. For a more careful treatment of turning token counts into money, see Estimate OpenAI API Costs with Python; to shrink the input side before you ever send it, Count Tokens in Python Before You Send is the companion piece.

Run this against a file with a hundred lines in it and the output already tells you something. Run it every morning and the week-on-week shape tells you much more.

Step 5 — Alert when a threshold trips

A dashboard nobody opens is a diary, not a monitor. Add a short script that compares yesterday's numbers to limits you chose and pushes a message when one is breached. Any endpoint that accepts a JSON body works — a chat webhook, an incident tool, or a tiny service of your own.

import os

import httpx
import pandas as pd
from dotenv import load_dotenv

load_dotenv()

THRESHOLDS = {          # starting points only — tune each to what your product tolerates
    "error_rate": 0.02,
    "p95_latency_ms": 8000,
    "spend": 25.0,
}


def check(daily: pd.DataFrame) -> list[str]:
    latest = daily.iloc[-1]
    return [
        f"{name} is {latest[name]:.3f} (limit {limit})"
        for name, limit in THRESHOLDS.items()
        if name in latest and latest[name] > limit
    ]


def notify(breaches: list[str]) -> None:
    url = os.getenv("ALERT_WEBHOOK_URL")
    if not url or not breaches:
        return
    httpx.post(url, json={"text": "AI API alert: " + "; ".join(breaches)}, timeout=10)


if __name__ == "__main__":
    # `daily` is the dataframe built in Step 4.
    from rollup import daily

    problems = check(daily)
    for problem in problems:
        print("BREACH:", problem)
    notify(problems)

Set the limits from your own baseline rather than from a round number that feels serious. Run the rollup for a normal week, look at the numbers, and set each threshold somewhere your ordinary days sit comfortably below. A limit that fires twice a week trains everyone to ignore it, and an alert that everyone ignores is worse than none, because it feels like coverage.

Two thresholds pay for themselves fastest. A spend limit catches a runaway loop the same day rather than at the end of the month — pair it with a hard ceiling as described in Set a Monthly AI API Spending Limit. An error-rate limit catches a bad deploy or a provider incident while you can still roll back.

Field reference

These are the fields worth writing on every line, and what each one buys you later.

Troubleshooting

ValueError: Trailing data when calling pd.read_json. You omitted lines=True, so pandas tried to read the whole file as one JSON document and choked on the second line. Add lines=True.

KeyError: 'input_tokens' in the rollup. Every row in that file is an error row, so the token fields were never written. The df.get(col) calls in Step 4 already guard against this; if you removed them, restore them or add df = df.reindex(columns=[...], fill_value=None) before aggregating.

AttributeError: 'NoneType' object has no attribute 'prompt_tokens'. You passed stream=True, and a streamed response carries no usage object unless you ask for it. Either drop streaming in the wrapper or guard with usage = getattr(response, "usage", None) and log None token counts.

Every log line appears twice. Another library, or your web framework, called logging.basicConfig() and attached a handler to the root logger. logger.propagate = False in ai_logging.py is the fix; confirm it did not get lost in a refactor.

Log lines are valid JSON but pandas reads ts as text. Some rows were written before you added the formatter and use a different timestamp shape. pd.to_datetime(df["ts"], utc=True, errors="coerce") will null the stragglers instead of failing the whole run.

When to use this vs. alternatives

• Use JSON lines plus pandas when one team owns the app, the volume is thousands of calls a day rather than millions, and you want zero new vendors. It is the cheapest thing that answers the four questions that matter, and it runs anywhere Python runs.
• Use a hosted observability platform when several people need live dashboards, when you want paging with on-call rotations, or when you need to trace one request across multiple services. You get search and retention policies for free; you pay per gigabyte ingested, so keep redacting prompts either way.
• Use your provider's usage dashboard as a cross-check on spend, never as your monitor. It knows nothing about your users, your request ids, or which feature made the call, and it typically lags by hours.

If your app is a web service, log the request id at the edge as well so a single identifier ties the HTTP request to the model call — Turn a Python AI Script into an API with FastAPI shows where that hook goes. And once the logs show which accounts drive volume, Rate-Limit AI API Calls in a SaaS with Python turns that knowledge into a limit.

Start with the wrapper from Step 2 and nothing else. One function, one JSON line, standard output. You will have useful data by tomorrow, and the rollup and the alert are twenty minutes of work each once the lines exist. Back to Deploying Python AI Apps.

Related guides

• Manage API Keys Safely in Production — keep the credentials out of the logs you just started writing.
• Turn a Python AI Script into an API with FastAPI — where to attach the request id in a web service.
• Estimate OpenAI API Costs with Python — turn the token counts in your logs into a reliable money figure.
• Fix the 429 Rate-Limit Error in Python — what to do about the error type your logs will surface most often.
• Schedule Python AI Jobs with GitHub Actions — run the rollup, the pruner and the alert on a timer.