AI 应用的日志设计——记什么才有分析价值
2026/4/30大约 8 分钟
AI 应用的日志设计——记什么才有分析价值
适读人群：AI工程师、做AI应用的后端开发者 | 阅读时长：约14分钟 | 核心价值：设计有实际分析价值的AI应用日志系统，在合规和可调试性之间取得平衡
有一次，我们的 AI 问答系统在某个时间段突然回答质量大幅下降，用户投诉激增。
我去查日志，发现什么都查不了。
我们记了请求时间、HTTP 状态码、接口响应时间。但是：
- Prompt 是什么？没记。
- 模型用了多少 Token？没记。
- 召回的是哪些文档？没记。
- 用户最后觉得这个回答有没有用？没记。
我只知道「有请求，有响应，响应时间正常」。但质量为什么下降，毫无线索。
最后排查了三天，靠抽样重放了几百个历史请求，才发现是知识库里混入了一批错误文档，导致召回了大量噪音。如果当时有完整的日志，半个小时就能定位到。
从那以后，我把 AI 应用的日志系统当成和代码逻辑同等重要的基础设施来设计。
AI 应用的日志为什么不能套用 Web 日志
普通 Web 应用的日志关注的是：系统做了什么。
请求来了、数据库查了什么、返回了什么 HTTP 状态码。这些足够了，因为 Web 应用的行为是确定性的：给定相同的请求，应该得到相同的结果。出了问题，通常是代码 bug 或者基础设施故障。
AI 应用的行为是非确定性的，而且影响结果的因素更多：
- 模型版本变了
- Prompt 改了
- 知识库内容变了
- 用户输入稍微不同，结果差很多
- 同一个输入，今天的输出和昨天的不一样
要分析 AI 应用的问题，你需要记录的是「这次 AI 推理发生了什么」，而不只是「这次 HTTP 请求发生了什么」。
AI 应用必须记录的核心字段
我把必须记录的内容分成四类，缺了任何一类都会有盲区。
类别一：完整的 Prompt 上下文
这是最重要的，也是最容易被忽略的。很多团队觉得「Prompt 太长了，记下来太占空间」，然后选择不记，然后出了问题什么都查不了。
必须记录：
- System Prompt（带版本号）
- 用户输入（原始的，不处理）
- 如果有 RAG，记录检索出来的文档片段列表（包括每个文档的 ID 和相似度分数）
- 最终拼装好的完整 Prompt（可选，但非常有助于调试）
类别二：模型调用参数和结果
- 使用的模型 ID（带版本号，比如
gpt-4o-2024-11-20） - Temperature、Top-p 等推理参数
- 输入 Token 数、输出 Token 数（分开记）
- 模型的原始输出（不经过任何后处理的）
- 后处理后的结构化输出（如果有的话）
- 推理耗时（模型调用的时间，不含前后的处理）
类别三：性能数据
- 端到端响应时间（用户视角）
- 各子步骤的耗时分布（向量检索耗时、模型推理耗时、后处理耗时）
- 是否触发了 Retry（如果有的话，记录 Retry 次数和原因）
类别四：用户反馈
这是很多团队忽略但极其有价值的数据：
- 用户是否给了明确反馈（点赞/踩）
- 用户是否对同一个问题进行了追问（隐含对上一个回答不满意）
- 用户是否很快关闭了会话（可能觉得没用）
这些隐式和显式的反馈，是你判断 AI 质量的最直接数据源。
完整的日志方案代码
说了这么多，直接给代码。
import json
import uuid
import time
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import Optional, List, Dict, Any
import logging
logger = logging.getLogger(__name__)
@dataclass
class RetrievedDoc:
"""RAG 检索出来的文档记录"""
doc_id: str
chunk_id: str
similarity_score: float
content_preview: str # 前200字符，用于日志可读性
doc_title: Optional[str] = None
@dataclass
class AICallLog:
"""单次 AI 调用的完整日志"""
# 基础标识
trace_id: str = field(default_factory=lambda: str(uuid.uuid4()))
session_id: Optional[str] = None
user_id: Optional[str] = None # 注意：需要脱敏处理
request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
# Prompt 相关
system_prompt_version: Optional[str] = None # 比如 "qa_v2.3"
user_input: Optional[str] = None
user_input_length: int = 0
retrieved_docs: List[RetrievedDoc] = field(default_factory=list)
full_prompt_tokens_estimate: int = 0
# 模型调用
model_id: Optional[str] = None # 完整版本号
temperature: Optional[float] = None
max_tokens: Optional[int] = None
input_tokens: int = 0
output_tokens: int = 0
total_tokens: int = 0
raw_model_output: Optional[str] = None
parsed_output: Optional[Dict] = None
# 性能
retrieval_latency_ms: int = 0
model_latency_ms: int = 0
postprocess_latency_ms: int = 0
total_latency_ms: int = 0
retry_count: int = 0
# 结果状态
success: bool = True
error_type: Optional[str] = None
error_message: Optional[str] = None
# 反馈（异步更新）
user_feedback: Optional[str] = None # "positive" / "negative" / None
feedback_timestamp: Optional[str] = None
class AILogger:
"""AI 应用日志记录器"""
def __init__(
self,
log_sink, # 日志输出目标（文件、ES、ClickHouse 等）
pii_masker=None, # 个人信息脱敏处理器
sample_rate: float = 1.0 # 采样率，高并发场景可以降低
):
self.log_sink = log_sink
self.pii_masker = pii_masker
self.sample_rate = sample_rate
def log_ai_call(self, log: AICallLog):
"""记录一次 AI 调用"""
import random
if random.random() > self.sample_rate:
return
# 脱敏处理
log_dict = asdict(log)
if self.pii_masker:
log_dict = self.pii_masker.mask(log_dict)
self.log_sink.write(log_dict)
def update_feedback(self, request_id: str, feedback: str):
"""用户反馈异步写入（反馈通常在请求之后才产生）"""
feedback_record = {
"request_id": request_id,
"user_feedback": feedback,
"feedback_timestamp": datetime.utcnow().isoformat(),
"record_type": "feedback_update"
}
self.log_sink.write(feedback_record)在实际请求处理中使用：
async def handle_qa_request(
user_input: str,
session_id: str,
user_id: str,
ai_logger: AILogger
) -> dict:
log = AICallLog(
session_id=session_id,
user_id=anonymize_user_id(user_id), # 立即脱敏
user_input=user_input,
user_input_length=len(user_input),
system_prompt_version="qa_v2.3",
model_id="gpt-4o-2024-11-20",
temperature=0.1
)
total_start = time.time()
try:
# 步骤一：向量检索
retrieval_start = time.time()
raw_results = vectorstore.similarity_search_with_score(user_input, k=5)
log.retrieval_latency_ms = int((time.time() - retrieval_start) * 1000)
log.retrieved_docs = [
RetrievedDoc(
doc_id=doc.metadata["doc_id"],
chunk_id=doc.metadata["chunk_id"],
similarity_score=round(score, 4),
content_preview=doc.page_content[:200],
doc_title=doc.metadata.get("title")
)
for doc, score in raw_results
]
# 步骤二：构建 Prompt 并调用模型
context = "\n\n".join([doc.page_content for doc, _ in raw_results])
prompt = build_qa_prompt(user_input, context)
model_start = time.time()
response = openai_client.chat.completions.create(
model=log.model_id,
messages=[{"role": "user", "content": prompt}],
temperature=log.temperature
)
log.model_latency_ms = int((time.time() - model_start) * 1000)
# 记录 Token 用量
log.input_tokens = response.usage.prompt_tokens
log.output_tokens = response.usage.completion_tokens
log.total_tokens = response.usage.total_tokens
log.raw_model_output = response.choices[0].message.content
# 步骤三：后处理
postprocess_start = time.time()
parsed = parse_and_validate_output(log.raw_model_output)
log.postprocess_latency_ms = int((time.time() - postprocess_start) * 1000)
log.parsed_output = parsed
log.success = True
return {"answer": parsed["answer"], "request_id": log.request_id}
except Exception as e:
log.success = False
log.error_type = type(e).__name__
log.error_message = str(e)
raise
finally:
log.total_latency_ms = int((time.time() - total_start) * 1000)
ai_logger.log_ai_call(log)隐私合规和可调试性的平衡
记录完整 Prompt 和用户输入，会引出隐私合规的问题。怎么平衡？
原则：记录必要信息，脱敏后再存
不是不记，而是「记，但做好脱敏」。
class PIIMasker:
"""个人信息脱敏处理器"""
def mask(self, log_dict: dict) -> dict:
"""对日志里的敏感字段做脱敏"""
masked = log_dict.copy()
# 用户 ID 做哈希（保留分析能力，但无法反查真实用户）
if masked.get("user_id"):
masked["user_id"] = self._hash_id(masked["user_id"])
# 用户输入脱敏（姓名、手机号、身份证等）
if masked.get("user_input"):
masked["user_input"] = self._mask_pii_in_text(masked["user_input"])
# 模型输出脱敏
if masked.get("raw_model_output"):
masked["raw_model_output"] = self._mask_pii_in_text(
masked["raw_model_output"]
)
return masked
def _mask_pii_in_text(self, text: str) -> str:
import re
# 手机号：138****8888
text = re.sub(r'1[3-9]\d{9}', lambda m: m.group()[:3] + '****' + m.group()[-4:], text)
# 身份证：4位省份码 + 屏蔽 + 4位尾码
text = re.sub(r'\d{6}(19|20)\d{2}\d{4}\d{3}[\dX]',
lambda m: m.group()[:4] + '**********' + m.group()[-4:], text)
# 邮箱：前2位 + *** + @ 后面保留
text = re.sub(r'[a-zA-Z0-9._%+-]{2,}@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
lambda m: m.group()[:2] + '***@' + m.group().split('@')[1], text)
return text
def _hash_id(self, user_id: str) -> str:
import hashlib
return hashlib.sha256(user_id.encode()).hexdigest()[:16]日志的分级存储
不是所有日志都需要长期保存全量内容：
热数据（最近7天）：存完整日志，包括完整 Prompt 和输出
用于实时故障排查
温数据（7-30天）：只存摘要信息（Token数、延迟、成功/失败、用户反馈）
去掉 Prompt 和模型输出的完整文本
用于趋势分析
冷数据（30天以上）：只保留聚合统计（按天、按模型版本）
用于成本分析和长期趋势这个分级方案可以在存储成本和可调试性之间取得平衡——真正需要调试的问题通常在 7 天内发现，长期数据主要用于统计分析。
有了日志后能做什么分析
建立了这套日志之后，可以回答这些问题：
质量监控：用户反馈的正负比率随时间是否有变化？哪类 Query 的负反馈率最高？
-- 每天的用户满意度趋势
SELECT
DATE(timestamp) as date,
COUNT(*) as total_calls,
SUM(CASE WHEN user_feedback = 'positive' THEN 1 ELSE 0 END) as positive,
SUM(CASE WHEN user_feedback = 'negative' THEN 1 ELSE 0 END) as negative,
ROUND(SUM(CASE WHEN user_feedback = 'positive' THEN 1 ELSE 0 END) * 100.0 /
NULLIF(SUM(CASE WHEN user_feedback IN ('positive','negative') THEN 1 ELSE 0 END), 0), 2) as positive_rate
FROM ai_call_logs
WHERE timestamp >= NOW() - INTERVAL '30 days'
GROUP BY DATE(timestamp)
ORDER BY date成本分析：每个功能模块的 Token 消耗分布，找出成本最高的场景做优化。
延迟分析：P50、P95、P99 延迟的分布，定位瓶颈在检索层还是模型层。
故障排查：出了问题，能精确回放当时的 Prompt 和召回文档，快速定位原因。
AI 应用的可观测性不是可选项，是你知道系统在干什么的唯一方式。日志系统设计好了，出了问题排查快，想优化有数据，质量回归能及时发现——这些价值在真实生产环境里会一遍遍地证明它值得前期的投入。