To monitor LLM calls in a Python app, record the same few things for every call: how long it took (and, for streamed answers, how long until the first token arrived), the input and output tokens the provider reports, what the call cost, whether it failed or was retried, which model served it, and which user request it belonged to. The OpenAI and Anthropic Python SDKs already return most of this on every response. The work is capturing it the same way everywhere and keeping it next to the rest of the request.
This guide covers what goes wrong with LLM calls in production, how to capture each of these signals in plain Python, and how the Atatus Python agent records most of them without a wrapper around each call. The code uses the official openai and anthropic packages, but the same approach works for any provider.
Table of Contents
- What goes wrong with LLM calls in production
- What to record for every LLM call
- How to capture latency, tokens and errors in Python
- How to measure time to first token
- How to work out the cost of each call
- Retries, timeouts and rate limits
- Chains and agents: several calls per request
- Should you store prompts and responses?
- How to monitor LLM calls in Python with Atatus
- An LLM monitoring checklist
- Frequently asked questions
What goes wrong with LLM calls in production
An LLM call is an external API request, so everything that applies to monitoring external requests applies here too. A few things make it harder than a typical API call:
- Long and uneven response times. The model writes its answer token by token, so long answers take longer, and the same endpoint can be fast for one prompt and slow for the next. With streaming, users notice the wait for the first words most.
- Cost that changes without a code change. You pay per token. A longer prompt template, more retrieved documents or a growing chat history raises the cost of every call, and nothing errors.
- Rate limits and overloads. Providers return HTTP 429 when you go over your rate limits, and Anthropic returns 529 when its API is overloaded. Both SDKs retry these for you, so often the only sign is a slower request.
- Long waits before a timeout. Both SDKs default to a 10-minute timeout and retry timeouts too, so a hung call can hold a user request for a long time.
- Failures that return HTTP 200. An answer cut off at the token limit, a refusal, or output that isn't the JSON you asked for all arrive as a successful response.
- Several calls per request. A retrieval or agent flow can make embedding, search, model and tool calls for one question. When it is slow, you need to know which step.
What to record for every LLM call
- Duration, measured around the SDK call. For streamed calls, also the time to the first token.
- Model and provider. Keep the model you asked for and the model the response reports, which can be a specific snapshot.
- Tokens. Input and output tokens from the response's
usage, plus cached tokens, which are billed at a different rate, and reasoning tokens where the provider reports them. - Cost, worked out from the tokens and your price table.
- Outcome. Success or the error type, the HTTP status code, and the finish reason (
finish_reasonfor OpenAI Chat Completions,stop_reasonfor Anthropic). - The provider's request ID, for when you need the provider's support.
- Context from your app: the feature that made the call, and the request or trace ID that ties it to the rest of the request.
User and customer IDs belong on log lines and trace attributes, where you can search by them, not in metric labels, where every distinct value creates a new time series. High cardinality explains why.
How to capture latency, tokens and errors in Python
A small wrapper around the SDK call is enough to start. This one times an OpenAI Chat Completions call, reads the token counts from response.usage, records the error type and status code when the call fails, and writes one JSON log line either way:
import json
import logging
import time
import openai
from openai import OpenAI
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("llm")
client = OpenAI(timeout=30.0, max_retries=2)
def chat(messages, model, feature):
record = {"provider": "openai", "model": model, "feature": feature}
start = time.perf_counter()
try:
response = client.chat.completions.create(model=model, messages=messages)
except openai.APIStatusError as e: # 4xx and 5xx, including RateLimitError (429)
record.update(outcome="error", error=type(e).__name__,
status=e.status_code, request_id=e.request_id)
raise
except openai.APIConnectionError as e: # network errors, and APITimeoutError
record.update(outcome="error", error=type(e).__name__)
raise
else:
usage = response.usage # can be None on OpenAI-compatible gateways
details = usage.prompt_tokens_details if usage else None
record.update(
outcome="ok",
response_model=response.model,
finish_reason=response.choices[0].finish_reason, # "length" = cut off
input_tokens=usage.prompt_tokens if usage else None,
cached_tokens=details.cached_tokens if details else None,
output_tokens=usage.completion_tokens if usage else None,
request_id=response._request_id,
)
return response
finally:
record["duration_ms"] = round((time.perf_counter() - start) * 1000)
log.info(json.dumps(record))The same wrapper for Anthropic reads slightly different fields. message.usage reports uncached input tokens, cache reads and cache writes separately, and stop_reason is "max_tokens" when the answer was cut off and "refusal" when the model declined:
from anthropic import Anthropic
anthropic_client = Anthropic(timeout=30.0, max_retries=2)
def chat_anthropic(messages, model, feature):
record = {"provider": "anthropic", "model": model, "feature": feature}
start = time.perf_counter()
message = anthropic_client.messages.create(
model=model, max_tokens=1024, messages=messages
)
usage = message.usage
record.update(
outcome="ok",
duration_ms=round((time.perf_counter() - start) * 1000),
response_model=message.model,
stop_reason=message.stop_reason,
input_tokens=usage.input_tokens, # uncached input only
cache_read_tokens=usage.cache_read_input_tokens or 0,
cache_write_tokens=usage.cache_creation_input_tokens or 0,
output_tokens=usage.output_tokens,
request_id=message._request_id,
)
log.info(json.dumps(record))
return messageErrors follow the same pattern, so wrap the call in the same try block after import anthropic: anthropic.APIStatusError carries status_code and request_id, anthropic.RateLimitError is the 429 case, and anthropic.APIConnectionError covers network errors and timeouts. The full list of error classes is in the OpenAI Python library README and the Anthropic Python SDK docs.
Two details matter when you add tokens up. OpenAI counts cached tokens inside prompt_tokens and also reports them in prompt_tokens_details.cached_tokens, while Anthropic's input_tokens leaves them out: total input is input_tokens + cache_read_input_tokens + cache_creation_input_tokens, treating either cache field as 0 when it is None. And on OpenAI reasoning models, the hidden reasoning tokens are part of the output count (completion_tokens_details.reasoning_tokens shows how many) and are billed as output tokens.
How to measure time to first token
When you stream the answer, measure two durations: from the request to the first piece of text, which is the wait the user sees, and from the request to the end of the stream, which is how long the call holds your worker.
Streaming also changes how you get token counts. A streamed OpenAI chat completion carries no usage unless you ask for it with stream_options={"include_usage": True}. The usage then arrives in one extra chunk at the end, with an empty choices list:
def stream_chat(messages, model):
start = time.perf_counter()
first_token_at = None
usage = None
parts = []
stream = client.chat.completions.create(
model=model,
messages=messages,
stream=True,
stream_options={"include_usage": True},
)
for chunk in stream:
if chunk.usage: # only the last chunk carries usage
usage = chunk.usage
if chunk.choices and chunk.choices[0].delta.content:
if first_token_at is None:
first_token_at = time.perf_counter()
parts.append(chunk.choices[0].delta.content)
end = time.perf_counter()
ttft_ms = round((first_token_at - start) * 1000) if first_token_at else None
total_ms = round((end - start) * 1000)
return "".join(parts), usage, ttft_ms, total_msWith Anthropic, pass stream=True and read the events. Text arrives in content_block_delta events, input tokens in message_start and output tokens in message_delta:
def stream_anthropic(messages, model):
start = time.perf_counter()
first_token_at = None
input_tokens = output_tokens = 0
parts = []
stream = anthropic_client.messages.create(
model=model, max_tokens=1024, messages=messages, stream=True
)
for event in stream:
if event.type == "message_start":
input_tokens = event.message.usage.input_tokens # uncached input only
elif event.type == "message_delta":
output_tokens = event.usage.output_tokens
elif event.type == "content_block_delta" and event.delta.type == "text_delta":
if first_token_at is None:
first_token_at = time.perf_counter()
parts.append(event.delta.text)
end = time.perf_counter()
ttft_ms = round((first_token_at - start) * 1000) if first_token_at else None
total_ms = round((end - start) * 1000)
return "".join(parts), input_tokens, output_tokens, ttft_ms, total_msThe SDK also has a messages.stream helper, but the Atatus agent doesn't support it yet, as the Atatus section explains.
How to work out the cost of each call
The cost of a call is each kind of token multiplied by its price, added up. Providers charge different rates for input and output tokens, and often different rates again for tokens read from or written to a prompt cache. Prices also change, so keep them in one table that you update from the provider's pricing page (OpenAI, Anthropic) instead of hard-coding them in each call site:
# USD per million tokens, copied from the provider's pricing page.
# Key by the model name you pass in the request. Add "cache_write" for
# models that bill cache writes, such as OpenAI's GPT-5.6 and later.
PRICES = {
# "model-name": {"input": ..., "cached_input": ..., "cache_write": ..., "output": ...},
}
def openai_cost(model, usage):
price = PRICES[model]
details = usage.prompt_tokens_details
cached = (details.cached_tokens or 0) if details else 0
written = (details.cache_write_tokens or 0) if details else 0
uncached = usage.prompt_tokens - cached - written
return (
uncached * price["input"]
+ cached * price["cached_input"]
+ written * price.get("cache_write", price["input"])
+ usage.completion_tokens * price["output"]
) / 1_000_000Put the cost on the same log line as the tokens, so you can total it by feature, model and customer, not only by day. When cost per request jumps after a deploy, compare input tokens per request before and after to find the prompt or context change behind it.
Retries, timeouts and rate limits
Both SDKs retry connection errors, 408, 409, 429 and 5xx responses twice by default, with exponential backoff of up to 8 seconds, or the wait the API asks for in Retry-After. The retries happen inside the one call you timed, so a slow call may be one slow response or three attempts. A few settings and habits keep this visible:
- Set the timeout yourself. The default is 10 minutes, and timed-out requests are retried. Pass
timeoutwhen you create the client, or per call withclient.with_options(timeout=...), based on how long your users can wait. It limits each connect or read, not the whole call, and each retry starts it again, so the worst case is roughlytimeout × (max_retries + 1)plus backoff. - Set
max_retrieson purpose. Retries help with short rate-limit spikes, but when a user is waiting, a quick clear error can be better than a long wait. - Log the retries. Both SDKs log each retry at INFO level through Python's standard
loggingmodule, on theopenaiandanthropicloggers. Set those loggers to INFO, with a handler configured, so retries show up in your logs:logging.basicConfig() # or your own handlers logging.getLogger("openai").setLevel(logging.INFO) logging.getLogger("anthropic").setLevel(logging.INFO) - Count rate limits by model. A steady rise in 429s means you are near your limit for that model, and the retries are adding latency for your users.
- Count the 200s that failed. Treat
finish_reason == "length",stop_reason in ("max_tokens", "model_context_window_exceeded", "refusal")and output that fails your JSON parsing as errors. Otherwise the error rate looks healthy while users get broken answers.
Chains and agents: several calls per request
When one user request makes several LLM calls, you need to see them together: which step took the time, how many calls there were, and what the whole request cost.
- Attach the request or trace ID to every call. Then you can add up tokens and cost per request and put the calls in order. A trace does this for you: each call is a span inside the request, next to its database queries and HTTP calls. Logs, traces and metrics covers how the three fit together.
- Count LLM calls per request. An agent that repeats the same tool call, or a retry loop around a parser, shows up as a request with far more calls than the flow needs.
- Watch input tokens per turn. With Chat Completions and the Anthropic Messages API, a chat sends its whole history with every turn, so input tokens grow as the conversation does. Trimming or summarizing old turns keeps that growth in check.
Should you store prompts and responses?
Prompt and response text is the most useful data when you debug a bad answer, and the most sensitive. It holds whatever your users typed, plus the documents and records you put into the context. Decide before you turn it on:
- Do you need the text for every call, or only for errors and a sample of the rest?
- Can you mask personal data, such as emails, phone numbers and IDs, before it is stored?
- Who can read it, and how long is it kept?
Token counts, timing, model and errors carry no prompt text, so you can always record those even when you leave the content out.
How to monitor LLM calls in Python with Atatus
For your own app's calls, LLM observability in Atatus gets its data from the Python agent, which records each call without a wrapper in your code. LLM monitoring was added in Python agent 2.1.0.
- Install the agent with the
llmobsextra. The extra adds the OpenTelemetry-based instrumentation for the LLM libraries:pip install "atatus[llmobs]" - Turn on LLM monitoring and tracing. Both are off by default. The agent sends LLM calls to Atatus as trace spans, so set
TRACINGas well asLLMOBS, or the LLM pages stay empty. In FastAPI:
The same keys work in the# main.py from fastapi import FastAPI import atatus from atatus.contrib.starlette import create_client, Atatus # Initialize Atatus before app = FastAPI() atatus_client = atatus.get_client() if atatus_client is None: atatus_client = create_client({ "APP_NAME": "support-assistant", "LICENSE_KEY": "lic_apm_xxxxxxx", "TRACING": True, "LLMOBS": True, }) app = FastAPI() app.add_middleware(Atatus, client=atatus_client)ATATUSsettings dict in Django and inapp.config["ATATUS"]in Flask, or you can set the environment variablesATATUS_TRACING=trueandATATUS_LLMOBS=true. The Python agent docs cover the agent setup for each framework. - Restart the app and send a few requests.
The agent instruments the supported libraries that are installed in your app. That covers the openai and anthropic SDKs, Amazon Bedrock and SageMaker calls made with boto3, Vertex AI through google-cloud-aiplatform, Groq, Ollama, Replicate and Hugging Face Transformers. It also covers LangChain and LangGraph, LlamaIndex, CrewAI and the MCP Python SDK, and the vector databases Chroma, Milvus, LanceDB and Marqo. Cohere below version 6, Mistral AI below version 1, Together AI below version 2 and the older ibm-watson-machine-learning package are covered too, but newer versions of those libraries are skipped without an error, so check that their calls appear. Calls made while your app handles a web request are recorded as spans in that request's trace.
A few limits apply to Anthropic and streaming today. The agent doesn't support Anthropic's messages.stream helper: with LLM monitoring on it raises AttributeError, so use messages.create(stream=True) as shown above. Streamed Anthropic calls carry no token counts, failed Anthropic calls don't appear on the LLM pages, and OpenAI streams report tokens only with include_usage.
Open LLM in the Atatus menu to see the data:
- Overview lists each project with its LLM Trace Count, Error Rate, Input Tokens, Output Tokens, Total Tokens and Total Cost.
- Explorer shows the Trace Error Rate, Trace Duration (p95), Estimated Cost, Token Usage and LLM Calls for the selected time range, with a cost chart, a latency chart (average, P75, P90 and P95), traces over time, and errors by error type. You can narrow it down by project, model provider and model name.
- LLM Traces lists the calls with their input and output, duration, total tokens and kind, such as chat or completion. You can filter by model provider, model name, status, status code, service and more. Opening a call shows its model, provider, total tokens and total cost, with the input and output messages and the cost of each.
Prompt and response text is recorded by default. To keep the text out of Atatus, set the environment variable TRACELOOP_TRACE_CONTENT=false in the app's environment. The LLM provider, LangChain and LlamaIndex instrumentations read this variable. The CrewAI, MCP and vector database instrumentations don't, so their inputs and results are still sent. Token counts, model and duration are still recorded.
An LLM monitoring checklist
- Every LLM call records its duration, model, provider, input and output tokens, outcome and the provider's request ID.
- Streamed calls record time to first token and total duration, and OpenAI streams use
include_usage. - Cost is worked out from tokens, cached tokens included, with a price table you keep up to date.
- Timeouts and
max_retriesare set on purpose, and retries show up in your logs. - 429s, 5xx errors and timeouts are counted by model.
- Cut-off answers, refusals and unparsable output count as failures.
- Each call carries the request or trace ID, so you can see calls, tokens and cost per request.
- You have decided whether to store prompt and response text, who can read it and how long it is kept.
Frequently asked questions
How do I track OpenAI token usage in Python?
Read it from the response. A Chat Completions response has response.usage.prompt_tokens, completion_tokens and total_tokens, and a Responses API response has response.usage.input_tokens, output_tokens and total_tokens. Log them with the model and the feature that made the call, so you can total usage by feature, model and customer.
Why is usage missing when I stream OpenAI responses?
Streamed chat completions don't include usage unless you ask for it. Pass stream_options={"include_usage": True}, and the usage arrives in a final chunk with an empty choices list. Check chunk.usage on each chunk and keep the one that isn't empty.
How do I calculate the cost of an LLM API call?
Multiply each kind of token by its price and add them up: uncached input, cached input, cache writes where they apply, and output, which includes reasoning tokens on OpenAI reasoning models. Take the prices from the provider's pricing page and keep them in one table, because they differ by model and change over time.
How do I measure time to first token in Python?
Record the time before you send a streaming request and again when the first chunk with text arrives. The difference is the time to first token. Keep the time at the end of the stream too, for the total duration.
Is it safe to log LLM prompts and responses?
Only if you treat them as user data. Prompts can contain personal data and the documents you put in the context, so mask what you can, limit who can read it and keep it no longer than you need. Tokens, timing, model and errors can be recorded without the text.
Atatus records the LLM calls your Python services make with their model, tokens, estimated cost and duration, alongside the traces of the requests that made them. See LLM observability for the full view, and Python performance monitoring for the rest of the app.
