企业官网建设流程全解析

RAG 延迟保障 SLA：用 P50/P95/P99 分位数设定并监控检索延迟目标

一、深度引言与场景痛点

大家好，我是赵咕咕。

先讲一个真实的故事。我们线上 RAG 服务的延迟监控面板上，"平均响应时间"是 350ms——看起来挺不错的。但用户投诉说"有时候搜个问题要好几十秒"。我去查日志，发现每 1000 个请求里，有 5 个花了 15 秒以上。

平均值把这 5 个"慢请求"淹没了。

平均值掩盖了长尾问题，而长尾问题恰恰是用户体验最差的部分。用户不会记住"平均 350ms"，但会记住"有一次我等了快一分钟"。

今天我们来聊聊怎么用**分位数（P50/P95/P99）**来设定 RAG 检索的延迟 SLA，并构建监控体系。

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

2.1 为什么分位数比平均值有用

平均值 = 所有请求延迟之和 / 请求总数。只要大部分请求够快，少数极慢请求对平均值影响不大。

分位数不同。P99 = 3000ms 的意思是：99% 的请求在 3000ms 内完成，但有 1% 的请求超过了 3000ms。这 1% 就是你要解决的长尾问题。

flowchart LR A[请求延迟分布] --> B[排序: 慢 → 快] B --> C[P50 = 中位数<br/>50% 用户感受到的延迟] B --> D[P95 = 第95分位<br/>只有 5% 请求超过此值] B --> E[P99 = 第99分位<br/>只有 1% 请求超过此值] C --> F[典型用户体验] D --> G[SLA 核心指标] E --> H[长尾监控 / 异常检测]

2.2 RAG 检索延迟的来源

在 RAG 系统中，单次检索的端到端延迟包含多个环节：

sequenceDiagram participant C as 客户端 participant G as API 网关 participant Q as Query 改写服务 participant E as Embedding 服务 participant V as 向量数据库 participant R as Re-Ranker participant L as LLM 生成 C->>G: HTTP 请求 G->>Q: Query 改写 (20-80ms) Q-->>G: 改写后的 Query G->>E: 文本向量化 (30-150ms) E-->>G: 向量 G->>V: 向量检索 (10-50ms) V-->>G: Top-K 文档 G->>R: 重排序 (50-200ms) R-->>G: 重排结果 G->>L: 上下文合成 (500-3000ms) L-->>G: 生成答案 G-->>C: HTTP 响应 Note over G,L: LLM 生成是最大瓶颈

可以看到，延迟大头在 LLM 生成环节。但检索链路（Embedding + 向量检索 + 重排序）也需要严格管控——这部分延迟叠加起来也有 100-400ms。

2.3 SLA 目标设定

典型的 RAG 系统 SLA 目标：

注意：这里说的是检索延迟（不含 LLM 生成），因为 LLM 生成的延迟可控性较低。

三、生产级代码实现

3.1 分位数延迟统计器

import asyncio import time import heapq from collections import defaultdict from dataclasses import dataclass, field @dataclass class LatencyStats: """延迟统计，使用堆排序计算近似分位数。""" samples: list[float] = field(default_factory=list) max_samples: int = 10000 def record(self, latency_ms: float) -> None: if len(self.samples) >= self.max_samples: # 超出上限时做蓄水池采样，保留统计代表性 import random idx = random.randint(0, len(self.samples)) if idx < self.max_samples: self.samples[idx] = latency_ms else: self.samples.append(latency_ms) def percentile(self, p: float) -> float: """计算第 p 分位数（p ∈ [0, 1]）。""" if not self.samples: return 0.0 sorted_samples = sorted(self.samples) k = (len(sorted_samples) - 1) * p f = int(k) c = k - f if f + 1 >= len(sorted_samples): return sorted_samples[-1] return sorted_samples[f] * (1 - c) + sorted_samples[f + 1] * c @property def p50(self) -> float: return self.percentile(0.50) @property def p95(self) -> float: return self.percentile(0.95) @property def p99(self) -> float: return self.percentile(0.99) @property def avg(self) -> float: return sum(self.samples) / len(self.samples) if self.samples else 0.0 class LatencyTracker: """异步延迟追踪器，按操作类型分别统计。""" def __init__(self): self._stats: dict[str, LatencyStats] = defaultdict(LatencyStats) self._lock = asyncio.Lock() async def record( self, operation: str, latency_ms: float ) -> None: async with self._lock: self._stats[operation].record(latency_ms) async def get_stats(self) -> dict[str, dict]: async with self._lock: return { op: { "p50": s.p50, "p95": s.p95, "p99": s.p99, "avg": s.avg, "count": len(s.samples), } for op, s in self._stats.items() } async def check_sla( self, sla_targets: dict[str, dict[str, float]] ) -> list[str]: """检查是否超越 SLA 阈值，返回告警列表。""" alerts = [] stats = await self.get_stats() for op, targets in sla_targets.items(): if op not in stats: continue s = stats[op] for pct, threshold in targets.items(): actual = s.get(f"p{pct[1:]}", 0.0) if actual > threshold: alerts.append( f"{op} P{pct[1:]}={actual:.0f}ms > " f"阈值 {threshold:.0f}ms" ) return alerts

3.2 在检索链路中集成延迟追踪

import functools from typing import Callable, Awaitable def track_latency( tracker: LatencyTracker, operation: str ): """延迟追踪装饰器。""" def decorator(func: Callable[..., Awaitable]): @functools.wraps(func) async def wrapper(*args, **kwargs): start = time.monotonic() try: result = await func(*args, **kwargs) return result finally: elapsed = (time.monotonic() - start) * 1000 await tracker.record(operation, elapsed) return wrapper return decorator # 使用示例 class RAGSearchPipeline: def __init__(self, tracker: LatencyTracker): self.tracker = tracker @track_latency # 报错：缺少参数 — 应使用闭包 async def dummy(self): ... async def search(self, query: str) -> list[dict]: t0 = time.monotonic() # 1. Embedding t1 = time.monotonic() embedding = await self._get_embedding(query) await self.tracker.record("embedding", (time.monotonic() - t1) * 1000) # 2. 向量检索 t2 = time.monotonic() docs = await self._vector_search(embedding) await self.tracker.record("vector_search", (time.monotonic() - t2) * 1000) # 3. 重排序 t3 = time.monotonic() ranked = await self._rerank(query, docs) await self.tracker.record("rerank", (time.monotonic() - t3) * 1000) # 4. 整体延迟 await self.tracker.record( "total_retrieval", (time.monotonic() - t0) * 1000 ) return ranked

3.3 SLA 告警器

class SLAAlertManager: """定期检查 SLA 并触发告警。""" def __init__( self, tracker: LatencyTracker, sla_targets: dict, alert_channel, # 如 Slack Webhook / 邮件 check_interval: int = 60, ): self.tracker = tracker self.sla = sla_targets self.channel = alert_channel self.interval = check_interval async def run(self) -> None: SLA_TARGETS = { "embedding": {"p95": 200, "p99": 500}, "vector_search": {"p95": 100, "p99": 300}, "rerank": {"p95": 300, "p99": 600}, "total_retrieval": {"p95": 800, "p99": 2000}, } self.sla = SLA_TARGETS while True: try: alerts = await self.tracker.check_sla(self.sla) if alerts: await self._send_alert(alerts) except Exception: pass # 告警本身不应该影响主流程 await asyncio.sleep(self.interval) async def _send_alert(self, alerts: list[str]) -> None: # 实际实现中这里发送到 Slack / PagerDuty for alert in alerts: print(f"[SLA ALERT] {alert}")

四、边界分析与架构权衡

4.1 采样率与精度

记录所有请求的分位数需要消耗内存。对于高流量服务（QPS > 1000），可以使用以下策略：

• 采样：只记录 10% 的请求用于分位数计算。对于 P50/P95 精度影响不大，对 P99 影响稍大。
• 近似算法：使用 t-digest 或 HdrHistogram 等近似分位数算法，内存占用 O(k)，k 是精度参数。
• 滚动窗口：只保留最近 1 小时的样本，旧数据自动丢弃。

4.2 多个 SLA 目标的优先级

如果 P95 达标但 P99 超标了，应该怎么排优先级？

P95 优先修复，因为它影响更大用户群。P99 超标通常只是因为"个别超慢请求"，可能是网络抖动或 GC 暂停——这些也需要修复，但优先级低于 P95 的全局性问题。

4.3 延迟与成本的权衡

降低 P99 通常比降低 P50 贵好几个数量级。从 P50 200ms 降到 100ms 可能只需要换个 Embedding 模型。但把 P99 从 2000ms 降到 1000ms，可能需要做请求复制（hedged requests）、增加超时重试、甚至换硬件。

在设定 SLA 目标时，要考虑成本。不要追 P99.999——那通常是给支付系统用的，RAG 系统 P99 < 2s 已经足够好了。

五、总结

平均值是监控的最大敌人。它把长尾问题完美地隐藏起来，让你的监控面板看起来很漂亮，但用户的体验却很糟糕。

RAG 延迟 SLA 的正确姿势：

1. 用分位数替代平均值：P50 看典型体验，P95 做 SLA 核心，P99 看长尾。
2. 分层监控：区分 Embedding 延迟、检索延迟、重排序延迟、LLM 生成延迟——每个环节独立设 SLA。
3. 自动告警：别等人肉看面板，让系统在 SLA 被突破时自动告警。
4. 接受现实：P99 永远比 P95 难优化，合理设定目标，控制成本。

我把这套方案落地后，最直观的变化是：SLA 达标率从"不知道多少"变成了清楚的 99.2%。哪些环节慢、慢多少、需要优化什么，都有了数据支撑。再也不用跟老板说"可能是 XXX 慢"——我可以直接拿出 P99 的说："是重排序环节慢了，P99 超标了 200ms，给我两天优化。"

下一篇预告：检索质量在线监控，让你的 RAG 系统不会悄悄"变笨"。