企业官网建设流程全解析

上周排查一个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开销和工具执行结果。具体痛点可以整理成下面这张表：

这些痛点叠加在一起，意味着AI应用的日志必须是一行一个JSON对象，并且带有关联ID。可以说，AI应用是结构化日志落地最典型的场景之一，这也是我为什么强烈建议团队哪怕不做链路追踪系统，也要先把结构化日志和request_id这两件事做扎实。只要这两件事落地，后续接ELK、接OpenTelemetry、做成本分析都会非常顺。

2. 结构化日志是什么：让日志从“给人看”变成“给系统看”

2.1 先定义日志字段，别让日志继续是纯文本

所谓结构化日志，简单说就是每条日志不再是一段自然语言，而是一个有固定字段的对象，通常以JSON Lines的形式输出，也就是每行一个独立JSON对象。这样做有几个非常直接的好处：日志分析系统可以直接解析字段做索引；可以在Elasticsearch里按字段过滤，比如level=error、request_id=xxx；还可以在Kibana里按字段做聚合统计，比如按llm_provider统计调用次数、按prompt_tokens求和。

我在O01项目里定义的统一日志字段大纲如下：

• timestamp：ISO8601格式，精确到毫秒，所有服务统一用UTC或统一带时区偏移，避免跨服务时对不上时间。
• level：DEBUG、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_ascii=False)

这里有几个细节值得说明。手动指定timestamp而不是用record.created格式化，是为了统一格式并避免有些日志库默认输出带小数秒的浮点时间戳，后续在日志平台里排序不够友好。RESERVED集合用来过滤logging内部属性，你可以把自定义属性通过extra参数传给logger.info("...", extra={"request_id": "123"})，Formatter会把它们合并进JSON的顶层字段。ensure_ascii=False是为了让中文日志在采集端和终端里都能直接读，不会变成一串\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的流转有三个关键节点：

1. 入口生成：在FastAPI中间件里，先从请求头X-Request-ID中读取，如果没有就自动生成。之所以优先读请求头，是为了支持调用方透传自己的追踪ID，方便前端和后端对账。
2. 内部透传：生成后要存到ContextVar里，让后续所有调用都能通过日志Filter自动读取。对Agent来说，每一次子任务、每一个工具调用都要看到这个ID。
3. 外部透传：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.Filter：

import 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(timeout=60) as client: resp = await client.post("https://llm-gateway.example.com/v1/chat/completions", json={"prompt": prompt}, headers=headers) 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对应trace，task_id对应span。提前把命名和字段统一好，后面切入OpenTelemetry时只需要把task_id替换成真正的span_id即可，业务代码几乎不用动。

6. 常见问题与排查技巧实录

6.1 问题速查表

日志改造和运行过程中，我整理了下面这些高频问题，基本覆盖了O01项目遇到的大部分坑。

这里特别啰嗦一句：不要把生产日志直接写到本地磁盘再定期删除。日志文件无限增长是ELK场景里最不想看到的事，而容器场景下stdout加采集器是更稳妥的方案。如果实在要落盘，用RotatingFileHandler，maxBytes=10MB，backupCount=5，再配合外部采集。

6.2 借助request_id还原一次完整对话链路

实际排查AI回答质量问题时，我很少去翻原始前端界面的聊天记录，而是直接从日志平台里按request_id拉取整条链路。比如用户报“某个问题回答得很奇怪”，前端把响应头里的X-Request-ID发过来，我就在Kibana里输request_id: "abc123"搜索，能看到：

1. 用户原始prompt的长度和内容摘要
2. Agent规划的步骤
3. 每一轮LLM调用的模型、token数量、耗时
4. 工具调用传了什么参数、返回了什么结果
5. 最后一次组装回答用了哪些上下文

通过这个时间线，基本能判断问题是出在检索阶段、模型阶段还是工具阶段。举例来说，如果工具返回结果正常但最终回答质量差，那大概率是prompt拼接逻辑有问题；如果第一轮LLM调用就报超时，那就要查模型网关的限流策略。这个排查思路在事件驱动日志体系下极其好用。

在生产环境没有Kibana时，Linux命令行也可以直接分析。假设日志文件是app.log，每条是JSON行：

grep '"request_id":"abc123"' app.log | jq -r '.timestamp + " " + .logger + " " + .event'

jnq不一定单独装，这里想要说明的是：只要有结构化字段，就能用管道工具快速排序、筛选、统计。用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这个编号，也可以成为你日志改造之路的第一步。