AI 应用的日志、Tracing 与可观测性入门

模型调用成功，并不意味着 AI 应用运行正常：答案可能为空，耗时可能突然升高，或者某次工具调用失败后被重试多次。没有可观测性时，我们只能看到用户的一句“怎么这么慢”。本篇只解决一个问题：如何为一次 AI 请求记录可关联、可脱敏、能解释问题的最小观测信息。示例只使用 Python 标准库和离线模拟模型，不需要密钥或网络。

三个概念分别解决什么问题

日志（log）是某个时间点发生的事件，例如“开始调用模型”“模型返回错误”。日志适合定位单个异常，但如果每行都没有请求标识，多用户并发时很难知道它们属于哪一次请求。

Trace是一条请求的完整链路。它通常有一个 trace_id，并由一个或多个 Span 组成，例如“准备提示词”“调用模型”“解析结果”。本篇不引入第三方 tracing SDK，而是用一个 Trace 对象模拟最小结构：记录开始时间、结束时间、状态和事件。

可观测性不是“打印更多内容”，而是让程序能从输出推断内部状态。对 AI 应用来说，至少要能回答：哪次请求、调用哪个模型、耗时多久、输入输出大致多大、是否重试、最后为什么结束。完整提示词和回答可能含有隐私，不应默认写进普通日志。

先确定观测字段

字段应服务于排错，而不是复制业务数据。一个实用的最小集合如下：

trace_id：关联同一请求的日志；
model：模型名称或内部别名；
duration_ms：请求耗时；
input_chars、output_chars：规模指标，不记录原文；
status：ok、error 或 timeout；
error_type：异常类型，便于统计；
attempt：当前尝试次数。

trace_id 应由应用生成并贯穿调用链，而不是让模型返回。生产系统还可以把这些字段送入日志平台，再按模型、状态和耗时分组观察。不要把 API 密钥、Authorization 请求头、完整用户输入或模型原文作为日志字段。

用标准库建立请求级日志

Python 官方 logging 模块支持 Logger、Handler 和 Formatter；extra 可以向一条日志附加上下文。为了让示例可直接运行，下面使用 JSON 输出，且把 Trace ID 作为显式字段传入。这样日志可以被命令行查看，也容易被日志系统解析。

import json
import logging
import time
import uuid
from dataclasses import dataclass, field


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        item = {
            "time": self.formatTime(record, "%Y-%m-%dT%H:%M:%S%z"),
            "level": record.levelname,
            "event": record.getMessage(),
            "trace_id": getattr(record, "trace_id", "-"),
        }
        for name in ("model", "duration_ms", "input_chars", "output_chars",
                     "status", "error_type", "attempt"):
            if hasattr(record, name):
                item[name] = getattr(record, name)
        return json.dumps(item, ensure_ascii=False)


logger = logging.getLogger("ai_app")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)


@dataclass
class Trace:
    trace_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    started: float = field(default_factory=time.perf_counter)
    events: list[str] = field(default_factory=list)

    def elapsed_ms(self) -> int:
        return round((time.perf_counter() - self.started) * 1000)

perf_counter() 适合测量耗时，因为它用于计算经过的时间，而不是展示墙上时钟。Trace.events 在这个最小示例中只保存在内存里；如果要跨服务传递，就应把 trace_id 放入请求上下文或标准 tracing 系统，而不能依赖全局变量。全局变量在并发请求下会互相覆盖。

包装一次模型调用

先用离线函数模拟模型，确保观测代码可以独立验证。真实 SDK 接入时，只需把 fake_model 换成调用函数，并保留开始、成功、异常三个记录点。

def fake_model(prompt: str) -> str:
    time.sleep(0.01)  # 模拟网络和模型处理耗时
    return f"已处理：{prompt[:20]}"


def generate(prompt: str, model: str = "demo-model") -> str:
    trace = Trace()
    common = {"trace_id": trace.trace_id, "model": model, "attempt": 1}
    logger.info("model_call_started", extra={**common, "input_chars": len(prompt)})
    try:
        answer = fake_model(prompt)
        logger.info(
            "model_call_finished",
            extra={**common, "output_chars": len(answer),
                   "duration_ms": trace.elapsed_ms(), "status": "ok"},
        )
        return answer
    except TimeoutError as exc:
        logger.warning(
            "model_call_failed",
            extra={**common, "duration_ms": trace.elapsed_ms(),
                   "status": "timeout", "error_type": type(exc).__name__},
        )
        raise
    except Exception as exc:
        logger.exception(
            "model_call_failed",
            extra={**common, "duration_ms": trace.elapsed_ms(),
                   "status": "error", "error_type": type(exc).__name__},
        )
        raise


if __name__ == "__main__":
    print(generate("请把这句话改写得更清楚。"))

开始和结束日志共享同一个 trace_id，因此可以把它们拼成一次调用。结束日志记录字符数而非原文，既能发现输入过长、输出为空等问题，也降低敏感信息泄露风险。logger.exception 会附带堆栈，适合错误日志；但堆栈本身也可能包含路径或参数，进入集中式平台前仍应检查脱敏策略。

从日志到可用的 Trace

当一次请求不只是模型调用，还包含检索、重排和工具执行时，可以为每个阶段记录 Span。最简单的做法是让每个阶段都接收同一个 Trace，并用事件名区分：retrieval_started、retrieval_finished、model_call_finished。每个 Span 至少需要 name、start、duration_ms 和 status。

实际工程中可采用 OpenTelemetry 等标准方案，让 Web 服务、数据库和模型调用自动串联，并将 Trace 导出到后端。无论使用什么 SDK，原则不变：Trace ID 负责关联，Span 负责分段，指标负责聚合。不要把“有 Trace”误认为“已经可观测”；还要为错误率、P95 延迟、每次请求 token 数或费用设置监控和告警。

常见问题

为什么不直接打印完整提示词？ 因为提示词可能含个人信息、业务机密或注入内容。调试时也应使用截断、哈希、字段白名单或经过授权的临时采样。

日志级别该怎么选？ 正常生命周期用 INFO，可恢复但值得关注的情况用 WARNING，带堆栈的失败用 ERROR。DEBUG 可以记录更多诊断细节，但不要在生产环境无控制地开启。

只记录总耗时够吗？ 不够。总耗时无法说明慢在检索、模型、工具还是重试。按阶段记录 Span，才能找到瓶颈。

如何验证字段没有漏记？ 为日志格式写测试，模拟成功、超时和普通异常三条路径，断言都有 trace_id、status 和 duration_ms，并断言输出中不存在密钥或原始用户文本。

小结

AI 应用的最小可观测性可以从标准库开始：为每次请求生成 trace_id，记录模型调用的开始、结束、状态和耗时，用 JSON 结构化输出，并主动避免记录敏感原文。随着链路变长，再把检索、工具和模型拆成 Span，配合错误率、延迟和成本指标。可观测性不是上线后的附加功能，而是验证重试、上下文裁剪和 Agent 停止条件是否按预期工作的基础。

搜索时光笔记