企业官网建设流程全解析

刚开始接手AI应用开发的那阵子，我跟不少人一样，习惯在关键位置塞几个print，看起来简单直接，跑完看控制台输出就行。直到有一次线上用户反馈“机器人答非所问”，我翻了几百行终端日志，愣是找不到一条能串起用户请求全过程的记录——这时候我才意识到，AI应用日志如果继续用print堆，基本等于裸奔。

后来我把日志体系整体换成了结构化日志，用JSON格式记录，再用request_id把一次用户请求从进入网关、到意图识别、到工具调用、再到LLM生成和流式返回的所有环节串成一条完整的链路。改造完成后，平时排查问题从“猜”变成了“查”，点开日志平台直接按request_id过滤，几十秒就能定位到是模型调用超时、工具参数传错，还是提示词把意图带偏了。这篇文章就把这套方案的思路和落地过程完整拆开，聊聊为什么非换不可、字段怎么设计、request_id怎么穿透整个异步链路，以及我在实际改造里踩过的一些坑。

1. 为什么AI应用日志是一条生死线

1.1 print在三层压力下早已崩盘

先把话说透：print不是不能用，它有自己的适用场景——本地调试一个几十行的脚本，跑一遍输出结果，够了。但AI应用和传统CRUD应用有个本质区别：它的每一次响应背后都是一条很长的链路，链路里任何一环出错，最终表现到用户端可能只是一句“答非所问”或者干脆超时。

print的第一个问题是它没有任何结构。你输出一个dict，控制台打出来是{'role': 'assistant', 'content': '你好'}，看着还行，但进了日志系统就是一行字符串，没法按字段筛选。第二个问题是它没有级别。调试信息、警告、错误全混在一起，生产环境也没法只保留WARNING以上级别。第三个问题也是最要命的——它没有关联标识。用户A的请求和用户B的请求在同一段代码里执行，print出来的内容交错在一起，你根本分不清哪几行属于同一个用户，哪几行属于同一次调用。说白了，print是“能看见”，不是“能排查”。

我见过不少AI应用团队,LLM调用、向量检索、工具函数全混在代码里，出事之后只能全量拉日志人工找。运气好时几分钟能对上号，运气不好时——比如流式输出中途断了，或者某个工具调用返回了异常结构——一天都定位不了问题。这种开发方式在小规模Demo阶段能忍，一旦用户量上来、链路复杂化，print日志就是定时炸弹。

1.2 AI应用日志的几个独特诉求

AI应用的日志模式和传统Web应用相比，有很明显的差异，这也是为什么不能照搬老方案的原因。传统Web日志关心的是“谁在什么时间访问了哪个接口，返回什么状态码”，记录对象以HTTP请求为单位；而AI应用日志要回答的问题远不止这些。

• 模型信息：这次调用用的是哪个模型？模型版本是什么？请求时传了哪些参数——temperature、max_tokens、top_p是多少？
• 性能指标：从发起LLM调用到收到第一个token花了多久？整个流式输出总共多久？token消耗了多少？缓存有没有命中？
• 工具调用：Agent在思考过程中决定调用哪个工具？工具入参是什么？返回结果截断后长什么样？
• 链路完整性：一次用户提问可能会触发多轮LLM调用和多个工具调用，这些环节之间如何关联？
• 成本和计费：按token计费的大模型，每次请求消耗的prompt_tokens和completion_tokens，得记录在案，否则月底对账都是糊涂账。

这些信息有一个共同点：它们都是高度结构化的。模型名称、token数、耗时、参数这些都是明确的字段，用print拼字符串去记录只会让日志变成一锅粥。所以结构化日志不是为了让日志“好看”，而是为了把这些关键字段变成可以筛选、聚合、统计的结构化数据，这才是它真正的价值。

1.3 结构化日志加request_id到底解决什么

一句话总结：结构化日志让每一行日志都变成机器可读的数据，request_id让属于同一次请求的所有日志行拥有同一个标记。两者一结合，你在日志平台里输入一个request_id，整条链路就像抽出一根线一样清晰呈现。

举个例子。用户问：“帮我查一下上海明天天气，顺便把长三角几个主要城市明天的温度做个对比。”这条请求在Agent架构里可能会依次触发意图识别、地点实体抽取、调用天气API、多次LLM推理生成对比表格。没有request_id时，这些环节产生的几十条日志散落在海量记录里，人工排查等于大海捞针。有request_id后，你只需要搜一个ID，就能看到：意图识别候选取了哪个、天气API返回了什么、哪一次LLM调用出现了延迟、最终回复是在哪个处理阶段被截断的。这种可观测性，直接决定你维护一个AI应用时的效率和信心。

2. 结构化日志：先把每一行日志变成“一行JSON”

2.1 结构化日志到底是什么

结构化日志的核心思想非常朴素：日志不只给人看，更要让机器能读。实现上最常见的方式就是把日志输出成JSON，每条日志是一条独立的JSON对象，用key-value的方式记录所有信息，而不是人拼的字符串。

{ "timestamp": "2025-01-15T10:23:45.123Z", "level": "INFO", "logger": "app.agent", "message": "llm call started", "request_id": "req_8f3a2b1c", "session_id": "sess_9d2e", "user_id": "u_7788", "model": "gpt-4o-mini", "params": { "temperature": 0.7, "max_tokens": 1024 }, "duration_ms": 0 }

这种格式的好处是显而易见的：Loki、Elasticsearch、ClickHouse这些日志系统可以按字段索引和过滤；告警规则可以直接基于某个字段的值触发；Kibana和Grafana能直接做可视化面板。相比之下，print输出的{"role": "assistant", "content": "..."}其实也是一个Python对象，但它离开了运行环境就只是一段字符串，不能按role或content去查。

这里需要强调一个原则：结构化的对象要在日志产生的源头就构造好，而不是事后正则解析。我看到有些团队先输出人类友好的日志，再写脚本去正则抽字段，这是本末倒置，不仅解析容易出错，还白白增加维护成本。

2.2 字段设计：先想清楚要查什么

设计字段之前，先问自己一个问题：将来出了线上事故，你最可能用什么条件去搜索日志？我的实践经验是，字段要围绕“可查询需求”设计，而不是把能想到的字段全塞进去，否则日志体积爆炸，查询也慢。

我梳理了一套AI应用日志字段标准，分为几个维度：

这套字段在初期不需要一次全部铺满，但request_id、session_id、model、duration_ms这几项建议从第一天就规范起来，因为后期改字段涉及所有埋点位置的变动，成本不低。

实际落地时还有一个容易被忽略的问题：字段名统一。同一个模型名在不同日志里，一会儿叫model一会儿叫model_name，将来查询和做面板的时候就会发现没法聚合，只能写一堆别名脚本。团队内部定一个字段规范，比任何技术方案都重要。

2.3 用Python logging实现JSON输出

市面上有不少结构化日志库，比如structlog。我用过一段时间，功能确实强大，支持Processor链、可定制渲染器。但对于多数团队来说，标准库logging加一个自定义Formatter就能完全覆盖需求，少一个第三方依赖，排查问题也少一个维度。

先看一个完整的JSON Formatter实现。这段代码的核心思路是继承logging.Formatter，把日志记录的所有属性收集成字典，再json.dumps。

import json import logging import time import uuid from contextvars import ContextVar # 用于在线程和异步任务间传递request_id request_id_var: ContextVar[str] = ContextVar("request_id", default="") class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: # 从LogRecord提取基础字段 log_data = { "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created)) + f".{int(record.msecs):03d}Z", "level": record.levelname, "logger": record.name, "message": record.getMessage(), } # 如果LogRecord有额外属性就合并进来 if hasattr(record, "extra_data"): log_data.update(record["extra_data"]) # 注入request_id request_id = request_id_var.get() if request_id: log_data["request_id"] = request_id # 异常堆栈单独处理 if record.exc_info: log_data["exc_info"] = self.formatException(record.exc_info) return json.dumps(log_data, ensure_ascii=False) def setup_logging(level=logging.INFO): handler = logging.StreamHandler() handler.setFormatter(JsonFormatter()) root = logging.getLogger() root.handlers = [handler] root.setLevel(level)

然后用法非常简单：

import logging logger = logging.getLogger("app.agent") logger.info("llm call started", extra={"extra_data": { "model": "gpt-4o-mini", "params": {"temperature": 0.7}, }})

输出就是一行合法JSON。这里有一个关键细节：Python标准库的logging给record附加自定义字段要通过extra参数，而且如果你传的extra里没有message等保留字段，会报KeyError。直接用一个嵌套的extra_data字段收集所有附加信息，可以避开这个问题，逻辑上也更清晰。

2.4 日志分级和采样：别让日志成本失控

结构化日志很好，但副作用也直接：日志量变大了。一个AI应用每次响应可能产生20到30条结构化日志，每条约0.5KB到2KB，一天100万用户请求的情况下，日志存储成本非常可观。所以在落地结构化日志的同时，分级和采样必须一起考虑。

我采用的策略是三层分级。

• DEBUG：只在本地调试或开启debug开关时输出，包含prompt完整体、工具返回的完整结果。这类日志信息量最大但有效期限短，通常跑通之后就不会再看，所以生产环境默认关闭。
• INFO：记录关键链路节点，不包含敏感信息。比如请求开始、意图识别结果、LLM调用开始与结束、工具调用参数与结果摘要、最终回复完成。
• WARNING/ERROR：记录异常和失败，包括超时、重试、解析失败、API异常、生成内容截断。这些日志要带尽可能多的上下文——request_id、session_id、失败URL、错误码。

采样策略上，INFO级别的日志建议全量保留，这是排查问题的关键数据源，别省。但如果某条日志字段特别大，比如完整prompt或完整工具返回，要做截断处理——字段超过某个长度只保留前N个字符，并用truncated: true标记。DEBUG日志在生产环境默认关闭，需要排查特定用户时再按比例或按用户白名单开启。

还有一个容易踩的坑：日志输出本身增加了IO开销。在高并发场景下，如果每个日志都同步写文件或同步发到日志收集器，会对接口延迟有明显影响。建议日志模块内部使用队列加异步消费者来发送，或者直接用FileHandler加延迟刷盘的方式，避免日志IO阻塞业务线程。

3. request_id贯穿：把散落的日志串成一条链路

3.1 request_id不是玄学，是一个身份证号

把request_id理解成快递单号就很好懂。你寄一个包裹，中途经过好几个中转站，每个站点都会扫一下单号。只要单号一致，随时可以查出包裹到了哪里。AI应用里的request_id就是这个单号，它被生成于请求到达的入口，然后一路传递给后续所有环节，每个环节产生的日志都把它带上。

request_id的生成标准很简单：全局唯一，尽量不要用自增数字。分布式环境下用UUID是最省心的方案，虽然字符串长一点，但唯一性有保障，并且可以网上去重。我常用的格式是带前缀的UUID，比如req_8f3a2b1c9d4e4f7ab2c1d0e3f4a5b6c7，这样在日志里一眼就能认出这是req开头的字段，也方便在日志系统里做字段过滤。

这里有个设计细节值得说：request_id必须在应用入口生成，越早越好。它代表的是一次完整的用户请求，而不是某一个内部函数调用。如果在一个函数内部才生成，你只能关联到该函数产生的日志，链路就断了。

3.2 在FastAPI中生成并传递request_id

以FastAPI为例，最干净的做法是写一个HTTP中间件，在请求进入路由处理之前生成request_id，存到ContextVar里，等响应结束后清理。

import uuid from starlette.middleware.base import BaseHTTPMiddleware from contextvars import ContextVar request_id_var: ContextVar[str] = ContextVar("request_id", default="") class RequestIDMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): # 优先复用客户端传来的request_id，方便联调时追溯 request_id = request.headers.get("X-Request-ID") if not request_id: request_id = f"req_{uuid.uuid4().hex}" token = request_id_var.set(request_id) request.state.request_id = request_id try: response = await call_next(request) response.headers["X-Request-ID"] = request_id return response finally: request_id_var.reset(token)

中间件里有几个细节要交代。第一个是允许客户端传入request_id。联调场景下，调用方希望用自己的ID体系来关联请求日志，所以我们要先读X-Request-ID请求头，有就用客户的，没有才自动生成。第二个是在响应头里写回request_id，这样下游调用方拿到响应就能知道这次请求的日志可以从哪里查。这个习惯很多团队没有，但做对外API时特别有用，客户报障直接给ID，双方查日志都是同一把钥匙。

3.3 contextvars：异步世界里的“隐形背包”

request_id有了，怎么让它在异步任务里不丢？这是整套方案里最技术性的一个点。FastAPI里路由函数可能是串行的，但AI应用里到处都是异步调用——httpx.AsyncClient调LLM API、多个工具函数并行执行、后台任务收集结果。如果request_id只是存在一个普通模块变量里，异步任务之间各自独立执行，很可能一个任务顺手改了这个全局变量，另一个任务读到的就变成别人的ID了。

Python的contextvars模块就是专门解决这个问题的。它的设计思路可以理解成给每个异步任务发一个背包，包里放着request_id，无论这个任务被await到哪一步，它读到的都是自己那份值。不同任务的背包互不干扰。

import contextvars import asyncio request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="") async def llm_call(prompt: str): # 这里读到的request_id属于当前调用链 current_rid = request_id_var.get() logger.info("llm call start", extra={"extra_data": {"request_id": current_rid}}) await asyncio.sleep(1) return "response" async def handle_request(): # 在入口处设置request_id request_id_var.set("req_test_123") await llm_call("hello")

在FastAPI中间件里request_id_var.set()之后，同一个请求上下文内的所有异步任务都会自动继承这个值，无需手动传递参数——前提是你用了async函数并且在同一个task生命周期内。这里最关键的一条：不要在异步任务里用线程本地存储。threading.local()在异步场景下完全不适用，因为多个协程可能共享同一个线程，但它们的请求ID完全不同。contextvars的语义才是异步场景下的正确选择。

3.4 跨进程、跨任务时request_id怎么传

request_id不能只在Python进程内转，它还要随请求链路传到外部服务。比如你的应用调用了另一个微服务或者向量数据库的API，下游服务的日志里如果也能带上同一个request_id，整个分布式链路就是贯通的。

HTTP协议里有一个约定俗成的字段：X-Request-ID。上游把你的request_id放进这个请求头，下游服务的中