尧图网站建设 
上周排查一个AI Agent告警时我盯着终端里滚动的print日志完全无从下手同一个用户问题触发了四次模型调用、两次工具调用、一次向量检索但所有输出都混成一团分不清先后、找不出关联更别提复现那条真正导致异常的历史链路。那次之后我在内部把“AI应用日志改造”单独立了一个项目代号O01核心就两件事把print换成结构化日志用request_id把一次请求牵扯到的所有调用串起来。这篇内容就是O01项目的完整回顾包括为什么必须改造、怎么设计字段、如何在FastAPI里落地request_id中间件以及后续做日志采集和时间线还原的实战经验适合正在开发AI Agent、RAG应用或者还在用print调大模型的团队参考。1. 先从print说起为什么AI应用的日志治理是刚需1.1 print日志在单机调试时还能凑合一旦进入Agent场景就立刻失控很多人习惯在代码里随手写print(调用LLM完成结果, resp)本地跑一个脚本、调一次接口这种方式勉强能看。可AI应用和传统Web接口有个本质区别一次用户请求会触发一段动态编排链。模型可能要经过多轮ReAct循环每轮里都要调用LLM、判断是否调用工具、再把工具结果拼回去交给模型继续推理。这个过程中涉及的网络调用、token计算、prompt构造、返回解析每一步都在消耗时间和成本。当这段链路里满是print时问题就暴露了多条请求并发时print输出会交错在一起不同模块之间的日志没有统一级别调试信息、错误信息混在stdout里更麻烦的是缺少关联标识你无法回答“这条日志属于哪个用户、哪次会话、哪一轮推理”。我见过一个团队排查线上Agent反馈问题时把几十个进程的print日志全拉到本地靠肉眼和正则搜关键词最后发现搜到的内容属于三个不同用户的两百多条记录这种情况不改造日志可观测性基本为零。1.2 大模型应用特有的五个日志痛点做AI应用之后我对日志的需求和传统后端不太一样。传统后端最关心接口响应时间、数据库慢查询、异常栈AI应用还要多关注模型调用链、提示词内容、token开销和工具执行结果。具体痛点可以整理成下面这张表痛点表现影响调用链长一次请求可能包含多次LLM调用和工具调用没有路由标识就很难串联时间线异步与并发Agent经常使用异步任务、后台线程、流式传输print输出乱序无法对应上下文载荷体积大一份prompt动辄几千token响应也很长刷屏严重日志文件快速膨胀敏感信息多prompt中可能包含用户个人信息、内部知识片段直接落盘会带来数据合规风险成本需要计量每个请求的token使用量和调用次数是核心指标非结构化文本无法做后续聚合统计这些痛点叠加在一起意味着AI应用的日志必须是一行一个JSON对象并且带有关联ID。可以说AI应用是结构化日志落地最典型的场景之一这也是我为什么强烈建议团队哪怕不做链路追踪系统也要先把结构化日志和request_id这两件事做扎实。只要这两件事落地后续接ELK、接OpenTelemetry、做成本分析都会非常顺。2. 结构化日志是什么让日志从“给人看”变成“给系统看”2.1 先定义日志字段别让日志继续是纯文本所谓结构化日志简单说就是每条日志不再是一段自然语言而是一个有固定字段的对象通常以JSON Lines的形式输出也就是每行一个独立JSON对象。这样做有几个非常直接的好处日志分析系统可以直接解析字段做索引可以在Elasticsearch里按字段过滤比如levelerror、request_idxxx还可以在Kibana里按字段做聚合统计比如按llm_provider统计调用次数、按prompt_tokens求和。我在O01项目里定义的统一日志字段大纲如下timestampISO8601格式精确到毫秒所有服务统一用UTC或统一带时区偏移避免跨服务时对不上时间。levelDEBUG、INFO、WARNING、ERROR用于过滤和告警。logger产生日志的模块名比如agent.planner、agent.tool_executor、llm.client。message人类可读的事件描述比如“llm_call_start”“tool_execution_success”。request_id贯穿整条链路的请求标识这是全文最关键的一个字段。session_id、user_id如果应用有会话概念这两个字段可以帮助定位用户维度问题。meta可扩展对象存放provider、model、token用量、耗时等指标。实际落地时不用一步到位但timestamp、level、logger、message、request_id这五个字段建议从第一天就固定下来。后面的session_id、user_id、meta可以视业务灵活加上。字段名也要尽量保持全小写下划线风格避免后续在Elasticsearch里出现大小写混乱的映射冲突。2.2 Python里用logging json formatter落地我选的方案与原因Python生态里日志方案不少常见的有三种原生logging配合自写JSON Formatter、structlog、python-json-logger。我最终选了原生logging加自写Formatter原因很实际一是团队项目依赖越少越好AI应用本身已经有一堆包了没必要为日志再增加一个强依赖二是原生logging在框架集成上最稳FastAPI、Celery、Django都能无缝衔接三是自写Formatter只有几十行代码完全可控后续要加脱敏、加特殊字段也很方便。下面这段代码是我在O01项目里用的JSONFormatter核心实现import json import logging from datetime import datetime class JsonFormatter(logging.Formatter): RESERVED set( (name, msg, args, levelname, levelno, pathname, filename, module, exc_info, exc_text, stack_info, lineno, funcName, created, msecs, relativeCreated, thread, threadName, processName, process, taskName) ) def format(self, record: logging.LogRecord) - str: record.message record.getMessage() if record.exc_info: record.exc_text self.formatException(record.exc_info) data { timestamp: datetime.utcnow().isoformat() Z, level: record.levelname, logger: record.name, message: record.message, } for key, value in record.__dict__.items(): if key not in self.RESERVED and not key.startswith(_): data[key] value if record.exc_text: data[exc_text] record.exc_text return json.dumps(data, ensure_asciiFalse)这里有几个细节值得说明。手动指定timestamp而不是用record.created格式化是为了统一格式并避免有些日志库默认输出带小数秒的浮点时间戳后续在日志平台里排序不够友好。RESERVED集合用来过滤logging内部属性你可以把自定义属性通过extra参数传给logger.info(..., extra{request_id: 123})Formatter会把它们合并进JSON的顶层字段。ensure_asciiFalse是为了让中文日志在采集端和终端里都能直接读不会变成一串\uXXXX。配置方面我推荐用logging.config.dictConfig统一管理而不是在代码里到处logger.setLevel。应用启动时加载一份YAML或字典配置控制台Handler默认输出到stdout而不是文件这个在后面讲容器化集中采集时很重要。2.3 Node.js和Go同样能落地原则是一样的如果是Node.js侧的项目我建议直接考虑pino它的性能和JSON输出能力都很成熟如果项目已经用了bunyan也可以用但维护力度相对弱一些。Go项目的做法是使用slog标准库的JSONHandler或者zerolog。原则没有变化每条日志必须是一个结构化对象必须能稳定解析并预留一个字段存放链路ID。技术栈再怎么换字段设计和透传规则才是核心这部分跨语言是一致的。3. request_id贯穿一条链路一个ID把这串调用串起来3.1 为什么非要有request_id相当于给请求发一个“快递单号”你网购一件商品中间要经过仓库、干线运输、本地配送等多个环节快递公司靠一个快递单号把所有这些环节关联起来。它不需要在各个环节之间共享复杂的上下文只需要把同一个单号打在每一站。AI应用里的request_id就是这个快递单号。当一个用户在前端发送一条消息后端Agent开始规划、检索、调用模型、执行工具过程中可能还会向外部服务发起HTTP请求这些都像是货物的不同环节。只有让每个环节的记录都携带同一个request_id你才能从日志平台里一次性捞出这条请求的完整时间线。没有这个ID所有日志就是散落一地的线索排查问题只能靠猜。3.2 生成、透传与返回请求ID的生命周期管理request_id生成规则我一般建议使用UUID4的hex形式或者带节点信息的雪花ID。生产环境如果只有一个服务UUID够用如果请求会横跨多个服务采用类似{service}-{timestamp}-{random}的格式能更快看出源头比如api-20240115-8f3a1b。整个ID的流转有三个关键节点入口生成在FastAPI中间件里先从请求头X-Request-ID中读取如果没有就自动生成。之所以优先读请求头是为了支持调用方透传自己的追踪ID方便前端和后端对账。内部透传生成后要存到ContextVar里让后续所有调用都能通过日志Filter自动读取。对Agent来说每一次子任务、每一个工具调用都要看到这个ID。外部透传Agent调用外部模型API、工具服务时要把ID放到HTTP头里一起发出去这样外部服务的日志也能串上。响应返回时把request_id写进响应头前端报错时直接把ID贴给开发定位效率能提升一倍。下面这段代码是我在FastAPI里用的中间件import uuid from contextvars import ContextVar from starlette.middleware.base import BaseHTTPMiddleware request_id_var: ContextVar[str] ContextVar(request_id, default-) class RequestIDMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): request_id request.headers.get(X-Request-ID) if not request_id: request_id uuid.uuid4().hex token request_id_var.set(request_id) try: response await call_next(request) response.headers[X-Request-ID] request_id return response finally: request_id_var.reset(token)这段代码有几个容易踩坑的地方。ContextVar.set()之后必须reset()否则在异步复用线程时会导致ID串号。中间件必须放在所有可能产生日志的依赖之前加载否则前面已产生的日志带不上ID。token的作用是支持嵌套修改比如某些服务内部想把子请求ID替换成更细粒度的ID重置时可以恢复之前的ID。接下来要把request_id注入到每一条日志记录。方式是定义一个logging.Filterimport logging from .context import request_id_var class RequestIdFilter(logging.Filter): def filter(self, record: logging.LogRecord) - bool: record.request_id request_id_var.get() return True把这个Filter加到所有Handler之后每条日志Formatter就能从record里拿到request_id并写入JSON。如果没有提前设置request_id_var.get()默认返回-不会抛异常。3.3 异步任务和多线程request_id最容易丢的地方FastAPI的后台任务BackgroundTasks和Celery异步任务是request_id丢失的高发区。因为ContextVar只在线程上下文中生效新的线程或进程不会自动继承父线程的ContextVar。如果你在请求里创建一个后台任务去处理后续逻辑任务里打出的日志记录request_id很可能就是默认的-。解决办法也很直接进入任务函数后手动把request_id重新设置回去。比如FastAPI的BackgroundTasks可以这样处理from contextvars import copy_context def run_in_background(func, *args, **kwargs): ctx copy_context() def wrapper(): ctx.run(func, *args, **kwargs) return wrapper # 在接口里 background_tasks.add_task(run_in_background(process_result, request_id_var.get(), data))当然更朴素的方案是把request_id作为参数传给后台任务函数函数开头直接request_id_var.set(request_id)。这种方式直观、清晰性能上也不会有什么损失。如果你用的是Celery则需要把request_id放进task的kwargs在task执行入口统一set。否则你会发现后台任务的日志和你线上请求完全对不上排查能力会大打折扣。4. 一个FastAPI LLM调用链的完整改造示例4.1 项目结构先看清楚要改动哪些地方为了能直接照着改我模拟一个极简AI服务FastAPI接收用户消息调用一个LLM函数必要时调用一个外部工具函数最后返回结果。全部逻辑压缩到一个项目里目录结构是这样ai_app/ ├── main.py # FastAPI入口中间件和路由 ├── logging_conf.py # logging字典配置包含JSON Formatter和Filter ├── context.py # request_id的ContextVar定义 ├── agent.py # Agent编排逻辑LLM调用、工具调用 └── llm_client.py # 模拟LLM HTTP调用这种结构不复杂但它能覆盖绝大多数AI应用日志改造会遇到的核心问题日志配置要独立、中间件要统一管理ID、Agent编排逻辑要利用日志记录关键事件。4.2 代码实现从日志配置到中间件再到业务调用先看logging_conf.py它负责构建统一的JSON输出管线import logging.config LOGGING_CONFIG { version: 1, disable_existing_loggers: False, formatters: { json: { (): logging_conf.JsonFormatter, } }, filters: { request_id: { (): logging_conf.RequestIdFilter, } }, handlers: { console: { class: logging.StreamHandler, formatter: json, level: INFO, filters: [request_id], stream: ext://sys.stdout, } }, root: { level: INFO, handlers: [console], } } def setup_logging(): logging.config.dictConfig(LOGGING_CONFIG)这里disable_existing_loggers设置为False很重要。FastAPI和uvicorn内部有自己的logger如果设置成True那些logger会被禁用很多框架日志会消失对排错反而更不利。输出到stdout而不是文件是因为在Docker和Kubernetes环境里把日志写到磁盘再让采集器去读文件相比直接stdout多了一层无效IO而且容器重建后文件就丢了。然后是main.py把中间件和日志配置接起来from fastapi import FastAPI, Request from context import request_id_var from logging_conf import setup_logging setup_logging() app FastAPI() app.middleware(http) async def request_id_middleware(request: Request, call_next): request_id request.headers.get(X-Request-ID) or uuid.uuid4().hex token request_id_var.set(request_id) try: response await call_next(request) response.headers[X-Request-ID] request_id return response finally: request_id_var.reset(token) app.post(/chat) async def chat(payload: dict): prompt payload[prompt] result await run_agent(prompt) return {result: result, request_id: request_id_var.get()}agent.py里的关键调用我加了结构化日志方便后续观察完整链路import logging import time logger logging.getLogger(agent) async def run_agent(prompt: str): logger.info(agent_start, extra{event: agent_start, prompt_len: len(prompt)}) start time.time() llm_resp await call_llm(prompt) logger.info( llm_call_done, extra{ event: llm_call_done, provider: openai-compatible, model: gpt-4o-mini, tokens_total: llm_resp.get(usage, {}).get(total_tokens, 0), latency_ms: int((time.time() - start) * 1000), }, ) if llm_resp.get(needs_tool): logger.info(tool_call_start, extra{event: tool_call_start, tool: calculator}) tool_result await run_tool(llm_resp[tool_args]) logger.info(tool_call_done, extra{event: tool_call_done, tool: calculator, result_snippet: str(tool_result)[:200]}) final_resp await call_llm(f根据工具结果生成回答: {tool_result}) return final_resp[text] return llm_resp[text]这段日志设计里我特别想强调命名习惯。event字段对应一个语义事件比如llm_call_done、tool_call_start而不是随意写“调用完成”。编程上有一套叫event-driven logging的思路简单说就是给每个日志事件一个机器可读的名称后续做告警和统计都会方便很多。extra里放结构化指标字段比如tokens_total、latency_ms它们和message文本分离这样Kibana里可以做数值聚合而不是对着一句中文文本做全文检索。调用外部的llm_client.py时要注意把request_id透传到HTTP请求头里import httpx from context import request_id_var async def call_llm(prompt: str) - dict: request_id request_id_var.get() headers {Content-Type: application/json, X-Request-ID: request_id} # 这里用mock返回真实项目替换成你的模型网关地址 async with httpx.AsyncClient(timeout60) as client: resp await client.post(https://llm-gateway.example.com/v1/chat/completions, json{prompt: prompt}, headersheaders) resp.raise_for_status() return resp.json()这样外部服务只要也支持读取X-Request-ID两边日志就能通过同一个ID关联起来。很多模型网关会自己生成内部trace id但你仍然要在业务侧保留自己的request_id作为约定不能被网关的内部ID覆盖掉。4.3 三条真实踩坑记录日志改造过程中最容易翻车的地方结构化日志改造本身不难真正磨人的是那些看起来很小的坑。我列三个印象最深的第一个坑日志重复。setup_logging()如果被调用两次或者框架启动时又加载了一遍配置root logger下面会挂多个Handler同一条日志被打印多次。解决方法是每次配置前先logging.shutdown()或者判断logging.getLogger().handlers是否为空再配置。很多框架的启动器也会默认加载logging配置在FastAPI里我用lifespan事件只调用一次setup_logging()。第二个坑异常打印丢失上下文。有些同学用logger.error(f调用失败: {e})结果只有一句话没有堆栈。改用logger.exception(llm_call_failed, extra{...})它会自动带上当前异常的堆栈信息。注意logger.exception只能在except块内使用它内部会读取sys.exc_info()在except外调用只会得到一行NoneType: None。第三个坑脱敏。AI请求里的prompt可能包含用户手机号、身份证等敏感内容。我建议在JsonFormatter里加一个sanitize函数对message和prompt字段做正则替换比如电话号码中间四位用星号替代密钥长度只保留末尾四位。日志系统的价值必须建立在合规的前提下一旦线下数据泄漏被审计发现日志改造得再好也没用。5. 日志采集与延伸从零散日志到完整可观测性5.1 用Filebeat把结构化日志采集到ELK一旦应用以JSON Lines格式输出日志采集就变得非常轻松。Filebeat是最常见的轻量级采集器之一它读文件或stdout解析JSON然后发送到Elasticsearch或Logstash。需要注意现在Elastic官方推荐用filestream类型替代旧的log类型但很多存量系统还在用log类型我实际配置时会根据ES版本选择。一个简化版的filebeat.yml如下filebeat.inputs: - type: filestream id: ai-app-logs enabled: true paths: - /var/log/ai-app/*.json parsers: - ndjson: target: - multiline: pattern: ^\{ negate: true match: after output.elasticsearch: hosts: [http://elasticsearch:9200] index: ai-app-logs-%{yyyy.MM.dd}ndjson解析器让Filebeat把JSON字段直接展平到输出文档的根级别这样Elasticsearch里可以直接通过request_id.keyword做过滤。后面那个multiline配置很重要AI应用里一条日志可能因为包含换行而断成多段用“每行只有JSON对象开头才算新日志”的规则就能把堆栈信息完整合并成一条记录。不加这个配置多行异常栈会被拆得乱七八糟排查错误时缺行少尾非常痛苦。5.2 从request_id升级到OpenTelemetry Trace的衔接路线request_id解决了“一次请求的日志怎么串起来”的问题但当你需要查询跨服务、跨进程的完整调用关系或者需要看某个外部API的耗时分布时request_id就不够用了。这时候标准做法是引入OpenTelemetry的Trace。Trace里有一个全链路唯一的trace_id你可以把它当成更高级的request_id在日志里同时记录request_id和trace_id两者并不冲突。在AI Agent场景下Trace的粒度会更细plan阶段记录Agent规划出来的步骤列表llm调用记model、prompt、response、token用量工具调用记工具名、参数、耗时、返回值摘要检索调用记向量库名称、命中条数、相似度分数这些Span和日志中的event一一对应实际上就是在日志基础上多了一层时间线和父子关系。我的建议是先把request_id做好不要一上来就上全链路Trace。因为Trace改造涉及SDK、Exporter、后端系统而request_id配合结构化日志已经能解决80%的线上排障问题。做到一定程度后再让request_id向trace_id演进迁移成本会低很多。5.3 在编排型Agent中request_id如何组织子任务现在很多项目开始做多智能体协作一个“主控Agent”会拆任务给“子Agent”执行。这种情况下让所有子任务都使用同一个request_id是可以的但不一定能满足细粒度追踪。我习惯在request_id基础上增加一个task_id或span_id字段表示当前是这次请求下的第几个子任务。主Agent发起子任务时生成task_id并把它放进日志的extra中这样既能从request_id维度看整体也能从task_id维度看某个子任务的执行情况。如果每个子Agent还涉及各自的LLM调用和工具调用那就可以把agent_name、subtask_name都作为结构化字段记下来。这个思路本质上是在为Trace做铺垫request_id对应tracetask_id对应span。提前把命名和字段统一好后面切入OpenTelemetry时只需要把task_id替换成真正的span_id即可业务代码几乎不用动。6. 常见问题与排查技巧实录6.1 问题速查表日志改造和运行过程中我整理了下面这些高频问题基本覆盖了O01项目遇到的大部分坑。症状可能原因解决方案日志重复打印setup_logging被多次调用root logger挂了多个Handler入口处只调用一次或先清理已有Handler日志里request_id全是-Filter没加进Handler或ContextVar设置后被reset检查dictConfig里filters配置确认中间件作用域日志乱序异步任务输出与请求上下文混在一起给日志加毫秒级时间戳按时间排序异步任务重新set request_id多行异常日志被拆开没有配置multiline规则Filebeat里用ndjson multiline合并日志里出现token明文未做脱敏处理Formatter中增加正则脱敏函数日志文件无限增长Handler是FileHandler且没有轮转开发环境用RotatingFileHandler生产环境输出stdout交给采集器这里特别啰嗦一句不要把生产日志直接写到本地磁盘再定期删除。日志文件无限增长是ELK场景里最不想看到的事而容器场景下stdout加采集器是更稳妥的方案。如果实在要落盘用RotatingFileHandlermaxBytes10MBbackupCount5再配合外部采集。6.2 借助request_id还原一次完整对话链路实际排查AI回答质量问题时我很少去翻原始前端界面的聊天记录而是直接从日志平台里按request_id拉取整条链路。比如用户报“某个问题回答得很奇怪”前端把响应头里的X-Request-ID发过来我就在Kibana里输request_id: abc123搜索能看到用户原始prompt的长度和内容摘要Agent规划的步骤每一轮LLM调用的模型、token数量、耗时工具调用传了什么参数、返回了什么结果最后一次组装回答用了哪些上下文通过这个时间线基本能判断问题是出在检索阶段、模型阶段还是工具阶段。举例来说如果工具返回结果正常但最终回答质量差那大概率是prompt拼接逻辑有问题如果第一轮LLM调用就报超时那就要查模型网关的限流策略。这个排查思路在事件驱动日志体系下极其好用。在生产环境没有Kibana时Linux命令行也可以直接分析。假设日志文件是app.log每条是JSON行grep request_id:abc123 app.log | jq -r .timestamp .logger .eventjnq不一定单独装这里想要说明的是只要有结构化字段就能用管道工具快速排序、筛选、统计。用jq把timestamp和event提取出来拼成一行可读时间线比在几千行原始JSON里翻找高效得多。6.3 关于“日志作用域”的一些心得很多日志库支持logger层级命名比如agent.llm、agent.tool、http.client。我建议严格规划logger作用域不要所有模块都用同一个root logger。作用域不只体现在logger name还体现在字段作用域像request_id这种全局字段放顶层像具体模型参数这种局部信息放extra的meta对象里。日志是给未来的你和其他同事看的作用域划分得越清晰检索时越省力。结语一次日志改造的实际收益O01项目做完到现在已经跑了几个月我最直观的感受不是排障更快而是团队终于能回答三类问题一次AI请求花了多少钱、为什么模型会给出某个回答、某一次线上异常到底发生在哪一步。这三个问题在print时代几乎无解在结构化日志加request_id落地之后都变成了普通的Kibana查询。最后再分享一个小技巧日志格式的字段顺序其实也有讲究。把timestamp放最前面、level放第二在终端直接看日志时扫一眼就能判断时间先后和严重程度。虽然JSON对象本身是无序的但业界通用做法还是保持重要字段靠前。这个细节可能不起眼却能明显改善日常开发时直接tail -f的阅读体验。如果你手头也有一个AI项目正在纠结要不要改造日志我建议别一次性追求完美。先给所有入口加上request_id中间件再把print替换成JSON日志然后处理异步任务透传最后再考虑采集和分析。每一次改动都能带来可感知的提升O01这个编号也可以成为你日志改造之路的第一步。 AI应用日志治理实战：结构化日志与request_id全链路追踪
上周排查一个AI Agent告警时我盯着终端里滚动的print日志完全无从下手同一个用户问题触发了四次模型调用、两次工具调用、一次向量检索但所有输出都混成一团分不清先后、找不出关联更别提复现那条真正导致异常的历史链路。那次之后我在内部把“AI应用日志改造”单独立了一个项目代号O01核心就两件事把print换成结构化日志用request_id把一次请求牵扯到的所有调用串起来。这篇内容就是O01项目的完整回顾包括为什么必须改造、怎么设计字段、如何在FastAPI里落地request_id中间件以及后续做日志采集和时间线还原的实战经验适合正在开发AI Agent、RAG应用或者还在用print调大模型的团队参考。1. 先从print说起为什么AI应用的日志治理是刚需1.1 print日志在单机调试时还能凑合一旦进入Agent场景就立刻失控很多人习惯在代码里随手写print(调用LLM完成结果, resp)本地跑一个脚本、调一次接口这种方式勉强能看。可AI应用和传统Web接口有个本质区别一次用户请求会触发一段动态编排链。模型可能要经过多轮ReAct循环每轮里都要调用LLM、判断是否调用工具、再把工具结果拼回去交给模型继续推理。这个过程中涉及的网络调用、token计算、prompt构造、返回解析每一步都在消耗时间和成本。当这段链路里满是print时问题就暴露了多条请求并发时print输出会交错在一起不同模块之间的日志没有统一级别调试信息、错误信息混在stdout里更麻烦的是缺少关联标识你无法回答“这条日志属于哪个用户、哪次会话、哪一轮推理”。我见过一个团队排查线上Agent反馈问题时把几十个进程的print日志全拉到本地靠肉眼和正则搜关键词最后发现搜到的内容属于三个不同用户的两百多条记录这种情况不改造日志可观测性基本为零。1.2 大模型应用特有的五个日志痛点做AI应用之后我对日志的需求和传统后端不太一样。传统后端最关心接口响应时间、数据库慢查询、异常栈AI应用还要多关注模型调用链、提示词内容、token开销和工具执行结果。具体痛点可以整理成下面这张表痛点表现影响调用链长一次请求可能包含多次LLM调用和工具调用没有路由标识就很难串联时间线异步与并发Agent经常使用异步任务、后台线程、流式传输print输出乱序无法对应上下文载荷体积大一份prompt动辄几千token响应也很长刷屏严重日志文件快速膨胀敏感信息多prompt中可能包含用户个人信息、内部知识片段直接落盘会带来数据合规风险成本需要计量每个请求的token使用量和调用次数是核心指标非结构化文本无法做后续聚合统计这些痛点叠加在一起意味着AI应用的日志必须是一行一个JSON对象并且带有关联ID。可以说AI应用是结构化日志落地最典型的场景之一这也是我为什么强烈建议团队哪怕不做链路追踪系统也要先把结构化日志和request_id这两件事做扎实。只要这两件事落地后续接ELK、接OpenTelemetry、做成本分析都会非常顺。2. 结构化日志是什么让日志从“给人看”变成“给系统看”2.1 先定义日志字段别让日志继续是纯文本所谓结构化日志简单说就是每条日志不再是一段自然语言而是一个有固定字段的对象通常以JSON Lines的形式输出也就是每行一个独立JSON对象。这样做有几个非常直接的好处日志分析系统可以直接解析字段做索引可以在Elasticsearch里按字段过滤比如levelerror、request_idxxx还可以在Kibana里按字段做聚合统计比如按llm_provider统计调用次数、按prompt_tokens求和。我在O01项目里定义的统一日志字段大纲如下timestampISO8601格式精确到毫秒所有服务统一用UTC或统一带时区偏移避免跨服务时对不上时间。levelDEBUG、INFO、WARNING、ERROR用于过滤和告警。logger产生日志的模块名比如agent.planner、agent.tool_executor、llm.client。message人类可读的事件描述比如“llm_call_start”“tool_execution_success”。request_id贯穿整条链路的请求标识这是全文最关键的一个字段。session_id、user_id如果应用有会话概念这两个字段可以帮助定位用户维度问题。meta可扩展对象存放provider、model、token用量、耗时等指标。实际落地时不用一步到位但timestamp、level、logger、message、request_id这五个字段建议从第一天就固定下来。后面的session_id、user_id、meta可以视业务灵活加上。字段名也要尽量保持全小写下划线风格避免后续在Elasticsearch里出现大小写混乱的映射冲突。2.2 Python里用logging json formatter落地我选的方案与原因Python生态里日志方案不少常见的有三种原生logging配合自写JSON Formatter、structlog、python-json-logger。我最终选了原生logging加自写Formatter原因很实际一是团队项目依赖越少越好AI应用本身已经有一堆包了没必要为日志再增加一个强依赖二是原生logging在框架集成上最稳FastAPI、Celery、Django都能无缝衔接三是自写Formatter只有几十行代码完全可控后续要加脱敏、加特殊字段也很方便。下面这段代码是我在O01项目里用的JSONFormatter核心实现import json import logging from datetime import datetime class JsonFormatter(logging.Formatter): RESERVED set( (name, msg, args, levelname, levelno, pathname, filename, module, exc_info, exc_text, stack_info, lineno, funcName, created, msecs, relativeCreated, thread, threadName, processName, process, taskName) ) def format(self, record: logging.LogRecord) - str: record.message record.getMessage() if record.exc_info: record.exc_text self.formatException(record.exc_info) data { timestamp: datetime.utcnow().isoformat() Z, level: record.levelname, logger: record.name, message: record.message, } for key, value in record.__dict__.items(): if key not in self.RESERVED and not key.startswith(_): data[key] value if record.exc_text: data[exc_text] record.exc_text return json.dumps(data, ensure_asciiFalse)这里有几个细节值得说明。手动指定timestamp而不是用record.created格式化是为了统一格式并避免有些日志库默认输出带小数秒的浮点时间戳后续在日志平台里排序不够友好。RESERVED集合用来过滤logging内部属性你可以把自定义属性通过extra参数传给logger.info(..., extra{request_id: 123})Formatter会把它们合并进JSON的顶层字段。ensure_asciiFalse是为了让中文日志在采集端和终端里都能直接读不会变成一串\uXXXX。配置方面我推荐用logging.config.dictConfig统一管理而不是在代码里到处logger.setLevel。应用启动时加载一份YAML或字典配置控制台Handler默认输出到stdout而不是文件这个在后面讲容器化集中采集时很重要。2.3 Node.js和Go同样能落地原则是一样的如果是Node.js侧的项目我建议直接考虑pino它的性能和JSON输出能力都很成熟如果项目已经用了bunyan也可以用但维护力度相对弱一些。Go项目的做法是使用slog标准库的JSONHandler或者zerolog。原则没有变化每条日志必须是一个结构化对象必须能稳定解析并预留一个字段存放链路ID。技术栈再怎么换字段设计和透传规则才是核心这部分跨语言是一致的。3. request_id贯穿一条链路一个ID把这串调用串起来3.1 为什么非要有request_id相当于给请求发一个“快递单号”你网购一件商品中间要经过仓库、干线运输、本地配送等多个环节快递公司靠一个快递单号把所有这些环节关联起来。它不需要在各个环节之间共享复杂的上下文只需要把同一个单号打在每一站。AI应用里的request_id就是这个快递单号。当一个用户在前端发送一条消息后端Agent开始规划、检索、调用模型、执行工具过程中可能还会向外部服务发起HTTP请求这些都像是货物的不同环节。只有让每个环节的记录都携带同一个request_id你才能从日志平台里一次性捞出这条请求的完整时间线。没有这个ID所有日志就是散落一地的线索排查问题只能靠猜。3.2 生成、透传与返回请求ID的生命周期管理request_id生成规则我一般建议使用UUID4的hex形式或者带节点信息的雪花ID。生产环境如果只有一个服务UUID够用如果请求会横跨多个服务采用类似{service}-{timestamp}-{random}的格式能更快看出源头比如api-20240115-8f3a1b。整个ID的流转有三个关键节点入口生成在FastAPI中间件里先从请求头X-Request-ID中读取如果没有就自动生成。之所以优先读请求头是为了支持调用方透传自己的追踪ID方便前端和后端对账。内部透传生成后要存到ContextVar里让后续所有调用都能通过日志Filter自动读取。对Agent来说每一次子任务、每一个工具调用都要看到这个ID。外部透传Agent调用外部模型API、工具服务时要把ID放到HTTP头里一起发出去这样外部服务的日志也能串上。响应返回时把request_id写进响应头前端报错时直接把ID贴给开发定位效率能提升一倍。下面这段代码是我在FastAPI里用的中间件import uuid from contextvars import ContextVar from starlette.middleware.base import BaseHTTPMiddleware request_id_var: ContextVar[str] ContextVar(request_id, default-) class RequestIDMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): request_id request.headers.get(X-Request-ID) if not request_id: request_id uuid.uuid4().hex token request_id_var.set(request_id) try: response await call_next(request) response.headers[X-Request-ID] request_id return response finally: request_id_var.reset(token)这段代码有几个容易踩坑的地方。ContextVar.set()之后必须reset()否则在异步复用线程时会导致ID串号。中间件必须放在所有可能产生日志的依赖之前加载否则前面已产生的日志带不上ID。token的作用是支持嵌套修改比如某些服务内部想把子请求ID替换成更细粒度的ID重置时可以恢复之前的ID。接下来要把request_id注入到每一条日志记录。方式是定义一个logging.Filterimport logging from .context import request_id_var class RequestIdFilter(logging.Filter): def filter(self, record: logging.LogRecord) - bool: record.request_id request_id_var.get() return True把这个Filter加到所有Handler之后每条日志Formatter就能从record里拿到request_id并写入JSON。如果没有提前设置request_id_var.get()默认返回-不会抛异常。3.3 异步任务和多线程request_id最容易丢的地方FastAPI的后台任务BackgroundTasks和Celery异步任务是request_id丢失的高发区。因为ContextVar只在线程上下文中生效新的线程或进程不会自动继承父线程的ContextVar。如果你在请求里创建一个后台任务去处理后续逻辑任务里打出的日志记录request_id很可能就是默认的-。解决办法也很直接进入任务函数后手动把request_id重新设置回去。比如FastAPI的BackgroundTasks可以这样处理from contextvars import copy_context def run_in_background(func, *args, **kwargs): ctx copy_context() def wrapper(): ctx.run(func, *args, **kwargs) return wrapper # 在接口里 background_tasks.add_task(run_in_background(process_result, request_id_var.get(), data))当然更朴素的方案是把request_id作为参数传给后台任务函数函数开头直接request_id_var.set(request_id)。这种方式直观、清晰性能上也不会有什么损失。如果你用的是Celery则需要把request_id放进task的kwargs在task执行入口统一set。否则你会发现后台任务的日志和你线上请求完全对不上排查能力会大打折扣。4. 一个FastAPI LLM调用链的完整改造示例4.1 项目结构先看清楚要改动哪些地方为了能直接照着改我模拟一个极简AI服务FastAPI接收用户消息调用一个LLM函数必要时调用一个外部工具函数最后返回结果。全部逻辑压缩到一个项目里目录结构是这样ai_app/ ├── main.py # FastAPI入口中间件和路由 ├── logging_conf.py # logging字典配置包含JSON Formatter和Filter ├── context.py # request_id的ContextVar定义 ├── agent.py # Agent编排逻辑LLM调用、工具调用 └── llm_client.py # 模拟LLM HTTP调用这种结构不复杂但它能覆盖绝大多数AI应用日志改造会遇到的核心问题日志配置要独立、中间件要统一管理ID、Agent编排逻辑要利用日志记录关键事件。4.2 代码实现从日志配置到中间件再到业务调用先看logging_conf.py它负责构建统一的JSON输出管线import logging.config LOGGING_CONFIG { version: 1, disable_existing_loggers: False, formatters: { json: { (): logging_conf.JsonFormatter, } }, filters: { request_id: { (): logging_conf.RequestIdFilter, } }, handlers: { console: { class: logging.StreamHandler, formatter: json, level: INFO, filters: [request_id], stream: ext://sys.stdout, } }, root: { level: INFO, handlers: [console], } } def setup_logging(): logging.config.dictConfig(LOGGING_CONFIG)这里disable_existing_loggers设置为False很重要。FastAPI和uvicorn内部有自己的logger如果设置成True那些logger会被禁用很多框架日志会消失对排错反而更不利。输出到stdout而不是文件是因为在Docker和Kubernetes环境里把日志写到磁盘再让采集器去读文件相比直接stdout多了一层无效IO而且容器重建后文件就丢了。然后是main.py把中间件和日志配置接起来from fastapi import FastAPI, Request from context import request_id_var from logging_conf import setup_logging setup_logging() app FastAPI() app.middleware(http) async def request_id_middleware(request: Request, call_next): request_id request.headers.get(X-Request-ID) or uuid.uuid4().hex token request_id_var.set(request_id) try: response await call_next(request) response.headers[X-Request-ID] request_id return response finally: request_id_var.reset(token) app.post(/chat) async def chat(payload: dict): prompt payload[prompt] result await run_agent(prompt) return {result: result, request_id: request_id_var.get()}agent.py里的关键调用我加了结构化日志方便后续观察完整链路import logging import time logger logging.getLogger(agent) async def run_agent(prompt: str): logger.info(agent_start, extra{event: agent_start, prompt_len: len(prompt)}) start time.time() llm_resp await call_llm(prompt) logger.info( llm_call_done, extra{ event: llm_call_done, provider: openai-compatible, model: gpt-4o-mini, tokens_total: llm_resp.get(usage, {}).get(total_tokens, 0), latency_ms: int((time.time() - start) * 1000), }, ) if llm_resp.get(needs_tool): logger.info(tool_call_start, extra{event: tool_call_start, tool: calculator}) tool_result await run_tool(llm_resp[tool_args]) logger.info(tool_call_done, extra{event: tool_call_done, tool: calculator, result_snippet: str(tool_result)[:200]}) final_resp await call_llm(f根据工具结果生成回答: {tool_result}) return final_resp[text] return llm_resp[text]这段日志设计里我特别想强调命名习惯。event字段对应一个语义事件比如llm_call_done、tool_call_start而不是随意写“调用完成”。编程上有一套叫event-driven logging的思路简单说就是给每个日志事件一个机器可读的名称后续做告警和统计都会方便很多。extra里放结构化指标字段比如tokens_total、latency_ms它们和message文本分离这样Kibana里可以做数值聚合而不是对着一句中文文本做全文检索。调用外部的llm_client.py时要注意把request_id透传到HTTP请求头里import httpx from context import request_id_var async def call_llm(prompt: str) - dict: request_id request_id_var.get() headers {Content-Type: application/json, X-Request-ID: request_id} # 这里用mock返回真实项目替换成你的模型网关地址 async with httpx.AsyncClient(timeout60) as client: resp await client.post(https://llm-gateway.example.com/v1/chat/completions, json{prompt: prompt}, headersheaders) resp.raise_for_status() return resp.json()这样外部服务只要也支持读取X-Request-ID两边日志就能通过同一个ID关联起来。很多模型网关会自己生成内部trace id但你仍然要在业务侧保留自己的request_id作为约定不能被网关的内部ID覆盖掉。4.3 三条真实踩坑记录日志改造过程中最容易翻车的地方结构化日志改造本身不难真正磨人的是那些看起来很小的坑。我列三个印象最深的第一个坑日志重复。setup_logging()如果被调用两次或者框架启动时又加载了一遍配置root logger下面会挂多个Handler同一条日志被打印多次。解决方法是每次配置前先logging.shutdown()或者判断logging.getLogger().handlers是否为空再配置。很多框架的启动器也会默认加载logging配置在FastAPI里我用lifespan事件只调用一次setup_logging()。第二个坑异常打印丢失上下文。有些同学用logger.error(f调用失败: {e})结果只有一句话没有堆栈。改用logger.exception(llm_call_failed, extra{...})它会自动带上当前异常的堆栈信息。注意logger.exception只能在except块内使用它内部会读取sys.exc_info()在except外调用只会得到一行NoneType: None。第三个坑脱敏。AI请求里的prompt可能包含用户手机号、身份证等敏感内容。我建议在JsonFormatter里加一个sanitize函数对message和prompt字段做正则替换比如电话号码中间四位用星号替代密钥长度只保留末尾四位。日志系统的价值必须建立在合规的前提下一旦线下数据泄漏被审计发现日志改造得再好也没用。5. 日志采集与延伸从零散日志到完整可观测性5.1 用Filebeat把结构化日志采集到ELK一旦应用以JSON Lines格式输出日志采集就变得非常轻松。Filebeat是最常见的轻量级采集器之一它读文件或stdout解析JSON然后发送到Elasticsearch或Logstash。需要注意现在Elastic官方推荐用filestream类型替代旧的log类型但很多存量系统还在用log类型我实际配置时会根据ES版本选择。一个简化版的filebeat.yml如下filebeat.inputs: - type: filestream id: ai-app-logs enabled: true paths: - /var/log/ai-app/*.json parsers: - ndjson: target: - multiline: pattern: ^\{ negate: true match: after output.elasticsearch: hosts: [http://elasticsearch:9200] index: ai-app-logs-%{yyyy.MM.dd}ndjson解析器让Filebeat把JSON字段直接展平到输出文档的根级别这样Elasticsearch里可以直接通过request_id.keyword做过滤。后面那个multiline配置很重要AI应用里一条日志可能因为包含换行而断成多段用“每行只有JSON对象开头才算新日志”的规则就能把堆栈信息完整合并成一条记录。不加这个配置多行异常栈会被拆得乱七八糟排查错误时缺行少尾非常痛苦。5.2 从request_id升级到OpenTelemetry Trace的衔接路线request_id解决了“一次请求的日志怎么串起来”的问题但当你需要查询跨服务、跨进程的完整调用关系或者需要看某个外部API的耗时分布时request_id就不够用了。这时候标准做法是引入OpenTelemetry的Trace。Trace里有一个全链路唯一的trace_id你可以把它当成更高级的request_id在日志里同时记录request_id和trace_id两者并不冲突。在AI Agent场景下Trace的粒度会更细plan阶段记录Agent规划出来的步骤列表llm调用记model、prompt、response、token用量工具调用记工具名、参数、耗时、返回值摘要检索调用记向量库名称、命中条数、相似度分数这些Span和日志中的event一一对应实际上就是在日志基础上多了一层时间线和父子关系。我的建议是先把request_id做好不要一上来就上全链路Trace。因为Trace改造涉及SDK、Exporter、后端系统而request_id配合结构化日志已经能解决80%的线上排障问题。做到一定程度后再让request_id向trace_id演进迁移成本会低很多。5.3 在编排型Agent中request_id如何组织子任务现在很多项目开始做多智能体协作一个“主控Agent”会拆任务给“子Agent”执行。这种情况下让所有子任务都使用同一个request_id是可以的但不一定能满足细粒度追踪。我习惯在request_id基础上增加一个task_id或span_id字段表示当前是这次请求下的第几个子任务。主Agent发起子任务时生成task_id并把它放进日志的extra中这样既能从request_id维度看整体也能从task_id维度看某个子任务的执行情况。如果每个子Agent还涉及各自的LLM调用和工具调用那就可以把agent_name、subtask_name都作为结构化字段记下来。这个思路本质上是在为Trace做铺垫request_id对应tracetask_id对应span。提前把命名和字段统一好后面切入OpenTelemetry时只需要把task_id替换成真正的span_id即可业务代码几乎不用动。6. 常见问题与排查技巧实录6.1 问题速查表日志改造和运行过程中我整理了下面这些高频问题基本覆盖了O01项目遇到的大部分坑。症状可能原因解决方案日志重复打印setup_logging被多次调用root logger挂了多个Handler入口处只调用一次或先清理已有Handler日志里request_id全是-Filter没加进Handler或ContextVar设置后被reset检查dictConfig里filters配置确认中间件作用域日志乱序异步任务输出与请求上下文混在一起给日志加毫秒级时间戳按时间排序异步任务重新set request_id多行异常日志被拆开没有配置multiline规则Filebeat里用ndjson multiline合并日志里出现token明文未做脱敏处理Formatter中增加正则脱敏函数日志文件无限增长Handler是FileHandler且没有轮转开发环境用RotatingFileHandler生产环境输出stdout交给采集器这里特别啰嗦一句不要把生产日志直接写到本地磁盘再定期删除。日志文件无限增长是ELK场景里最不想看到的事而容器场景下stdout加采集器是更稳妥的方案。如果实在要落盘用RotatingFileHandlermaxBytes10MBbackupCount5再配合外部采集。6.2 借助request_id还原一次完整对话链路实际排查AI回答质量问题时我很少去翻原始前端界面的聊天记录而是直接从日志平台里按request_id拉取整条链路。比如用户报“某个问题回答得很奇怪”前端把响应头里的X-Request-ID发过来我就在Kibana里输request_id: abc123搜索能看到用户原始prompt的长度和内容摘要Agent规划的步骤每一轮LLM调用的模型、token数量、耗时工具调用传了什么参数、返回了什么结果最后一次组装回答用了哪些上下文通过这个时间线基本能判断问题是出在检索阶段、模型阶段还是工具阶段。举例来说如果工具返回结果正常但最终回答质量差那大概率是prompt拼接逻辑有问题如果第一轮LLM调用就报超时那就要查模型网关的限流策略。这个排查思路在事件驱动日志体系下极其好用。在生产环境没有Kibana时Linux命令行也可以直接分析。假设日志文件是app.log每条是JSON行grep request_id:abc123 app.log | jq -r .timestamp .logger .eventjnq不一定单独装这里想要说明的是只要有结构化字段就能用管道工具快速排序、筛选、统计。用jq把timestamp和event提取出来拼成一行可读时间线比在几千行原始JSON里翻找高效得多。6.3 关于“日志作用域”的一些心得很多日志库支持logger层级命名比如agent.llm、agent.tool、http.client。我建议严格规划logger作用域不要所有模块都用同一个root logger。作用域不只体现在logger name还体现在字段作用域像request_id这种全局字段放顶层像具体模型参数这种局部信息放extra的meta对象里。日志是给未来的你和其他同事看的作用域划分得越清晰检索时越省力。结语一次日志改造的实际收益O01项目做完到现在已经跑了几个月我最直观的感受不是排障更快而是团队终于能回答三类问题一次AI请求花了多少钱、为什么模型会给出某个回答、某一次线上异常到底发生在哪一步。这三个问题在print时代几乎无解在结构化日志加request_id落地之后都变成了普通的Kibana查询。最后再分享一个小技巧日志格式的字段顺序其实也有讲究。把timestamp放最前面、level放第二在终端直接看日志时扫一眼就能判断时间先后和严重程度。虽然JSON对象本身是无序的但业界通用做法还是保持重要字段靠前。这个细节可能不起眼却能明显改善日常开发时直接tail -f的阅读体验。如果你手头也有一个AI项目正在纠结要不要改造日志我建议别一次性追求完美。先给所有入口加上request_id中间件再把print替换成JSON日志然后处理异步任务透传最后再考虑采集和分析。每一次改动都能带来可感知的提升O01这个编号也可以成为你日志改造之路的第一步。