Python 日志与调试：AI 应用的可观测基础

本文适合谁：熟悉 Java log4j/SLF4J 的工程师，想建立 AI 应用可观测性的开发者。读完本篇，你能调试 LLM 的输入输出、记录 token 消耗、用结构化日志支撑生产环境问题排查。

AI 应用出问题时，最难回答的问题是：LLM 收到的 prompt 到底是什么？这次调用花了多少 token？是网络超时还是模型返回了错误？

普通 print() 调试在生产环境里没有意义：没有时间戳、没有级别、不知道来自哪个文件、无法开关、无法聚合。结构化日志和正确的异常处理才是 AI 应用可观测性的基础。

1.1 logging 模块的核心概念

Python 日志级别从 DEBUG 到 CRITICAL 及各类 Handler 的分发关系

Python 标准库的 logging 模块围绕四个概念展开：

Logger：记录日志的入口，通过名称区分来源（logging.getLogger("my_app.llm")）
Handler：决定日志输出到哪里（控制台、文件、远程服务）
Formatter：决定日志的格式（时间戳、级别、消息）
Level：日志级别（DEBUG / INFO / WARNING / ERROR / CRITICAL），低于设定级别的日志不输出

python

import logging
import sys

def setup_logging(level: str = "INFO") -> None:
    """
    配置全局日志格式
    在应用入口（main.py）调用一次，所有模块共享此配置
    """
    # 创建格式化器：时间戳 + 级别 + logger名称 + 消息
    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    # 控制台 Handler：INFO 及以上输出到 stdout
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(formatter)

    # 文件 Handler：DEBUG 及以上写入文件，用于排查问题
    file_handler = logging.FileHandler("app.log", encoding="utf-8")
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

    # 配置根 Logger
    root_logger = logging.getLogger()
    root_logger.setLevel(logging.DEBUG)  # 根级别设最低，各 Handler 自行过滤
    root_logger.addHandler(console_handler)
    root_logger.addHandler(file_handler)

# 在各模块里，用模块路径命名 Logger
# 名称层级结构让日志可以按模块过滤
logger = logging.getLogger(__name__)  # __name__ = "myapp.llm.client"

logger.debug("调试信息，生产环境不显示")
logger.info("正常操作日志")
logger.warning("需要关注但不影响运行")
logger.error("错误，需要处理")
logger.critical("严重错误，系统可能无法继续")

1.2 结构化日志（JSON 格式）

文本日志在数量少时可以直接阅读，但接入 ELK（Elasticsearch + Logstash + Kibana，一套开源日志收集、存储和可视化平台）或云日志服务时，JSON 格式的结构化日志才能被高效查询和分析。

python

import logging
import json
import time
import traceback
from typing import Any

class JSONFormatter(logging.Formatter):
    """
    将日志记录格式化为 JSON，每行一个 JSON 对象
    便于 ELK、CloudWatch、StackDriver 等日志系统解析
    """

    def format(self, record: logging.LogRecord) -> str:
        log_data: dict[str, Any] = {
            "timestamp": self.formatTime(record, self.datefmt),
            "level":     record.levelname,
            "logger":    record.name,
            "message":   record.getMessage(),
            "file":      f"{record.filename}:{record.lineno}",
        }

        # 附加到 LogRecord 的额外字段（通过 extra= 参数传入）
        for key, value in record.__dict__.items():
            if key.startswith("ctx_"):  # 约定：以 ctx_ 开头的是业务上下文
                log_data[key[4:]] = value  # 去掉前缀 ctx_

        # 如果有异常，附加堆栈信息
        if record.exc_info:
            log_data["exception"] = {
                "type":    record.exc_info[0].__name__,
                "message": str(record.exc_info[1]),
                "traceback": traceback.format_exception(*record.exc_info),
            }

        return json.dumps(log_data, ensure_ascii=False)

# 配置 JSON 格式日志
def setup_json_logging() -> None:
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JSONFormatter())
    logging.getLogger().addHandler(handler)
    logging.getLogger().setLevel(logging.DEBUG)

1.3 AI 应用日志最佳实践

AI 应用的日志需要记录普通应用不关心的内容：发给 LLM 的 prompt、模型返回的内容、token 消耗量、响应时间。这些信息对于成本控制和问题排查至关重要。

python

import logging
import time
from typing import Any
from openai import OpenAI
from openai.types.chat import ChatCompletion

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

class LLMClient:
    """
    封装 LLM 调用，自动记录每次调用的详细日志
    生产环境中这些日志是排查问题的主要依据
    """

    def __init__(self, model: str = "gpt-4o"):
        self.client = OpenAI()
        self.model  = model

    def chat(
        self,
        messages: list[dict],
        request_id: str | None = None,
        **kwargs,
    ) -> ChatCompletion:
        start_time = time.monotonic()

        # 记录请求日志：发送前先记，方便排查超时问题
        logger.info(
            "LLM request started",
            extra={
                "ctx_request_id": request_id,
                "ctx_model":      self.model,
                "ctx_messages":   messages,          # 完整 prompt
                "ctx_params":     kwargs,
            }
        )

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                **kwargs,
            )

            elapsed = time.monotonic() - start_time
            usage   = response.usage

            # 记录成功日志：包含 token 消耗和耗时
            logger.info(
                "LLM request completed",
                extra={
                    "ctx_request_id":      request_id,
                    "ctx_model":           self.model,
                    "ctx_elapsed_seconds": round(elapsed, 3),
                    "ctx_prompt_tokens":   usage.prompt_tokens,
                    "ctx_completion_tokens": usage.completion_tokens,
                    "ctx_total_tokens":    usage.total_tokens,
                    "ctx_response":        response.choices[0].message.content,
                }
            )
            return response

        except Exception as e:
            elapsed = time.monotonic() - start_time
            # 记录失败日志：exc_info=True 附加完整堆栈
            logger.error(
                "LLM request failed",
                exc_info=True,
                extra={
                    "ctx_request_id":      request_id,
                    "ctx_model":           self.model,
                    "ctx_elapsed_seconds": round(elapsed, 3),
                    "ctx_error_type":      type(e).__name__,
                }
            )
            raise  # 重新抛出，不在这里吞掉异常

输出的 JSON 日志示例，便于在 Kibana 里过滤查询：

json

{
  "timestamp": "2024-01-15 14:32:01",
  "level": "INFO",
  "logger": "ai_app.llm",
  "message": "LLM request completed",
  "request_id": "req-abc123",
  "model": "gpt-4o",
  "elapsed_seconds": 1.234,
  "prompt_tokens": 156,
  "completion_tokens": 89,
  "total_tokens": 245
}

与 Java log4j/slf4j 的对比：

维度	Python logging	Java slf4j + log4j2
配置方式	代码配置或 `logging.config.dictConfig()`	`log4j2.xml` / `logback.xml`
结构化日志	需要自定义 Formatter	原生支持 JSON Layout
MDC（Mapped Diagnostic Context，映射诊断上下文，一种把请求 ID 等上下文信息附加到每条日志的机制）	用 `LoggerAdapter` 或 `extra=` 参数	`MDC.put("requestId", id)`
异步日志	需要 `QueueHandler`	`<AsyncLogger>` 配置
性能	够用，高吞吐场景考虑 `loguru`	log4j2 AsyncLogger 性能极高
第三方增强	`loguru`（更简洁的 API）	`logstash-logback-encoder`

1.4 调试技巧

1.4.1 pdb 和 breakpoint()

Python 3.7+ 内置 breakpoint() 函数，在代码里插入就能进入调试器（pdb，Python 自带的命令行调试工具，可以暂停程序、逐行执行、查看变量值）：

python

def process_llm_response(response: dict) -> str:
    """解析 LLM 响应，提取结构化数据"""
    choices = response.get("choices", [])

    # 在这里设置断点，检查 choices 的实际内容
    breakpoint()  # 等价于 import pdb; pdb.set_trace()

    if not choices:
        return ""
    return choices[0].get("message", {}).get("content", "")

常用 pdb 命令：

命令	作用
`n` (next)	执行下一行（不进入函数内部）
`s` (step)	执行下一步（进入函数内部）
`c` (continue)	继续执行直到下一个断点
`p expr`	打印表达式的值
`pp expr`	美化打印（适合 dict/list）
`l` (list)	显示当前位置的代码
`q` (quit)	退出调试器
`w` (where)	显示调用栈

PYTHONBREAKPOINT=0 环境变量可以全局禁用所有 breakpoint() 调用，在生产环境设置这个变量是一种保险措施。

1.4.2 调试异步代码

asyncio 应用的调试需要特别注意：

python

import asyncio
import logging

# 开启 asyncio 调试模式：检测阻塞调用、未等待的协程
asyncio.get_event_loop().set_debug(True)
logging.getLogger("asyncio").setLevel(logging.DEBUG)

async def fetch_llm(prompt: str) -> str:
    # 在异步函数里，breakpoint() 依然有效（Python 3.11+ 有 asyncio 感知）
    breakpoint()
    await asyncio.sleep(0.1)
    return "result"

1.5 异常处理模式

1.5.1 自定义异常类

AI 应用的错误来源多样：网络超时、模型返回错误、解析失败、token 超限。用自定义异常类型区分这些错误，让错误处理代码更清晰：

python

class AIAppError(Exception):
    """AI 应用基础异常，所有自定义异常继承于此"""
    pass

class LLMError(AIAppError):
    """LLM 调用相关错误的基类"""
    def __init__(self, message: str, model: str | None = None, request_id: str | None = None):
        super().__init__(message)
        self.model      = model
        self.request_id = request_id

class LLMTimeoutError(LLMError):
    """LLM 调用超时"""
    def __init__(self, model: str, timeout_seconds: float):
        super().__init__(
            f"LLM call timed out after {timeout_seconds}s",
            model=model
        )
        self.timeout_seconds = timeout_seconds

class LLMParseError(LLMError):
    """LLM 响应解析失败"""
    def __init__(self, model: str, raw_response: str):
        super().__init__("Failed to parse LLM response", model=model)
        self.raw_response = raw_response

class TokenLimitError(LLMError):
    """Token 数量超出限制"""
    def __init__(self, model: str, token_count: int, limit: int):
        super().__init__(
            f"Token count {token_count} exceeds limit {limit}",
            model=model
        )
        self.token_count = token_count
        self.limit       = limit

1.5.2 异常链（Exception Chaining）

Python 的 raise ... from ... 语法可以保留原始异常的上下文（即最初引发错误的原始原因），同时包装成更有语义的异常，调试时能看到完整的错误因果链：

python

import json
from openai import APITimeoutError, RateLimitError

def call_llm_with_retry(messages: list[dict], model: str = "gpt-4o") -> str:
    """调用 LLM，将第三方异常转换为应用自定义异常"""
    client = OpenAI()
    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            timeout=30.0,
        )
        raw = response.choices[0].message.content

        try:
            return json.loads(raw)
        except json.JSONDecodeError as e:
            # raise ... from e 保留原始 JSONDecodeError 作为 __cause__
            # 调试时可以看到完整的错误链
            raise LLMParseError(model=model, raw_response=raw) from e

    except APITimeoutError as e:
        raise LLMTimeoutError(model=model, timeout_seconds=30.0) from e

    except RateLimitError as e:
        # 保留原始错误，同时提供应用层面有意义的错误信息
        raise LLMError("Rate limit exceeded, please retry later", model=model) from e

异常链的输出示例：

code

LLMParseError: Failed to parse LLM response

The above exception was the direct cause of the following exception:

json.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

完整的日志与可观测性架构：

1.6 小结

logging 模块的 Logger / Handler / Formatter 三层结构让日志输出灵活可配：开发环境输出到控制台，生产环境输出 JSON 到 stdout（标准输出，程序默认的文字输出通道）让 k8s（容器编排平台）收集，测试环境可以把 Level 设为 DEBUG 看到所有细节。

AI 应用的日志要记录三类核心信息：发给 LLM 的完整 prompt、模型返回的原始内容、以及每次调用的 token 消耗和耗时。没有这些，成本分析和问题排查都无从下手。

breakpoint() 是最轻量的调试方式，不需要 IDE。复杂的异步应用结合 VS Code 的 debugpy 调试更高效。自定义异常类加上异常链（raise ... from ...），让错误信息在保留原始上下文的同时，对调用方更有意义。

1.7 如何调试 LLM 的输入输出

AI 应用最常见的问题："LLM 为什么给出这个答案？"要回答这个问题，需要看到 LLM 实际收到了什么 prompt，以及返回了什么内容。

1.7.1 方法一：在代码里打印（最简单）

python

import logging
import json
from openai import OpenAI

logger = logging.getLogger(__name__)
client = OpenAI()

def debug_llm_call(messages: list[dict], **kwargs) -> str:
    """包装 LLM 调用，开发环境打印完整的输入输出。"""

    # 打印发送给 LLM 的完整 prompt
    logger.debug("=== LLM INPUT ===")
    for i, msg in enumerate(messages):
        logger.debug(f"[{i}] {msg['role']}: {msg['content'][:200]}...")

    response = client.chat.completions.create(
        messages=messages,
        model=kwargs.get("model", "gpt-4o"),
        **{k: v for k, v in kwargs.items() if k != "model"}
    )

    content = response.choices[0].message.content
    usage = response.usage

    # 打印 LLM 的回复和 token 消耗
    logger.debug("=== LLM OUTPUT ===")
    logger.debug(f"回复: {content[:200]}...")
    logger.debug(f"Token消耗: prompt={usage.prompt_tokens}, "
                 f"completion={usage.completion_tokens}, "
                 f"total={usage.total_tokens}")

    return content

1.7.2 方法二：用 LangSmith 追踪（生产推荐）

python

import os
from langsmith import traceable

# 设置环境变量开启追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"

@traceable(name="llm-debug-call")
def traced_llm_call(prompt: str) -> str:
    """
    加了 @traceable 后，LangSmith 会自动记录：
    - 发送给 LLM 的完整 messages
    - LLM 返回的内容
    - Token 消耗
    - 耗时
    在 LangSmith UI（smith.langchain.com）可以查看完整链路
    """
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

1.7.3 方法三：启用 OpenAI SDK 的调试日志

python

import logging

# 开启 OpenAI SDK 的详细日志（包含完整的 HTTP 请求和响应）
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("openai").setLevel(logging.DEBUG)
logging.getLogger("httpcore").setLevel(logging.DEBUG)

# 运行时会打印类似：
# DEBUG openai._base_client:_base_client.py - Request options: {'method': 'post', 'url': '/chat/completions', ...}
# DEBUG openai._base_client:_base_client.py - Response: 200 {"id": "chatcmpl-...", "choices": [...]}

调试技巧对比：

场景	推荐方法	原因
开发环境快速调试	`logger.debug()` 打印	简单直接
复杂多步骤 Agent 调试	LangSmith `@traceable`	可视化调用树
排查 HTTP 层问题（超时、格式）	OpenAI SDK DEBUG 日志	看到原始请求/响应
生产环境监控	JSON 结构化日志 + ELK	可查询、可告警

1.8 小结

工具/实践	作用	Java 对应
`logging.getLogger(__name__)`	按模块命名 logger	`LoggerFactory.getLogger(getClass())`
`JSONFormatter`	结构化日志	`JsonLayout` (Logback)
`LLMClient` 封装	统一记录 prompt/response/token	`@Around` AOP 切面
`breakpoint()`	命令行调试器	Eclipse/IDEA 断点
`raise ... from ...`	异常链，保留原始错误	`throw new Xxx(e)`
LangSmith `@traceable`	可视化 LLM 调用链路	Zipkin/SkyWalking