Unstructured logs (“LLM call failed”) are useless in production. Structured logs with consistent fields let you filter, aggregate, and alert in seconds. This lesson covers what an LLM request log should contain and how to ship it without leaking PII.
Explore each section of a production-ready LLM request log:
Every LLM request log should contain these fields — each answering a specific operational question.
from pydantic import BaseModel
from typing import Literal
class LlmRequestLog(BaseModel):
# Identity — for correlation across systems
request_id: str # UUID — correlate with API gateway
trace_id: str # Links all spans in one user interaction
user_id_hash: str # SHA-256 of user_id — NEVER raw PII
tenant_id: str | None
# Request — what went in
model: str
operation: str # "chat" | "rag" | "summarization"
input_tokens: int
prompt_hash: str # Hash — for regression detection without PII
cache_hit: bool
# Response — what came out
output_tokens: int
latency_ms: float
status: Literal["success", "error", "timeout", "rate_limited"]
error_type: str | None
# Cost — for budget tracking
cost_usd: float
escalated: bool # True if model was upgraded mid-request
fallback_used: boolfrom pydantic import BaseModel
from typing import Literal
import time, uuid, hashlib
class LlmRequestLog(BaseModel):
# Identity
request_id: str # UUID — correlate with API gateway logs
trace_id: str # Links all spans in one user interaction
user_id_hash: str # SHA-256 of user_id — NEVER raw PII
tenant_id: str | None
# Request
model: str
operation: str # "chat" | "rag" | "summarization" | "classification"
input_tokens: int
prompt_hash: str # Hash of prompt — for regression detection without storing PII
cache_hit: bool
# Response
output_tokens: int
latency_ms: float
status: Literal["success", "error", "timeout", "rate_limited"]
error_type: str | None
# Quality signals
model_tier: Literal["haiku", "sonnet", "opus"]
escalated: bool # True if cascade upgraded the model
fallback_used: bool
# Cost
cost_usd: float
def make_request_log(
user_id: str,
response,
latency_ms: float,
**kwargs,
) -> LlmRequestLog:
return LlmRequestLog(
request_id=str(uuid.uuid4()),
trace_id=kwargs.get("trace_id", str(uuid.uuid4())[:8]),
user_id_hash=hashlib.sha256(user_id.encode()).hexdigest(),
tenant_id=kwargs.get("tenant_id"),
model=response.model,
operation=kwargs.get("operation", "unknown"),
input_tokens=response.usage.input_tokens,
prompt_hash=hashlib.sha256(kwargs.get("prompt", "").encode()).hexdigest()[:12],
cache_hit=kwargs.get("cache_hit", False),
output_tokens=response.usage.output_tokens,
latency_ms=latency_ms,
status="success",
error_type=None,
model_tier=get_tier(response.model),
escalated=kwargs.get("escalated", False),
fallback_used=kwargs.get("fallback_used", False),
cost_usd=compute_cost(response.model, response.usage),
)import logging
import json
import sys
# Configure a JSON logger — works with CloudWatch, Datadog, Splunk, Grafana Loki
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_dict = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"message": record.getMessage(),
}
if hasattr(record, "llm_log"):
log_dict.update(record.llm_log)
return json.dumps(log_dict)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
logger = logging.getLogger("llm")
logger.addHandler(handler)
logger.setLevel(logging.INFO)
def log_llm_request(request_log: LlmRequestLog) -> None:
extra = {"llm_log": request_log.model_dump()}
if request_log.status == "error":
logger.error("LLM request failed", extra=extra)
elif request_log.latency_ms > 5000:
logger.warning("LLM request slow", extra=extra)
else:
logger.info("LLM request", extra=extra)-- P95 latency by operation (last 1 hour)
SELECT
operation,
PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY latency_ms) AS p95_latency_ms,
COUNT(*) AS request_count,
ROUND(AVG(cost_usd)::numeric, 6) AS avg_cost
FROM llm_request_logs
WHERE timestamp > NOW() - INTERVAL '1 hour'
AND status = 'success'
GROUP BY operation
ORDER BY p95_latency_ms DESC;
-- Error rate by model tier (for alerting: alert if > 1%)
SELECT
model_tier,
COUNT(*) FILTER (WHERE status = 'error') AS errors,
COUNT(*) AS total,
ROUND(100.0 * COUNT(*) FILTER (WHERE status = 'error') / COUNT(*), 2) AS error_rate_pct
FROM llm_request_logs
WHERE timestamp > NOW() - INTERVAL '15 minutes'
GROUP BY model_tier;
-- Cache hit rate by tenant
SELECT
tenant_id,
ROUND(100.0 * SUM(cache_hit::int) / COUNT(*), 1) AS cache_hit_rate_pct,
SUM(cost_usd) AS total_cost,
SUM(cost_usd) FILTER (WHERE NOT cache_hit) AS cache_miss_cost
FROM llm_request_logs
GROUP BY tenant_id
ORDER BY total_cost DESC;Contains PII. Log prompt_hash instead. Store raw prompts only in an encrypted, access-controlled store for debugging.
May contain PII reflected from user input. Log response_hash and output_tokens instead.
Obvious. Audit your logger config — accidentally logged secrets are the #1 API key leak vector.
SHA-256 hash them. Use the hash for correlation; store the mapping in a separate, access-controlled table.
Golden rule: log what you can query, not what you can read.Every field in your log schema should answer a specific operational question: “What is the p95 latency for RAG queries?”, “Which tenant has the highest error rate?”, “How much did we spend on sonnet today?” If a field doesn't answer a question you'd actually ask, don't log it.
Lesson 4 uses these logs to detect quality drift — automatically catching when your LLM is getting worse over time without anyone noticing.