当前位置：首页 > news >正文

检索质量在线监控：实时检测索引漂移、召回率和精度下降的异常告警

news 2026/7/12 16:30:38

检索质量在线监控：实时检测索引漂移、召回率和精度下降的异常告警

一、深度引言与场景痛点

大家好，我是赵咕咕。

RAG 系统最可怕的问题不是突然挂掉，而是慢慢变笨。今天能搜到的东西，下周搜不到。三周前准确的回答，现在全是幻觉。用户不会告诉你——他们只是默默地走了，你的留存曲线一天比一天难看。

问题出在"索引漂移"。随着新文档的加入和旧文档的更新，你的向量索引在不知不觉中"偏移"了。原来用户问"A 产品的价格"，能搜到正确的价目表文档。但新的相似文档被索引后，向量空间的分布变了——同一个 query 的向量不再指向最相关的文档。

今天我们聊聊怎么实时监控检索质量，在它悄悄变差之前就发现问题。

二、底层机制与原理深度剖析

2.1 检索质量的三维指标

检索质量可以从三个维度来衡量：

召回率（Recall）：应该被检索到的相关文档，真正被检索到了多少？召回率下降意味着"漏掉了重要信息"。

精确率（Precision）：检索到的文档中，真正相关的有多少？精确率下降意味着"混进了无关信息"。

排序质量（NDCG）：最相关的文档是否排在最前面？排序质量下降意味着 LLM 可能看到了错误的内容当作上下文。

2.2 在线监控的架构

在线监控的核心挑战是：你没有标注数据来判断"是否相关"。解决方案是用统计手段间接检测：

flowchart TB A[线上请求流量] --> B{采样策略} B -->|5% 流量| C[采样请求] B -->|95% 流量| D[正常服务] C --> E[Query 聚类分析] C --> F[检索结果分布分析] C --> G[置信度分数变化] E --> H[检测 Query 分布漂移] F --> I[检测结果同质化] G --> J[检测检索信心下降] H --> K{异常检测引擎} I --> K J --> K K -->|发现异常| L[触发告警 + 记录快照] K -->|正常| M[更新基线]

2.3 具体检测方法

Query 分布漂移检测：对比当前时段和上一个时段的 Query embedding 分布（用 KL 散度或 Wasserstein 距离）。如果分布显著变化，说明用户的提问方式改变了，可能需要更新索引。

检索结果同质化检测：如果 top_k 结果的向量之间余弦相似度越来越高，说明检索结果在"塌缩"——无论问什么，返回的都是差不多的文档。

置信度分数变化：向量检索返回的相似度分数。如果整体分数趋势性下降，说明索引质量在恶化。

flowchart LR A[采样请求] --> B[提取特征] B --> C[与历史基线对比] C --> D{漂移检测} D -->|KL散度 > 阈值| E[Query分布告警] D -->|相似度均值下降 > 10%| F[检索信心告警] D -->|top_k 互相似度上升| G[结果同质化告警] D -->|正常| H[更新滑动窗口]

三、生产级代码实现

3.1 漂移检测器

import asyncio import numpy as np from collections import deque from dataclasses import dataclass, field from typing import Any @dataclass class DriftWindow: """滑动窗口存储采样数据。""" embeddings: deque = field(default_factory=lambda: deque(maxlen=1000)) similarity_scores: deque = field(default_factory=lambda: deque(maxlen=1000)) def add(self, embedding: list[float], score: float) -> None: self.embeddings.append(np.array(embedding)) self.similarity_scores.append(score) @property def mean_score(self) -> float: if not self.similarity_scores: return 0.0 return float(np.mean(list(self.similarity_scores))) @property def embedding_center(self) -> np.ndarray: if not self.embeddings: return np.array([]) return np.mean(list(self.embeddings), axis=0) class RetrievalQualityMonitor: """检索质量在线监控器。""" def __init__( self, window_size: int = 1000, drift_threshold: float = 0.3, score_drop_threshold: float = 0.1, homogeneity_threshold: float = 0.85, ): self._current = DriftWindow() self._baseline = DriftWindow() self._window_size = window_size self._drift_threshold = drift_threshold self._score_drop_threshold = score_drop_threshold self._homogeneity_threshold = homogeneity_threshold self._lock = asyncio.Lock() async def record( self, query_embedding: list[float], top_k_scores: list[float], top_k_embeddings: list[list[float]], ) -> None: """记录一次检索的采样数据。""" async with self._lock: avg_score = np.mean(top_k_scores) if top_k_scores else 0.0 self._current.add(query_embedding, avg_score) if self._baseline.embeddings and len(self._baseline.embeddings) < 500: # 还在建立基线阶段，把所有数据也当基线 self._baseline.add(query_embedding, avg_score) async def check_drift(self) -> dict[str, Any]: """检查是否发生漂移。""" async with self._lock: if len(self._current.embeddings) < 100: return {"status": "insufficient_data"} alerts = [] # 检查 1：相似度分数是否趋势性下降 current_mean = self._current.mean_score baseline_mean = self._baseline.mean_score if baseline_mean > 0: score_drop = (baseline_mean - current_mean) / baseline_mean if score_drop > self._score_drop_threshold: alerts.append({ "type": "score_drop", "baseline": baseline_mean, "current": current_mean, "drop_ratio": score_drop, }) # 检查 2：Query embedding 分布漂移（简化版 KL 近似） if self._baseline.embedding_center.size > 0 and self._current.embedding_center.size > 0: drift = self._cosine_distance( self._baseline.embedding_center, self._current.embedding_center, ) if drift > self._drift_threshold: alerts.append({ "type": "query_drift", "distance": drift, }) return { "status": "alert" if alerts else "normal", "alerts": alerts, "current_mean_score": current_mean, "baseline_mean_score": baseline_mean, "sample_count": len(self._current.embeddings), } async def update_baseline(self) -> None: """将当前窗口设为新的基线。""" async with self._lock: self._baseline = DriftWindow() for emb in self._current.embeddings: self._baseline.embeddings.append(emb) for score in self._current.similarity_scores: self._baseline.similarity_scores.append(score) @staticmethod def _cosine_distance(a: np.ndarray, b: np.ndarray) -> float: cosine_sim = np.dot(a, b) / ( np.linalg.norm(a) * np.linalg.norm(b) + 1e-8 ) return float(1.0 - cosine_sim)

3.2 在线监控循环

import logging logger = logging.getLogger(__name__) class RetrievalGuard: """在检索链路中嵌入质量监控。""" def __init__( self, monitor: RetrievalQualityMonitor, embedding_service, sample_rate: float = 0.05, ): self.monitor = monitor self.embedder = embedding_service self.sample_rate = sample_rate async def wrap_search( self, query: str, searcher, ) -> tuple[list[dict], dict[str, Any]]: """包装检索调用，同时采集监控数据。""" import random results = await searcher(query) # 按采样率记录 if random.random() < self.sample_rate: try: query_emb = await self.embedder.embed(query) top_scores = [ r.get("score", 0.0) for r in results[:5] ] await self.monitor.record( query_embedding=query_emb, top_k_scores=top_scores, top_k_embeddings=[], # 可省略以减少开销 ) except Exception: logger.debug("监控采样失败（不影响主流程）") return results, {} class QualityAlertLoop: """定期检查检索质量并告警。""" def __init__( self, monitor: RetrievalQualityMonitor, check_interval: int = 300, ): self.monitor = monitor self.interval = check_interval async def run(self) -> None: while True: await asyncio.sleep(self.interval) try: result = await self.monitor.check_drift() if result["status"] == "alert": for alert in result["alerts"]: logger.warning( "检索质量异常: type=%s, detail=%s", alert["type"], alert, ) # 发送告警通知 await self._notify(result) except Exception as e: logger.error("质量检查异常: %s", e) async def _notify(self, result: dict) -> None: # 接入实际的告警通道 pass