外观
从静默掉点到提前 48 小时告警：LLM 漂移巡检实战
问题背景：LLM 线上效果为什么会悄悄变差
传统服务的监控通常围绕 QPS、延迟、错误率和资源利用率展开，但 LLM 应用即使这些指标全部正常，业务效果也可能持续衰减。典型现象是：客服机器人答非所问的比例上升、代码助手补全采纳率下降、RAG 引用过期文档增多、智能体工作流中途放弃率提高。根因往往不是服务宕机，而是线上输入分布发生了漂移：新用户提出了训练基准中没有的问题，业务规则变更后提示词上下文失效，模型供应商小版本升级改变了 JSON 遵循能力，或者检索库中新增了大量低质量文档。
因此，LLM 运维不能只看“有没有报错”，还要持续回答三个问题：今天的输入和上线验收时是否相似？模型输出结构和质量是否稳定？核心业务效果是否出现统计显著的衰减？这正是漂移监控、llm-observability 与自动巡检体系要解决的问题。
方案对比：四类 LLM 漂移监控路径
常见方案可以分为四类。第一类是人工抽检，成本低但发现滞后，适合早期日均请求量较小的应用；第二类是规则与嵌入漂移监控，能够覆盖输入长度、语言、意图、语义簇变化，适合作为日常第一道防线；第三类是在线 LLM 裁判或离线评测，准确性更高，但存在额外模型成本和评分波动；第四类是业务反馈闭环，例如采纳、点赞、转人工、退款、工单重开等信号，最贴近真实价值，但通常存在标签延迟。
推荐采用“输入分布日检 + 黄金集定时评测 + 业务指标周巡检”的组合，而不是单独依赖 LLM 裁判。输入监控负责发现流量结构变化，黄金集负责发现版本回归，业务指标负责确认真实效果衰减。
监控指标设计：先定义可比较的基线
漂移监控的第一步不是接告警，而是冻结一份可信基线。基线应包含三部分：上线时通过验收的黄金评测集、连续 7 到 14 天的生产请求样本、每个核心场景的业务效果指标。黄金集不求大，但要覆盖高频意图、边界问题、安全拒答、格式约束和历史故障案例。生产样本建议按用户、时间和意图分层采样，避免被高频机器人流量淹没。
输入侧至少监控五类特征：文本长度分布、语言与语种、意图或主题占比、嵌入向量语义距离、敏感词与高风险输入比例。输出侧监控 JSON 解析成功率、字段完整率、拒答率、工具调用成功率、引用覆盖率。效果侧则监控自动评分、规则判分、用户采纳率、人工纠错率和转人工率。
实操步骤：用 Python 搭建每日漂移巡检任务
下面示例使用嵌入向量做语义漂移，用 PSI（Population Stability Index）衡量分布变化，并用黄金集计算格式成功率与拒答率。示例不依赖特定厂商，可将 fake_embed 替换为 OpenAI、通义、混元或本地 embedding 服务。
python
import json
import math
from collections import Counter
from pathlib import Path
# 可直接运行：python drift_patrol.py
# 依赖：仅标准库；生产环境可将 fake_embed 替换为真实 embedding 接口
def fake_embed(text: str, dim: int = 16) -> list[float]:
"""用字符哈希构造稳定的伪向量，便于本地演示。"""
vec = [0.0] * dim
for ch in text:
vec[ord(ch) % dim] += 1.0
total = sum(vec) or 1.0
return [v / total for v in vec]
def cosine(a: list[float], b: list[float]) -> float:
dot = sum(x * y for x, y in zip(a, b))
na = math.sqrt(sum(x * x for x in a))
nb = math.sqrt(sum(y * y for y in b))
return dot / (na * nb + 1e-9)
def semantic_drift(baseline_texts: list[str], current_texts: list[str]) -> float:
"""计算当前样本到基线中心的平均余弦距离，值越高表示语义偏移越大。"""
base_vectors = [fake_embed(t) for t in baseline_texts]
cur_vectors = [fake_embed(t) for t in current_texts]
dim = len(base_vectors[0])
center = [sum(v[i] for v in base_vectors) / len(base_vectors) for i in range(dim)]
distances = [1 - cosine(v, center) for v in cur_vectors]
return sum(distances) / len(distances)
def psi(expected: list[float], actual: list[float], bins: int = 10) -> float:
"""PSI 小于 0.1 稳定，0.1-0.25 轻微漂移，大于 0.25 显著漂移。"""
lo, hi = min(expected + actual), max(expected + actual) + 1e-6
edges = [lo + (hi - lo) * i / bins for i in range(bins + 1)]
def proportions(values):
cnt = Counter()
for v in values:
idx = min(int((v - lo) / (hi - lo) * bins), bins - 1)
cnt[idx] += 1
return [(cnt[i] + 0.5) / (len(values) + 0.5 * bins) for i in range(bins)]
p_exp, p_act = proportions(expected), proportions(actual)
return sum((a - e) * math.log(a / e) for a, e in zip(p_act, p_exp))
def evaluate_outputs(outputs: list[dict]) -> dict:
"""巡检输出质量：JSON 格式、必填字段、拒答率。"""
total = len(outputs)
valid_json = sum(1 for x in outputs if x.get("answer") is not None)
has_source = sum(1 for x in outputs if x.get("sources"))
refusal = sum(1 for x in outputs if "无法回答" in str(x.get("answer", "")))
return {
"format_success_rate": round(valid_json / total, 4),
"source_coverage": round(has_source / total, 4),
"refusal_rate": round(refusal / total, 4),
}
def patrol(baseline, current, outputs) -> dict:
base_len = [float(len(x)) for x in baseline]
cur_len = [float(len(x)) for x in current]
report = {
"length_psi": round(psi(base_len, cur_len), 4),
"semantic_drift": round(semantic_drift(baseline, current), 4),
"quality": evaluate_outputs(outputs),
"alerts": [],
}
if report["length_psi"] > 0.25:
report["alerts"].append("输入长度分布显著漂移，请检查新流量入口")
if report["semantic_drift"] > 0.18:
report["alerts"].append("语义中心偏移，请按意图聚类抽样复核")
if report["quality"]["format_success_rate"] < 0.97:
report["alerts"].append("结构化输出成功率下降，请检查模型版本或提示词")
return report
if __name__ == "__main__":
baseline_requests = ["如何重置密码", "订单为什么还没发货", "发票抬头怎么修改"]
current_requests = ["请帮我分析数据库慢查询", "这段 Java 空指针怎么修复", "生成压测脚本并解释结果"]
current_outputs = [
{"answer": "建议先查看执行计划", "sources": ["doc-1"]},
{"answer": "无法回答", "sources": []},
{"answer": "可以使用 wrk 压测", "sources": ["doc-8"]},
]
print(json.dumps(patrol(baseline_requests, current_requests, current_outputs), ensure_ascii=False, indent=2))生产化时，这段任务应由调度系统每日执行，并将结果写入 Prometheus、ClickHouse 或日志平台。建议暴露四类指标：llm_input_length_psi、llm_semantic_drift_score、llm_output_format_success_rate、llm_business_adopt_rate。每个指标都要携带 app_name、scenario、model_version、prompt_version、retriever_version 标签，否则告警后无法快速定位。
告警策略：从统计异常到可执行工单
漂移不等于故障，不能把所有统计波动都升级为电话告警。更合理的做法是分级：P1 表示核心业务指标连续两个统计周期下降，例如采纳率从 72% 降到 61%，同时格式成功率跌破 95%；P2 表示输入 PSI 或语义距离超过阈值，但效果指标尚未明显下降；P3 表示单个新意图占比升高，进入观察看板即可。
告警内容必须附带上钻路径：异常时间段、受影响场景、模型与提示词版本、Top 漂移问题样例、与基线的差异、最近变更记录。只发送“漂移分数过高”会让值班同学无法处理。自动化工单可以建议三种动作：运行黄金集回归、按新问题簇扩充评测集、回滚最近的提示词或模型版本。
评测集维护：让巡检体系持续有效
很多团队搭建监控后两周就失去信任，原因是评测集长期不更新。黄金集应采用“核心集稳定、增量集滚动”的维护方式：核心集覆盖不能被修改的基础能力，增量集来自真实线上失败案例和新业务规则。每次新增样本都要标注期望行为、判定方式和业务优先级，避免把有争议的用户反馈直接当成标准答案。
对于无法用规则判定的开放式回答，可以使用 LLM 裁判，但裁判提示词必须固定版本，并同时输出分数、理由和不确定性。更稳妥的方式是将裁判结果用于抽样排序，而不是单独触发 P1。
避坑总结：漂移巡检最常见的 5 个问题
只监控均值，不监控分布和分场景。 常见错误现象是整体采纳率稳定，但某个高价值场景已经暴跌，原因是低价值流量增长稀释了问题。修正方法是按场景、用户分层和模型版本拆分指标，并同时观察 P50、P95 与占比变化。
PSI 直接用于高维 embedding。 有些文章把每一维 embedding 都单独算 PSI，导致告警噪声极大。高维向量应先计算到基线中心的距离、做聚类占比比较，或使用降维后的主题分布；PSI 更适合长度、Token 数、延迟、分数等一维指标。
黄金集混入线上原始日志后反复“修答案”。 这会让评测分数看起来一直变好，却无法发现回归。应将历史故障案例冻结为不可随意修改的回归样本，新案例走评审流程，并记录修改原因。
模型输出正常但业务效果下降，却只排查模型。 RAG 和智能体应用中，检索召回为空、工具超时被吞、知识库权限变更、上游页面传错用户类型，都会表现为“答案变差”。告警必须关联 prompt、retriever、tool、model 四类版本。
告警阈值一次配死后无人校准。 大促、节假日、新产品发布都会带来自然流量变化。建议阈值基于滚动基线和同期数据动态生成，P2 告警先进入抽样复核；连续误报应进入阈值复盘，而不是简单关闭监控。
一套可靠的 LLM 巡检体系，本质上是把“模型能力”纳入和接口、数据库、消息队列同等重要的生产资产管理。输入分布监控帮助团队尽早发现变化，黄金集评测验证版本稳定性，业务反馈确认真实效果衰减。三者联动，才能在用户大规模感知之前完成定位、回滚和修复。
免责声明
本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。