Python 的默认日志与 logging 模块

一、问题

1.1 由一道日志题引发的学习

1. Python 默认日志有什么缺陷？

2. logging 模块在 AI 服务端生产日志中应该如何分级配置？

原参考答案：

默认缺陷：日志无链路 ID、无时间戳格式化、文件不切割、无法区分业务/模型/接口日志、线上故障无法溯源。
生产分级配置：
DEBUG：本地开发，打印模型入参、向量召回详情；
INFO：生产默认级别，记录接口调用、知识库更新、会话创建；
WARNING：模型调用超时、检索召回量不足、内存水位偏高；
ERROR：接口崩溃、向量库连接中断、模型推理失败，绑定链路 ID 告警。

这个问题看起来是在问“日志级别怎么用”，但背后其实牵涉到服务端开发中非常核心的一组能力：

  • print()、默认日志、logging 模块之间的差异
  • 日志级别的设计原则
  • Logger、Handler、Formatter、Filter 的职责
  • 日志格式、日志文件、日志切割
  • 异常堆栈记录
  • 请求链路 ID / request_id / trace_id
  • 模块化日志与不同业务日志分流
  • AI / RAG 服务中的模型调用日志、向量检索日志、接口日志
  • 结构化日志与线上可观测性
  • 日志安全：敏感信息脱敏、避免泄露 token 和隐私数据

因此，本文不只是为了回答这道面试题，而是以它为入口，系统梳理 Python logging 模块从基础使用到生产实践的完整知识。

1.2 简要回答

Python 中最简单的“默认日志”通常指两类：

  1. 使用 print() 直接输出信息；
  2. 直接调用 logging.warning()、logging.error() 等模块级函数，而没有做完整日志配置。

它们在小脚本中够用，但在生产服务中问题很多：

问题 说明
缺少统一格式 不一定有时间、模块名、日志级别、进程线程信息
默认级别较高 默认 root logger 级别是 WARNING，DEBUG、INFO 默认不会输出
难以定位请求 没有 request_id / trace_id，无法串起一次请求的完整链路
不适合文件管理 默认不负责日志文件切割、归档、保留策略
不区分业务模块 接口、模型、向量库、数据库日志混在一起，不利于检索
异常信息可能不完整 如果只打印错误消息而不打印堆栈，线上很难定位根因
不便于接入告警 无结构化字段，不利于 ELK、Loki、云日志平台、APM 分析

生产环境中应该使用 logging 建立统一日志体系。AI 服务端常见分级可以这样设计：

级别 使用场景 AI 服务端示例
DEBUG 开发调试、详细内部状态 模型入参摘要、prompt 片段、向量召回详情、SQL 参数
INFO 正常业务事件 接口调用、会话创建、知识库更新、模型调用成功、任务完成
WARNING 可恢复但需要关注的问题 模型调用超时后重试、召回数量不足、内存水位偏高、限流接近阈值
ERROR 当前请求或任务失败 接口异常、向量库连接失败、模型推理失败、数据库写入失败
CRITICAL 系统级严重故障 核心依赖不可用、服务无法启动、配置缺失导致系统整体不可用

一句话总结：

默认日志适合临时调试，logging 才适合工程化管理；生产日志必须做到有级别、有格式、有链路 ID、有文件策略、有异常堆栈、有模块区分、有安全边界。

二、为什么这个问题值得深入理解？

2.1 日志不是“随便打印一下”

初学阶段，我们经常这样调试：

print("进入函数")
print("user_id=", user_id)
print("result=", result)

这种方式在本地小脚本里很直接，但一旦进入真实服务端场景，就会暴露问题：

  • 输出格式不统一；
  • 不知道日志来自哪个模块；
  • 不知道是普通信息、警告还是错误；
  • 多个请求并发时日志交织在一起；
  • 线上服务重启后终端输出可能丢失；
  • 无法按级别过滤；
  • 无法方便接入日志平台；
  • 无法设置日志文件大小、保留天数和切割策略。

服务端日志的目的不是“给开发者看一眼”，而是为了在问题发生时回答几个关键问题：

  1. 什么时候发生的？
  2. 哪个请求发生的？
  3. 哪个用户 / 会话 / 任务触发的？
  4. 执行到了哪个模块？
  5. 外部依赖是否正常？
  6. 错误堆栈是什么？
  7. 问题影响范围多大？
  8. 是否需要告警和人工介入？

2.2 AI 服务端为什么更依赖日志？

普通 Web 服务已经需要日志，AI 服务端更需要日志。

因为 AI / RAG / Agent 服务往往包含更多不稳定因素：

  • 大模型接口可能超时、限流、返回异常；
  • prompt 较长，问题定位依赖上下文；
  • RAG 检索可能召回不足、召回错误、相似度过低；
  • 向量数据库、对象存储、关系数据库都可能成为故障点；
  • 一个用户请求可能经过多个阶段：鉴权、检索、重排、拼接 prompt、模型推理、后处理、存储会话；
  • 同一个错误可能不是代码 bug，而是模型输出不稳定、数据质量差、第三方服务波动。

如果没有良好的日志，一句“模型回答不对”几乎无法定位。

一个比较完整的 AI 服务端日志，至少应该能看到：

  • request_id：这一次请求的唯一标识；
  • user_id / session_id：哪个用户、哪个会话；
  • kb_id：使用了哪个知识库；
  • query 摘要：用户问题的大致内容；
  • retrieval_count：召回了多少条；
  • top_score：最高相似度是多少；
  • model：调用了哪个模型；
  • latency_ms：模型耗时多少；
  • input_tokens / output_tokens：token 消耗；
  • error：失败原因；
  • stacktrace：异常堆栈。

三、默认日志、print() 与 logging 的差异

3.1 print() 的特点

print() 是最简单的输出方式：

name = "Tom"
print("create user", name)

它的优点是简单直接，但它不是专业日志系统。

对比点 print()
是否有日志级别 没有
是否能按级别过滤 不能
是否有统一格式 需要自己拼接
是否能输出到多个目的地 不方便
是否适合日志切割 不适合
是否适合异常堆栈 需要手动处理
是否适合线上服务 不推荐

print() 适合：

  • 学习语法；
  • 临时验证变量；
  • 一次性脚本的简单输出。

不适合：

  • Web 服务；
  • 后台任务；
  • 长期运行进程；
  • 需要排查线上问题的系统。

3.2 Python 默认 logging 行为

如果直接使用 logging 模块级函数：

import logging

logging.debug("debug message")
logging.info("info message")
logging.warning("warning message")
logging.error("error message")

通常只会看到：

WARNING:root:warning message
ERROR:root:error message

原因是：

  1. 默认 root logger 的日志级别是 WARNING；
  2. DEBUG 和 INFO 低于 WARNING，默认不会输出；
  3. 默认输出到标准错误流；
  4. 默认格式比较简单，类似 LEVEL:logger_name:message。

也就是说，Python 的默认 logging 并不是不能用，而是默认配置太简单，只能满足最基础的控制台输出。

3.3 默认日志的缺陷

默认日志在生产中主要缺陷如下：

缺陷 影响
没有标准化时间格式 难以按时间排查问题
没有链路 ID 多个请求日志混在一起，无法追踪单次请求
没有模块区分 不知道日志来自接口、模型、数据库还是向量检索
没有文件切割 长期运行后日志文件可能无限增长
没有结构化字段 日志平台不容易检索和统计
没有异常堆栈 只能看到错误描述，看不到错误发生位置
没有敏感信息控制 容易把 token、密码、用户隐私写入日志
没有不同环境策略 开发、测试、生产可能需要不同级别和输出方式

3.4 print()、默认 logging、完整 logging 配置对比

能力 print() 默认 logging 完整 logging 配置
日志级别 无 有，默认 WARNING 可自定义
输出格式 手动拼接 默认简单格式 可统一配置
输出位置 标准输出 标准错误 控制台、文件、远程服务等
模块区分 手动写 默认 root getLogger(__name__)
文件切割 无 无 可用 rotating handler
异常堆栈 手动处理 支持但需正确使用 统一规范
链路 ID 无 无 可通过 Filter 注入
日志平台接入 不方便 较弱 适合结构化采集
生产可维护性 差 一般 强

四、logging 基础使用方法

4.1 五个常用日志级别

Python 标准库 logging 内置了五个常用级别：

import logging

logging.debug("调试信息")
logging.info("普通信息")
logging.warning("警告信息")
logging.error("错误信息")
logging.critical("严重错误")

级别从低到高：

DEBUG < INFO < WARNING < ERROR < CRITICAL

日志系统会根据配置的阈值决定哪些日志可以输出。

例如配置为 INFO 时：

  • DEBUG 不输出；
  • INFO、WARNING、ERROR、CRITICAL 输出。

4.2 使用 basicConfig() 快速配置

最简单的配置方式是 logging.basicConfig()：

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)

logging.debug("debug message")
logging.info("service started")
logging.warning("memory usage is high")
logging.error("database connection failed")

可能输出：

2026-07-09 10:00:00,123 | INFO | root | service started
2026-07-09 10:00:00,124 | WARNING | root | memory usage is high
2026-07-09 10:00:00,125 | ERROR | root | database connection failed

常见格式字段：

字段 含义
%(asctime)s 日志时间
%(levelname)s 日志级别名称
%(name)s logger 名称
%(module)s 模块名
%(funcName)s 函数名
%(lineno)d 行号
%(message)s 日志消息
%(process)d 进程 ID
%(threadName)s 线程名

4.3 输出到文件

import logging

logging.basicConfig(
    filename="app.log",
    filemode="a",
    encoding="utf-8",
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)

logging.info("application started")
logging.error("something failed")

说明：

  • filename="app.log"：写入文件；
  • filemode="a"：追加模式；
  • encoding="utf-8"：避免中文乱码；
  • level=logging.INFO：输出 INFO 及以上级别。

不过，basicConfig(filename=...) 只适合简单程序。生产服务更推荐使用 Handler 或 dictConfig()。

4.4 推荐使用 getLogger(__name__)

在真实项目中，不推荐所有地方都直接写：

logging.info("message")

更推荐每个模块创建自己的 logger：

import logging

logger = logging.getLogger(__name__)


def create_user(user_id: int):
    logger.info("create user, user_id=%s", user_id)

__name__ 是当前模块名。

如果文件结构如下：

project/
  api/user.py
  service/chat.py
  service/retrieval.py

那么不同模块的 logger 名可能是：

api.user
service.chat
service.retrieval

这样日志中就能看出消息来自哪个模块。

4.5 为什么推荐占位符，而不是 f-string？

日志中经常看到这种写法：

logger.info("user login, user_id=%s", user_id)

而不是：

logger.info(f"user login, user_id={user_id}")

原因是 logging 会延迟格式化。只有当这条日志真的需要输出时，才会把参数拼到字符串里。

这在 DEBUG 日志很多、生产环境又关闭 DEBUG 时尤其有意义。

推荐写法：

logger.debug("retrieval result: query=%s, docs=%s", query, docs)

不推荐：

logger.debug(f"retrieval result: query={query}, docs={docs}")

当然，对于普通业务代码来说，性能差异通常不是最大问题，但使用占位符是更符合 logging 习惯的写法。

4.6 记录异常堆栈

错误日志最重要的不是一句“出错了”，而是完整堆栈。

不推荐：

try:
    1 / 0
except ZeroDivisionError as e:
    logger.error("calculate failed: %s", e)

这只能看到错误信息，看不到完整调用链。

推荐：

try:
    1 / 0
except ZeroDivisionError:
    logger.exception("calculate failed")

logger.exception() 只能在 except 块中使用，它会自动带上当前异常堆栈。

等价写法：

try:
    1 / 0
except ZeroDivisionError:
    logger.error("calculate failed", exc_info=True)

输出中会包含类似：

Traceback (most recent call last):
  File "app.py", line 10, in <module>
    1 / 0
ZeroDivisionError: division by zero

自动包含：
出错文件名
出错代码行号
完整函数调用栈
异常类型 + 异常信息

线上排障时，这个堆栈非常关键。

五、logging 核心组件模型

logging 模块的强大之处在于它不是简单打印，而是一套可扩展的日志处理流程。

一次日志调用大致经过：

业务代码
  ↓
Logger
  ↓
Filter
  ↓
Handler
  ↓
Formatter
  ↓
控制台 / 文件 / 远程服务

5.1 Logger：日志入口

Logger 是业务代码直接使用的对象。

import logging

logger = logging.getLogger("service.chat")

logger.info("chat request received")
logger.error("chat request failed")

它负责判断：

  • 当前日志级别是否应该处理；
  • 这条日志交给哪些 handler；
  • 是否向父 logger 传播。

5.2 Handler：日志输出目的地

Handler 决定日志输出到哪里。

常见 handler：

Handler 作用
StreamHandler 输出到控制台
FileHandler 输出到文件
RotatingFileHandler 按文件大小切割
TimedRotatingFileHandler 按时间切割
QueueHandler 放入队列，适合异步或多进程场景
SMTPHandler 通过邮件发送日志
SysLogHandler 输出到 syslog

示例：同时输出到控制台和文件：

import logging

logger = logging.getLogger("app")
logger.setLevel(logging.INFO)

formatter = logging.Formatter(
    "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)

console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)

file_handler = logging.FileHandler("app.log", encoding="utf-8")
file_handler.setLevel(logging.ERROR)
file_handler.setFormatter(formatter)

logger.addHandler(console_handler)
logger.addHandler(file_handler)

logger.info("this message goes to console")
logger.error("this message goes to console and file")

这里：

  • INFO 日志只进入控制台；
  • ERROR 日志同时进入控制台和文件。

5.3 Formatter：日志格式

Formatter 决定日志最终长什么样。

formatter = logging.Formatter(
    fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

示例输出：

2026-07-09 10:00:00 | INFO     | service.chat | ask:42 | chat success

生产中建议至少包含：

  • 时间；
  • 级别；
  • logger 名称；
  • 模块 / 函数 / 行号；
  • request_id；
  • message。

5.4 Filter：过滤或补充上下文

Filter 可以决定一条日志是否输出，也可以给日志记录补充字段。

例如给每条日志加上 request_id：

import logging

class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = "req-001"
        return True


logger = logging.getLogger("app")
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
handler.addFilter(RequestIdFilter())
handler.setFormatter(logging.Formatter(
    "%(asctime)s | %(levelname)s | %(request_id)s | %(name)s | %(message)s"
))

logger.addHandler(handler)
logger.info("hello")

输出：

2026-07-09 10:00:00,123 | INFO | req-001 | app | hello

后面我们会用 contextvars 实现真正的请求级 request_id。

5.5 LogRecord：一条日志事件

当你调用：

logger.info("user login, user_id=%s", user_id)

logging 内部会创建一个 LogRecord 对象。

它里面包含：

  • 日志级别；
  • 日志内容；
  • logger 名；
  • 文件路径；
  • 行号；
  • 函数名；
  • 时间戳；
  • 线程、进程信息；
  • 异常信息；
  • 自定义字段。

Formatter 最终就是从 LogRecord 中取字段，拼成字符串。

5.6 日志传播与重复打印问题

Logger 是有层级的。

例如：

service
service.chat
service.retrieval

service.chat 是 service 的子 logger。

默认情况下，子 logger 的日志会向父 logger 传播，这叫 propagate。

如果你同时给子 logger 和 root logger 都加了 handler，就可能出现重复输出。

示例：

import logging

root = logging.getLogger()
root.setLevel(logging.INFO)
root.addHandler(logging.StreamHandler())

logger = logging.getLogger("service.chat")
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())

logger.info("hello")

可能输出两次。

解决方式之一：

logger.propagate = False

更常见的实践是：

  • 应用入口统一配置 root logger；
  • 各模块只 getLogger(__name__)，不要随便加 handler；
  • 如果某个 logger 单独配置 handler，再考虑关闭 propagate。

六、日志级别应该如何设计？

6.1 DEBUG：调试细节

DEBUG 用于开发和问题排查，内容通常较详细。

适合记录：

  • 函数入参；
  • 中间计算结果；
  • SQL 参数；
  • prompt 片段；
  • 向量召回详情；
  • 重排前后的文档列表；
  • 调用外部接口前后的原始响应摘要。

示例：

logger.debug(
    "retrieval detail, query=%s, kb_id=%s, top_k=%s, scores=%s",
    query,
    kb_id,
    top_k,
    scores,
)

注意：生产环境通常不长期打开 DEBUG，因为：

  • 日志量巨大；
  • 影响性能；
  • 可能泄露敏感信息；
  • 增加日志存储成本。

6.2 INFO：正常业务事件

INFO 用于记录系统正常运行中的关键事件。

适合记录：

  • 服务启动成功；
  • 用户登录；
  • 接口调用；
  • 会话创建；
  • 知识库更新；
  • 文件上传完成；
  • 模型调用成功；
  • 后台任务完成。

示例：

logger.info(
    "chat completed, user_id=%s, session_id=%s, model=%s, latency_ms=%s",
    user_id,
    session_id,
    model,
    latency_ms,
)

生产环境通常默认使用 INFO 级别。

6.3 WARNING：异常苗头，但系统还能继续

WARNING 表示发生了不符合预期的情况，但系统仍然可以继续运行。

适合记录：

  • 模型调用超时，但重试成功；
  • 向量检索召回数量不足；
  • 相似度低于阈值；
  • 缓存未命中率异常升高；
  • 内存水位偏高；
  • 第三方接口响应变慢；
  • 用户输入过长，被截断处理。

示例：

if len(docs) < min_docs:
    logger.warning(
        "retrieval result too few, query=%s, kb_id=%s, expected=%s, actual=%s",
        query,
        kb_id,
        min_docs,
        len(docs),
    )

WARNING 的重点是：需要关注，但不一定代表当前请求失败。

6.4 ERROR：当前操作失败

ERROR 表示某个请求、任务或操作失败，需要定位原因。

适合记录：

  • 接口处理失败；
  • 数据库写入失败；
  • 向量库连接中断；
  • 模型推理失败；
  • 文件解析失败；
  • 消息队列消费失败。

示例：

try:
    result = call_llm(prompt)
except TimeoutError:
    logger.exception("llm call failed, model=%s", model)
    raise

注意：不是所有“不符合预期”都应该用 ERROR。

例如：

  • 用户输错密码：通常是 INFO 或 WARNING；
  • 请求参数校验失败：通常是 INFO 或 WARNING；
  • 检索没有结果但系统能正常返回兜底回答：通常是 WARNING；
  • 第三方接口失败导致当前请求无法完成：应该是 ERROR。

6.5 CRITICAL：系统级严重故障

CRITICAL 表示系统处于非常严重的异常状态。

适合记录：

  • 服务无法启动；
  • 核心配置缺失；
  • 数据库整体不可用；
  • 模型服务全局不可用；
  • 磁盘空间耗尽；
  • 消息队列完全不可连接。

示例：

try:
    load_required_config()
except Exception:
    logger.critical("service startup failed: required config missing", exc_info=True)
    raise

CRITICAL 通常应该触发告警。

6.6 AI 服务端日志级别建议表

场景 推荐级别 示例
打印 prompt 调试片段 DEBUG prompt built, tokens=1200
向量召回文档得分 DEBUG retrieval scores=[0.91, 0.83]
用户发起对话请求 INFO chat request received
模型调用成功 INFO llm call success, latency_ms=820
知识库更新完成 INFO kb updated, file_count=10
检索结果为空 WARNING retrieval empty
模型调用超时但重试成功 WARNING llm timeout, retry success
向量库连接失败 ERROR vector db connection failed
模型推理失败 ERROR llm generation failed
服务启动失败 CRITICAL application startup failed

七、生产环境配置：从简单到可维护

7.1 控制台 + 文件双输出

生产中常见需求：

  • 控制台输出给容器平台采集；
  • 文件输出用于本机保留和排查；
  • 不同 handler 可以有不同级别。
import logging
from logging.handlers import RotatingFileHandler


def setup_logging():
    logger = logging.getLogger("app")
    logger.setLevel(logging.INFO)

    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)

    file_handler = RotatingFileHandler(
        filename="app.log",
        maxBytes=10 * 1024 * 1024,
        backupCount=5,
        encoding="utf-8",
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(formatter)

    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    return logger


logger = setup_logging()
logger.info("application started")

7.2 按大小切割：RotatingFileHandler

如果日志一直写入同一个文件，文件可能越来越大。

RotatingFileHandler 可以按大小切割：

from logging.handlers import RotatingFileHandler

file_handler = RotatingFileHandler(
    filename="app.log",
    maxBytes=100 * 1024 * 1024,
    backupCount=10,
    encoding="utf-8",
)

含义：

  • maxBytes=100 * 1024 * 1024：单个日志文件最大 100MB；
  • backupCount=10：最多保留 10 个历史文件；
  • 超过大小后会生成 app.log.1、app.log.2 等文件。

适合：

  • 日志量比较均匀；
  • 希望控制单文件大小；
  • 单机部署或传统 VM 部署。

7.3 按时间切割：TimedRotatingFileHandler

如果希望每天生成一个日志文件，可以使用 TimedRotatingFileHandler。

from logging.handlers import TimedRotatingFileHandler

file_handler = TimedRotatingFileHandler(
    filename="app.log",
    when="midnight",
    interval=1,
    backupCount=14,
    encoding="utf-8",
)

含义：

  • when="midnight"：每天午夜切割；
  • interval=1：每 1 天切割一次；
  • backupCount=14：保留最近 14 天日志。

适合：

  • 按天排查问题；
  • 日志平台按日期归档；
  • 运维习惯按天管理日志。

7.4 用 dictConfig() 集中管理配置

当项目变大时，不建议到处手动创建 handler。

更推荐用 logging.config.dictConfig() 集中配置。

import logging
import logging.config

LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "standard": {
            "format": "%(asctime)s | %(levelname)-8s | %(name)s | %(funcName)s:%(lineno)d | %(message)s",
            "datefmt": "%Y-%m-%d %H:%M:%S",
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "standard",
        },
        "app_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "INFO",
            "formatter": "standard",
            "filename": "app.log",
            "maxBytes": 10485760,
            "backupCount": 5,
            "encoding": "utf-8",
        },
        "error_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "ERROR",
            "formatter": "standard",
            "filename": "error.log",
            "maxBytes": 10485760,
            "backupCount": 10,
            "encoding": "utf-8",
        },
    },
    "loggers": {
        "app": {
            "handlers": ["console", "app_file", "error_file"],
            "level": "INFO",
            "propagate": False,
        },
        "app.retrieval": {
            "handlers": ["console", "app_file"],
            "level": "INFO",
            "propagate": False,
        },
        "app.llm": {
            "handlers": ["console", "app_file", "error_file"],
            "level": "INFO",
            "propagate": False,
        },
    },
    "root": {
        "handlers": ["console"],
        "level": "WARNING",
    },
}


def setup_logging():
    logging.config.dictConfig(LOGGING_CONFIG)


setup_logging()

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

这个配置的特点：

  • app 日志写入控制台、普通文件、错误文件；
  • ERROR 及以上级别额外进入 error.log；
  • app.retrieval 和 app.llm 可以单独调整级别和输出位置；
  • root 保持 WARNING，避免第三方库输出过多低级别日志。

7.5 不同环境使用不同级别

常见环境策略：

环境 推荐级别 说明
本地开发 DEBUG 方便看细节
测试环境 INFO 或 DEBUG 根据问题排查需要调整
预发环境 INFO 尽量接近生产
生产环境 INFO 默认记录关键事件
生产临时排障 短时间开启 DEBUG 排查后及时恢复

示例：

import logging
import os

LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO").upper()

logging.basicConfig(
    level=getattr(logging, LOG_LEVEL, logging.INFO),
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)

八、AI / RAG 服务端日志实战模板

下面用一个简化的 RAG 问答流程演示如何记录日志。

流程包括：

  1. 接收用户问题；
  2. 检索知识库；
  3. 构造 prompt；
  4. 调用模型；
  5. 返回答案；
  6. 记录成功或失败日志。

8.1 普通函数版示例

import logging
import time

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


def retrieve_documents(query: str, kb_id: str, top_k: int = 5):
    # 这里用模拟数据代替真实向量检索
    docs = [
        {"doc_id": "doc-1", "score": 0.91, "content": "Python logging 用于记录日志"},
        {"doc_id": "doc-2", "score": 0.84, "content": "Handler 决定日志输出位置"},
    ]

    logger.debug(
        "retrieval detail, query=%s, kb_id=%s, top_k=%s, docs=%s",
        query,
        kb_id,
        top_k,
        docs,
    )

    if len(docs) < top_k:
        logger.warning(
            "retrieval result less than top_k, query=%s, kb_id=%s, expected=%s, actual=%s",
            query,
            kb_id,
            top_k,
            len(docs),
        )

    return docs


def call_llm(prompt: str, model: str):
    # 这里用模拟结果代替真实模型调用
    time.sleep(0.1)
    return {
        "answer": "logging 模块可以通过 Logger、Handler、Formatter 实现生产级日志。",
        "input_tokens": 320,
        "output_tokens": 80,
    }


def answer_question(user_id: str, session_id: str, query: str, kb_id: str):
    start = time.perf_counter()
    model = "example-llm"

    logger.info(
        "chat request received, user_id=%s, session_id=%s, kb_id=%s",
        user_id,
        session_id,
        kb_id,
    )

    try:
        docs = retrieve_documents(query=query, kb_id=kb_id)
        prompt = build_prompt(query, docs)
        result = call_llm(prompt, model=model)

        latency_ms = round((time.perf_counter() - start) * 1000, 2)

        logger.info(
            "chat completed, user_id=%s, session_id=%s, kb_id=%s, model=%s, "
            "retrieval_count=%s, input_tokens=%s, output_tokens=%s, latency_ms=%s",
            user_id,
            session_id,
            kb_id,
            model,
            len(docs),
            result["input_tokens"],
            result["output_tokens"],
            latency_ms,
        )

        return result["answer"]

    except Exception:
        latency_ms = round((time.perf_counter() - start) * 1000, 2)
        logger.exception(
            "chat failed, user_id=%s, session_id=%s, kb_id=%s, model=%s, latency_ms=%s",
            user_id,
            session_id,
            kb_id,
            model,
            latency_ms,
        )
        raise


def build_prompt(query: str, docs: list[dict]) -> str:
    context = "\n".join(doc["content"] for doc in docs)
    return f"基于以下资料回答问题：\n{context}\n\n问题：{query}"

这个示例中：

  • 请求开始用 INFO；
  • 检索细节用 DEBUG；
  • 召回不足用 WARNING；
  • 成功完成用 INFO；
  • 异常失败用 logger.exception()。

8.2 模型调用日志应该记录什么？

模型调用建议记录：

字段 含义
request_id 请求链路 ID
user_id 用户 ID
session_id 会话 ID
model 模型名称
latency_ms 调用耗时
input_tokens 输入 token 数
output_tokens 输出 token 数
retry_count 重试次数
status 成功或失败
error_type 错误类型

示例：

logger.info(
    "llm call success, model=%s, input_tokens=%s, output_tokens=%s, latency_ms=%s",
    model,
    input_tokens,
    output_tokens,
    latency_ms,
)

失败时：

try:
    response = call_llm(prompt)
except TimeoutError:
    logger.warning("llm timeout, model=%s, retry_count=%s", model, retry_count)
    raise
except Exception:
    logger.exception("llm call failed, model=%s", model)
    raise

8.3 向量检索日志应该记录什么？

RAG 检索建议记录：

字段 含义
kb_id 知识库 ID
query 用户问题摘要
top_k 期望召回数量
retrieval_count 实际召回数量
top_score 最高相似度
threshold 相似度阈值
latency_ms 检索耗时

示例：

def log_retrieval_result(query: str, kb_id: str, docs: list[dict], threshold: float):
    top_score = max((doc.get("score", 0) for doc in docs), default=0)

    logger.info(
        "retrieval completed, kb_id=%s, query=%s, retrieval_count=%s, top_score=%.4f, threshold=%.4f",
        kb_id,
        query[:80],
        len(docs),
        top_score,
        threshold,
    )

    if not docs:
        logger.warning("retrieval empty, kb_id=%s, query=%s", kb_id, query[:80])
    elif top_score < threshold:
        logger.warning(
            "retrieval low score, kb_id=%s, query=%s, top_score=%.4f, threshold=%.4f",
            kb_id,
            query[:80],
            top_score,
            threshold,
        )

注意：真实生产中不一定要完整记录用户原始问题，尤其是涉及隐私数据时，应该做截断、脱敏或摘要化。

九、链路 ID / request_id 的实现方式

9.1 为什么需要 request_id？

在 Web 服务中，同一时间可能有很多请求同时执行。

如果日志是这样：

start chat
retrieval completed
start chat
llm call success
retrieval completed
llm call failed

你很难判断每行日志属于哪个请求。

如果加上 request_id：

req-001 | start chat
req-001 | retrieval completed
req-002 | start chat
req-001 | llm call success
req-002 | retrieval completed
req-002 | llm call failed

就能清楚串起一次请求的完整链路。

9.2 使用 contextvars 保存请求上下文

contextvars 适合在异步服务中保存请求级变量。

import contextvars
import logging
import uuid

request_id_var = contextvars.ContextVar("request_id", default="-")


class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True


def setup_logging():
    logger = logging.getLogger("app")
    logger.setLevel(logging.INFO)

    handler = logging.StreamHandler()
    handler.addFilter(RequestIdFilter())
    handler.setFormatter(logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(request_id)s | %(name)s | %(message)s"
    ))

    logger.addHandler(handler)
    return logger


logger = setup_logging()


def handle_request(user_id: str):
    request_id = str(uuid.uuid4())
    token = request_id_var.set(request_id)

    try:
        logger.info("request start, user_id=%s", user_id)
        do_something()
        logger.info("request finished, user_id=%s", user_id)
    except Exception:
        logger.exception("request failed, user_id=%s", user_id)
        raise
    finally:
        request_id_var.reset(token)


def do_something():
    logger.info("do something")

这样，同一次请求中的日志都会自动带上相同的 request_id。

9.3 FastAPI 中的 request_id 思路

如果是 FastAPI，可以通过中间件设置 request_id。

下面示例需要安装 FastAPI，属于 Web 框架场景，不是 logging 标准库本身：

import contextvars
import logging
import uuid

from fastapi import FastAPI, Request

app = FastAPI()
request_id_var = contextvars.ContextVar("request_id", default="-")
logger = logging.getLogger("app")


@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    token = request_id_var.set(request_id)

    try:
        logger.info("http request start, method=%s, path=%s", request.method, request.url.path)
        response = await call_next(request)
        logger.info("http request end, status_code=%s", response.status_code)
        response.headers["X-Request-ID"] = request_id
        return response
    except Exception:
        logger.exception("http request failed")
        raise
    finally:
        request_id_var.reset(token)

实际项目中，RequestIdFilter 会从 request_id_var 读取值并注入日志。

十、结构化日志与可观测性

10.1 为什么需要结构化日志？

普通文本日志适合人看：

2026-07-09 10:00:00 | INFO | req-001 | app.chat | chat completed, user_id=1001, latency_ms=820

但日志平台更喜欢结构化字段：

{"time":"2026-07-09 10:00:00","level":"INFO","request_id":"req-001","logger":"app.chat","event":"chat_completed","user_id":"1001","latency_ms":820}

结构化日志的优势：

  • 可以按字段检索：request_id=req-001；
  • 可以统计：平均 latency_ms；
  • 可以聚合：按 model 分组看错误率；
  • 可以告警：level=ERROR 超过阈值；
  • 可以和链路追踪、指标系统关联。

10.2 简单 JSON Formatter 示例

下面用标准库实现一个简化版 JSON formatter：

import json
import logging
from datetime import datetime


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "time": datetime.fromtimestamp(record.created).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
            "message": record.getMessage(),
        }

        if hasattr(record, "request_id"):
            log_data["request_id"] = record.request_id

        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)

        return json.dumps(log_data, ensure_ascii=False)


logger = logging.getLogger("app")
logger.setLevel(logging.INFO)

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)

logger.info("service started")

输出类似：

{"time":"2026-07-09T10:00:00.123456","level":"INFO","logger":"app","module":"main","function":"<module>","line":30,"message":"service started"}

实际生产中也可以使用成熟的第三方结构化日志库，但理解标准库实现原理很重要。

10.3 日志、指标、链路追踪的关系

可观测性通常包括三类数据：

类型 作用 示例
日志 Logs 记录具体事件 某次请求失败堆栈
指标 Metrics 统计趋势和数值 QPS、错误率、P95 延迟
链路追踪 Traces 展示一次请求经过哪些服务 API → 检索 → 模型 → 数据库

日志不是万能的。

例如：

  • 想知道“过去 5 分钟错误率是否超过 5%”，更适合指标；
  • 想知道“一次请求到底卡在哪个服务”，更适合链路追踪；
  • 想知道“这次错误具体堆栈是什么”，更适合日志。

但日志是最基础、最容易落地的一环。

十一、常见坑与最佳实践

11.1 不要用 print() 替代生产日志

print() 不能提供级别、模块、格式、切割、异常堆栈等能力。

本地临时调试可以用，长期保留的服务端代码应使用 logging。

11.2 不要在库代码里随意 basicConfig()

如果你写的是一个被别人导入的模块或库，不应该在库内部主动配置全局日志。

不推荐：

# mylib.py
import logging

logging.basicConfig(level=logging.INFO)

推荐：

# mylib.py
import logging

logger = logging.getLogger(__name__)


def do_work():
    logger.info("work started")

让应用入口决定日志如何配置。

11.3 避免重复添加 handler

在某些场景中，setup_logging() 可能被调用多次。

如果每次都 addHandler()，日志会重复输出。

可以这样防御：

import logging


def setup_logging():
    logger = logging.getLogger("app")

    if logger.handlers:
        return logger

    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(
        "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
    ))
    logger.addHandler(handler)
    return logger

或者更推荐在应用启动入口只配置一次日志。

11.4 不要记录敏感信息

日志很容易被多人访问，也可能进入第三方日志平台。

不要记录：

  • 密码；
  • token；
  • API key；
  • 身份证号；
  • 银行卡号；
  • 完整手机号；
  • 用户隐私文本；
  • 未脱敏的 prompt 和模型输出。

示例：

def mask_phone(phone: str) -> str:
    if len(phone) != 11:
        return "***"
    return phone[:3] + "****" + phone[-4:]


logger.info("user login, phone=%s", mask_phone("13812345678"))

对于 AI 服务，尤其要注意用户输入可能包含隐私信息，不要无脑记录完整 query、prompt 和回答。

11.5 异常日志要带堆栈

不推荐：

except Exception as e:
    logger.error("failed: %s", e)

推荐：

except Exception:
    logger.exception("operation failed")

或者：

except Exception:
    logger.error("operation failed", exc_info=True)

11.6 日志内容要有业务语义

不好的日志：

logger.info("success")
logger.error("failed")

问题是：什么成功？什么失败？谁的请求？哪个模块？

更好的日志：

logger.info(
    "kb update success, kb_id=%s, file_count=%s, user_id=%s",
    kb_id,
    file_count,
    user_id,
)

logger.exception(
    "kb update failed, kb_id=%s, file_name=%s, user_id=%s",
    kb_id,
    file_name,
    user_id,
)

好的日志应该能回答：

  • 谁触发的？
  • 做了什么？
  • 结果如何？
  • 关键参数是什么？
  • 失败原因是什么？
  • 如何继续定位？

11.7 不要把所有问题都打成 ERROR

如果把参数错误、用户取消、检索为空、缓存未命中都打成 ERROR，真正的系统故障会被淹没。

判断标准：

问题 推荐级别
用户输入不合法 INFO / WARNING
查询无结果但有兜底返回 WARNING
第三方接口偶发超时但重试成功 WARNING
当前请求失败 ERROR
系统核心能力不可用 CRITICAL

11.8 第三方库日志要控制级别

有些第三方库可能输出大量日志。

可以单独设置：

import logging

logging.getLogger("urllib3").setLevel(logging.WARNING)
logging.getLogger("httpx").setLevel(logging.WARNING)

这样可以避免生产日志被无关细节刷屏。

11.9 多进程 / 高并发日志要谨慎

在多进程场景下，多个进程同时写同一个文件可能带来竞争问题。

常见方案：

  • 容器化部署中直接输出到 stdout，由平台采集；
  • 使用 QueueHandler + QueueListener 集中写日志；
  • 使用外部日志采集 Agent；
  • 避免多个进程同时写同一个普通文件。

简化版队列日志思路：

import logging
import logging.handlers
import queue

log_queue = queue.Queue()

queue_handler = logging.handlers.QueueHandler(log_queue)
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(logging.Formatter(
    "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
))

listener = logging.handlers.QueueListener(log_queue, stream_handler)
listener.start()

logger = logging.getLogger("app")
logger.setLevel(logging.INFO)
logger.addHandler(queue_handler)

logger.info("message from application")

listener.stop()

这个例子只是说明思路：业务线程把日志放入队列，由 listener 统一处理输出。

十二、完整示例：一个可复用的生产日志配置

下面给出一个相对完整的标准库版本，适合作为中小型项目起点。

import contextvars
import logging
import logging.config
import uuid

request_id_var = contextvars.ContextVar("request_id", default="-")


class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()
        return True


LOGGING_CONFIG = {
    "version": 1,
    "disable_existing_loggers": False,
    "filters": {
        "request_id": {
            "()": RequestIdFilter,
        }
    },
    "formatters": {
        "standard": {
            "format": (
                "%(asctime)s | %(levelname)-8s | %(request_id)s | "
                "%(name)s | %(funcName)s:%(lineno)d | %(message)s"
            ),
            "datefmt": "%Y-%m-%d %H:%M:%S",
        }
    },
    "handlers": {
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "standard",
            "filters": ["request_id"],
        },
        "app_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "INFO",
            "formatter": "standard",
            "filters": ["request_id"],
            "filename": "app.log",
            "maxBytes": 10485760,
            "backupCount": 5,
            "encoding": "utf-8",
        },
        "error_file": {
            "class": "logging.handlers.RotatingFileHandler",
            "level": "ERROR",
            "formatter": "standard",
            "filters": ["request_id"],
            "filename": "error.log",
            "maxBytes": 10485760,
            "backupCount": 10,
            "encoding": "utf-8",
        },
    },
    "loggers": {
        "app": {
            "handlers": ["console", "app_file", "error_file"],
            "level": "INFO",
            "propagate": False,
        }
    },
    "root": {
        "handlers": ["console"],
        "level": "WARNING",
    },
}


def setup_logging():
    logging.config.dictConfig(LOGGING_CONFIG)


def handle_request(user_id: str):
    request_id = str(uuid.uuid4())
    token = request_id_var.set(request_id)
    logger = logging.getLogger("app.chat")

    try:
        logger.info("request start, user_id=%s", user_id)
        logger.info("chat completed, user_id=%s", user_id)
    except Exception:
        logger.exception("request failed, user_id=%s", user_id)
        raise
    finally:
        request_id_var.reset(token)


if __name__ == "__main__":
    setup_logging()
    handle_request("user-001")

这个配置具备：

  • 统一格式；
  • request_id 注入；
  • 控制台输出；
  • 普通日志文件；
  • 错误日志文件；
  • 文件大小切割；
  • 模块化 logger；
  • 异常堆栈支持。

十三、如何回答更专业？

13.1 问题一：Python 默认日志有什么缺陷？

可以这样回答：

Python 默认日志配置比较简单。默认 root logger 的级别是 WARNING，低于 WARNING 的 DEBUG 和 INFO 默认不会输出；默认格式通常只包含级别、logger 名和消息，不适合生产排障。
如果只用 print() 或未配置的 logging，会缺少统一时间格式、模块名、函数行号、request_id、异常堆栈、日志切割和结构化字段。
在线上服务中，这会导致无法按请求链路追踪问题，也无法区分接口日志、业务日志、模型日志、数据库日志和向量检索日志，更不方便接入日志平台和告警系统。
因此生产环境应该基于 logging 配置 logger、handler、formatter、filter，并结合日志级别、文件切割、链路 ID 和异常堆栈形成完整日志体系。

13.2 问题二：AI 服务端生产日志如何分级配置？

可以这样回答：

AI 服务端一般生产默认使用 INFO 级别，本地开发或临时排障时开启 DEBUG。
DEBUG 用于记录模型入参摘要、prompt 构造细节、向量召回结果、相似度分数等调试信息，但生产中要注意脱敏和控制日志量。
INFO 用于记录正常业务事件，比如接口调用、会话创建、知识库更新、模型调用成功、token 消耗、耗时等。
WARNING 用于记录可恢复但需要关注的问题，比如模型调用超时后重试成功、检索召回量不足、相似度过低、缓存异常、内存水位偏高。
ERROR 用于记录当前请求或任务失败的问题，比如接口崩溃、向量库连接中断、模型推理失败、数据库写入失败，并且应该带上 request_id 和异常堆栈。
CRITICAL 用于核心服务不可用、启动失败、关键配置缺失等系统级严重故障，通常需要触发告警。
同时日志中应包含 request_id、user_id、session_id、model、kb_id、latency、token 数等关键字段，方便后续检索、追踪和告警。

十四、围绕这个问题还应该举一反三掌握什么？

14.1 异常处理机制

日志和异常处理关系非常紧密。

应该掌握：

  • try-except-else-finally；
  • 捕获具体异常，而不是无脑捕获 Exception；
  • raise 重新抛出异常；
  • 自定义异常；
  • logger.exception() 记录堆栈。

14.2 文件 IO 与上下文管理

日志最终经常写入文件，因此需要理解：

  • 文件打开模式；
  • 编码问题；
  • 文件句柄释放；
  • with 上下文管理；
  • 文件大小增长和切割策略。

14.3 服务端可观测性

日志只是可观测性的一部分，还应该了解：

  • Metrics：QPS、错误率、P95 延迟；
  • Tracing：链路追踪；
  • Alerting：告警规则；
  • Dashboard：监控面板；
  • OpenTelemetry：统一可观测性标准。

14.4 安全与合规

日志不是越详细越好。

必须注意：

  • 敏感字段脱敏；
  • 控制 prompt 和模型输出记录范围；
  • 避免记录 access token、API key；
  • 日志访问权限控制；
  • 日志保留周期；
  • 用户隐私和合规要求。

14.5 性能与成本

日志也有成本：

  • 字符串格式化成本；
  • 磁盘 IO 成本；
  • 日志采集和存储成本；
  • 日志平台索引成本；
  • 高并发下锁竞争或队列积压。

因此生产中需要：

  • 合理设置日志级别；
  • 避免无意义高频日志；
  • 对大字段截断；
  • 必要时做采样；
  • 使用异步或队列日志方案。

十五、总结

Python 的默认日志能力适合简单脚本，但不适合生产服务。

真正的生产级日志体系应该至少具备：

  1. 日志级别清晰：DEBUG、INFO、WARNING、ERROR、CRITICAL 各司其职。
  2. 格式统一：时间、级别、模块、函数、行号、消息结构一致。
  3. 模块区分：通过 getLogger(__name__) 区分不同代码模块。
  4. 输出可控：控制台、文件、错误文件、远程日志平台按需配置。
  5. 文件可管理：使用 RotatingFileHandler 或 TimedRotatingFileHandler 防止无限增长。
  6. 异常可定位：错误日志必须带堆栈。
  7. 链路可追踪：通过 request_id / trace_id 串起一次请求。
  8. 字段可检索：关键字段结构化，便于日志平台查询。
  9. 内容有边界：避免敏感信息泄露。
  10. 适应业务场景：AI 服务端要重点记录模型、检索、token、耗时、知识库、会话等信息。

最后可以用一句话收束：

logging 不只是 Python 的一个标准库模块，而是服务端工程中定位问题、理解系统运行状态、支撑告警和可观测性的基础设施。能否设计好日志，往往体现了一个开发者是否具备生产环境思维。

# python # 面试
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

别再只聊大模型！AI的下一场革命，藏在“身体”里

avatar 亚马逊云科技技术品牌专区
cover

别再只聊大模型！人工智能的下一场革命，藏在“身体”里

avatar 亚马逊云科技技术品牌专区
cover

别再只聊大模型！AI的下一场革命，藏在“身体”里

avatar 亚马逊云科技技术品牌专区