尧图网站建设 
1. 为什么 AI 应用的日志不能再靠 print 硬扛刚接触 AI 应用开发那会儿我和大多数人一样调试全靠print。模型返回了什么、请求卡在哪一步、token 消耗了多少统统print到终端里看一眼。本地跑个 demo 没问题可一旦把服务部署到线上多个用户同时发请求终端里的输出就像春运火车站的大屏——密密麻麻、互相穿插根本分不清哪条日志属于哪次请求。更别提排查线上问题了用户说“我这边回答错了”你连他当时发的是什么 prompt 都对不上号。这个项目要解决的核心问题就一个让 AI 应用的日志从“能看”变成“能查、能追、能分析”。具体做法是两条腿走路——用结构化日志替代裸print让每条日志都带字段、可被机器解析用request_id贯穿一次请求的完整生命周期让散落在各处的日志能串成一条线。这套方案适合所有正在做 AI 应用开发的人不管你是刚入门的新手还是已经上线了服务、被日志问题折磨过的老手都能直接抄作业。我先把结论摆在这儿print不是不能用而是它只适合“临时看一眼”的场景。一旦你的 AI 应用涉及到多轮对话、工具调用、流式输出、异步任务print就会变成技术债。结构化日志加 request_id 这套组合本质上是在给你的应用装一套“黑匣子”出问题时能快速还原现场平时还能做统计分析。下面我从设计思路开始一步步拆给你看。2. 整体设计思路与方案选型2.1 从 print 到结构化日志到底改变了什么print输出的是给人看的纯文本比如用户提问: 今天天气怎么样。这条信息人眼能读懂但机器读不懂。你想统计“今天有多少次请求是关于天气的”只能靠字符串匹配脆弱且容易出错。结构化日志的核心区别在于每条日志是一个带字段的数据结构通常用 JSON 格式输出比如{ timestamp: 2025-01-15T10:23:45.123Z, level: INFO, request_id: req-a1b2c3d4, event: llm_request_start, model: gpt-4o, prompt_tokens: 128, user_id: u_9527 }这条日志里request_id是贯穿字段event是事件类型model、prompt_tokens、user_id是业务字段。有了这些字段你可以直接用日志平台做聚合查询比如“查所有levelERROR且modelgpt-4o的请求”或者“统计每个用户的平均 token 消耗”。这就是结构化日志的价值——它让日志从“文本”变成了“数据”。我选 JSON 作为输出格式理由有三点。第一JSON 是自描述的字段名和值一一对应不需要额外维护解析规则。第二几乎所有日志采集工具Filebeat、Fluentd、Logstash都原生支持 JSON 解析接入成本低。第三JSON 在 Python 里有成熟的序列化库性能开销可控。相比之下用keyvalue这种格式虽然更紧凑但遇到嵌套结构就不好处理了AI 应用里嵌套的 metadata 很常见所以 JSON 更合适。2.2 request_id 为什么是贯穿链路的“主键”一次 AI 请求往往不是单一操作而是由多个步骤组成的链路接收用户输入 → 构造 prompt → 调用模型 API → 处理流式响应 → 调用工具函数 → 生成最终回复。如果每个步骤都独立打日志出问题时你看到的就是一堆互不关联的记录根本不知道哪几条属于同一次请求。request_id的作用就是给这一次请求分配一个唯一标识在链路的每个环节都带上它。这样无论日志散落在多少个文件、多少个服务里只要按request_id过滤就能还原出完整的执行轨迹。这跟数据库里的主键是一个道理——它是串联所有相关记录的那根线。生成request_id我推荐用 UUID4简单可靠碰撞概率可以忽略。如果你想要更短的 ID可以用时间戳加随机数的组合但要注意在高并发下保证唯一性。我实测下来UUID4 的前 8 位在单机场景下已经足够区分但跨服务追踪时建议保留完整 UUID避免极端情况下的碰撞。2.3 方案选型的几个关键取舍在动手之前有几个选型问题需要先想清楚这些取舍会直接影响后续的实现方式。第一个取舍用标准库 logging 还是第三方库。Python 标准库的logging模块功能足够配合json序列化就能输出结构化日志零依赖。第三方库比如structlog提供了更优雅的 API 和更丰富的处理器但引入了额外依赖。我的建议是如果你的项目已经用了structlog或loguru继续用如果是新项目且团队对依赖敏感标准库完全够用。这个项目我用标准库实现因为它的可移植性最好你复制过去就能跑。第二个取舍同步写日志还是异步写。AI 应用的请求延迟本来就高模型推理动辄几秒日志写入的几毫秒开销相对可以忽略。但如果你的 QPS 很高同步写日志可能成为瓶颈。这时候可以用队列加后台线程的方式异步写但会增加复杂度。我的经验是日吞吐量在十万条以下同步写完全没问题超过这个量级再考虑异步。第三个取舍日志输出到文件还是标准输出。容器化部署的场景下推荐输出到标准输出stdout由容器运行时负责收集。传统虚拟机部署则输出到文件配合 Filebeat 采集。这个项目两种都支持通过配置切换。3. 核心细节解析与实操要点3.1 结构化日志的字段设计规范字段设计是结构化日志的地基设计得好后续查询分析事半功倍设计得乱日志就是一堆垃圾数据。我总结了一套 AI 应用日志的字段规范分成三类。第一类是通用字段每条日志都必须有字段名类型说明timestampstringISO 8601 格式带毫秒和时区levelstringDEBUG/INFO/WARNING/ERROR/CRITICALrequest_idstring请求唯一标识贯穿链路eventstring事件类型用下划线命名servicestring服务名微服务架构下必填loggerstring日志记录器名称通常是模块名第二类是业务字段根据事件类型动态添加。比如 LLM 调用事件带上model、prompt_tokens、completion_tokens、latency_ms工具调用事件带上tool_name、tool_input、tool_output。第三类是上下文字段比如user_id、session_id、trace_id。这些字段不一定每条日志都有但在需要关联用户行为时非常关键。注意字段名统一用蛇形命名snake_case不要混用驼峰。字段值尽量用基本类型避免嵌套过深。如果确实需要嵌套控制在两层以内否则查询时会很痛苦。我踩过的一个坑是早期把整个 prompt 原文塞进日志字段结果日志文件暴涨一天几个 G。后来改成只记录 prompt 的哈希值和长度需要看原文时再去专门的存储里查。日志里不要放超大文本这是血泪教训。3.2 request_id 的生成与传递机制request_id的生成时机很关键。它必须在请求进入应用的第一时间生成早于任何业务逻辑。在 Web 框架里通常用中间件middleware来实现。以 FastAPI 为例一个请求进来中间件先执行生成request_id然后把它注入到请求上下文里后续所有日志都从这个上下文取。传递机制有两种常见方案。方案一是显式传递把request_id作为参数在函数间传递。这种方式直观但侵入性强每个函数都要加参数容易漏。方案二是上下文变量用 Python 的contextvars模块把request_id存到一个全局可访问的上下文变量里日志记录器自动读取。这种方式对业务代码零侵入我强烈推荐。contextvars的原理是给每个执行上下文比如每个请求的协程维护一份独立的变量副本互不干扰。这正好契合异步框架的并发模型。你只需要在中间件里set一次后续在任何地方get都能拿到当前请求的request_id。import contextvars request_id_var contextvars.ContextVar(request_id, defaultNone) def get_request_id(): return request_id_var.get()这段代码定义了一个上下文变量默认值是None。中间件里调用request_id_var.set(new_id)设置日志过滤器里调用get_request_id()读取。就这么简单。3.3 日志记录器的封装与过滤器标准库的logging模块要输出结构化日志需要做两件事自定义 Formatter 和自定义 Filter。Formatter 负责把日志记录对象序列化成 JSON。它从record对象里提取字段组装成字典再json.dumps。这里有个细节record对象里有很多内置属性如name、levelname、pathname你需要区分哪些是内置的、哪些是你通过extra参数传进来的业务字段。我的做法是维护一个内置属性白名单白名单之外的都当作业务字段处理。Filter 负责注入request_id。它在每条日志被处理时执行从上下文变量里读取request_id塞进record对象。这样 Formatter 序列化时就能拿到它。import logging import json class JsonFormatter(logging.Formatter): def format(self, record): log_data { timestamp: self.formatTime(record), level: record.levelname, logger: record.name, message: record.getMessage(), } # 注入 request_id if hasattr(record, request_id): log_data[request_id] record.request_id # 注入业务字段 for key, value in record.__dict__.items(): if key not in RESERVED_ATTRS: log_data[key] value return json.dumps(log_data, ensure_asciiFalse)RESERVED_ATTRS是内置属性集合需要提前定义好。这个封装一次写好全项目复用。3.4 日志级别与采样策略AI 应用的日志量很容易失控尤其是 DEBUG 级别。我的建议是生产环境默认 INFO 级别DEBUG 只在排查问题时临时开启。INFO 级别记录关键节点比如请求开始、模型调用完成、工具调用、请求结束。DEBUG 级别记录详细参数比如完整的 prompt、模型原始响应。对于高频事件比如流式输出的每个 token不要逐条打日志而是采样或聚合。比如每 100 个 token 打一条进度日志或者只在流式结束时打一条汇总日志。这样既保留了可观测性又不会把日志系统压垮。实操心得给日志加上sampling_rate字段记录这条日志的采样率。这样在做统计时可以用采样率反推真实数量。比如采样率 0.1统计到 1000 条实际约 10000 条。4. 实操过程与核心环节实现4.1 环境准备与依赖安装这个项目用 Python 实现依赖很少。核心只需要标准库如果要跑 Web 服务示例需要 FastAPI 和 Uvicorn。pip install fastapi uvicorn日志采集部分如果部署在服务器上推荐用 Filebeat 采集日志文件。Filebeat 的安装这里不展开重点讲配置。它的核心配置是filebeat.inputs指定日志路径output.elasticsearch或output.logstash指定输出目标。Filebeat 会自动解析 JSON 格式的日志行把字段提取出来。4.2 日志模块的完整实现我把日志模块拆成三个文件context.py管理上下文变量formatter.py定义格式化器logger.py提供初始化函数。这样职责清晰便于维护。context.py里定义request_id_var和读写函数。formatter.py里定义JsonFormatter和RequestIdFilter。logger.py里提供setup_logging()函数配置根日志记录器添加处理器和过滤器。# logger.py import logging import sys from .formatter import JsonFormatter, RequestIdFilter def setup_logging(levellogging.INFO, outputstdout): logger logging.getLogger() logger.setLevel(level) logger.handlers.clear() if output stdout: handler logging.StreamHandler(sys.stdout) else: handler logging.FileHandler(app.log) handler.setFormatter(JsonFormatter()) handler.addFilter(RequestIdFilter()) logger.addHandler(handler) return logger调用setup_logging()后全项目的日志都会走这套配置。业务代码里只需要logger logging.getLogger(__name__)然后正常打日志即可。4.3 在 FastAPI 中集成 request_id 中间件中间件是注入request_id的最佳位置。它在请求进入时生成 ID在请求结束时清理上下文保证不同请求之间不串号。from fastapi import FastAPI, Request import uuid from .context import request_id_var app FastAPI() app.middleware(http) async def add_request_id(request: Request, call_next): req_id request.headers.get(X-Request-ID) or str(uuid.uuid4()) request_id_var.set(req_id) response await call_next(request) response.headers[X-Request-ID] req_id return response这里有个细节优先从请求头X-Request-ID读取如果客户端传了就用客户端的没传就自己生成。这样做的好处是支持跨服务追踪——上游服务生成的request_id可以透传到下游。响应头里也带上request_id方便前端排查问题时提供给后端。4.4 在 AI 调用链路中打日志有了基础设施业务代码里打日志就很自然了。我在 LLM 调用的关键节点都埋了日志。请求开始时打一条llm_request_start带上模型名和 prompt 长度。模型返回后打一条llm_request_end带上 token 消耗和延迟。如果调用失败打llm_request_error带上错误码和错误信息。import logging import time logger logging.getLogger(__name__) async def call_llm(prompt, model): start time.time() logger.info(llm_request_start, extra{ event: llm_request_start, model: model, prompt_length: len(prompt), }) try: response await llm_client.chat(prompt, model) latency int((time.time() - start) * 1000) logger.info(llm_request_end, extra{ event: llm_request_end, model: model, prompt_tokens: response.usage.prompt_tokens, completion_tokens: response.usage.completion_tokens, latency_ms: latency, }) return response except Exception as e: logger.error(llm_request_error, extra{ event: llm_request_error, model: model, error_type: type(e).__name__, error_message: str(e), }) raise注意extra参数里的event字段它和日志消息分开消息是给人看的简短描述event是给机器用的分类标签。这样查询时可以用event:llm_request_end精确过滤。4.5 日志采集与查询配置日志写到文件或标准输出后需要采集到日志平台才能发挥价值。我用 Filebeat 采集输出到 Elasticsearch用 Kibana 查询。Filebeat 的配置关键点有两个一是开启 JSON 解析二是设置request_id为 keyword 类型方便聚合。filebeat.inputs: - type: filestream paths: - /var/log/ai-app/*.log parsers: - ndjson: target: add_error_key: truendjson解析器会把每行 JSON 展开成字段。target: 表示字段平铺到根层级不嵌套。这样request_id就是顶层字段查询时直接request_id: xxx即可。在 Kibana 里我常用的查询有这么几个。查某次请求的完整链路request_id: req-a1b2c3d4按时间排序。查所有错误level: ERROR。查慢请求event: llm_request_end and latency_ms 5000。查某个模型的调用量按model字段做聚合。5. 常见问题与排查技巧实录5.1 日志里 request_id 是 null 怎么办这是最常见的问题原因通常是日志在中间件设置request_id之前就打了。比如应用启动时的初始化日志或者中间件之外的代码。解决办法是给request_id一个默认值比如system表示非请求触发的日志。还有一种情况是异步任务里丢了上下文。contextvars在asyncio.create_task创建的新任务里默认是空的需要手动复制上下文。用contextvars.copy_context()复制当前上下文再传给新任务。import asyncio import contextvars ctx contextvars.copy_context() asyncio.create_task(some_task(), contextctx)这个坑我在做异步工具调用时踩过排查了半天才发现是上下文没传过去。5.2 日志量太大导致磁盘打满AI 应用的日志量确实容易失控尤其是把完整 prompt 和响应都记进去的时候。我的应对策略分三层。第一层是控制字段大文本只记哈希和长度不记原文。第二层是控制级别生产环境用 INFODEBUG 按需开启。第三层是日志轮转用RotatingFileHandler或TimedRotatingFileHandler限制单文件大小和保留天数。from logging.handlers import RotatingFileHandler handler RotatingFileHandler( app.log, maxBytes100 * 1024 * 1024, # 100MB backupCount10, )这样最多占用 1GB 磁盘超过就自动清理最旧的文件。5.3 结构化日志影响性能怎么优化有人担心 JSON 序列化拖慢应用。实测下来单条日志的序列化开销在微秒级相比模型推理的秒级延迟可以忽略。但如果你的日志量真的很大可以从这几个方面优化。一是延迟序列化用logging的LogRecord先缓存真正输出时才序列化。二是异步写入用队列加后台线程。三是减少不必要的字段字段越少序列化越快。我做过一个压测单机每秒写 5 万条结构化日志CPU 占用增加约 15%。对于绝大多数 AI 应用来说这个开销完全可以接受。5.4 常见问题速查表问题现象可能原因排查方法解决方案request_id 为 null日志早于中间件执行检查日志时间戳与请求开始时间设置默认值或调整中间件顺序日志无法被采集格式不是标准 JSON用jq验证日志行检查 Formatter 输出字段查询不到字段类型是 text 而非 keyword查看索引映射修改索引模板设为 keyword异步任务日志丢失上下文未传递检查任务创建方式用 copy_context 复制上下文日志文件不轮转Handler 配置错误检查 Handler 类型改用 RotatingFileHandler中文乱码序列化未指定编码查看日志文件编码json.dumps(ensure_asciiFalse)5.5 几个我踩过的坑和独家技巧坑一extra 参数覆盖内置字段。如果你在extra里传了message或level这种内置字段名会直接报错。解决办法是业务字段加前缀比如biz_message或者维护一个保留字段列表做校验。坑二日志顺序错乱。多线程或多进程写同一个文件时日志可能交错。解决办法是用QueueHandler加QueueListener把日志写入串行化。技巧一给日志加 trace_id 和 span_id。如果你的应用有分布式追踪request_id可以作为trace_id每个步骤生成span_id这样能和追踪系统打通。技巧二用日志做实时告警。在日志平台配置规则比如“5 分钟内 ERROR 日志超过 10 条”就触发告警。这比等用户投诉再排查主动得多。技巧三定期分析日志找优化点。我每周会跑一次日志分析看哪些 prompt 的 token 消耗最高、哪些模型的延迟最大、哪些错误最频繁。这些数据直接指导了后续的优化方向。6. 从日志到可观测性的延伸结构化日志加 request_id 只是可观测性的第一步。当你把这套基础设施搭好之后会发现它能延伸出很多有价值的应用。比如成本分析。每条 LLM 调用日志都带 token 消耗按user_id聚合就能算出每个用户的成本按model聚合就能对比不同模型的性价比。我们团队就是靠这个数据把一部分简单任务从大模型切到了小模型成本降了六成。再比如质量监控。给模型响应打一个质量评分字段记录在日志里就能追踪模型输出的质量趋势。如果某个时间段评分下降可能是 prompt 模板出了问题或者模型版本更新导致的。还有用户行为分析。通过session_id把同一用户的多次请求串起来能看出用户的使用路径和偏好。这些数据对产品迭代很有参考价值。我个人在实际操作中的体会是日志这件事前期多花一小时设计字段后期能省十小时排查时间。很多团队觉得打日志是小事随便print一下就行等到线上出问题才发现日志根本不够用。结构化日志加 request_id 这套方案投入不大但回报是长期的。你不需要一次性做到完美先把request_id贯穿起来再把关键节点的日志结构化逐步迭代就行。最后分享一个小技巧在开发环境把日志输出成带颜色的可读格式生产环境输出 JSON。这样开发时看着舒服生产时机器好解析。用同一个 Formatter 接口根据环境变量切换实现即可代码改动很小。 AI应用日志实战：结构化日志与request_id全链路追踪
1. 为什么 AI 应用的日志不能再靠 print 硬扛刚接触 AI 应用开发那会儿我和大多数人一样调试全靠print。模型返回了什么、请求卡在哪一步、token 消耗了多少统统print到终端里看一眼。本地跑个 demo 没问题可一旦把服务部署到线上多个用户同时发请求终端里的输出就像春运火车站的大屏——密密麻麻、互相穿插根本分不清哪条日志属于哪次请求。更别提排查线上问题了用户说“我这边回答错了”你连他当时发的是什么 prompt 都对不上号。这个项目要解决的核心问题就一个让 AI 应用的日志从“能看”变成“能查、能追、能分析”。具体做法是两条腿走路——用结构化日志替代裸print让每条日志都带字段、可被机器解析用request_id贯穿一次请求的完整生命周期让散落在各处的日志能串成一条线。这套方案适合所有正在做 AI 应用开发的人不管你是刚入门的新手还是已经上线了服务、被日志问题折磨过的老手都能直接抄作业。我先把结论摆在这儿print不是不能用而是它只适合“临时看一眼”的场景。一旦你的 AI 应用涉及到多轮对话、工具调用、流式输出、异步任务print就会变成技术债。结构化日志加 request_id 这套组合本质上是在给你的应用装一套“黑匣子”出问题时能快速还原现场平时还能做统计分析。下面我从设计思路开始一步步拆给你看。2. 整体设计思路与方案选型2.1 从 print 到结构化日志到底改变了什么print输出的是给人看的纯文本比如用户提问: 今天天气怎么样。这条信息人眼能读懂但机器读不懂。你想统计“今天有多少次请求是关于天气的”只能靠字符串匹配脆弱且容易出错。结构化日志的核心区别在于每条日志是一个带字段的数据结构通常用 JSON 格式输出比如{ timestamp: 2025-01-15T10:23:45.123Z, level: INFO, request_id: req-a1b2c3d4, event: llm_request_start, model: gpt-4o, prompt_tokens: 128, user_id: u_9527 }这条日志里request_id是贯穿字段event是事件类型model、prompt_tokens、user_id是业务字段。有了这些字段你可以直接用日志平台做聚合查询比如“查所有levelERROR且modelgpt-4o的请求”或者“统计每个用户的平均 token 消耗”。这就是结构化日志的价值——它让日志从“文本”变成了“数据”。我选 JSON 作为输出格式理由有三点。第一JSON 是自描述的字段名和值一一对应不需要额外维护解析规则。第二几乎所有日志采集工具Filebeat、Fluentd、Logstash都原生支持 JSON 解析接入成本低。第三JSON 在 Python 里有成熟的序列化库性能开销可控。相比之下用keyvalue这种格式虽然更紧凑但遇到嵌套结构就不好处理了AI 应用里嵌套的 metadata 很常见所以 JSON 更合适。2.2 request_id 为什么是贯穿链路的“主键”一次 AI 请求往往不是单一操作而是由多个步骤组成的链路接收用户输入 → 构造 prompt → 调用模型 API → 处理流式响应 → 调用工具函数 → 生成最终回复。如果每个步骤都独立打日志出问题时你看到的就是一堆互不关联的记录根本不知道哪几条属于同一次请求。request_id的作用就是给这一次请求分配一个唯一标识在链路的每个环节都带上它。这样无论日志散落在多少个文件、多少个服务里只要按request_id过滤就能还原出完整的执行轨迹。这跟数据库里的主键是一个道理——它是串联所有相关记录的那根线。生成request_id我推荐用 UUID4简单可靠碰撞概率可以忽略。如果你想要更短的 ID可以用时间戳加随机数的组合但要注意在高并发下保证唯一性。我实测下来UUID4 的前 8 位在单机场景下已经足够区分但跨服务追踪时建议保留完整 UUID避免极端情况下的碰撞。2.3 方案选型的几个关键取舍在动手之前有几个选型问题需要先想清楚这些取舍会直接影响后续的实现方式。第一个取舍用标准库 logging 还是第三方库。Python 标准库的logging模块功能足够配合json序列化就能输出结构化日志零依赖。第三方库比如structlog提供了更优雅的 API 和更丰富的处理器但引入了额外依赖。我的建议是如果你的项目已经用了structlog或loguru继续用如果是新项目且团队对依赖敏感标准库完全够用。这个项目我用标准库实现因为它的可移植性最好你复制过去就能跑。第二个取舍同步写日志还是异步写。AI 应用的请求延迟本来就高模型推理动辄几秒日志写入的几毫秒开销相对可以忽略。但如果你的 QPS 很高同步写日志可能成为瓶颈。这时候可以用队列加后台线程的方式异步写但会增加复杂度。我的经验是日吞吐量在十万条以下同步写完全没问题超过这个量级再考虑异步。第三个取舍日志输出到文件还是标准输出。容器化部署的场景下推荐输出到标准输出stdout由容器运行时负责收集。传统虚拟机部署则输出到文件配合 Filebeat 采集。这个项目两种都支持通过配置切换。3. 核心细节解析与实操要点3.1 结构化日志的字段设计规范字段设计是结构化日志的地基设计得好后续查询分析事半功倍设计得乱日志就是一堆垃圾数据。我总结了一套 AI 应用日志的字段规范分成三类。第一类是通用字段每条日志都必须有字段名类型说明timestampstringISO 8601 格式带毫秒和时区levelstringDEBUG/INFO/WARNING/ERROR/CRITICALrequest_idstring请求唯一标识贯穿链路eventstring事件类型用下划线命名servicestring服务名微服务架构下必填loggerstring日志记录器名称通常是模块名第二类是业务字段根据事件类型动态添加。比如 LLM 调用事件带上model、prompt_tokens、completion_tokens、latency_ms工具调用事件带上tool_name、tool_input、tool_output。第三类是上下文字段比如user_id、session_id、trace_id。这些字段不一定每条日志都有但在需要关联用户行为时非常关键。注意字段名统一用蛇形命名snake_case不要混用驼峰。字段值尽量用基本类型避免嵌套过深。如果确实需要嵌套控制在两层以内否则查询时会很痛苦。我踩过的一个坑是早期把整个 prompt 原文塞进日志字段结果日志文件暴涨一天几个 G。后来改成只记录 prompt 的哈希值和长度需要看原文时再去专门的存储里查。日志里不要放超大文本这是血泪教训。3.2 request_id 的生成与传递机制request_id的生成时机很关键。它必须在请求进入应用的第一时间生成早于任何业务逻辑。在 Web 框架里通常用中间件middleware来实现。以 FastAPI 为例一个请求进来中间件先执行生成request_id然后把它注入到请求上下文里后续所有日志都从这个上下文取。传递机制有两种常见方案。方案一是显式传递把request_id作为参数在函数间传递。这种方式直观但侵入性强每个函数都要加参数容易漏。方案二是上下文变量用 Python 的contextvars模块把request_id存到一个全局可访问的上下文变量里日志记录器自动读取。这种方式对业务代码零侵入我强烈推荐。contextvars的原理是给每个执行上下文比如每个请求的协程维护一份独立的变量副本互不干扰。这正好契合异步框架的并发模型。你只需要在中间件里set一次后续在任何地方get都能拿到当前请求的request_id。import contextvars request_id_var contextvars.ContextVar(request_id, defaultNone) def get_request_id(): return request_id_var.get()这段代码定义了一个上下文变量默认值是None。中间件里调用request_id_var.set(new_id)设置日志过滤器里调用get_request_id()读取。就这么简单。3.3 日志记录器的封装与过滤器标准库的logging模块要输出结构化日志需要做两件事自定义 Formatter 和自定义 Filter。Formatter 负责把日志记录对象序列化成 JSON。它从record对象里提取字段组装成字典再json.dumps。这里有个细节record对象里有很多内置属性如name、levelname、pathname你需要区分哪些是内置的、哪些是你通过extra参数传进来的业务字段。我的做法是维护一个内置属性白名单白名单之外的都当作业务字段处理。Filter 负责注入request_id。它在每条日志被处理时执行从上下文变量里读取request_id塞进record对象。这样 Formatter 序列化时就能拿到它。import logging import json class JsonFormatter(logging.Formatter): def format(self, record): log_data { timestamp: self.formatTime(record), level: record.levelname, logger: record.name, message: record.getMessage(), } # 注入 request_id if hasattr(record, request_id): log_data[request_id] record.request_id # 注入业务字段 for key, value in record.__dict__.items(): if key not in RESERVED_ATTRS: log_data[key] value return json.dumps(log_data, ensure_asciiFalse)RESERVED_ATTRS是内置属性集合需要提前定义好。这个封装一次写好全项目复用。3.4 日志级别与采样策略AI 应用的日志量很容易失控尤其是 DEBUG 级别。我的建议是生产环境默认 INFO 级别DEBUG 只在排查问题时临时开启。INFO 级别记录关键节点比如请求开始、模型调用完成、工具调用、请求结束。DEBUG 级别记录详细参数比如完整的 prompt、模型原始响应。对于高频事件比如流式输出的每个 token不要逐条打日志而是采样或聚合。比如每 100 个 token 打一条进度日志或者只在流式结束时打一条汇总日志。这样既保留了可观测性又不会把日志系统压垮。实操心得给日志加上sampling_rate字段记录这条日志的采样率。这样在做统计时可以用采样率反推真实数量。比如采样率 0.1统计到 1000 条实际约 10000 条。4. 实操过程与核心环节实现4.1 环境准备与依赖安装这个项目用 Python 实现依赖很少。核心只需要标准库如果要跑 Web 服务示例需要 FastAPI 和 Uvicorn。pip install fastapi uvicorn日志采集部分如果部署在服务器上推荐用 Filebeat 采集日志文件。Filebeat 的安装这里不展开重点讲配置。它的核心配置是filebeat.inputs指定日志路径output.elasticsearch或output.logstash指定输出目标。Filebeat 会自动解析 JSON 格式的日志行把字段提取出来。4.2 日志模块的完整实现我把日志模块拆成三个文件context.py管理上下文变量formatter.py定义格式化器logger.py提供初始化函数。这样职责清晰便于维护。context.py里定义request_id_var和读写函数。formatter.py里定义JsonFormatter和RequestIdFilter。logger.py里提供setup_logging()函数配置根日志记录器添加处理器和过滤器。# logger.py import logging import sys from .formatter import JsonFormatter, RequestIdFilter def setup_logging(levellogging.INFO, outputstdout): logger logging.getLogger() logger.setLevel(level) logger.handlers.clear() if output stdout: handler logging.StreamHandler(sys.stdout) else: handler logging.FileHandler(app.log) handler.setFormatter(JsonFormatter()) handler.addFilter(RequestIdFilter()) logger.addHandler(handler) return logger调用setup_logging()后全项目的日志都会走这套配置。业务代码里只需要logger logging.getLogger(__name__)然后正常打日志即可。4.3 在 FastAPI 中集成 request_id 中间件中间件是注入request_id的最佳位置。它在请求进入时生成 ID在请求结束时清理上下文保证不同请求之间不串号。from fastapi import FastAPI, Request import uuid from .context import request_id_var app FastAPI() app.middleware(http) async def add_request_id(request: Request, call_next): req_id request.headers.get(X-Request-ID) or str(uuid.uuid4()) request_id_var.set(req_id) response await call_next(request) response.headers[X-Request-ID] req_id return response这里有个细节优先从请求头X-Request-ID读取如果客户端传了就用客户端的没传就自己生成。这样做的好处是支持跨服务追踪——上游服务生成的request_id可以透传到下游。响应头里也带上request_id方便前端排查问题时提供给后端。4.4 在 AI 调用链路中打日志有了基础设施业务代码里打日志就很自然了。我在 LLM 调用的关键节点都埋了日志。请求开始时打一条llm_request_start带上模型名和 prompt 长度。模型返回后打一条llm_request_end带上 token 消耗和延迟。如果调用失败打llm_request_error带上错误码和错误信息。import logging import time logger logging.getLogger(__name__) async def call_llm(prompt, model): start time.time() logger.info(llm_request_start, extra{ event: llm_request_start, model: model, prompt_length: len(prompt), }) try: response await llm_client.chat(prompt, model) latency int((time.time() - start) * 1000) logger.info(llm_request_end, extra{ event: llm_request_end, model: model, prompt_tokens: response.usage.prompt_tokens, completion_tokens: response.usage.completion_tokens, latency_ms: latency, }) return response except Exception as e: logger.error(llm_request_error, extra{ event: llm_request_error, model: model, error_type: type(e).__name__, error_message: str(e), }) raise注意extra参数里的event字段它和日志消息分开消息是给人看的简短描述event是给机器用的分类标签。这样查询时可以用event:llm_request_end精确过滤。4.5 日志采集与查询配置日志写到文件或标准输出后需要采集到日志平台才能发挥价值。我用 Filebeat 采集输出到 Elasticsearch用 Kibana 查询。Filebeat 的配置关键点有两个一是开启 JSON 解析二是设置request_id为 keyword 类型方便聚合。filebeat.inputs: - type: filestream paths: - /var/log/ai-app/*.log parsers: - ndjson: target: add_error_key: truendjson解析器会把每行 JSON 展开成字段。target: 表示字段平铺到根层级不嵌套。这样request_id就是顶层字段查询时直接request_id: xxx即可。在 Kibana 里我常用的查询有这么几个。查某次请求的完整链路request_id: req-a1b2c3d4按时间排序。查所有错误level: ERROR。查慢请求event: llm_request_end and latency_ms 5000。查某个模型的调用量按model字段做聚合。5. 常见问题与排查技巧实录5.1 日志里 request_id 是 null 怎么办这是最常见的问题原因通常是日志在中间件设置request_id之前就打了。比如应用启动时的初始化日志或者中间件之外的代码。解决办法是给request_id一个默认值比如system表示非请求触发的日志。还有一种情况是异步任务里丢了上下文。contextvars在asyncio.create_task创建的新任务里默认是空的需要手动复制上下文。用contextvars.copy_context()复制当前上下文再传给新任务。import asyncio import contextvars ctx contextvars.copy_context() asyncio.create_task(some_task(), contextctx)这个坑我在做异步工具调用时踩过排查了半天才发现是上下文没传过去。5.2 日志量太大导致磁盘打满AI 应用的日志量确实容易失控尤其是把完整 prompt 和响应都记进去的时候。我的应对策略分三层。第一层是控制字段大文本只记哈希和长度不记原文。第二层是控制级别生产环境用 INFODEBUG 按需开启。第三层是日志轮转用RotatingFileHandler或TimedRotatingFileHandler限制单文件大小和保留天数。from logging.handlers import RotatingFileHandler handler RotatingFileHandler( app.log, maxBytes100 * 1024 * 1024, # 100MB backupCount10, )这样最多占用 1GB 磁盘超过就自动清理最旧的文件。5.3 结构化日志影响性能怎么优化有人担心 JSON 序列化拖慢应用。实测下来单条日志的序列化开销在微秒级相比模型推理的秒级延迟可以忽略。但如果你的日志量真的很大可以从这几个方面优化。一是延迟序列化用logging的LogRecord先缓存真正输出时才序列化。二是异步写入用队列加后台线程。三是减少不必要的字段字段越少序列化越快。我做过一个压测单机每秒写 5 万条结构化日志CPU 占用增加约 15%。对于绝大多数 AI 应用来说这个开销完全可以接受。5.4 常见问题速查表问题现象可能原因排查方法解决方案request_id 为 null日志早于中间件执行检查日志时间戳与请求开始时间设置默认值或调整中间件顺序日志无法被采集格式不是标准 JSON用jq验证日志行检查 Formatter 输出字段查询不到字段类型是 text 而非 keyword查看索引映射修改索引模板设为 keyword异步任务日志丢失上下文未传递检查任务创建方式用 copy_context 复制上下文日志文件不轮转Handler 配置错误检查 Handler 类型改用 RotatingFileHandler中文乱码序列化未指定编码查看日志文件编码json.dumps(ensure_asciiFalse)5.5 几个我踩过的坑和独家技巧坑一extra 参数覆盖内置字段。如果你在extra里传了message或level这种内置字段名会直接报错。解决办法是业务字段加前缀比如biz_message或者维护一个保留字段列表做校验。坑二日志顺序错乱。多线程或多进程写同一个文件时日志可能交错。解决办法是用QueueHandler加QueueListener把日志写入串行化。技巧一给日志加 trace_id 和 span_id。如果你的应用有分布式追踪request_id可以作为trace_id每个步骤生成span_id这样能和追踪系统打通。技巧二用日志做实时告警。在日志平台配置规则比如“5 分钟内 ERROR 日志超过 10 条”就触发告警。这比等用户投诉再排查主动得多。技巧三定期分析日志找优化点。我每周会跑一次日志分析看哪些 prompt 的 token 消耗最高、哪些模型的延迟最大、哪些错误最频繁。这些数据直接指导了后续的优化方向。6. 从日志到可观测性的延伸结构化日志加 request_id 只是可观测性的第一步。当你把这套基础设施搭好之后会发现它能延伸出很多有价值的应用。比如成本分析。每条 LLM 调用日志都带 token 消耗按user_id聚合就能算出每个用户的成本按model聚合就能对比不同模型的性价比。我们团队就是靠这个数据把一部分简单任务从大模型切到了小模型成本降了六成。再比如质量监控。给模型响应打一个质量评分字段记录在日志里就能追踪模型输出的质量趋势。如果某个时间段评分下降可能是 prompt 模板出了问题或者模型版本更新导致的。还有用户行为分析。通过session_id把同一用户的多次请求串起来能看出用户的使用路径和偏好。这些数据对产品迭代很有参考价值。我个人在实际操作中的体会是日志这件事前期多花一小时设计字段后期能省十小时排查时间。很多团队觉得打日志是小事随便print一下就行等到线上出问题才发现日志根本不够用。结构化日志加 request_id 这套方案投入不大但回报是长期的。你不需要一次性做到完美先把request_id贯穿起来再把关键节点的日志结构化逐步迭代就行。最后分享一个小技巧在开发环境把日志输出成带颜色的可读格式生产环境输出 JSON。这样开发时看着舒服生产时机器好解析。用同一个 Formatter 接口根据环境变量切换实现即可代码改动很小。