企业官网建设流程全解析

这标题看着像一句口号，但背后的东西值得掰开揉碎讲清楚。我做 AI 应用后端也有两三年了，从最早全链路 print 定位问题，到被线上事故逼着重构日志体系，中间踩的坑能写满一个记事本。今天这篇文章就围绕"结构化日志 + request_id 贯穿"这套组合方案，把为什么、怎么做、踩过哪些坑，一次性说透。

1. 为什么 AI 应用日志不能再靠 print

1.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 里？因为结构化的价值在于可过滤、可聚合、可告警。你可以用event=vector_search AND latency_ms>500这样的条件秒级定位慢检索；也可以按request_id一键聚合所有关联日志；还能直接对接监控系统做 P99 延迟告警。这些都是 print + 字符串做不到的。

2.2 标准库 logging 改造：最小可用方案

有人一上来就上 Loguru、Structlog，其实最稳妥的做法是先把 Python 标准库logging玩明白。我当时的改造分三步走。

第一步：定义 JSONFormatter

import 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_ascii=False)

第二步：定义 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_name=service_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，平台统一采集。这样日志格式、采集策略、存储策略彻底解耦，你改日志字段不需要动采集配置。

提示：如果你必须自己落盘，请用ConcurrentRotatingFileHandler（cloghandler 包），或者按进程 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 context：

import 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_provider（OpenAI/Claude/本地模型）、model_name、stream等字段打出来，否则你复盘时连"哪家模型出的问题"都分不清。

3.3 检索链路：RAG 场景下的埋点

RAG 应用里，日志最该关注的是召回质量和延迟。每个检索步骤都要有 event：

logger.info("embedding_generate", latency_ms=45, tokens_input=120) logger.info("vector_search", top_k=10, hits=8, latency_ms=156) logger.info("rerank", candidates=10, selected_count=3, latency_ms=28)

配合 request_id，你就能回答这样的问题：用户 ID 为u_1024的请求里，哪一次向量检索超过 500ms？哪一次 rerank 后结果太少导致最终回答拉胯？

实操心得：打印检索出的 top 结果 ID 和分数也很有用。整个数组太长时只打印前 5 个，但 score 一定要带，这是判断召回质量最直接的证据。

3.4 流式输出与 SSE

流式输出是 AI 应用的常态。很多人刚开始用 SSE（Server-Sent Events）时，只在生成完毕后打一行日志，结果一旦中间断流就完全不知道断在哪。

我的做法是：每推送 N 个 token 或每隔固定时间，打一条轻量日志：

logger.info("token_chunk", stream_position=1024, queue_size=0.03, elapsed_ms=8900)

这看起来有点"噪音"，但在定位"用户说没收到完整回复，但服务端显示已完成"这类问题时，作用巨大。你需要确认：是生成侧中断？还是 SSE 推送侧断连？还是客户端解析出错？有了流式日志，这些都能快速区分。

4. 线上排查方法论：从 request_id 到根因

4.1 典型问题：用户反馈"回答不对"

用户在前端看到错误提示，点击"复制 request_id"发给过来。你可以直接在日志平台里where request_id = "req_xxx"，把全部链路捞出来。然后按顺序检查：

1. 有没有鉴权或参数错误？401/422 这类会直接结束链路。
2. **检索阶段正常吗？**看vector_search的 hits 数。如果原本 top_k=5，结果 hits=1，说明知识库里没召回相关内容，回答自然不准。
3. rerank 之后，有没有把高分内容截断？如果上下文长度超限，截断策略可能把关键内容切掉了。日志里 prompt_tokens 和 max_tokens 一眼就能看出来。
4. 大模型返回质量如何？看模型名、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 分钟内event=llm_call且http_status=429达到一定数量 -> 触发限流告警；
• event=vector_search的 P99 延迟超过 1s -> 触发检索服务告警；
• event=token_chunk的每秒推送速率异常低 -> 触发模型推理异常告警。

有了这些规则，很多线上问题根本不用户反馈，你就能提前发现并处理。

6. 常见问题排查清单

7. 最后的实操建议

花了一整篇讲这套日志体系，其实核心就三件事：结构化、request_id、闭环。

如果你厌倦了"print 打天下"，别一次性全改。先挑一个最核心的服务，只把入口和 LLM 调用链路的日志结构化，加上 request_id，跑一两周。你会很快体会到"排查问题不再是碰运气"的感觉。

再扎实一点的话，建议把你踩过的坑整理成团队规范，比如"日志字段命名约定""哪些事件必须打日志""敏感信息怎么处理"等。日志体系的价值，不在代码本身，而在它能不能真正帮你睡个安稳觉。

我自己的感受是：自从做完这套改造，再遇到用户反馈"AI 答错了"，我花 5 分钟就能定位到是检索问题、模型问题还是上下文管理问题。相比之下，只靠 print 时代那种"看日志靠缘分"的日子，效率完全不在一个层次。

如果你也打算尽快落地，出一个简单可行的 checklist：

1. 定好 JSON 日志格式、字段规范；
2. 入口中间件统一生成和注入 request_id；
3. 封装日志库，统一自动放入请求上下文字段；
4. 对 RAG、LLM 调用、SSE 等核心路径打点；
5. 加脱敏、采样、告警规则；
6. 用 request_id 串联起全链路。

这套东西做完，你的 AI 应用才算真正"裸奔可观测"，后续排查压力能小一大半。