Module 5: Structured Logging for AI Systems
7. Project: AI Logging System
Description
This is the integrative project for Module 5. You build a complete logging system for the sentiment analysis app from Module 4, integrating all the pieces: structlog with JSON output, correlation IDs with contextvars, token and cost tracking, reproducibility context, and the connection with the guardrails. When you finish, you have a system that lets you answer any question about what happened in your app using only the logs.
AI Logging System Architecture
┌─────────────────────────────────────────────────────────────┐
│ Request /analyze │
└────────────────────────────┬────────────────────────────────┘
│
┌──────────────▼──────────────┐
│ RequestTracingMiddleware │
│ → get/generate request_id │
│ → bind_contextvars() │
│ → log request_started │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ GuardrailsPipeline │
│ → sanitize input │ → WARNING if injection
│ → check injection │ → WARNING if PII redacted
│ → redact PII │ → INFO if input modified
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ call_llm_with_logging() │
│ → INFO: tokens, cost │
│ → WARNING: high latency │
│ → ERROR: API failure │
│ → DEBUG: prompt (optional) │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ Output validation │
│ → WARNING if invalid JSON │
│ → WARNING if fallback used │
└──────────────┬──────────────┘
│
┌──────────────▼──────────────┐
│ RequestTracingMiddleware │
│ → log request_completed │
│ → total cost, duration │
│ → clear_contextvars() │
└─────────────────────────────┘
All logs have request_id automatically.
Output: JSON Lines to stdout and/or file.
Project structure
src/
├── logging_config.py ← Central structlog configuration
├── tracing.py ← request_id with contextvars
├── middleware.py ← FastAPI middleware for tracing
├── cost_calculator.py ← Cost calculation per model
├── llm_wrapper.py ← LLM client wrapper with logging
├── reproducibility.py ← LLMCallContext for reproducibility
├── guardrails/
│ └── pipeline.py ← Already implemented in Module 4 (with logging)
└── app/
└── main.py ← FastAPI app with middleware and endpoints
scripts/
├── analyze_logs.py ← Cost and metrics analysis from logs
└── reproduce_from_logs.py ← Reproduce request from logs
tests/
├── unit/
│ ├── test_logging_config.py
│ ├── test_tracing.py
│ ├── test_cost_calculator.py
│ └── test_llm_wrapper.py
└── integration/
└── test_logging_integration.py
Step 1: Central configuration
# src/logging_config.py
import logging
import os
import sys
import hashlib
from typing import Any, MutableMapping
import structlog
# ─── Pricing per 1M tokens ────────────────────────────────────────────────
MODEL_PRICING = {
"gpt-4o-mini": {"input": 0.150, "output": 0.600},
"gpt-4o": {"input": 2.500, "output": 10.000},
"gpt-4-turbo": {"input": 10.000, "output": 30.000},
"gpt-4": {"input": 30.000, "output": 60.000},
"gpt-3.5-turbo": {"input": 0.500, "output": 1.500},
}
DEFAULT_PRICING = {"input": 0.150, "output": 0.600}
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
"""Calculate the cost in USD of an LLM call."""
pricing = MODEL_PRICING.get(model)
if pricing is None:
for key in MODEL_PRICING:
if model.startswith(key):
pricing = MODEL_PRICING[key]
break
pricing = pricing or DEFAULT_PRICING
return round(
(input_tokens / 1_000_000) * pricing["input"] +
(output_tokens / 1_000_000) * pricing["output"],
8
)
# ─── Custom processors ──────────────────────────────────────────────────
def add_app_metadata(logger, method_name, event_dict):
"""Add application metadata to all logs."""
event_dict["app"] = os.getenv("APP_NAME", "sentiment-analyzer")
event_dict["version"] = os.getenv("APP_VERSION", "0.1.0")
event_dict["env"] = os.getenv("ENVIRONMENT", "development")
return event_dict
def sanitize_sensitive_fields(logger, method_name, event_dict):
"""Redact sensitive fields before logging."""
REDACT = {"password", "api_key", "secret", "authorization", "token"}
HASH = {"user_id", "email"}
for field in list(event_dict.keys()):
field_lower = field.lower()
if any(s in field_lower for s in REDACT):
event_dict[field] = "[REDACTED]"
elif any(p in field_lower for p in HASH):
event_dict[field] = hashlib.sha256(
str(event_dict[field]).encode()
).hexdigest()[:12]
return event_dict
def truncate_long_values(logger, method_name, event_dict, max_len=500):
"""Truncate long strings to avoid huge logs."""
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)}chars]"
return event_dict
# ─── Main configuration ──────────────────────────────────────────────
def configure_logging(
env: str = None,
log_level: int = None,
log_file: str = None
) -> None:
"""
Configure structlog for the given environment.
Args:
env: "production" | "development" | "testing"
log_level: Log level override
log_file: If specified, in addition to stdout, write to this file
"""
env = env or os.getenv("ENVIRONMENT", "development")
log_level = log_level or getattr(
logging,
os.getenv("LOG_LEVEL", "INFO").upper(),
logging.INFO
)
use_json = env in ("production", "staging")
# Configure Python's standard logging
handlers = [logging.StreamHandler(sys.stdout)]
if log_file:
from logging.handlers import TimedRotatingFileHandler
os.makedirs(os.path.dirname(log_file) or ".", exist_ok=True)
handlers.append(TimedRotatingFileHandler(
log_file, when="midnight", backupCount=30, encoding="utf-8"
))
logging.basicConfig(handlers=handlers, format="%(message)s", level=log_level)
shared_processors = [
structlog.contextvars.merge_contextvars,
add_app_metadata,
sanitize_sensitive_fields,
structlog.processors.add_log_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.stdlib.add_logger_name,
structlog.processors.StackInfoRenderer(),
]
if use_json:
processors = shared_processors + [
truncate_long_values,
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer()
]
else:
processors = shared_processors + [
structlog.dev.ConsoleRenderer(colors=True)
]
structlog.configure(
processors=processors,
wrapper_class=structlog.make_filtering_bound_logger(log_level),
context_class=dict,
logger_factory=structlog.PrintLoggerFactory(),
cache_logger_on_first_use=True,
)Step 2: Tracing with contextvars
# src/tracing.py
import uuid
import contextvars
from typing import Optional
_request_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
"request_id", default=None
)
def generate_request_id() -> str:
return uuid.uuid4().hex[:8]
def set_request_id(request_id: str) -> None:
_request_id_var.set(request_id)
structlog.contextvars.bind_contextvars(request_id=request_id)
def get_request_id() -> Optional[str]:
return _request_id_var.get()
def get_or_create_request_id() -> str:
current = _request_id_var.get()
if current is None:
current = generate_request_id()
set_request_id(current)
return current
def clear_request_id() -> None:
_request_id_var.set(None)
structlog.contextvars.clear_contextvars()Step 3: FastAPI middleware
# src/middleware.py
import time
import structlog
from starlette.middleware.base import BaseHTTPMiddleware
from src.tracing import generate_request_id, set_request_id, clear_request_id
log = structlog.get_logger()
class RequestTracingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
request_id = (
request.headers.get("X-Request-ID") or
generate_request_id()
)
set_request_id(request_id)
start = time.time()
log.info(
"request_started",
method=request.method,
path=request.url.path
)
try:
response = await call_next(request)
duration_ms = (time.time() - start) * 1000
log.info(
"request_completed",
status_code=response.status_code,
duration_ms=round(duration_ms, 1),
path=request.url.path
)
response.headers["X-Request-ID"] = request_id
return response
except Exception as e:
log.error(
"request_failed",
error_type=type(e).__name__,
duration_ms=round((time.time() - start) * 1000, 1)
)
raise
finally:
clear_request_id()Step 4: LLM wrapper with complete logging
# src/llm_wrapper.py
import time
import hashlib
import json
import structlog
from typing import Optional, Tuple, Any
from src.logging_config import calculate_cost
log = structlog.get_logger()
HIGH_COST_USD = 0.05
HIGH_LATENCY_MS = 10_000
HIGH_TOKENS = 10_000
def call_llm(
client,
model: str,
messages: list,
temperature: float = 0.0,
max_tokens: int = 500,
seed: Optional[int] = None,
debug_mode: bool = False,
**kwargs
) -> Tuple[Any, dict]:
"""
Make an LLM call with complete logging.
Returns (response, metrics_dict).
"""
start = time.time()
messages_hash = _hash_messages(messages)
# Pre-flight: warn if the input looks very long
total_chars = sum(len(m.get("content", "")) for m in messages)
if total_chars > 40_000:
log.warning(
"large_input_pre_call",
estimated_tokens=total_chars // 4,
messages_hash=messages_hash
)
# DEBUG: show prompt (only if explicitly enabled)
if debug_mode:
last_content = messages[-1].get("content", "") if messages else ""
log.debug(
"llm_prompt",
model=model,
temperature=temperature,
seed=seed,
prompt_preview=last_content[:500]
)
try:
response = client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
seed=seed,
**kwargs
)
duration_ms = (time.time() - start) * 1000
cost_usd = calculate_cost(
model,
response.usage.prompt_tokens,
response.usage.completion_tokens
)
metrics = {
"model": model,
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens,
"cost_usd": cost_usd,
"duration_ms": round(duration_ms, 1),
"finish_reason": response.choices[0].finish_reason,
"messages_hash": messages_hash,
"seed": seed,
"temperature": temperature,
}
# System fingerprint to detect model changes
if hasattr(response, "system_fingerprint") and response.system_fingerprint:
metrics["system_fingerprint"] = response.system_fingerprint
# INFO: call metrics (always)
log.info("llm_call_completed", **metrics)
# Alerts
if response.choices[0].finish_reason == "length":
log.warning("response_truncated", max_tokens=max_tokens, **metrics)
if duration_ms > HIGH_LATENCY_MS:
log.warning("high_latency", **metrics)
if cost_usd > HIGH_COST_USD:
log.warning("high_cost", **metrics)
if response.usage.prompt_tokens > HIGH_TOKENS:
log.warning("high_input_tokens", **metrics)
# DEBUG: full response
if debug_mode:
raw = response.choices[0].message.content
log.debug("llm_response",
response_preview=(raw or "")[:300],
response_hash=hashlib.sha256(
(raw or "").encode()
).hexdigest()[:12]
)
return response, metrics
except Exception as e:
duration_ms = (time.time() - start) * 1000
last_content = messages[-1].get("content", "") if messages else ""
log.error(
"llm_call_failed",
error_type=type(e).__name__,
error_msg=str(e)[:200],
model=model,
duration_ms=round(duration_ms, 1),
messages_hash=messages_hash,
prompt_preview=last_content[:300] # More context on error
)
raise
def _hash_messages(messages: list) -> str:
content = json.dumps(messages, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(content.encode()).hexdigest()[:12]Step 5: Integrated FastAPI app
# src/app/main.py
import os
import structlog
from fastapi import FastAPI, Request
from pydantic import BaseModel
from src.logging_config import configure_logging
from src.middleware import RequestTracingMiddleware
from src.llm_wrapper import call_llm
from src.guardrails.pipeline import GuardrailsPipeline, PUBLIC_API_CONFIG
# Configure logging at startup
configure_logging(
env=os.getenv("ENVIRONMENT", "development"),
log_file="logs/app.json" if os.getenv("LOG_TO_FILE") else None
)
log = structlog.get_logger()
# Create app and register middleware
app = FastAPI(title="AI Sentiment Analyzer with Logging")
app.add_middleware(RequestTracingMiddleware)
# Guardrails pipeline instance
_pipeline: GuardrailsPipeline = None
def get_pipeline() -> GuardrailsPipeline:
global _pipeline
if _pipeline is None:
_pipeline = GuardrailsPipeline(
config=PUBLIC_API_CONFIG,
openai_client=_get_openai_client()
)
return _pipeline
def _get_openai_client():
from openai import OpenAI
return OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class AnalyzeRequest(BaseModel):
text: str
debug: bool = False # Enable debug logging for this request
class AnalyzeResponse(BaseModel):
sentiment: str
score: float
request_id: str
cost_usd: float
duration_ms: float
@app.post("/analyze", response_model=AnalyzeResponse)
async def analyze_endpoint(body: AnalyzeRequest, request: Request):
"""Analyze the sentiment of a text."""
from src.tracing import get_request_id
import time
request_id = get_request_id()
start = time.time()
log.info("analyze_requested", text_length=len(body.text))
# Run guardrails (from Module 4)
pipeline = get_pipeline()
pipeline_result = pipeline.process(
body.text,
request_id=request_id,
debug_mode=body.debug
)
if pipeline_result.blocked:
log.warning(
"request_blocked_by_guardrails",
reason=pipeline_result.block_reason,
guardrail=pipeline_result.blocked_by
)
from fastapi import HTTPException
raise HTTPException(400, detail=pipeline_result.block_reason)
# Log activated guardrails
if pipeline_result.activated_guardrails:
log.info(
"guardrails_applied",
activated=pipeline_result.activated_guardrails
)
# Call the LLM with complete logging
client = _get_openai_client()
SENTIMENT_PROMPT = """Analyze the sentiment of the following text.
Respond with a JSON object: {"sentiment": "positive|negative|neutral|mixed", "score": 0.0}
Score: 1.0 = very positive, -1.0 = very negative, 0.0 = neutral.
Text: {text}"""
messages = [
{"role": "system", "content": "You are a sentiment analysis expert."},
{"role": "user", "content": SENTIMENT_PROMPT.format(
text=pipeline_result.processed_input
)}
]
response, metrics = call_llm(
client=client,
model="gpt-4o-mini",
messages=messages,
temperature=0.0,
max_tokens=200,
seed=42,
debug_mode=body.debug
)
# Parse result
import json
raw = response.choices[0].message.content.strip()
try:
result = json.loads(raw)
sentiment = result["sentiment"]
score = float(result["score"])
except (json.JSONDecodeError, KeyError, ValueError):
log.warning(
"output_parse_failed",
raw_preview=raw[:200],
messages_hash=metrics["messages_hash"]
)
sentiment = "unknown"
score = 0.0
total_duration_ms = (time.time() - start) * 1000
log.info(
"analyze_completed",
sentiment=sentiment,
score=score,
total_cost_usd=metrics["cost_usd"],
total_duration_ms=round(total_duration_ms, 1)
)
return AnalyzeResponse(
sentiment=sentiment,
score=score,
request_id=request_id,
cost_usd=metrics["cost_usd"],
duration_ms=total_duration_ms
)
@app.get("/health")
async def health():
log.info("health_check")
return {"status": "ok"}Step 6: Logging system tests
# tests/unit/test_llm_wrapper.py
import json
import io
import pytest
import structlog
from unittest.mock import MagicMock
def make_mock_response(model="gpt-4o-mini", input_tokens=245, output_tokens=87,
content='{"sentiment": "positive", "score": 0.8}',
finish_reason="stop"):
mock = MagicMock()
mock.usage.prompt_tokens = input_tokens
mock.usage.completion_tokens = output_tokens
mock.usage.total_tokens = input_tokens + output_tokens
mock.choices[0].message.content = content
mock.choices[0].finish_reason = finish_reason
mock.system_fingerprint = "fp_test123"
return mock
@pytest.fixture(autouse=True)
def json_log_output():
"""Capture JSON logs for each test."""
output = io.StringIO()
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.processors.add_log_level,
structlog.processors.JSONRenderer()
],
logger_factory=structlog.PrintLoggerFactory(file=output),
cache_logger_on_first_use=False
)
yield output
structlog.contextvars.clear_contextvars()
def read_logs(output: io.StringIO) -> list:
return [json.loads(l) for l in output.getvalue().strip().split("\n") if l.strip()]
class TestCallLLM:
def test_logs_tokens_and_cost(self, json_log_output):
"""Verify that tokens and cost are logged."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response()
structlog.contextvars.bind_contextvars(request_id="test-001")
_, metrics = call_llm(mock_client, "gpt-4o-mini",
[{"role": "user", "content": "test"}])
logs = read_logs(json_log_output)
completed = next(l for l in logs if l["event"] == "llm_call_completed")
assert completed["input_tokens"] == 245
assert completed["output_tokens"] == 87
assert "cost_usd" in completed
assert completed["cost_usd"] > 0
assert completed["request_id"] == "test-001"
def test_warns_on_truncated_response(self, json_log_output):
"""If finish_reason is 'length', it must log a WARNING."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response(
finish_reason="length"
)
call_llm(mock_client, "gpt-4o-mini", [{"role": "user", "content": "test"}])
logs = read_logs(json_log_output)
warnings = [l for l in logs if l.get("level") == "warning"]
assert any(l["event"] == "response_truncated" for l in warnings)
def test_no_prompt_in_info_logs(self, json_log_output):
"""The prompt must not appear in INFO-level logs."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response()
secret_text = "My credit card is 4111-1111-1111-1111"
call_llm(mock_client, "gpt-4o-mini",
[{"role": "user", "content": secret_text}],
debug_mode=False)
output = json_log_output.getvalue()
assert "4111-1111-1111-1111" not in output
def test_includes_system_fingerprint(self, json_log_output):
"""The system_fingerprint must appear in the log."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response()
call_llm(mock_client, "gpt-4o-mini", [{"role": "user", "content": "test"}])
logs = read_logs(json_log_output)
completed = next(l for l in logs if l["event"] == "llm_call_completed")
assert completed.get("system_fingerprint") == "fp_test123"
class TestCostCalculator:
def test_gpt4o_mini_pricing(self):
"""Verify the cost calculation for gpt-4o-mini."""
from src.logging_config import calculate_cost
# 1M input = $0.15, 1M output = $0.60
cost = calculate_cost("gpt-4o-mini", 1_000_000, 0)
assert abs(cost - 0.15) < 0.001
def test_cost_returned_in_metrics(self):
"""call_llm must return cost in metrics dict."""
from src.llm_wrapper import call_llm
mock_client = MagicMock()
mock_client.chat.completions.create.return_value = make_mock_response(
input_tokens=500, output_tokens=200
)
_, metrics = call_llm(mock_client, "gpt-4o-mini",
[{"role": "user", "content": "test"}])
assert "cost_usd" in metrics
assert metrics["cost_usd"] > 0Step 7: Analysis script
# scripts/analyze_logs.py
"""
Analyze the app's JSON Lines logs and generate a report.
Usage: python scripts/analyze_logs.py [log_file] [date_prefix]
"""
import json
import sys
from collections import defaultdict
def load_jsonl(path: str) -> list:
logs = []
with open(path) as f:
for line in f:
line = line.strip()
if line:
try:
logs.append(json.loads(line))
except json.JSONDecodeError:
pass
return logs
def report(logs: list, date_prefix: str = None):
if date_prefix:
logs = [l for l in logs if l.get("timestamp", "").startswith(date_prefix)]
llm_logs = [l for l in logs if l.get("event") == "llm_call_completed"]
error_logs = [l for l in logs if l.get("level") == "error"]
guard_logs = [l for l in logs if l.get("event") == "guardrail_activated"]
warn_logs = [l for l in logs if l.get("level") == "warning"]
print("=" * 60)
print("AI LOGGING SYSTEM — REPORT")
if date_prefix:
print(f"Date filter: {date_prefix}")
print("=" * 60)
# Summary
print(f"\n📊 TOTALS")
print(f" LLM calls: {len(llm_logs)}")
print(f" Errors: {len(error_logs)}")
print(f" Warnings: {len(warn_logs)}")
print(f" Guardrail events: {len(guard_logs)}")
# Cost
if llm_logs:
total = sum(l.get("cost_usd", 0) for l in llm_logs)
tokens = sum(l.get("total_tokens", 0) for l in llm_logs)
avg = total / len(llm_logs)
top = max(llm_logs, key=lambda l: l.get("cost_usd", 0))
print(f"\n💰 COST")
print(f" Total: ${total:.6f}")
print(f" Avg per call: ${avg:.8f}")
print(f" Total tokens: {tokens:,}")
print(f" Most expensive: ${top.get('cost_usd', 0):.6f}")
print(f" → request_id: {top.get('request_id', 'N/A')}")
print(f" → tokens: {top.get('total_tokens', 0):,}")
# Performance
if llm_logs:
durations = [l.get("duration_ms", 0) for l in llm_logs]
avg_d = sum(durations) / len(durations)
slow = [l for l in llm_logs if l.get("duration_ms", 0) > 5000]
print(f"\n⚡ PERFORMANCE")
print(f" Avg LLM latency: {avg_d:.0f}ms")
print(f" Slow (>5s): {len(slow)}")
# Errors
if error_logs:
etypes = defaultdict(int)
for l in error_logs:
etypes[l.get("error_type", "unknown")] += 1
print(f"\n❌ ERRORS ({len(error_logs)})")
for et, cnt in sorted(etypes.items(), key=lambda x: -x[1]):
print(f" {et}: {cnt}")
# Guardrails
if guard_logs:
gtypes = defaultdict(int)
for l in guard_logs:
gtypes[l.get("guardrail_type", "unknown")] += 1
print(f"\n🛡️ GUARDRAILS ({len(guard_logs)})")
for gt, cnt in sorted(gtypes.items(), key=lambda x: -x[1]):
print(f" {gt}: {cnt}")
print("\n" + "=" * 60)
if __name__ == "__main__":
log_file = sys.argv[1] if len(sys.argv) > 1 else "logs/app.json"
date_prefix = sys.argv[2] if len(sys.argv) > 2 else None
logs = load_jsonl(log_file)
report(logs, date_prefix)Verification: what you should see when done
# Start the app
ENVIRONMENT=development python -m uvicorn src.app.main:app --reload
# In another terminal, make a request
curl -X POST http://localhost:8000/analyze \
-H "Content-Type: application/json" \
-d '{"text": "The product is great but shipping was slow"}'
# Response includes request_id:
# {
# "sentiment": "mixed",
# "score": 0.2,
# "request_id": "a1b2c3d4", ← For debugging
# "cost_usd": 0.0000888,
# "duration_ms": 1240.5
# }
# Logs in console (development mode):
# [info] request_started path=/analyze
# [info] llm_call_completed model=gpt-4o-mini input_tokens=156 cost_usd=0.0000888
# [info] analyze_completed sentiment=mixed score=0.2
# [info] request_completed status_code=200 duration_ms=1240.5
# With ENVIRONMENT=production, logs would be JSON Lines:
# {"event": "request_started", "request_id": "a1b2c3d4", "path": "/analyze", ...}
# {"event": "llm_call_completed", "request_id": "a1b2c3d4", "model": "gpt-4o-mini", ...}
# ...
# To search for a specific request:
# jq 'select(.request_id == "a1b2c3d4")' logs/app.json | jq -s 'sort_by(.timestamp)'
# To see the total cost for the day:
# python scripts/analyze_logs.py logs/app.json 2024-01-15Project checklist
[ ] src/logging_config.py implemented
[ ] configure_logging() works with dev and production env
[ ] Custom processors: sanitize_sensitive_fields, add_app_metadata
[ ] JSON output in production, ConsoleRenderer in development
[ ] src/tracing.py implemented
[ ] generate_request_id() generates unique IDs
[ ] set_request_id() binds in contextvars AND structlog context
[ ] get_request_id() returns the ID of the current context
[ ] src/middleware.py implemented
[ ] Extracts/generates request_id
[ ] Logs request_started and request_completed
[ ] Adds X-Request-ID to the response
[ ] src/llm_wrapper.py implemented
[ ] Logs tokens, cost, duration in INFO
[ ] WARNING alerts for high latency and high cost
[ ] ERROR with more context than INFO
[ ] Does not expose prompt in INFO
[ ] Integrated app
[ ] Middleware registered
[ ] call_llm() used in the endpoint
[ ] Guardrails logged correctly
[ ] Tests pass
[ ] test_logs_tokens_and_cost
[ ] test_warns_on_truncated_response
[ ] test_no_prompt_in_info_logs
[ ] Scripts work
[ ] analyze_logs.py generates report
[ ] jq queries return expected results