日志、错误与可观察性

学习目标：让日志成为运行时诊断接口

本节结束时，你能把 JavaScript/TypeScript 的 console.log 迁移到 Python logging，并解释 logger、handler、formatter 和日志级别各自做什么。你会为一次数据 job 设计稳定的结构化上下文，保留异常链和 traceback，避免记录 prompt、token 或原始用户文本，并用运行输出或测试验证日志真的能支持排错，而不是只让终端更热闹。

日志不是程序的第二个返回值，也不是越多越好。它是运行时的诊断接口：在不暂停服务的情况下，告诉我们哪个 job、哪个数据版本、哪个阶段做了什么、用了多久、处理了多少条、为什么失败。AI 数据管线尤其需要这些事实来区分输入漂移、模型服务超时、代码回归和数据脱敏错误。

从 JS/TS 迁移的心智模型：一条日志会经过四层

JS/TS 中 console.info 通常立即写到 stdout；Python 的 logger.info 先创建一条 LogRecord，再由 logger 的级别决定是否接受，接着沿 logger 层级传播给 handler，最后由 formatter 把记录渲染成文本或 JSON。一个 logger 没有 handler 不代表调用失败，只代表这条记录可能没有可见输出；库代码因此不应在 import 时擅自调用 basicConfig。

try {
const result = await run(input);
console.info("run complete", { requestId, count: result.length });
} catch (error) {
console.error("run failed", { requestId, error });
throw error;
}

logger.info("run_complete", extra={"request_id": request_id, "count": len(result)})
try:
  result = run(item)
except Exception:
  logger.exception("run_failed", extra={"request_id": request_id})
  raise

1. logger、handler 与 formatter 的分工

logger 是业务代码拿到的入口，通常用 logging.getLogger(__name__)；handler 决定写到哪里，例如控制台、文件或采集器；formatter 决定时间、级别、logger 名称、消息和上下文怎样排列。logger 的级别是第一道过滤，handler 还可以有自己的级别，因此“设置了 INFO 仍看不到 DEBUG”可能是任一层过滤造成的。propagate=True 时，子 logger 还会把记录交给父 logger，重复 handler 就会产生重复行。

import json
import logging
import sys

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        event = {
            "time": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        for name in ("job_id", "stage", "input_version", "rows", "elapsed_ms"):
            if hasattr(record, name):
                event[name] = getattr(record, name)
        if record.exc_info:
            event["exception"] = self.formatException(record.exc_info)
        return json.dumps(event, ensure_ascii=False)

def configure_logging() -> None:
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.setLevel(logging.INFO)
    root.handlers.clear()
    root.addHandler(handler)

这里的 extra 字段会被放到 LogRecord 上，所以字段名必须是 formatter 预期的名字，且不要覆盖 message、levelname 等保留属性。清空 handler 只适合应用入口的单次配置，不应由可复用库在 import 时执行。若项目使用 JSON 日志库，也仍然要先定义事件字段、敏感字段和异常策略，再选择实现。

2. 日志级别、异常链与 traceback

DEBUG 适合本地复现某个样本的细节，INFO 记录正常阶段，WARNING 记录已隔离但任务可继续的坏行或重试，ERROR 表示当前操作失败。logger.exception 只能在 except 块中正确附带当前异常的 traceback；它不会自动修复异常，也不应该被用来掩盖错误。若要增加业务语义，使用 raise DataReadError(...) from exc，这样新的异常说明“哪一步失败”，__cause__ 仍保留底层 JSON 或网络原因。

import json
import logging

logger = logging.getLogger(__name__)

class BadEvent(ValueError):
    pass

def parse_event(line: str, line_number: int) -> dict[str, object]:
    try:
        value = json.loads(line)
    except json.JSONDecodeError as exc:
        raise BadEvent(f"invalid JSON at line {line_number}") from exc
    if not isinstance(value, dict) or "id" not in value:
        raise BadEvent(f"missing id at line {line_number}")
    return value

def read_event(line: str, line_number: int, job_id: str):
    try:
        return parse_event(line, line_number)
    except BadEvent:
        logger.warning(
            "event_rejected",
            extra={"job_id": job_id, "stage": "parse", "line_number": line_number},
        )
        raise

内部解析函数只在它能补充上下文时包装异常；拥有 job 生命周期的入口负责记录一次完整 traceback。不要在每层都 logger.exception 后再抛出，否则同一个坏样本会产生多份堆栈。日志消息可以写事件名，详细的字段放在结构化上下文中，便于采集器按 job_id、stage 和 line_number 搜索。

3. 结构化上下文：extra、LoggerAdapter 与脱敏

如果每个调用点都手写 extra={"job_id": ...}，很快会漏字段。LoggerAdapter 可以为一组日志预绑定上下文；请求服务则可绑定 request ID、model version 和 tenant 的安全标识。上下文应是低基数、稳定、可索引的字段，例如版本号、计数和阶段；不要把整份输入、Authorization header 或模型回复放进去。需要调试单条样本时，记录不可逆的哈希或内部样本 ID，并遵守保留期限。

同一个字段名还要保持类型稳定：rows 总是整数，elapsed_ms 总是数值，input_version 总是字符串。否则查询系统会把同一个字段拆成多个类型，告警和报表都会变得不可靠。结构化不等于一定要 JSON；关键是事件和字段能被机器稳定解析。

4. 运行验证：输入、输出和可测试行为

可以用固定输入运行 prepare_job：输入 3 行，2 行合法、1 行坏 JSON，预期输出统计为 input=3 kept=2 rejected=1；stderr 中出现三个字段稳定的事件：开始、坏行 warning、完成。日志测试可以使用 pytest 的 caplog，断言 job_id、stage、rows 和 elapsed_ms 存在，同时断言原始文本和 token 不在日志中。

验证顺序是先确认 logger 名称和级别，再确认 handler 数量，再模拟一条异常并检查 traceback，最后检查真实采集格式。若日志只有消息没有上下文，通常是 extra 没有传给正确的 logger 或 formatter 没有读取字段；若输出重复，检查 handler 是否在测试 setup 中被添加了多次以及 propagate 是否仍为 True。

常见错误、排错与调试路径

看不到日志时，先问四个问题：调用点是否真的执行；logger 和 handler 的有效级别是什么；入口是否配置了 handler；输出是否被测试 runner 或容器重定向。看到重复日志时，列出 logging.getLogger().handlers 和目标 logger 的 handlers，检查是否重复配置。看到乱码或 JSON 解析失败时，检查 formatter 是否把异常对象直接塞入 JSON，应该先调用 formatException。

日志突然暴涨通常是把每条样本的完整 payload 以 INFO 写出，或在重试循环里没有限制 warning。调试 AI 服务时记录请求开始、结束、状态码、延迟、重试次数和模型版本即可；把 prompt 长度、输入哈希和输出 schema 摘要作为安全替代。若异常链丢失，检查是否写成 raise NewError(str(exc)) 而不是 raise NewError(...) from exc。

练习：为数据 job 增加诊断上下文

任务是让一次清洗任务在开始、阶段完成和失败时记录 job_id、输入记录数、保留数、丢弃数、数据版本和耗时；坏 JSON 记录 warning 并带行号，无法写出时记录一次完整异常后继续向上抛出。验收日志中不出现原始文本、token 或 Authorization，且同一异常不会被多层重复记录。

logger = logging.getLogger(__name__)
started = time.perf_counter()
logger.info("prepare_started", extra={"job_id": job_id, "stage": "prepare"})
try:
  stats = prepare_job(job_id, rows)
except OSError:
  logger.exception("prepare_failed", extra={"job_id": job_id, "stage": "write"})
  raise
else:
  logger.info("prepare_finished", extra={"job_id": job_id, **stats,
      "elapsed_ms": round((time.perf_counter() - started) * 1000)})

与 AI 数据管线、模型和服务连接

数据抽取、特征生成、训练评估、embedding 批处理和在线推理都需要同一套可观察字段：job 或 request ID、输入 schema 版本、模型版本、阶段、批次大小、延迟、重试次数、接受/拒绝统计。这样当模型指标下降时，可以沿着 ID 回看是字段缺失率增加、SQL 样本变化、HTTP 429 变多，还是模型本身改变。

对推理服务，日志还要区分用户可见错误和内部诊断：返回给客户端的是安全的错误码，内部日志保留异常链和上游状态；对数据任务，把坏记录放入可审核的隔离输出，而不是在日志中复制整条记录。稳定的结构化上下文可以被指标系统聚合成 p95 延迟、拒绝率和重试率，也能为 AI 数据审计提供证据。

小结

Python logging 的核心是 logger 产生事件、handler 选择去处、formatter 定义输出契约、级别控制噪声，异常链保留原因，结构化上下文支持检索。迁移自 JS/TS 时，先定义要回答的问题和脱敏边界，再选择实现；运行验证必须能从输入、输出和 traceback 证明这条日志对调试真的有用。