上个月我们线上一个问答助手突然开始答非所问用户截图一张接一张甩过来。我把服务器翻了个底朝天看到的只有一堆print混在 stdout 里的日志有正常请求、有某个内部函数调试留下的废话、还有一条孤零零的异常堆栈。那一刻我真的很想砸键盘——AI应用日志要是还在用print打天下出了问题就是拿着手电筒在煤堆里找针。后来我们把日志体系整个换掉核心就两件事结构化日志 request_id贯穿。改完之后再排查线上问题基本从猜变成了看一条请求从进来到返回每个环节花了多久、调了哪个模型、用了多少token、工具传了什么参数全部在日志里串成一条线。这篇文章就把我这次改造的完整思路和踩坑过程写出来不光是代码还包括为什么这么做、上线后遇到了哪些坑。1. 先把旧账算清楚AI应用里的print日志为什么寸步难行1.1 一条AI请求背后藏着的是一条长链路传统Web接口的调用链相对短请求进来、查个库、算个结果、返回大部分时候一条堆栈就能定位问题。AI应用完全不是这么回事尤其到了LLM应用、RAG、Agent这个层面一次用户提问背后往往是一整条流水线入口网关鉴权、限流、会话恢复编排层决定是直接回复还是走工具、走检索上下文组装把多轮对话历史拼进Prompt检索链路Query改写、Embedding、向量库召回、重排LLM调用可能有重试、可能有流式输出、可能要触发多次模型调用工具执行Agent场景里模型会决定调用某个函数返回结果还要回填再交给模型二次总结后处理脱敏、安全过滤、缓存写入返回并持久化会话。任何一个环节出问题用户看到的表象可能都是回答变傻了。我举一个真实发生过的例子用户先问帮我查一下杭州明天的天气然后紧接着问那适合穿什么衣服第二轮回答完全没有参考第一轮的信息。表面上看是上下文丢失但print日志根本看不出是会话存储过期了、是检索没召回、还是Prompt模板把历史截断了——因为print只是告诉你这行代码跑过了它不携带业务语义。传统单体Web开发里print虽然难看但勉强能靠看执行顺序来猜问题。AI应用这个链路长度下靠猜是完全不现实的。你需要的是一个能回答这条请求到底经历了什么的日志体系而不是某行代码执行了的记录。1.2 print做不到的三件事我和不少团队聊过大家一开始都觉得先print顶着等上线了再改。但真上线就傻眼因为print在AI应用场景下至少有三件事根本做不到第一没有级别。调试用的细节输出、普通流程日志、真正的错误全部混在一起。线上环境不可能把debug信息全打出来否则日志量直接爆炸但你也没有办法只挑错误看——因为print没有warning、error之分。第二没有结构化字段。print输出的是一行自然语言人眼能看懂的部分机器看不懂。你没法在ELK、Loki或者ClickHouse里对日志做字段检索没法写查询所有request_id为xxx的日志这种操作也没法基于token数超过阈值做监控告警。日志进了日志平台只能全文检索效率极低且误差很大。第三没法把同一请求关联起来。这是最致命的一点。十几个用户同时在线每个请求都经过检索、模型调用、工具调用print出来几十行你怎么知道哪几行属于同一个用户你只能靠时间戳硬猜。一旦并发稍高、日志交叠基本上等于没有日志。我可以用一个表格对比一下能力print日志现状结构化日志 request_id日志级别无全混在一起debug/info/warning/error分层字段检索仅全文搜索按request_id、model、耗时等字段精确检索请求关联靠时间戳猜同一条request_id串起全链路监控告警无法实现可按字段聚合、告警数据消费人眼阅读可直接接入日志平台和监控系统1.3 什么时候print仍然值得用我得说清楚print不是罪只有print才是问题。在下面这些场景我仍然会用它本地一次性脚本、临时排查某一小段逻辑、以及刚开始调试Prompt时想快速看一眼模型的原始输入输出。这种场景开个交互式环境print一下比折腾日志框架快得多。我的建议是给它划一个明确边界本地开发随便用但一旦代码要进CI、要上测试环境print必须被拦下来。我们团队现在会在CI里加一个简单的检查禁止print(出现在核心代码里强制走logger。强制手段比自觉管用得多。2. 结构化日志落地JSON格式和关键字段设计2.1 为什么是JSON而不是漂亮的对齐文本很多人改结构化日志第一反应是用loguru把日志输出得好看点比如对齐时间、加颜色、用|分割字段。这一步有用但远远不够。真正值钱的不是人类可读性而是机器可消费性。试想两行日志文本格式2025-01-01 10:00:00 INFO request 123 done cost 500ms model gpt-4oJSON格式{ts:2025-01-01T10:00:00Z,level:INFO,request_id:123,msg:request done,cost_ms:500,model:gpt-4o}文本格式人眼看着挺好但你想写一条告警所有耗时超过2秒的请求监控系统根本拿不到cost这个字段。JSON格式就不一样了打入ELK或者Loki之后字段直接可以用于检索、聚合、画图。Loki虽然主打文本标签但JSON里用json解析器提取字段也是常规操作ELK那边更不用说了自带JSON解析。所以我的建议是结构化日志不是看起来整齐而是每个业务关键信息都是独立字段让日志平台能直接消费。2.2 字段设计原则哪些字段是必须的字段不是越多越好每多一个字段日志写入和数据平台的成本都会增加。但下面这张表里的字段我建议在AI应用场景下至少都具备字段示例作用ts2025-01-01T10:00:00.123Z精确到毫秒的时间levelINFO / WARNING / ERROR日志级别loggerretriever / llm / gateway来源模块request_ida1b2c3d4一次用户请求的唯一ID贯穿全链路session_ids-8f9a0s多轮会话ID跨请求关联msgllm call finished人类可读的说明errorConnectionError异常类型错误时才有stack堆栈文本错误堆栈只在ERROR级别输出除了这些通用字段AI应用还需要一些专属字段比如模型名、token用量、耗时等这个我在第4章专门展开。这里有个很关键的点业务字段不要塞进msg里。我看到很多团队虽然用了JSON但把所有变化的信息都拼成一个字符串塞进msg比如msg: modelgpt-4o tokens100 cost500ms这等于又退回文本日志了。loguru里应该利用**kwargs成为json的独立字段而不是拼字符串。2.3 代码改造loguru加自定义JSON sink如果你用的是Python最省事的方案是loguru。它不需要像标准logging那样配置一堆Handler和Formatter一个add()搞定。我们的做法是写一个自定义sink把日志转成JSONimport json import sys from datetime import datetime, timezone from loguru import logger def json_sink(message): record message.record entry { ts: datetime.now(timezone.utc).isoformat(), level: record[level].name, logger: record[name], msg: record[message], request_id: record[extra].get(request_id, -), session_id: record[extra].get(session_id, -), } # 把调用时传入的额外字段展开 extra record[extra] for k, v in extra.items(): if k not in (request_id, session_id): entry[k] v if record[exception]: entry[error] str(record[exception].value) entry[stack] .join(record[exception].format_exception()) print(json.dumps(entry, ensure_asciiFalse), filesys.stdout, flushTrue) logger.remove() logger.add(json_sink, levelINFO)使用的时候特别注意自定义sink里我们用print来输出这个print是刻意保留的——我们要把JSON行写到stdout给容器日志收集器捞走而不是写进loguru自己的处理器。日志输出方向的统一我在第5章会再讲。用的时候业务代码里的调用方式是这样logger.info(llm call finished, modelgpt-4o, cost_ms500, total_tokens128)loguru会把model、cost_ms、total_tokens放进record[extra]我们的json_sink就能把它们变成JSON字段。这样日志从产生的那一刻就是结构化的不是靠后面解析。提示如果你团队用的是标准logging不想引入loguru可以用python-json-logger这个库自定义JsonFormatter效果类似。loguru的优势是上手快、额外字段传参方便缺点是高并发下默认sink有序列化和锁开销压测时要关注。我们在QPS过千之后切回了标准logging JsonFormatter那是因为要精确控制性能和队列中小项目前期用loguru完全足够。3. request_id贯穿中间件播种链路里生长3.1 入口中间件请求来了先埋IDrequest_id必须从请求一进门就种下去。我们的项目是FastAPI在中间件里做这件事import uuid from contextvars import ContextVar from fastapi import Request request_id_var: ContextVar[str] ContextVar(request_id, default-) app.middleware(http) async def request_id_middleware(request: Request, call_next): # 网关如果已经生成了request_id就透传否则自己生成 rid request.headers.get(x-request-id) or uuid.uuid4().hex[:12] token request_id_var.set(rid) try: response await call_next(request) response.headers[x-request-id] rid return response finally: request_id_var.reset(token)三个细节值得说优先透传网关的ID。如果你们Nginx或者API网关已经统一生成了request_id这里就别再生成新的否则客户端看到的是两套ID下游排查时会很混乱。set/reset配对。中间件里用了token和reset确保这个协程结束后ContextVar能恢复原样不会污染连接池复用的下一个请求。如果你只set不reset在高并发复用场景下会出现request_id串号这坑我踩过。响应头里回写。把request_id返回给客户端用户反馈问题时只要把ID发过来开发人员直接拿ID去日志平台捞全链路效率高非常多。3.2 contextvars是关键异步环境下的传递机制为什么用ContextVar而不是全局变量、甚至不直接塞在request对象里到处传这要从Python的异步并发模型说起。ContextVar最核心的价值是每个协程都有自己独立的上下文副本A协程设置的值不会影响B协程两个并发的请求可以各自持有自己的request_id互不干扰。全局变量做不到这点塞在request对象里呢你还要把request一层层传进LLM调用、检索器这显然不现实。但ContextVar有一个非常经典的坑我详细说一下。Python的asyncio.create_task会拷贝当前上下文所以你在子协程任务里能拿到request_id。但是asyncio.to_thread或run_in_executor这类线程池调用不会自动携带协程的上下文子线程里request_id_var.get()只会拿到默认值。async def main(): request_id_var.set(req-123) # 能拿到 req-123 task asyncio.create_task(worker()) await task # 拿不到 req-123线程里打印出 - await asyncio.to_thread(worker2) def worker2(): print(request_id_var.get()) # - def worker_manual(rid: str): request_id_var.set(rid) print(request_id_var.get()) # req-123 # 推荐显式传参并手动set await asyncio.to_thread(worker_manual, request_id_var.get())解决办法就是上面的worker_manual写法在线程函数内部手动set一次。别偷懒在线程池场景里想靠ContextVar自动传递是等不到的。3.3 下游服务传递让所有环节都喊同一个IDrequest_id不仅要在自己进程里贯穿还要跟着调用往下游走。两个场景场景一调用LLM服务。如果你用的是OpenAI SDK可以直接在client里配置统一的headerfrom openai import AsyncOpenAI client AsyncOpenAI( api_key..., base_url..., default_headers{x-request-id: request_id_var.get()} )这样下游的模型网关、自建推理服务打日志时也能带上同一个request_id。模型层如果报错你能在自己日志里看到是哪个请求触发的、下游报了什么错。场景二自研服务之间的HTTP调用。建议封装一个公共的函数统一注入import httpx def build_headers(extra: dict | None None) - dict: headers { x-request-id: request_id_var.get(), Content-Type: application/json, } if extra: headers.update(extra) return headers async def call_retriever(query: str): async with httpx.AsyncClient(timeout10) as client: resp await client.post( http://retriever/query, headersbuild_headers(), json{query: query} ) return resp.json()记住一个原则内部服务之间只要有一个环节没传request_id这条链路就断在那一环。比传了但格式不对更可怕的是默认不传只在遇到问题时临时加那排查成本依然很高。服务间调用的header从第一天就统一约定好。4. AI应用专属日志token、耗时与工具调用记录4.1 必须记录的AI专属字段通用字段解决的是这条日志属于谁AI应用还需要一组专属字段来解决这次模型调用发生了什么。model定位是不是某个模型的问题。我们线上混着多个模型有些场景某模型会偶发输出异常没有这个字段你连哪条日志涉及哪个模型都分不清。prompt_tokens/completion_tokens/total_tokens这组数据不止是算账用的排查上下文是不是被塞满了为什么突然变慢时非常有价值。一个会话如果total_tokens持续接近模型上下文上限不用等用户投诉你就能预判要出问题了。latency_ms模型调用耗时。如果你是流式输出还要区分首字延迟和总耗时两者代表的问题完全不一样。temperature/top_p等参数复现问题的时候这些参数决定了模型的随机性。你不记录下来用户说刚才同样的输入给我不同的答案你根本没法判断是正常的随机还是Bug。tool_callsAgent场景下模型决定调用哪个工具、传了什么参数。工具调用的参数错误是AI胡言乱语的一大来源必须记录。retry_count第几次调用才成功。重试说明下游有抖动这个字段是稳定性监控的重要输入。4.2 用装饰器统一包装LLM调用我们给所有LLM调用包了一层统一的装饰器让日志记录逻辑集中在一个地方而不是在每个调用点手写import functools import time def log_llm_call(func): functools.wraps(func) async def wrapper(*args, **kwargs): start time.perf_counter() try: resp await func(*args, **kwargs) usage getattr(resp, usage, None) logger.info( llm_call_finished, modelkwargs.get(model) or getattr(resp, model, -), prompt_tokensusage.prompt_tokens if usage else 0, completion_tokensusage.completion_tokens if usage else 0, total_tokensusage.total_tokens if usage else 0, latency_msint((time.perf_counter() - start) * 1000), retry_countgetattr(func, retry_count, 0), ) return resp except Exception as e: logger.warning( llm_call_failed, modelkwargs.get(model, -), latency_msint((time.perf_counter() - start) * 1000), errorstr(e), retry_countgetattr(func, retry_count, 0), ) raise return wrapper log_llm_call async def call_llm(model: str, messages: list[dict], **kwargs): ...这个装饰器有几个隐藏的心机异常也记日志而且是在raise之前否则错误详情会丢失耗时不管成败都记因为失败的慢请求和正常的慢请求同样重要usage字段用getattr兜底因为不同模型提供商的返回结构不一样有的没有usage字段。4.3 检索与工具调用的证据链RAG应用里最让人头疼的问题是模型没有基于检索到的内容回答。这时候日志里必须有证据链Query是什么、召回了哪些chunk、每个chunk的相似度分数是多少、最终拼进Prompt的是哪几段。没有这些你只能猜是检索问题还是模型问题。我们的检索日志大致长这样logger.info( retriever_hit, queryquery, top_ktop_k, hit_chunk_ids[c[chunk_id] for c in results], hit_scores[round(c[score], 4) for c in results], )注意这里记录的是chunk_id而不是完整内容原因有二一是chunk内容可能很长体积太大二是这些内容往往来自用户上传的文档可能有敏感信息。日志里只保留ID内容在需要时再去向量库和原始文档里捞。工具调用同理记录的是函数名、入参、出参摘要、耗时logger.info( tool_call, tool_nameget_weather, tool_input{city: 杭州, date: 2025-01-01}, tool_output_summary晴, 5~12℃, latency_ms230, )出参必须用摘要而不是完整输出。一次工具调用可能返回几十KB数据全部进日志会让日志平台内存崩溃而且很多工具返回值本身就是用户的业务数据完整记录有泄露风险。摘要控制在几十字内够人判断这次调用是不是合理即可。5. 上线之后被日志坑过的几个瞬间排错实录5.1 contextvars丢失流式输出场景的request_id断链我们上线后第一次排查故障就遇到了request_id断链。现象是同一条请求前面的日志都有request_id到了流式输出的日志request_id突然变成了-。排查链路是这样的先怀疑流式生成器是新的协程任务于是检查asyncio.create_task的上下文拷贝——发现生成器里能拿到。再怀疑是StreamingResponse在FastAPI内部跑在不同的执行上下文中用代码打点确认确实是生成器内部的request_id丢了。这个问题卡了我们很久最终方案是在流式生成器内部用外层传入的request_id重新set一次。from fastapi.responses import StreamingResponse async def generate_stream(rid: str): token request_id_var.set(rid) try: # 这里所有logger.info都会正确带上request_id async for chunk in call_llm_stream(...): yield chunk finally: request_id_var.reset(token) app.post(/chat) async def chat(request: Request): rid request_id_var.get() return StreamingResponse(generate_stream(rid))这个教训的核心是ContextVar不是魔法凡是代码执行上下文发生变化的地方线程、回调、某些框架自建的任务都要亲自动手把ID种回去。我建议把这个当成代码审查的一个必查项凡是有新起协程或线程的地方都问一句request_id有没有丢。5.2 PII与敏感信息大模型输入输出日志可能踩红线这是所有AI应用日志里最容易被忽视、但后果最严重的问题。你把用户的完整Prompt和模型的完整输出打进日志表面上只是方便排查实际上等于把你的对话数据库明文放进了日志平台。客服机器人场景下模型输出里可能带着用户手机号、身份证号一旦日志平台权限管控不严就是安全事故。我们的脱敏方案是这样做的默认只记录Prompt的首尾各若干字符中间用...截断模型输出同样截断只在debug级别且临时开启时才记录完整内容敏感信息在进入日志前过一道脱敏函数把手机号、邮箱、身份证的正则匹配替换成***日志平台按角色控制字段可见权限核心日志只有线上oncall人员能看。代价是排查某些问题时会多一步开debug日志、复现一次的操作但换来的是合规底线。我不止一次遇到团队嫌麻烦不想做脱敏结果被安全审计点名。5.3 日志量与成本一条请求两百行日志的教训有一次我们给一个Agent应用做压测发现一天下来日志量涨了快一百倍日志平台账单直接惊掉下巴。分析后发现原因很蠢流式输出每个chunk都打了一行日志加上工具调用的完整入参出参一条请求轻松产生两百多行日志。后来我们定了一套分层策略info级别只记关键事件请求开始、结束、LLM调用摘要、工具调用摘要、检索命中摘要debug级别才记详情完整Prompt、完整输出、检索chunk全文线上默认开infodebug按需对个别request_id采样。采样还有一种做法按request_id尾号采样比如固定采样5%的请求打debug日志。这样既不影响大多数请求的性能又保留了部分全量数据用于统计。还有一个成本优化的点日志级别和日志量要设监控告警。一旦单条请求平均日志行数超过阈值立刻报警这能及时拦住某天代码合入后日志暴增这类问题。别看这个很基础很多事故都是等月底账单出来才发现的。最后说几句大实话这套改造做完之后我的切身体会是结构化日志和request_id真正解决的不是日志好不好看的问题而是AI应用可不可排查的问题。AI应用的故障形态跟传统Web完全不同它更多是行为异常而不是进程崩溃。回答走偏、上下文丢失、工具调用错误这些都是print永远无法定位的。只有把日志变成可检索、可关联、带证据链的数据你才有底气说这个AI应用是可维护的。如果你也准备动手改我的建议是别一次性铺开重写先拿一条核心链路做试点入口中间件种request_id、关键环节打JSON结构化日志、LLM调用包一个装饰器记录token和耗时。跑通了再铺到全业务。完了之后做一个实验随便挑一个线上请求的request_id看看你能否在一分钟内从日志里还原出这条请求从头到尾的全部经历。能说明合格不能继续补。