Module 5: Structured Logging for AI Systems

8. Module 5 Summary and Troubleshooting

Description

This module transformed your app from "print() as logging" into an observability system that lets you answer any question about your app in production using only the logs. This capsule consolidates everything you learned into a decision map, the most common errors with their solutions, and a production checklist.

What you built in this module

Before:
  AI app
  └── print(response)  ← "logging"
  
  Debugging: "I add prints and redeploy"
  Cost: "no idea, I see it on the end-of-month bill"
  Error in prod: "I don't know what happened"

After:
  AI app
  ├── RequestTracingMiddleware   → request_id in all logs
  ├── logging_config.py         → structlog with JSON output
  ├── tracing.py                → contextvars (async-safe)
  ├── llm_wrapper.py            → tokens, cost, latency, alerts
  └── guardrails/               → activations logged
  
  Debugging: "I search request_id, see full history in 30s"
  Cost: "jq 'cost' logs.json | sum"
  Error in prod: "logs show error_type, duration, model, hash"

Module decision map

What should I log in this event?
│
├─ Is it the start/end of a request?
│   └─ INFO: request_id, path, method, duration, status_code
│
├─ Is it an LLM call?
│   └─ INFO: model, input/output_tokens, cost_usd, duration_ms, finish_reason
│       + WARNING if: latency > 10s, cost > $0.05, finish_reason == "length"
│       + ERROR if: exception — with prompt_preview and messages_hash
│
├─ Is it a guardrail activation?
│   └─ WARNING: guardrail_type, action (blocked/modified), layer
│       NOT the full input (it may be an attack or contain PII)
│
├─ Is it a parsing/validation error?
│   └─ WARNING: raw_preview, messages_hash — to correlate with the call
│
└─ Is it a fatal error?
    └─ ERROR + exc_info=True — with the full stack trace in JSON

With what level?
  └─ DEBUG: full prompt, full response — ONLY if debug_mode=True
  └─ INFO: request summary, metrics (always in prod)
  └─ WARNING: anomalies that require attention but are not a failure
  └─ ERROR: something failed and requires intervention

Quick reference: the module's tools

Tool	What for	When to use
`structlog.contextvars.bind_contextvars()`	Bind request_id to the context	At the start of each request
`structlog.contextvars.clear_contextvars()`	Clean up when done	At the end of each request (in finally)
`structlog.contextvars.merge_contextvars`	Processor that injects the context	Always in the processors list
`structlog.processors.JSONRenderer()`	Output in JSON Lines	In production
`structlog.dev.ConsoleRenderer()`	Readable output	In development
`log.bind(key=value)`	Bind fields to a local logger	For subtasks with additional fields
`log.exception("event")`	Log error + stack trace	In except blocks

The 7 most common errors and their solutions

Error 1: request_id does not appear in some logs

# Symptom: some logs have request_id, others don't

# Cause A: The merge_contextvars processor is not configured
# Fix:
structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,  # ← First in the list
        ...
    ]
)

# Cause B: There is a logger created BEFORE configuring structlog
# Fix: always call configure_logging() BEFORE creating any logger
# and use cache_logger_on_first_use=True (the default)

# Cause C: The context was cleared before it was logged
# Fix: make sure clear_contextvars() is in the middleware's finally,
# not in the try

Error 2: Mixed-up logs in async (request_id from another request)

# Symptom: the request_id in a log belongs to another concurrent request

# Cause: contextvars is not used — a global variable is used instead
# ❌ INCORRECT:
_current_request_id = None  # Global variable (does NOT work in async)

def set_request_id(rid): 
    global _current_request_id
    _current_request_id = rid

# ✅ CORRECT: use contextvars
import contextvars
_request_id_var = contextvars.ContextVar("request_id", default=None)

def set_request_id(rid):
    _request_id_var.set(rid)
    structlog.contextvars.bind_contextvars(request_id=rid)

# contextvars is task-scoped in async, one coroutine's request_id does not
# contaminate another

Error 3: Logs are plain text in production, not JSON

# Symptom: in production the logs are: "2024-01-15 INFO request_completed"

# Cause: ENVIRONMENT is not set, or configure_logging is not called
# Fix:
# 1. Set in docker/k8s: ENVIRONMENT=production
# 2. Make sure to call configure_logging() before any log:
configure_logging()  # ← At the start of main.py or app factory

# Verify:
print(os.getenv("ENVIRONMENT"))  # Must be "production"

Error 4: cost_usd is 0 or None in the logs

# Symptom: LLM request logs with no cost or with cost 0

# Cause A: response.usage is None (some endpoints don't return it)
# Fix:
if response.usage:
    cost = calculate_cost(model, response.usage.prompt_tokens,
                          response.usage.completion_tokens)
else:
    # Estimate with tiktoken
    input_tokens = count_tokens_in_messages(messages, model)
    cost = calculate_cost(model, input_tokens, 500)  # Estimate
    log.warning("usage_not_available", estimated=True)

# Cause B: The model is not in the pricing table
# Fix: add the model to MODEL_PRICING or check the fallback

Error 5: Huge logs (>1MB per entry)

# Symptom: log files grow very fast, logs with MB in size

# Cause: Logging large objects (full prompt in INFO, 
# very long responses, fully serialized Python objects)

# Fix 1: Use the truncate_long_values processor
def truncate_long_values(logger, method_name, event_dict, max_len=500):
    for key, value in event_dict.items():
        if isinstance(value, str) and len(value) > max_len:
            event_dict[key] = value[:max_len] + f"...[{len(value)}]"
    return event_dict

# Fix 2: Only log what you need
# ❌ log.info("response", full_response=response.model_dump())  # Large object
# ✅ log.info("response", tokens=response.usage.total_tokens, cost=cost)

Error 6: PII or secrets in the logs

# Symptom: emails, API keys, or personal data appear in the logs

# Cause: logging the user input without sanitizing
# log.info("request", input=user_text)  # user_text may contain PII

# Fix 1: sanitize_sensitive_fields processor (already implemented)
# Fix 2: Never log the full input in INFO
# ✅ log.info("request", input_length=len(user_text), input_hash=hash(user_text))

# Fix 3: If you need to log the input (for debugging):
# → Only in ERROR or DEBUG mode
# → Only the first 200-300 chars
# → Redact PII first with the PII guardrail from Module 4

Error 7: structlog does not capture logs from external libraries

# Symptom: logs from httpx, openai, etc. appear in a different format
# or don't appear in the JSON logs

# Cause: structlog and Python's standard logging are different systems
# Fix: configure the bridge between the two

import logging
import structlog

# Configure Python's standard logging to use the same handler
logging.basicConfig(format="%(message)s", level=logging.INFO)

# Configure structlog to process stdlib logs too
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_log_level,
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    logger_factory=structlog.stdlib.LoggerFactory(),  # ← stdlib, not PrintLoggerFactory
)

Diagnostic tree

PROBLEM: "I can't debug an error in production"
│
├─ Do you have the request_id?
│   ├─ NO → Does the response include the X-Request-ID header?
│   │        ├─ NO → Add request_id to the response (middleware)
│   │        └─ YES → Ask the user to report that header
│   │
│   └─ YES → jq 'select(.request_id == "XXX")' logs.json | jq -s 'sort_by(.timestamp)'
│           ├─ Do you see all the steps of the request?
│           │   └─ NO → Are all events logged? Check for gaps in the sequence
│           └─ Do you see the error?
│               └─ NO → Is the log level low enough? Check LOG_LEVEL

PROBLEM: "The logs are not JSON in production"
│
└─ Is ENVIRONMENT=production set?
    ├─ NO → Set it in docker-compose/k8s env
    └─ YES → Is configure_logging() called before the first log?
             ├─ NO → Move configure_logging() to the start of the app
             └─ YES → Does the app use cache_logger_on_first_use=True and reuse old loggers?
                      └─ Fix: structlog.reset_defaults() at startup (only in tests)

PROBLEM: "I don't know how much I spent yesterday on OpenAI"
│
├─ Do the logs have cost_usd?
│   ├─ NO → Does call_llm return metrics with cost_usd?
│   │        ├─ NO → Implement calculate_cost() and log it in call_llm
│   │        └─ YES → But the wrapper isn't used. Use call_llm() everywhere
│   └─ YES → jq -s '[.[].cost_usd // 0] | add' logs.json
│             or python scripts/analyze_logs.py logs/app.json 2024-01-15

PROBLEM: "The cost_usd is incorrect"
│
└─ Are the prices in MODEL_PRICING up to date?
    └─ Check at openai.com/pricing
       ├─ Update MODEL_PRICING in logging_config.py
       └─ Add a test that validates known prices

Module 5 production checklist

CONFIGURATION
[ ] configure_logging() is called at the start of the app
[ ] ENVIRONMENT=production is set in production
[ ] LOG_LEVEL=INFO in production (DEBUG only if necessary)
[ ] LOG_FILE is configured if you want logs in a file
[ ] Log rotation configured (backupCount=30)

TRACING
[ ] RequestTracingMiddleware registered in the FastAPI app
[ ] All endpoints use contextvars for request_id
[ ] Response headers include X-Request-ID
[ ] clear_contextvars() in the middleware's finally

LOGGING IN THE CODE
[ ] All llm calls go through call_llm()
[ ] All INFO logs have request_id (verify with a test)
[ ] No API keys or PII in the logs
[ ] sanitize_sensitive_fields processor active

GUARDRAILS
[ ] Guardrail activations logged with WARNING
[ ] guardrail_type and action in each guardrail log
[ ] The full input is not logged in guardrail WARNING

COST TRACKING
[ ] MODEL_PRICING has all models in use
[ ] cost_usd appears in each LLM call log
[ ] high_cost alerts active
[ ] analyze_logs.py script available to the team

TESTS
[ ] test: request_id appears in all logs
[ ] test: prompt does not appear in INFO logs
[ ] test: cost_usd is in the LLM call logs
[ ] test: warnings for finish_reason='length'

Module vocabulary

Term	Definition
Structured logging	Logging that produces JSON data instead of free text, machine-queryable
JSON Lines (JSONL)	Format where each line is an independent JSON object, ideal for append-only logs
Correlation ID / request_id	Unique identifier that connects all logs in a request's lifecycle
contextvars	Python module that stores per-coroutine state (async-safe, thread-safe)
bind_contextvars()	structlog function to bind fields to the current coroutine's context
finish_reason	Reason the LLM stopped generating: "stop", "length", "content_filter"
system_fingerprint	Identifier of the OpenAI server/model version, useful for detecting changes
reproducibility context	The set of parameters (model, temperature, seed, prompt) needed to re-run an LLM call
seed	Parameter that asks the model to try to be reproducible for the same input
Log aggregation	System that centralizes logs from multiple instances or services (ELK, Loki, Datadog)

Connection with Module 6

Module 6 (Code Quality Patterns for AI) organizes all the code you've built so far. You have:

Tests (Module 2, 3)
Guardrails (Module 4)
Logging (Module 5)

The problem is that they're probably intertwined: main.py imports directly from guardrails, llm_wrapper depends on logging_config, and the tests are hard to isolate because everything is coupled.

Module 6 applies clean architecture to separate concerns:

Domain layer: the business logic (what does a positive sentiment mean?)
Application layer: the use cases (analize_sentiment)
Infrastructure layer: guardrails, logging, calls to the OpenAI API
Interface layer: FastAPI endpoints

This separation makes the code maintainable, testable, and extensible — the transition from "code that works" to "production code."

Additional module resources

structlog Documentation — The complete reference
structlog contextvars guide — Async logging
jq Manual — For ad-hoc queries over the logs
OpenTelemetry for Python — For advanced distributed tracing
Twelve-Factor App — Logs — The philosophy behind the module
OpenAI Pricing — Up-to-date prices for MODEL_PRICING
Loki (Grafana) — Log aggregation system optimized for JSON logs
Datadog APM — Enterprise observability with support for LLM apps

← Previous7. Project: AI Logging System Next →1. Introduction: Code Quality for AI