Bài 45: Logging cho AI app — gì cần log, gì không

Sau bài này bạn sẽ:

• ✅ Hiểu logging cho AI app cần capture gì thêm so với app thông thường
• ✅ Dùng được python-json-logger và structlog để ghi log dạng JSON có cấu trúc
• ✅ Viết FastAPI middleware log HTTP request kèm latency và request ID
• ✅ Áp dụng pattern log cho LLM call (token count, finish reason, latency)
• ✅ Biết field nào không được log (API key, PII) và cách mask
• ✅ Phân biệt logging với tracing, biết khi nào cần thêm LLM observability tool

App thông thường log để debug lỗi và theo dõi performance. AI app cần thêm nhiều loại thông tin khác:

Debug model behavior

Khi model trả output lạ hoặc sai, câu hỏi đầu tiên luôn là: "Input gửi lên là gì?". Không có log, bạn không tái hiện được lỗi. Điều này đặc biệt khó với LLM vì output là probabilistic — cùng input, khác lần chạy có thể khác output.

Audit trail cho regulated industry

Trong y tế (HIPAA), tài chính (SOX, PCI-DSS), hay legal, mỗi quyết định do AI tạo ra cần có audit trail: ai gửi input, lúc nào, model nào đưa ra output gì. Log là bằng chứng khi cần review.

Quality monitoring

Qua log accumulated, bạn phát hiện được pattern như: finish_reason ngày càng nhiều length (model bị truncate) thay vì stop; hoặc error rate tăng đột biến theo giờ cao điểm. Đây là tín hiệu cần điều tra trước khi user phàn nàn.

Cost tracking

Mỗi LLM request có giá tiền theo token. Log token_in và token_out mỗi request giúp tính cost thực tế theo user, theo feature, theo ngày — và phát hiện sớm khi có request anomaly (prompt injection dài bất thường).

Performance baseline

Latency LLM call rất biến động (500ms đến 10s). Log latency từng step giúp xác định bottleneck: chậm ở embedding, ở vector search, hay ở LLM generation. Bài 46 sẽ đi sâu vào metrics, nhưng log là nguồn dữ liệu gốc.

Trong một AI app điển hình có 3 lớp log với mục đích khác nhau:

Application logs

Các sự kiện ứng dụng: startup, shutdown, config load, lỗi hệ thống. Format thường là text hoặc JSON. Đây là log bạn đã quen từ app non-AI.

logger.info("Model loaded", extra={"model": "gpt-4o-mini", "load_time_ms": 230})
logger.error("Database connection failed", exc_info=True)

Request logs

Mỗi HTTP request vào API: method, path, status code, latency, request ID. Thường implement qua middleware ở tầng framework (FastAPI, Flask).

{"method": "POST", "path": "/inference", "status": 200, "latency_ms": 1423, "request_id": "req-abc123"}

AI-specific logs

Thông tin đặc thù của AI: model name/version, token count, finish reason, tool calls, embedding model, retrieval score. Đây là lớp không có trong app thường và cần thiết kế thêm.

{"event": "llm_call", "model": "gpt-4o-mini", "tokens_in": 245, "tokens_out": 178, "latency_ms": 1342, "finish_reason": "stop"}

Python có module logging trong standard library. Không cần cài thêm gì để bắt đầu:

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)

logger.info("Server started")
logger.warning("Config missing, using defaults")
logger.error("Failed to load model", exc_info=True)

Log levels

Python logging có 5 level theo thứ tự tăng dần:

• DEBUG — chi tiết nhất: giá trị biến, flow bên trong function. Không bật trong production.
• INFO — sự kiện bình thường: request đến, model load xong, task hoàn thành.
• WARNING — tình huống bất thường nhưng app vẫn chạy được: retry lần 2, config fallback.
• ERROR — lỗi xảy ra, operation cụ thể thất bại nhưng app vẫn sống.
• CRITICAL — lỗi nghiêm trọng, app có thể không tiếp tục được.

Chiến lược theo môi trường

Không dùng cùng 1 level cho mọi môi trường:

• Development: DEBUG — log mọi thứ để debug dễ.
• Staging: INFO — giống production nhưng có thể bật DEBUG khi cần.
• Production: INFO — log sự kiện quan trọng và ERROR trở lên. DEBUG quá nhiều gây tốn storage và chậm I/O.

Dùng __name__ làm logger name

logging.getLogger(__name__) tạo logger có tên theo module path (vd app.services.llm). Khi đọc log, bạn biết ngay dòng log đến từ module nào mà không cần trace thêm.

Không dùng root logger trực tiếp

Gọi logging.info(...) trực tiếp (root logger) sẽ in log từ tất cả thư viện (httpx, openai, sqlalchemy) vào stdout. Nên tạo logger theo module và cấu hình level riêng:

import logging

# Giảm noise từ thư viện bên ngoài
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("openai").setLevel(logging.WARNING)

# Logger cho code của mình
logger = logging.getLogger("app")
logger.setLevel(logging.INFO)

Log dạng text như "inference complete for user u123, latency 1342ms" khó parse khi cần filter, aggregate hoặc tìm kiếm. Structured logging ghi log dạng JSON — mỗi field là 1 key riêng biệt.

Cài thư viện

pip install python-json-logger

Setup

import logging
from pythonjsonlogger import jsonlogger

logger = logging.getLogger("app")
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    "%(asctime)s %(levelname)s %(name)s %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Log với extra fields
logger.info("inference_complete", extra={
    "user_id": "u123",
    "model": "gpt-4o-mini",
    "tokens_in": 245,
    "tokens_out": 178,
    "latency_ms": 1342,
})

Output:

{
  "asctime": "2026-05-27 10:23:41,512",
  "levelname": "INFO",
  "name": "app",
  "message": "inference_complete",
  "user_id": "u123",
  "model": "gpt-4o-mini",
  "tokens_in": 245,
  "tokens_out": 178,
  "latency_ms": 1342
}

JSON log dễ ingest vào bất kỳ log aggregator nào: ELK Stack (Elasticsearch + Logstash + Kibana), Datadog, Splunk, Grafana Loki. Khi có Kibana hay Datadog, bạn filter model:gpt-4o-mini AND latency_ms:>2000 ngay trong UI.

structlog là thư viện logging hiện đại hơn, được thiết kế từ đầu cho structured logging. API gọn hơn và hỗ trợ context binding tự động.

pip install structlog

import structlog

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer(),
    ]
)

logger = structlog.get_logger()

# Gọi trực tiếp với keyword args — không cần extra={}
logger.info("inference_complete", user_id="u123", model="gpt-4o-mini", latency_ms=1342)

Context binding

Một tính năng của structlog là bind context một lần, tự động gắn vào mọi log tiếp theo trong cùng request:

bound_logger = logger.bind(request_id="req-abc123", user_id="u456")

# Cả 2 dòng dưới tự động có request_id và user_id
bound_logger.info("retrieving_documents", collection="knowledge_base")
bound_logger.info("llm_call_start", model="gpt-4o-mini")

So sánh nhanh

Với project nhỏ hoặc dùng chung codebase với stdlib logging, python-json-logger đủ dùng. Với project mới, structlog ergonomic hơn.

Bài 4 đã đề cập async endpoint. Với logging, cần thêm middleware để log mỗi request mà không cần viết code log trong từng endpoint:

import time
import logging
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger("app.http")

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        request_id = request.headers.get("X-Request-Id", "")

        response = await call_next(request)

        latency_ms = (time.perf_counter() - start) * 1000
        logger.info("http_request", extra={
            "method": request.method,
            "path": request.url.path,
            "status": response.status_code,
            "latency_ms": round(latency_ms, 2),
            "request_id": request_id,
        })
        return response

app.add_middleware(LoggingMiddleware)

Lưu ý quan trọng

• Dùng time.perf_counter() thay vì time.time() cho đo latency — độ chính xác cao hơn.
• Không đọc request.body() trong middleware vì sau đó endpoint sẽ không đọc được body nữa (stream đã consumed). Nếu cần log body, dùng background task hoặc clone stream.
• Bỏ qua endpoint health check để tránh spam log từ load balancer (xem Section 16).

Exclude health check

SKIP_LOG_PATHS = {"/health", "/metrics", "/ping"}

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if request.url.path in SKIP_LOG_PATHS:
            return await call_next(request)

        start = time.perf_counter()
        # ... rest of logging logic

Khi AI app gọi nhiều service (vector DB, LLM API, SQL DB), log phân tán ở nhiều chỗ. Để ghép lại thành 1 request trace, mỗi request cần có unique ID gắn vào toàn bộ log liên quan.

Pattern cơ bản

import uuid
from fastapi import Request

def get_or_create_request_id(request: Request) -> str:
    return request.headers.get("X-Request-Id") or str(uuid.uuid4())

Khi client gửi kèm header X-Request-Id, giữ nguyên giá trị đó (end-to-end traceability). Nếu không có, tự sinh UUID mới. Request ID này phải được:

• Gắn vào tất cả log trong suốt vòng đời request.
• Truyền tiếp qua header khi gọi downstream service.
• Gắn vào error response để user (hay frontend) có thể report.

Dùng asgi-correlation-id

Thư viện asgi-correlation-id tự động inject request ID vào mọi log trong cùng async context — không cần truyền tay qua từng function:

pip install asgi-correlation-id

from asgi_correlation_id import CorrelationIdMiddleware

app.add_middleware(CorrelationIdMiddleware)
# Sau đó trong bất kỳ log nào trong request, correlation_id tự có

Truyền xuống LLM call

from asgi_correlation_id import correlation_id

def call_llm(messages):
    headers = {"X-Request-Id": correlation_id.get()}
    # Truyền vào metadata openai client (nếu SDK hỗ trợ)
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=messages,
        extra_headers=headers,
    )
    return response

Mỗi LLM call cần được wrap để capture đủ thông tin trước và sau khi gọi API:

import time
import logging
from asgi_correlation_id import correlation_id

logger = logging.getLogger("app.llm")

def call_llm(messages: list[dict], model: str = "gpt-4o-mini") -> object:
    start = time.perf_counter()
    req_id = correlation_id.get()

    try:
        response = openai_client.chat.completions.create(
            model=model,
            messages=messages,
        )

        latency_ms = (time.perf_counter() - start) * 1000
        logger.info("llm_call_success", extra={
            "request_id": req_id,
            "model": model,
            "tokens_in": response.usage.prompt_tokens,
            "tokens_out": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
            "latency_ms": round(latency_ms, 2),
            "finish_reason": response.choices[0].finish_reason,
            # Không log full prompt/response — xem Section 10
        })
        return response

    except Exception as e:
        latency_ms = (time.perf_counter() - start) * 1000
        logger.error("llm_call_failed", extra={
            "request_id": req_id,
            "model": model,
            "error_type": type(e).__name__,
            "error": str(e),
            "latency_ms": round(latency_ms, 2),
        }, exc_info=True)
        raise

finish_reason — tại sao quan trọng

OpenAI API trả về finish_reason cho mỗi response. Các giá trị phổ biến:

• stop — model tự dừng khi kết thúc câu trả lời. Trạng thái bình thường.
• length — response bị cắt ngang vì chạm giới hạn max_tokens. Cần tăng limit hoặc rút gọn prompt.
• tool_calls — model muốn gọi tool. Cần xử lý tiếp.
• content_filter — OpenAI filter chặn output. Cần xem lại prompt hoặc báo user.

Nếu finish_reason=length xuất hiện nhiều trong log, đó là dấu hiệu cần review cấu hình max_tokens.

Cần log

• Timestamp (ISO 8601, có timezone).
• request_id / correlation ID.
• user_id — nếu có consent của user và pháp lý cho phép.
• Service name + version.
• Model name + version (vd gpt-4o-mini-2024-07-18).
• Token count: tokens_in, tokens_out.
• Latency từng step (embedding, retrieval, LLM generation).
• finish_reason.
• Error type + stack trace khi có exception.
• Tool calls + result (truncated nếu quá dài).
• Inference hyperparameters: temperature, max_tokens, top_p.

Không nên log

• API key, password, bearer token — lỗi phổ biến nhất: key lọt vào log qua exception message. Phải sanitize trước khi log.
• PII trực tiếp: số CMND, số thẻ tín dụng, địa chỉ nhà, số điện thoại. Dùng mask (xem Section 11).
• Full prompt khi chứa PII của user: người dùng có thể dán CMND hay thông tin cá nhân vào prompt.
• System prompt đầy đủ: nhiều system prompt chứa business logic hoặc IP của công ty.
• Medical / legal content trong regulated environment nếu không có consent rõ ràng.

Truncate thay vì skip hoàn toàn

Đôi khi cần log một phần nội dung để debug nhưng không muốn log toàn bộ:

def truncate_content(text: str, max_chars: int = 200) -> str:
    if len(text) <= max_chars:
        return text
    return text[:100] + "...[truncated]..." + text[-100:]

logger.debug("llm_prompt_preview", extra={
    "prompt_preview": truncate_content(prompt),
    "prompt_length": len(prompt),
    "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest()[:16],
})

Hash của full content giúp correlation sau này nếu cần reproduce — mà không lưu raw content trong log.

Khi cần log nội dung user input (vd để debug), phải mask PII trước. Cách đơn giản là dùng regex để thay thế pattern nhạy cảm:

import re

SENSITIVE_PATTERNS = {
    "email": re.compile(r"\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b"),
    "phone_vn": re.compile(r"\b(0|\+84)[0-9]{8,9}\b"),
    "credit_card": re.compile(r"\b\d{13,16}\b"),
    "id_card_vn": re.compile(r"\b\d{9}(?:\d{3})?\b"),  # CMND 9 hoặc 12 số
}

def mask_sensitive(text: str) -> str:
    for field_name, pattern in SENSITIVE_PATTERNS.items():
        label = f"[{field_name.upper()}_REDACTED]"
        text = pattern.sub(label, text)
    return text

# Sử dụng
safe_input = mask_sensitive(user_input)
logger.info("user_query", extra={"query": safe_input})

Lưu ý về regex PII

Regex PII không bao giờ đủ 100% vì ngôn ngữ tự nhiên rất đa dạng. Đây là lớp phòng thủ thêm, không phải giải pháp toàn diện. Với system yêu cầu compliance cao (HIPAA, GDPR), cần xem xét dedicated PII detection tool như Microsoft Presidio.

Mask API key trong exception

API key hay lọt vào log qua exception message. Ví dụ khi OpenAI API trả về lỗi 401, message có thể chứa fragment của key:

import os

def sanitize_error_message(message: str) -> str:
    """Thay thế các API key trong error message."""
    api_key = os.getenv("OPENAI_API_KEY", "")
    if api_key and api_key in message:
        message = message.replace(api_key, "[API_KEY_REDACTED]")
    return message

try:
    response = openai_client.chat.completions.create(...)
except Exception as e:
    safe_msg = sanitize_error_message(str(e))
    logger.error("llm_error", extra={"error": safe_msg})

Log cần đi đến đúng chỗ tùy môi trường và scale:

stdout / stderr

Container thường log ra stdout/stderr. Docker và Kubernetes tự collect và có thể pipe sang log driver. Đây là cách đơn giản nhất và phù hợp với 12-Factor App principle ("Treat logs as event streams").

File với log rotation

Nếu không chạy trên container hoặc cần log file local:

import logging
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    "app.log",
    maxBytes=10 * 1024 * 1024,  # 10 MB mỗi file
    backupCount=5,               # Giữ 5 file cũ
)

Centralized log service

Khi có nhiều instance hoặc nhiều service, cần aggregated view:

• ELK Stack: Elasticsearch + Logstash + Kibana. Self-hosted, phức tạp nhưng miễn phí.
• Grafana Loki: Log aggregation nhẹ hơn ELK, tích hợp tốt với Grafana. Self-hosted hoặc Grafana Cloud.
• Datadog: SaaS, tốn tiền nhưng setup nhanh, query mạnh.
• Splunk: Phổ biến ở enterprise, giá cao.

Cloud native

• AWS CloudWatch Logs: tự động collect từ Lambda, ECS, EC2.
• GCP Cloud Logging: tích hợp với GKE, Cloud Run.
• Azure Monitor / Application Insights: cho Azure services.

Pattern phổ biến

App ghi log ra stdout → collector (Fluentd, Vector, Fluent Bit) collect từ container → forward đến centralized service. Collector chạy như DaemonSet trên Kubernetes.

Log service tính phí theo GB ingested. Với app có traffic cao, cần kiểm soát volume:

Sampling

Không nhất thiết phải log mọi request thành công ở INFO level:

import random

def should_log_request(sample_rate: float = 0.1) -> bool:
    """Log 10% request thành công, 100% lỗi."""
    return random.random() < sample_rate

# Trong middleware
if response.status_code >= 400 or should_log_request():
    logger.info("http_request", extra={...})

Luôn log 100% error, chỉ sample request thành công. Nếu muốn accurate latency metrics, dùng Prometheus (bài 49) thay vì dựa vào log.

Log retention policy

Không nên giữ tất cả log mãi mãi:

• Hot tier (Elasticsearch, Datadog): 7–30 ngày — tìm kiếm nhanh, giá cao.
• Cold tier (S3, GCS): 1–2 năm — archive, giá thấp nhưng query chậm.
• Compliance yêu cầu (HIPAA, SOX) có thể định nghĩa retention tối thiểu khác.

Log level control runtime

Tránh restart service để thay đổi log level. Có thể dùng endpoint nội bộ:

from fastapi import APIRouter, Depends
import logging

internal_router = APIRouter()

@internal_router.post("/admin/log-level")
def set_log_level(level: str):
    """Thay đổi log level không cần restart — chỉ expose nội bộ."""
    numeric_level = getattr(logging, level.upper(), None)
    if not isinstance(numeric_level, int):
        return {"error": f"Invalid level: {level}"}
    logging.getLogger("app").setLevel(numeric_level)
    return {"level": level.upper()}

Logging và tracing là 2 khái niệm khác nhau, dù đều liên quan đến observability:

OpenTelemetry

OpenTelemetry (OTel) là standard open-source cho observability: logs, metrics, traces dùng chung SDK và protocol (OTLP). Với Python:

pip install opentelemetry-sdk opentelemetry-instrumentation-fastapi

OTel auto-instrument FastAPI, thêm trace ID vào mỗi request, export sang Jaeger hoặc Tempo. Nếu đã dùng OTel tracing, bạn có thể tự động inject trace_id vào log để correlate giữa 2 hệ thống.

Bài này tập trung logging. Tracing sâu hơn sẽ được đề cập trong bài 49 (Prometheus + Grafana).

General-purpose logging không capture tốt ngữ cảnh LLM: chain của LangChain, multi-step agent, prompt templates. Có 2 tool phổ biến chuyên cho LLM observability:

LangSmith

Sản phẩm của LangChain, tích hợp tự nhiên với LangChain/LangGraph. Tự động capture: prompt, response, latency, token count, cost cho toàn bộ chain kể cả sub-call. UI hiển thị trace tree từng step. Có thể tạo evaluation dataset từ log thực tế và chạy automated scoring. Cần account LangSmith, có free tier.

Langfuse

Open source, có thể self-host hoặc dùng cloud. Hỗ trợ nhiều framework (LangChain, LlamaIndex, raw OpenAI SDK). SDK nhẹ, dễ tích hợp thủ công bằng decorator. Có eval, prompt management, dataset tích hợp trong cùng UI.

Cả 2 tool không thay thế logging thông thường — chúng là lớp observability bổ sung đặc thù cho LLM call. Application logs và request logs vẫn cần duy trì riêng.

Dùng print() thay vì logger

print() không có level, không có timestamp, không có formatter, không thể tắt theo môi trường. Khi deploy production, bạn không thể disable tất cả print() mà không sửa code.

# Sai
print(f"LLM response: {response.choices[0].message.content}")

# Đúng
logger.debug("llm_response_preview", extra={"preview": truncate_content(response.choices[0].message.content)})

Log interpolated string thay vì extra dict

# Sai — text khó parse, không structured
logger.info(f"User {user_id} called model {model} in {latency_ms}ms")

# Đúng — field riêng biệt, dễ query
logger.info("llm_call", extra={"user_id": user_id, "model": model, "latency_ms": latency_ms})

Log API key qua exception

# Nguy hiểm — nếu e chứa API key fragment
logger.error(f"OpenAI error: {e}")

# An toàn
logger.error("openai_error", extra={"error": sanitize_error_message(str(e))})

Log 100% full prompt/response

Mỗi LLM call có thể có prompt 2000 token (~6000 ký tự). Log toàn bộ với traffic 1000 req/s → vài GB log/giờ. Dùng truncate + hash như đề cập ở Section 10.

Quên timezone trong timestamp

Log không có timezone gây khó trace khi team ở nhiều múi giờ khác nhau hoặc server ở UTC nhưng bạn đọc log theo local time. Luôn dùng UTC hoặc ISO 8601 với timezone offset (2026-05-27T10:23:41+00:00).

Không exclude health check endpoint

Load balancer thường probe /health mỗi 5–10 giây. 1 instance 10 req/phút → 86,400 request/ngày log nhảm. Luôn exclude path này khỏi request logging middleware.

Logger không thread-safe khi dùng FileHandler

Python logging module đã thread-safe. Nhưng nếu dùng custom handler hoặc thư viện bên ngoài, kiểm tra thread safety. Với async FastAPI, dùng asyncio-safe logger hoặc background task để tránh block event loop khi ghi I/O.

• ✅ AI app cần 3 lớp log: application, request, AI-specific (token, latency, finish_reason)
• ✅ Dùng python-json-logger hoặc structlog để ghi log JSON có cấu trúc
• ✅ FastAPI middleware log mọi request tập trung — tránh log lặp trong từng endpoint
• ✅ Correlation ID giúp trace log xuyên suốt từ HTTP request đến LLM call
• ✅ Không log API key, PII trực tiếp — mask trước khi log
• ✅ Truncate prompt/response dài, log hash để tái tạo khi cần
• ✅ Log stdout → collector → centralized service là pattern chuẩn cho container
• ✅ Sampling request thành công, log 100% lỗi để kiểm soát volume
• ✅ LangSmith / Langfuse bổ sung observability riêng cho LLM chain — không thay thế logging thường

Bài 46: Metrics — latency, throughput, error rate — chuyển từ log sang metrics: cách đo và expose các chỉ số hiệu năng cho AI service.