loguru 完全指南

最后更新：2026-04-22

官方文档：https://loguru.readthedocs.io/en/stable/
适用版本：loguru 0.7+（2026-05-08 核实）

1. 基础概念

为什么使用 loguru

标准库 logging 需要手动创建 Logger、Handler、Formatter 三层对象，配置繁琐。loguru 用一个全局 logger 对象统一所有操作，开箱即用。

核心设计

loguru 只有一个全局 logger 对象（loguru.logger），所有输出目标（终端、文件、自定义 sink）通过 logger.add() 注册为 sink，每个 sink 独立配置级别、格式、轮转策略，logger.remove() 移除 sink。

默认 sink 是 sys.stderr，id 为 0。

2. 快速开始

from loguru import logger

# 开箱即用，默认输出到 stderr（带颜色）
logger.debug("调试信息")
logger.info("服务已启动")
logger.warning("磁盘空间不足")
logger.error("数据库连接失败")
logger.critical("进程崩溃")

# 记录异常（自动附带 traceback）
try:
    1 / 0
except ZeroDivisionError:
    logger.exception("捕获到异常")  # 等价于 logger.error + exc_info

默认输出格式：

2026-04-22 10:23:45.123 | DEBUG    | __main__:<module>:5 - 调试信息

3. logger.add() — 注册 sink

logger.add() 是 loguru 最核心的 API，用于添加输出目标。

函数签名

logger.add(sink, *, level="DEBUG", format=<默认格式>, filter=None,
           colorize=None, serialize=False, backtrace=True, diagnose=True,
           enqueue=False, catch=True,
           rotation=None, retention=None, compression=None,
           delay=False, watch=False, mode="a", buffering=1, encoding="utf8",
           errors="backslashreplace")

参数说明

返回值

logger.add() 返回一个整数 id，可传给 logger.remove(id) 来移除该 sink。

示例

from loguru import logger

# 移除默认 stderr sink
logger.remove(0)

# 添加终端 sink（仅 INFO 及以上）
logger.add(
    sys.stderr,
    level="INFO",
    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
    colorize=True,
)

# 添加文件 sink
file_id = logger.add(
    "logs/app.log",
    level="DEBUG",
    rotation="100 MB",       # 超过 100 MB 轮转
    retention="30 days",     # 保留 30 天
    compression="gz",        # 压缩旧文件
    enqueue=True,            # 异步写入（多进程安全）
    backtrace=True,
    diagnose=False,          # 生产环境关闭变量展示
)

4. 格式化（format 占位符）

内置占位符

颜色标签

format 字符串中可以嵌入颜色标签（仅终端 sink 有效）：

<red>红色</red>
<green>绿色</green>
<yellow>黄色</yellow>
<blue>蓝色</blue>
<magenta>洋红</magenta>
<cyan>青色</cyan>
<white>白色</white>
<bold>粗体</bold>
<underline>下划线</underline>
<level>使用当前级别的颜色</level>

自定义 format 函数

def formatter(record):
    # record 是包含所有字段的字典
    fmt = "{time:HH:mm:ss} | {level} | {message}\n"
    if record["exception"]:
        fmt += "{exception}\n"
    return fmt

logger.add(sys.stderr, format=formatter)

5. 日志级别

内置级别

loguru 比标准库多了 TRACE（比 DEBUG 更低）和 SUCCESS（介于 INFO 和 WARNING 之间）。

自定义级别

# add_level(name, no, color="", icon="")
logger.level("AUDIT", no=35, color="<yellow><bold>", icon="@")

# 使用自定义级别
logger.log("AUDIT", "用户 {} 登录", user_id)

logger.level() 参数：

6. 文件轮转（rotation）

rotation 参数决定何时创建新日志文件：

文件名支持 {time} 占位符，轮转后自动加时间戳区分：

logger.add("logs/app_{time:YYYY-MM-DD}.log", rotation="00:00")
# 生成：app_2026-04-21.log, app_2026-04-22.log ...

7. 文件保留（retention）

retention 参数决定旧日志文件保留多久：

logger.add(
    "logs/app.log",
    rotation="1 day",
    retention=7,          # 最多保留 7 个历史文件
    compression="gz",
)

8. 过滤器（filter）

字符串过滤（模块名前缀）

# 只输出来自 myapp 模块及其子模块的日志
logger.add("myapp.log", filter="myapp")

函数过滤

def only_errors(record):
    return record["level"].no >= 40

logger.add("errors.log", filter=only_errors)

filter 函数接收 record 字典，常用字段：

dict 过滤（按模块名映射级别）

logger.add(
    sys.stderr,
    filter={
        "myapp":         "DEBUG",   # myapp 及子模块输出 DEBUG 以上
        "myapp.db":      "WARNING", # myapp.db 只输出 WARNING 以上（更严格）
        "third_party":   False,     # 屏蔽该模块
        "":              "INFO",    # 其他模块输出 INFO 以上
    }
)

9. 上下文绑定（bind / contextualize）

bind() — 静态绑定

bind() 创建一个绑定了额外字段的新 logger 副本，原 logger 不受影响：

request_logger = logger.bind(request_id="abc-123", user_id=42)
request_logger.info("处理请求")
# 日志 extra 字段中包含 request_id 和 user_id

# 可在 format 中引用
logger.add(sys.stderr, format="{time} | {extra[request_id]} | {message}")

contextualize() — 上下文管理器绑定

适合在一个代码块内临时绑定上下文（如 FastAPI 请求处理中）：

with logger.contextualize(request_id="abc-123"):
    logger.info("进入请求处理")
    process_request()
    logger.info("请求处理完成")
# 退出 with 块后，request_id 自动清除

在异步场景下，contextualize() 基于 contextvars，天然隔离不同协程的上下文：

async def handle_request(request_id: str):
    with logger.contextualize(request_id=request_id):
        logger.info("开始处理")
        await do_work()
        logger.info("处理完成")

10. 异常捕获

logger.exception()

在 except 块中调用，自动附加当前异常的 traceback：

try:
    result = risky_operation()
except Exception:
    logger.exception("操作失败")   # 相当于 logger.error + 自动获取 exc_info

@logger.catch — 装饰器

自动捕获并记录函数内未处理的异常：

@logger.catch
def my_function():
    return 1 / 0  # 异常会被记录，然后重新抛出

# 也支持异步函数
@logger.catch
async def async_function():
    ...

logger.catch() 可作为装饰器也可作为上下文管理器：

with logger.catch():
    risky_operation()

logger.catch() 参数：

11. 结构化日志（序列化）

serialize=True 时，每条日志输出为一行 JSON，适合对接 ELK、Loki 等日志收集系统：

logger.add("logs/structured.log", serialize=True)
logger.bind(user_id=42).info("用户登录")

输出：

{"text": "2026-04-22 10:23:45.123 | INFO | __main__:<module>:3 - 用户登录\n", "record": {"elapsed": {"repr": "...", "seconds": 0.1}, "exception": null, "extra": {"user_id": 42}, "file": {"name": "example.py", "path": "/app/example.py"}, "function": "<module>", "level": {"icon": "ℹ️", "name": "INFO", "no": 20}, "line": 3, "message": "用户登录", "module": "example", "name": "__main__", "process": {"id": 1234, "name": "MainProcess"}, "thread": {"id": 140234, "name": "MainThread"}, "time": {"repr": "2026-04-22 10:23:45.123456+08:00", "timestamp": 1745283825.123}}}

12. 多进程安全

使用 enqueue=True 时，日志写入先进入线程安全队列，由后台线程统一写入文件，避免多进程并发写入导致日志交错：

logger.add("logs/app.log", enqueue=True)

在 multiprocessing 场景下，子进程不继承父进程的 logger 配置。推荐模式：

from loguru import logger
import multiprocessing

def worker(queue):
    # 子进程中重新配置 logger，将日志发送到队列
    logger.remove()
    logger.add(queue, serialize=True)
    logger.info("子进程工作中")

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    logger.add(queue, serialize=True)  # 主进程监听队列

    p = multiprocessing.Process(target=worker, args=(queue,))
    p.start()
    p.join()

13. 与标准库 logging 集成

已有项目大量使用标准库 logging（如 FastAPI、SQLAlchemy、uvicorn 的内部日志），可以用 InterceptHandler 将所有标准库日志重定向到 loguru：

import logging
from loguru import logger

class InterceptHandler(logging.Handler):
    def emit(self, record: logging.LogRecord) -> None:
        # 将标准库 level 映射到 loguru level
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno

        # 找到正确的调用栈帧，使文件名/行号显示准确
        frame, depth = logging.currentframe(), 0
        while frame and (depth == 0 or frame.f_code.co_filename == logging.__file__):
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(
            level, record.getMessage()
        )

# 接管所有标准库日志
logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

# 或者只接管特定库
for name in ("uvicorn", "uvicorn.error", "uvicorn.access", "sqlalchemy.engine"):
    logging.getLogger(name).handlers = [InterceptHandler()]
    logging.getLogger(name).propagate = False

14. 在 FastAPI 中使用

# config/logging.py
import sys
import logging
from loguru import logger

class InterceptHandler(logging.Handler):
    def emit(self, record):
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = record.levelno
        frame, depth = logging.currentframe(), 0
        while frame and (depth == 0 or frame.f_code.co_filename == logging.__file__):
            frame = frame.f_back
            depth += 1
        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())


def setup_logging(log_level: str = "INFO", json_logs: bool = False):
    # 移除默认 sink
    logger.remove()

    # 终端输出
    logger.add(
        sys.stderr,
        level=log_level,
        colorize=True,
        backtrace=True,
        diagnose=not json_logs,
    )

    # 文件输出（生产环境）
    if json_logs:
        logger.add(
            "logs/app.log",
            level=log_level,
            serialize=True,
            rotation="1 day",
            retention="30 days",
            compression="gz",
            enqueue=True,
            diagnose=False,
        )

    # 接管 uvicorn/fastapi 标准库日志
    for name in logging.root.manager.loggerDict:
        logging.getLogger(name).handlers = [InterceptHandler()]
        logging.getLogger(name).propagate = False
    logging.root.handlers = [InterceptHandler()]

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from loguru import logger
from config.logging import setup_logging

@asynccontextmanager
async def lifespan(app: FastAPI):
    setup_logging(log_level="INFO", json_logs=True)
    logger.info("应用启动")
    yield
    logger.info("应用关闭")

app = FastAPI(lifespan=lifespan)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    with logger.contextualize(
        path=request.url.path,
        method=request.method,
        client=request.client.host if request.client else None,
    ):
        logger.info("收到请求")
        response = await call_next(request)
        logger.info("请求完成，状态码 {}", response.status_code)
        return response

15. logger.opt() — 高级选项

logger.opt() 返回一个临时修改了选项的 logger，常用于框架内部：

logger.opt(depth=1).info("消息")       # 调整调用栈深度，使文件名/行号指向调用者
logger.opt(exception=True).error("出错") # 强制附加当前异常信息
logger.opt(lazy=True).debug("耗时操作: {}", lambda: compute())  # 懒求值，级别不满足时不执行 lambda
logger.opt(colors=True).info("<red>带颜色</red>的消息")         # 强制启用颜色标签解析
logger.opt(raw=True).info("原始字符串\n")                       # 绕过格式化直接输出
logger.opt(record=True).info("record={record}")                 # 将 record 字典注入消息

opt() 参数：

16. 最佳实践

项目统一配置入口

将 logger.remove() 和所有 logger.add() 集中在一个初始化函数中，在应用启动时调用一次。各模块只 from loguru import logger 直接使用，不做任何配置。

生产环境关闭 diagnose

diagnose=True 会在 traceback 中展示局部变量值，可能泄露密码、Token 等敏感信息。生产环境必须设置 diagnose=False。

文件 sink 始终开启 enqueue

即使当前是单进程，enqueue=True 也能避免 IO 阻塞影响主线程性能，代价极小。

用 bind() 传递业务上下文

不要在消息字符串中硬编码业务 ID，用 bind() 或 contextualize() 绑定，便于日志聚合系统按字段过滤：

# 不推荐
logger.info(f"处理订单 {order_id} 失败")

# 推荐
logger.bind(order_id=order_id).error("处理订单失败")

区分开发与生产配置

import os

def setup_logging():
    logger.remove()
    if os.getenv("ENV") == "production":
        logger.add("logs/app.log", serialize=True, level="INFO",
                   rotation="1 day", retention="14 days",
                   enqueue=True, diagnose=False)
    else:
        logger.add(sys.stderr, level="DEBUG", colorize=True, diagnose=True)

17. 常见陷阱

格式化字符串用 {} 而非 %s 或 f-string

loguru 使用 str.format() 风格的懒格式化，直到日志被实际处理时才求值，减少无效格式化开销：

# 不推荐：f-string 立即求值，即使日志级别不满足也会执行字符串拼接
logger.debug(f"结果：{expensive_function()}")

# 推荐：只有级别满足时才调用 str.format()
logger.debug("结果：{}", expensive_function())

# 更进一步：用 opt(lazy=True) 完全避免求值
logger.opt(lazy=True).debug("结果：{}", lambda: expensive_function())

不要在子进程中直接使用父进程配置的 logger

fork 后子进程继承了父进程的 logger 状态（包括文件句柄），多进程同时写同一文件会导致日志交错。解决方案：使用 enqueue=True，或在子进程初始化时重新调用 logger.remove() + logger.add()。

rotation 与文件名模板配合

如果文件名不含 {time} 但同时设置了 rotation，轮转后旧文件会被自动添加时间戳重命名。若文件名已含 {time} 则新文件按时间命名，无需额外处理。

remove() 传参而非不传参

logger.remove() 不传参数会移除所有 sink（包括默认 stderr）。通常应传入 logger.add() 返回的 id：

stderr_id = logger.add(sys.stderr)
# ...
logger.remove(stderr_id)   # 只移除这一个 sink

# 移除默认 sink
logger.remove(0)            # 默认 stderr 的 id 是 0

Windows 多进程与 enqueue

在 Windows 上使用 multiprocessing + enqueue=True 时，需确保 logger 配置在 if __name__ == "__main__": 块内，否则子进程导入时会重复执行配置代码。

最佳实践

应用启动时移除默认 sink 再添加自定义配置：loguru 默认向 stderr 输出，生产环境应先 logger.remove() 清除默认配置再添加文件和结构化 sink，避免日志重复输出。

用 serialize=True 输出结构化 JSON：生产环境日志要便于机器解析，serialize=True 将每条日志序列化为 JSON 行，直接接入 ELK、Grafana Loki 等日志系统：

logger.add("app.log", serialize=True, rotation="100 MB")

按日志级别分文件输出：错误日志单独文件便于告警和排查：

logger.add("app.log", level="INFO", rotation="1 day")
logger.add("error.log", level="ERROR", rotation="1 week", retention="30 days")

在 FastAPI/异步应用中用 enqueue=True：异步环境下 file IO 是同步阻塞，enqueue=True 将日志写入队列，由独立线程处理，不阻塞事件循环。

用 logger.contextualize() 绑定请求上下文：在中间件中绑定 request_id、user_id 等字段，该请求所有日志自动携带这些信息，无需每次手动传参。

常见陷阱

陷阱：多次调用 logger.add() 导致日志重复输出

现象： 每条日志输出两次或更多次。
原因： 每次导入模块或函数被多次调用时重复 logger.add()，而 loguru 的 sink 是累积的。
解决： 在应用启动入口集中配置一次，可用 logger.remove() + 单次 logger.add()；或用 if not logger._core.handlers: 守卫。

陷阱：loguru 与 logging 第三方库日志割裂

现象： FastAPI、SQLAlchemy 等库的日志通过 logging 模块输出，与 loguru 的格式不一致，难以集中管理。
原因： 第三方库使用标准 logging 模块，loguru 默认不接收它们的日志。
解决： 用 loguru.logger 的 PropagateHandler 将标准 logging 的日志转发到 loguru：

import logging
from loguru import logger

class InterceptHandler(logging.Handler):
    def emit(self, record):
        level = logger.level(record.levelname).name
        logger.opt(depth=6, exception=record.exc_info).log(level, record.getMessage())

logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)

陷阱：rotation 参数字符串格式错误静默失效

现象： 设置 rotation="500 mb" 后日志文件不轮转。
原因： loguru 的 rotation 大小写敏感，接受 "500 MB" 但不接受 "500 mb"（小写）。
解决： 始终用大写单位（"100 MB"、"1 GB"），或用时间格式（"1 day"、"00:00"）。

参见

asyncio异步编程完全指南
FastAPI完全指南