尧图网页设计 
刚开始接手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一次用户请求的唯一ID核心关联键基础标识session_id会话ID用于多轮对话上下文基础标识user_id用户标识便于按用户维度分析生命周期timestampISO 8601格式时间戳精确到毫秒生命周期duration_ms当前阶段的耗时用于性能统计操作实体model使用的模型名称如gpt-4o-mini操作实体tool_name调用的工具/函数名如get_weather关键参数params请求参数如temperature、max_tokens业务信息prompt_tokens输入token数用于成本核算业务信息completion_tokens输出token数业务信息cache_hit是否命中缓存排障时很关键环境属性environment如dev/staging/prod多环境日志混存时特别有用环境属性trace_id和OpenTelemetry联动时使用这套字段在初期不需要一次全部铺满但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_asciiFalse) def setup_logging(levellogging.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 freq_{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放进这个请求头下游服务的中 AI应用日志改造：结构化日志与request_id全链路追踪实践
刚开始接手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一次用户请求的唯一ID核心关联键基础标识session_id会话ID用于多轮对话上下文基础标识user_id用户标识便于按用户维度分析生命周期timestampISO 8601格式时间戳精确到毫秒生命周期duration_ms当前阶段的耗时用于性能统计操作实体model使用的模型名称如gpt-4o-mini操作实体tool_name调用的工具/函数名如get_weather关键参数params请求参数如temperature、max_tokens业务信息prompt_tokens输入token数用于成本核算业务信息completion_tokens输出token数业务信息cache_hit是否命中缓存排障时很关键环境属性environment如dev/staging/prod多环境日志混存时特别有用环境属性trace_id和OpenTelemetry联动时使用这套字段在初期不需要一次全部铺满但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_asciiFalse) def setup_logging(levellogging.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 freq_{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放进这个请求头下游服务的中