2. Logging Strategies for LLM

Description

Not everything should be logged — and not everything should be logged at the same level. This capsule defines the complete logging strategy for LLM apps: what to log at each level (DEBUG/INFO/WARNING/ERROR), why the full prompt shouldn't be logged in production by default, how to enable verbose logging on-demand for a specific request, and how to connect logging with the guardrails from the previous module.

The problem of logging without a strategy

# Two opposite mistakes:

# Mistake 1: Too little — "logging" with prints
print(f"Response: {result}")
# → No context, no correlation, no metrics

# Mistake 2: Too much — logging everything at every level
log.debug("Full prompt", prompt=full_prompt)  # In PRODUCTION, for every request
log.debug("Full response", response=full_response)
# → Enormous storage, exposed PII, unreadable logs, degraded performance
# With 1000 requests/day × 2KB of prompt = 2GB of logs/day

# The correct strategy: log enough to answer
# the most important questions, without the costs of logging everything

Log levels for AI: the definition

import logging
import structlog

# CRITICAL: The system can't continue
# - API budget completely exceeded
# - Total failure of the logging system
# - Corrupted configuration data

# ERROR: Something failed and requires immediate attention
# - The LLM API call failed and there's no fallback
# - Parsing the LLM output is impossible after retries
# - A security guardrail failed due to an internal error (not by detecting an attack)
# - Unhandled exception in the pipeline

# WARNING: Anomalous behavior that is NOT a failure but requires monitoring
# - The fallback model was used (the primary one didn't respond)
# - A guardrail activated and blocked a request (expected behavior, but monitor it)
# - Latency > 10s (high, but not an error)
# - A request's cost > $0.10 (anomalous)
# - The LLM's JSON needed multiple parse attempts

# INFO: Normal system state, business metrics
# - Request completed (with tokens, cost, duration)
# - Server start and stop
# - Configuration loaded

# DEBUG: Detailed information for development and debugging
# - Full prompt (only if enabled)
# - Full response
# - Complete stack of the guardrail pipeline
# - Exact parameters of the LLM call

Standard schema for each level

# src/logging_config.py

import structlog
import logging
import time
from typing import Any, Optional

def build_llm_request_log(
    request_id: str,
    model: str,
    input_tokens: int,
    output_tokens: int,
    duration_ms: float,
    cost_usd: float,
    endpoint: str = None,
    guardrails_activated: list = None
) -> dict:
    """Standard schema for the INFO log of a completed request."""
    return {
        "event": "llm_request_completed",
        "request_id": request_id,
        "model": model,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "total_tokens": input_tokens + output_tokens,
        "cost_usd": round(cost_usd, 8),
        "duration_ms": round(duration_ms, 1),
        "endpoint": endpoint,
        "guardrails_activated": guardrails_activated or [],
    }

def build_error_log(
    request_id: str,
    error_type: str,
    error_message: str,
    model: str = None,
    prompt_hash: str = None  # Hash of the prompt (not the full prompt)
) -> dict:
    """Standard schema for the ERROR log of a failed request."""
    return {
        "event": "llm_request_failed",
        "request_id": request_id,
        "error_type": error_type,
        "error_message": error_message[:200],  # Truncate to avoid enormous logs
        "model": model,
        "prompt_hash": prompt_hash,  # To correlate with the reproducibility log
    }

def build_guardrail_log(
    request_id: str,
    guardrail_type: str,
    action: str,  # "blocked", "modified", "allowed"
    layer: str = None,  # "pattern", "llm_judge", "regex"
    reason: str = None
) -> dict:
    """Standard schema for the WARNING log of an activated guardrail."""
    return {
        "event": "guardrail_activated",
        "request_id": request_id,
        "guardrail_type": guardrail_type,
        "action": action,
        "layer": layer,
        "reason": reason,
    }

The complete logging function for an LLM call

# src/llm_wrapper.py
import time
import hashlib
import structlog
from typing import Optional, Any

log = structlog.get_logger()

def call_llm_with_logging(
    client,
    model: str,
    messages: list,
    request_id: str,
    temperature: float = 0.0,
    max_tokens: int = 500,
    endpoint: str = None,
    debug_mode: bool = False,
    **kwargs
) -> Any:
    """
    Wrapper for LLM calls that includes complete logging.
    
    - INFO: always (tokens, cost, duration)
    - DEBUG: only if debug_mode=True (full prompt, full response)
    - WARNING: if there's a fallback or high latency
    - ERROR: if the call fails
    """
    bound_log = log.bind(request_id=request_id, endpoint=endpoint)
    start_time = time.time()
    
    # DEBUG: log the full prompt (only in debug mode)
    if debug_mode:
        prompt_content = messages[-1].get("content", "") if messages else ""
        bound_log.debug(
            "llm_prompt_sent",
            model=model,
            temperature=temperature,
            max_tokens=max_tokens,
            prompt_length=len(prompt_content),
            # Only log the first 500 chars of the prompt even in debug
            prompt_preview=prompt_content[:500]
        )
    
    # Calculate the prompt hash for reproducibility (always)
    prompt_hash = _hash_messages(messages)
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature,
            max_tokens=max_tokens,
            **kwargs
        )
        
        duration_ms = (time.time() - start_time) * 1000
        
        # Calculate cost
        from src.logging_config import PRICING
        cost_usd = _calculate_cost(model, response.usage.prompt_tokens, response.usage.completion_tokens)
        
        # INFO: completed-request log (always)
        bound_log.info(
            "llm_request_completed",
            model=model,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
            total_tokens=response.usage.total_tokens,
            duration_ms=round(duration_ms, 1),
            cost_usd=round(cost_usd, 8),
            prompt_hash=prompt_hash  # For reproducibility
        )
        
        # WARNING: if latency is high
        if duration_ms > 10_000:
            bound_log.warning(
                "high_latency_request",
                duration_ms=round(duration_ms, 1),
                model=model
            )
        
        # WARNING: if cost is high
        if cost_usd > 0.10:
            bound_log.warning(
                "high_cost_request",
                cost_usd=round(cost_usd, 6),
                model=model,
                total_tokens=response.usage.total_tokens
            )
        
        # DEBUG: log the full response (only in debug mode)
        if debug_mode:
            raw_response = response.choices[0].message.content
            bound_log.debug(
                "llm_response_received",
                response_length=len(raw_response or ""),
                response_preview=(raw_response or "")[:200],
                finish_reason=response.choices[0].finish_reason
            )
        
        return response
    
    except Exception as e:
        duration_ms = (time.time() - start_time) * 1000
        
        # ERROR: the LLM call failed
        bound_log.error(
            "llm_request_failed",
            error_type=type(e).__name__,
            error_message=str(e)[:200],
            model=model,
            duration_ms=round(duration_ms, 1),
            prompt_hash=prompt_hash
        )
        raise

def _hash_messages(messages: list) -> str:
    """Calculates a short hash of the prompt for reproducibility."""
    content = "".join(m.get("content", "") for m in messages)
    return hashlib.sha256(content.encode()).hexdigest()[:12]

def _calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    PRICES = {
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4": {"input": 30.00, "output": 60.00},
    }
    prices = PRICES.get(model, PRICES["gpt-4o-mini"])
    return (
        input_tokens / 1_000_000 * prices["input"] +
        output_tokens / 1_000_000 * prices["output"]
    )

On-demand debug logging: no redeploying to debug

# Enable debug logging for a specific request without affecting the others:

# Option 1: Special header (for APIs)
from fastapi import Request

async def get_debug_mode(request: Request) -> bool:
    """Only enable debug mode if the header is present AND there's a valid key."""
    debug_key = request.headers.get("X-Debug-Logging")
    valid_debug_key = os.getenv("DEBUG_LOGGING_KEY", "")
    return bool(valid_debug_key and debug_key == valid_debug_key)

@app.post("/analyze")
async def analyze(request: Request, body: AnalyzeRequest):
    debug_mode = await get_debug_mode(request)
    request_id = get_or_create_request_id(request)
    
    result = analyze_sentiment_with_guardrails(
        body.text,
        request_id=request_id,
        debug_mode=debug_mode
    )
    return result

# Option 2: Allowlist of request_ids for debug
DEBUG_REQUESTS = set()  # IDs that should be logged in debug

def should_debug(request_id: str) -> bool:
    return request_id in DEBUG_REQUESTS

# Usage: before processing a suspicious request, add its ID to the list

What you should NEVER log

# ❌ API keys
log.info("openai_configured", api_key=os.getenv("OPENAI_API_KEY"))

# ❌ Unredacted PII
log.info("request", input_text=user_text)  # If it contains names, emails, etc.

# ❌ Full prompt in INFO (production)
log.info("llm_call", prompt=full_prompt_string)  # 2KB × 1000 req/day = 2MB/day with no value

# ❌ Passwords or other secrets
log.debug("db_connection", password=db_password)

# ✅ What you SHOULD log:
log.info("request_completed",
    request_id=request_id,
    input_length=len(user_text),     # Length, not the content
    input_hash=hash(user_text),      # Hash to correlate, not the text
    model=model,
    tokens=response.usage.total_tokens,
    cost_usd=cost
)

Connecting guardrails with logging

# src/guardrails/pipeline.py (with logging integrated)
def _log(self, event: str, data: dict = None):
    """Log a guardrail activation with the appropriate level."""
    if not self.config.log_activations:
        return
    
    bound_log = log.bind(request_id=data.get("request_id", "unknown"))
    
    if event in ("injection_blocked", "content_filtered"):
        # WARNING: the guardrail blocked something — expected behavior but notable
        bound_log.warning(
            "guardrail_activated",
            guardrail_event=event,
            **{k: v for k, v in (data or {}).items() if k != "request_id"}
        )
    elif event in ("pii_redacted", "input_sanitized"):
        # INFO: normal pipeline modification
        bound_log.info(
            "guardrail_applied",
            guardrail_event=event,
            **{k: v for k, v in (data or {}).items() if k != "request_id"}
        )

Storage estimation

# Helps make decisions about what to log:

LOG_SIZE_ESTIMATES = {
    "INFO summary (no content)": "200-400 bytes",
    "WARNING guardrail": "300-500 bytes",
    "ERROR with hash": "400-600 bytes",
    "DEBUG with prompt preview (500 chars)": "800-1200 bytes",
    "DEBUG with full prompt (5K chars)": "5000-6000 bytes",
}

# With 1,000 requests/day:
DAILY_STORAGE = {
    "INFO only": "0.4 MB/day",
    "INFO + WARNING + ERROR": "0.6 MB/day",
    "Full DEBUG for all": "5-6 MB/day",
    "Full prompt in DEBUG for all": "50-60 MB/day",
}
# Conclusion: INFO for all is trivial. Full DEBUG only for specific requests.

Logging system tests

# tests/unit/test_logging.py
import pytest
import json
import io
import structlog

@pytest.fixture
def capture_logs():
    """Captures structlog logs to verify in tests."""
    output = io.StringIO()
    structlog.configure(
        processors=[structlog.processors.JSONRenderer()],
        logger_factory=structlog.PrintLoggerFactory(file=output)
    )
    yield output
    output.close()

def test_llm_request_logs_tokens_and_cost(capture_logs):
    """The completed-request log includes tokens and cost."""
    from src.llm_wrapper import call_llm_with_logging
    mock_client = create_mock_client()
    
    call_llm_with_logging(
        client=mock_client,
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "test"}],
        request_id="test-123"
    )
    
    logs = [json.loads(line) for line in capture_logs.getvalue().strip().split("\n") if line]
    completed_log = next((l for l in logs if l.get("event") == "llm_request_completed"), None)
    
    assert completed_log is not None
    assert "input_tokens" in completed_log
    assert "cost_usd" in completed_log
    assert "duration_ms" in completed_log
    assert completed_log["request_id"] == "test-123"

def test_no_full_prompt_in_info_logs(capture_logs):
    """The full prompt must NOT appear in INFO-level logs."""
    from src.llm_wrapper import call_llm_with_logging
    mock_client = create_mock_client()
    
    secret_prompt = "This is a very secret prompt with PII: juan@example.com"
    call_llm_with_logging(
        client=mock_client,
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": secret_prompt}],
        request_id="test-456",
        debug_mode=False  # Explicitly without debug
    )
    
    log_output = capture_logs.getvalue()
    assert secret_prompt not in log_output  # The full prompt must not appear
    assert "juan@example.com" not in log_output  # PII either

Exercises

Exercise 1: Define a log schema

For each scenario, write the log with the correct level and all the necessary fields:

1. The LLM responded in 12 seconds
2. The injection guardrail blocked a request
3. The LLM's JSON couldn't be parsed

# Scenario 1: High latency
log.warning("high_latency_request",
    request_id=request_id,
    duration_ms=12000,
    model="gpt-4o-mini",
    threshold_ms=10000
)

# Scenario 2: Guardrail activated
log.warning("guardrail_activated",
    request_id=request_id,
    guardrail_type="prompt_injection",
    action="blocked",
    layer="pattern",
    endpoint="/analyze"
)

# Scenario 3: Parse error
log.error("json_parse_failed",
    request_id=request_id,
    error_type="JSONDecodeError",
    response_preview=raw_response[:100],  # Preview, not the full thing
    model="gpt-4o-mini"
)

Exercise 2: Identify incorrect logs

What's wrong with these logs?

log.debug("processing", api_key=os.getenv("OPENAI_API_KEY"))
log.info("user_input", text=request.text)
log.info("llm_done")  # No fields

Exercise 3: Debug mode implementation

Implement a debug mode system that lets you enable verbose logging for a specific request using an X-Debug-Request-ID header:

# Global set of IDs in debug mode (in memory, lost on redeploy)
_debug_request_ids: set = set()

def enable_debug_for_request(request_id: str):
    """Enables debug mode for a specific request."""
    _debug_request_ids.add(request_id)

def is_debug_mode(request_id: str) -> bool:
    return request_id in _debug_request_ids

# In FastAPI:
@app.post("/debug/enable/{request_id}")
async def enable_debug(request_id: str, admin_key: str = Header()):
    if admin_key != os.getenv("ADMIN_KEY"):
        raise HTTPException(403)
    enable_debug_for_request(request_id)
    return {"message": f"Debug enabled for {request_id}"}

Summary

• INFO by default in production: tokens, cost, duration, request_id — no content
• DEBUG only on-demand: full prompt, full response — for specific requests
• WARNING: guardrail activated, fallback used, high latency, high cost
• ERROR: LLM failure without fallback, unhandled exception
• Never log: API keys, unredacted PII, full prompts in INFO
• Estimated storage: INFO summary = ~0.4MB/day/1000 requests — completely manageable

Additional resources

1. structlog Log Levels — Level configuration
2. Python Logging HOWTO — Base logging concepts in Python
3. GDPR and logs (ICO) — Legal considerations for PII in logs
4. The Art of Logging — Practical guide on what to log
5. 12 Factor — Logs — Logging philosophy in cloud-native apps