这标题看着像一句口号但背后的东西值得掰开揉碎讲清楚。我做 AI 应用后端也有两三年了从最早全链路 print 定位问题到被线上事故逼着重构日志体系中间踩的坑能写满一个记事本。今天这篇文章就围绕结构化日志 request_id 贯穿这套组合方案把为什么、怎么做、踩过哪些坑一次性说透。1. 为什么 AI 应用日志不能再靠 print1.1 print 的三大致命伤先说结论print 不是不能用而是只适合本地调试 代码还没跑起来的阶段。一旦服务上了线、流量进来、多个副本同时跑print 的问题就会无限放大。第一是没结构。print 输出的是字符串你想查某个字段只能用 grep 硬抠。如果日志里混着 JSON、普通文本、报错堆栈你会发现 grep 出来的东西根本没法看更别提拿去做监控告警。第二是没关联。用户一次请求从 API 网关进来经过鉴权、参数校验、RAG 检索、大模型调用、结果后处理再到流式返回中间可能横跨好几个服务、好几十个函数。print 是散的你只知道某个模块打了一行日志但不知道这行日志属于哪一次用户请求。出了问题你只能靠时间戳去猜猜错一次就可能多排查半小时。第三是性能损耗。在高并发下print 默认输出到 stdout遇到阻塞还会拖慢主线程。我见过因为日志量太大把磁盘打满、直接 OOM 的案例原因就是 print 没控制级别、没做采样、没做轮转。1.2 AI 应用的特殊性日志的颗粒度要求更高传统 Web 接口一次请求的日志量可能是几十行。AI 应用呢一次流式对话请求可能涉及 prompt 拼接、上下文截断、embedding 向量检索、rerank 排序、模型流式输出、工具调用链、多轮记忆管理……任何一个环节出问题都需要回溯完整链路。更麻烦的是LLM 的输出是概率性的同样的输入可能产生完全不同的结果。如果没有 request_id 把这轮对话的每一次内部调用串起来你想复现一次用户反馈的回答变差了基本等于大海捞针。所以AI 应用日志的第一性原理是可观测性不是看单条日志而是看整条链路。而链路的最小抓手就是一个贯穿始终的 request_id。2. 结构化日志 request_id 的整体设计2.1 一条日志的完整形态我理想中的日志长这样{ timestamp: 2025-07-14T10:32:11.234Z, request_id: req_8f3a2c91b5, service: rag_router, level: INFO, event: vector_search, module: retriever.vector_store, message: search completed, latency_ms: 156, top_k: 5, score: 0.87, user_id: u_1024 }字段不用多但必须统一。timestamp统一用 UTC ISO8601 格式request_id全局唯一event表示业务事件名module表示代码位置message是人类可读描述剩下的全是结构化字段。为什么不把东西全塞 message 里因为结构化的价值在于可过滤、可聚合、可告警。你可以用eventvector_search AND latency_ms500这样的条件秒级定位慢检索也可以按request_id一键聚合所有关联日志还能直接对接监控系统做 P99 延迟告警。这些都是 print 字符串做不到的。2.2 标准库 logging 改造最小可用方案有人一上来就上 Loguru、Structlog其实最稳妥的做法是先把 Python 标准库logging玩明白。我当时的改造分三步走。第一步定义 JSONFormatterimport json import logging import time class JsonFormatter(logging.Formatter): def format(self, record): log { timestamp: time.strftime(%Y-%m-%dT%H:%M:%S, time.gmtime(record.created)) f.{int(record.msecs):03d}Z, level: record.levelname, service: getattr(record, service, default_service), event: getattr(record, event, record.funcName), module: record.name, message: record.getMessage(), } # 把 extra 字段合并进来 extra getattr(record, extra, None) if extra: log.update(extra) if record.exc_info: log[exception] self.formatException(record.exc_info) return json.dumps(log, ensure_asciiFalse)第二步定义 Logger 封装import logging class AppLogger: def __init__(self, service_name): self.logger logging.getLogger(service_name) self.logger.setLevel(logging.INFO) handler logging.StreamHandler() handler.setFormatter(JsonFormatter(service_nameservice_name)) self.logger.addHandler(handler) def info(self, event, message, **kwargs): self.logger.info(message, extra{event: event, extra: kwargs}) def error(self, event, message, **kwargs): self.logger.error(message, extra{event: event, extra: kwargs})第三步request_id 传递关键点来了。request_id 必须在请求入口生成然后通过上下文传递到所有子模块。Python 里我推荐用contextvars它是线程安全的天然支持 asyncio 协程。import contextvars import uuid request_id_var contextvars.ContextVar(request_id, default-) def new_request_id(): return uuid.uuid4().hex def set_request_id(rid): request_id_var.set(rid) def get_request_id(): return request_id_var.get()然后在入口中间件里rid new_request_id() set_request_id(rid) try: # 处理请求 pass finally: set_request_id(-)再回到日志封装里把 request_id 自动注入def info(self, event, message, **kwargs): auto_extras {request_id: get_request_id()} kwargs.update(auto_extras) self.logger.info(message, extra{event: event, extra: kwargs})这样只要在同一协程链路上所有日志都会自动带上 request_id。你不需要在业务代码里手动传这个参数这是最大的效率提升。2.3 生产环境别用 logging 自带的 RotatingFileHandler很多人第一次配日志轮转时会直接用RotatingFileHandler。它在单进程下没问题但一旦服务用 uvicorn 多 worker 跑多个进程写同一个日志文件就会互相踩。你可能会遇到行被写一半、文件被错误 truncate 等诡异问题。我更推荐的做法是应用层只往 stdout 写 JSON 日志文件轮转、采集、存储交给 Docker 日志驱动或外部的日志采集器处理比如 filebeat、fluentd、vector。在 Kubernetes 环境里这也是标准姿势——容器把日志打到 stdout平台统一采集。这样日志格式、采集策略、存储策略彻底解耦你改日志字段不需要动采集配置。提示如果你必须自己落盘请用ConcurrentRotatingFileHandlercloghandler 包或者按进程 pid 分包写绝不要裸用基础版 RotatingFileHandler。3. AI 链路上的 request_id 贯穿实战3.1 FastAPI 中间件统一注入FastAPI 是目前做 AI 应用最常用的框架之一。我在中间件里做的事情很简单生成 request_id设置 contextvar同时顺手把响应头也带上X-Request-ID这样前端报错时可以直接把 request_id 反馈给我们。from fastapi import FastAPI, Request import logging app FastAPI() app.middleware(http) async def request_id_middleware(request: Request, call_next): rid new_request_id() set_request_id(rid) response await call_next(request) response.headers[X-Request-ID] rid return response但这里有个坑FastAPI 的中间件和路由处理函数之间contextvars的传递在大部分情况下是正常的但如果你的代码里开了新的asyncio.Task比如用了asyncio.create_task做后台任务contextvar 不会自动传播到新任务里request_id 就丢了。解决办法有两个一是尽量把日志同步执行二是在创建 Task 时手动 copy contextimport asyncio from contextvars import copy_context ctx copy_context() asyncio.create_task(run_after(ctx))这个细节非常容易踩我见过不少线上日志断链的事故最后追查出来都是 Task 惹的祸。3.2 LLM 调用链路prompt 和响应必须记录AI 应用日志和普通日志最不一样的地方是要记录模型输入输出。但你千万别把完整 prompt 和完整响应全量打出来——费空间、费带宽、还可能涉及数据隐私。我的习惯是分级处理DEBUG 级记录完整 prompt、完整响应仅开发环境开启INFO 级记录 prompt 长度、token 数、模型名称、延迟、温度参数WARNING 级记录截断事件、重试事件、安全过滤命中的内容摘要ERROR 级记录异常堆栈、失败的阶段、错误码这样既不丢关键信息也不会把日志压垮。另外大模型调用时的请求和响应之间有较长间隔一定要把llm_providerOpenAI/Claude/本地模型、model_name、stream等字段打出来否则你复盘时连哪家模型出的问题都分不清。3.3 检索链路RAG 场景下的埋点RAG 应用里日志最该关注的是召回质量和延迟。每个检索步骤都要有 eventlogger.info(embedding_generate, latency_ms45, tokens_input120) logger.info(vector_search, top_k10, hits8, latency_ms156) logger.info(rerank, candidates10, selected_count3, latency_ms28)配合 request_id你就能回答这样的问题用户 ID 为u_1024的请求里哪一次向量检索超过 500ms哪一次 rerank 后结果太少导致最终回答拉胯实操心得打印检索出的 top 结果 ID 和分数也很有用。整个数组太长时只打印前 5 个但 score 一定要带这是判断召回质量最直接的证据。3.4 流式输出与 SSE流式输出是 AI 应用的常态。很多人刚开始用 SSEServer-Sent Events时只在生成完毕后打一行日志结果一旦中间断流就完全不知道断在哪。我的做法是每推送 N 个 token 或每隔固定时间打一条轻量日志logger.info(token_chunk, stream_position1024, queue_size0.03, elapsed_ms8900)这看起来有点噪音但在定位用户说没收到完整回复但服务端显示已完成这类问题时作用巨大。你需要确认是生成侧中断还是 SSE 推送侧断连还是客户端解析出错有了流式日志这些都能快速区分。4. 线上排查方法论从 request_id 到根因4.1 典型问题用户反馈回答不对用户在前端看到错误提示点击复制 request_id发给过来。你可以直接在日志平台里where request_id req_xxx把全部链路捞出来。然后按顺序检查有没有鉴权或参数错误401/422 这类会直接结束链路。**检索阶段正常吗**看vector_search的 hits 数。如果原本 top_k5结果 hits1说明知识库里没召回相关内容回答自然不准。rerank 之后有没有把高分内容截断如果上下文长度超限截断策略可能把关键内容切掉了。日志里 prompt_tokens 和 max_tokens 一眼就能看出来。大模型返回质量如何看模型名、temperature、以及响应是否因 safety filter 被拦截。这套流程分分钟就能梳理完。4.2 定位延迟瓶颈我把各环节延迟都打到了结构化字段里然后按请求聚合SELECT request_id, MAX(CASE WHEN event vector_search THEN latency_ms END) AS search_ms, MAX(CASE WHEN event llm_call THEN latency_ms END) AS llm_ms FROM app_logs WHERE timestamp now() - interval 1 hour GROUP BY request_id直接在日志平台或者用 pandas 分析一下就能看到搜索阶段和 LLM 阶段的耗时占比。很多时候你以为问题出在模型调用其实是 embedding 接口或 vector store 网络超时拖慢了整体链路。4.3 快慢链路对比线上一批慢日志的复盘例子我之前处理过一个真实案例用户抱怨最近 AI 回答越来越慢。我把慢请求10s的日志拉出来发现 80% 的慢请求集中在同一个 request_id 前缀而它们的共同特征是vector_search耗时高达 3-5 秒。继续追查发现这个问题的根因不在检索本身而是因为业务高峰期知识库新增了一大批文档embedding 服务负载升高导致检索前的 embedding 生成从 50ms 飙到 4 秒。这个结论如果不借助结构化字段纯靠 print 文本很难在大量日志中聚焦。5. 日志体系落地的额外环节5.1 采样与降噪AI 应用日志量天然巨大尤其你为每条 stream 记录 token chunk 后。全量存储不现实尤其是高并发的线上服务。我的策略是默认保存所有 WARN/ERROR 日志INFO 日志按 request_id 采样成功且延迟低于 P95 的请求只存 10%异常和慢请求P99全存DEBUG 日志只在显式开启时输出绝不上生产。其实更精确的做法是尾部采样关键错误全采这类采样策略你在很多可观测性平台里都能配置关键是别所有日志都一股脑存。5.2 敏感信息脱敏AI 应用日志里最容易踩的法律红线就是用户隐私数据。比如 prompt 里带了用户的手机号、邮箱、身份证日志里如果全量记录出了事故就麻烦大了。我的做法是打日志前统一走一个脱敏函数把card_id、phone、email类型的字段替换成掩码。import re def mask_phone(s): return re.sub(r(?\d{3})\d(?\d{2}), *, str(s))更稳妥的是在日志框架层做一个统一的SanitizingFilter对所有结构化字段做正则匹配命中敏感模式就自动打码。这个成本很低但能规避 90% 以上的风险。注意错误详情里带着的大模型完整输出可能也会包含用户隐私内容。建议日志落库前对message字段也做一次检查。5.3 从日志到告警日志不只是用来事后查还要主动发现风险。我比较推荐基于结构化字段做规则告警5 分钟内eventllm_call且http_status429达到一定数量 - 触发限流告警eventvector_search的 P99 延迟超过 1s - 触发检索服务告警eventtoken_chunk的每秒推送速率异常低 - 触发模型推理异常告警。有了这些规则很多线上问题根本不用户反馈你就能提前发现并处理。6. 常见问题排查清单症状可能原因排查方式日志中 request_id 为空中间件未生效 / contextvar 未被设置检查入口处是否调用了 set_request_id兼容 async taskrequest_id 断链异步任务/子线程未继承 context用 copy_context 手动转发或改用框架辅助库JSON 日志 json 解析失败有非 JSON 输出混入 stdout搜索业务代码里的 print建议统一改造成 logger日志文件被多个进程写坏用了 RotatingFileHandler改 stdout 外部采集prompt 日志里出现敏感信息缺少脱敏步骤加 SinitizingFilter 或在日志调用前脱敏日志量太大存储爆炸全量采集 未采样做分层采样重点存慢/错日志trace 日志无序、难以串联未在请求上下文统一注入 request_id按上下文封装 logger统一调度流式输出中断无痕迹未记录 stream 位置每 N 个 token 打一条轻日志7. 最后的实操建议花了一整篇讲这套日志体系其实核心就三件事结构化、request_id、闭环。如果你厌倦了print 打天下别一次性全改。先挑一个最核心的服务只把入口和 LLM 调用链路的日志结构化加上 request_id跑一两周。你会很快体会到排查问题不再是碰运气的感觉。再扎实一点的话建议把你踩过的坑整理成团队规范比如日志字段命名约定哪些事件必须打日志敏感信息怎么处理等。日志体系的价值不在代码本身而在它能不能真正帮你睡个安稳觉。我自己的感受是自从做完这套改造再遇到用户反馈AI 答错了我花 5 分钟就能定位到是检索问题、模型问题还是上下文管理问题。相比之下只靠 print 时代那种看日志靠缘分的日子效率完全不在一个层次。如果你也打算尽快落地出一个简单可行的 checklist定好 JSON 日志格式、字段规范入口中间件统一生成和注入 request_id封装日志库统一自动放入请求上下文字段对 RAG、LLM 调用、SSE 等核心路径打点加脱敏、采样、告警规则用 request_id 串联起全链路。这套东西做完你的 AI 应用才算真正裸奔可观测后续排查压力能小一大半。
