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

RAG 延迟优化 checklist：30 个你可以在一个下午做完的降延迟措施

news 2026/7/29 15:22:07

RAG 延迟优化 checklist：30 个你可以在一个下午做完的降延迟措施

RAG 系统的延迟是用户流失的第一杀手。想象一下：你问一个问题，等了 5 秒还没反应，你会不会关掉页面？现实数据表明，RAG 的 P95 延迟每增加 1 秒，用户跳出率大约上升 15%。

好消息是，大部分 RAG 延迟问题不需要重构系统。30 个优化措施里，有 20 个以上的改动量在 10 行代码以内。这篇 checklist 帮你一个下午把延迟砍半。

一、深度引言与场景痛点

RAG 延迟不是单一数字，是多个阶段的累加。先拆开看清：

LLM 生成通常是最大的延迟来源，但检索链也有不少水分。优化的总原则是：先砍 LLM 生成时间，再压检索链延迟，最后用并发和缓存兜底。

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

这部分是延迟大户，但也是优化空间最大的。

限制 max_tokens。如果回答通常不超过 300 字，不要设 2000。每 100 个 token 大约增加 0.5-1 秒延迟。max_tokens=512比max_tokens=2048快 3-4 倍。
使用 streaming 输出。不要等全部生成完再返回。streaming 让用户在第一秒就看到文字，心理延迟感降低 60% 以上。
换更快的模型。GPT-4 比 GPT-3.5-Turbo 慢 3-5 倍。如果你的场景不需要深度推理，Haiku 或 GPT-3.5 足够了。
减少 Prompt 长度。每减少 500 个 token 的 Prompt，LLM 生成延迟约降低 0.3-0.5 秒。精简系统 Prompt，移除冗余的格式说明。
关闭不必要的参数。temperature=0时模型做 greedy decoding，比 sampling 模式快 10-20%。
使用 prompt caching。Anthropic 和部分 OpenAI 模型支持 Prompt 缓存。固定的系统 Prompt 部分可以缓存，每次调用只消耗增量 token。
缩短上下文窗口。通过max_context_tokens限制送给模型的上下文长度。检索返回的 20 个 chunk 不需要全部喂给模型。
使用专用的轻量级模型。对于简单的分类、提取任务，用分类模型代替生成模型。分类模型延迟通常在 50-200ms。
批量处理离线任务。对于不需要实时响应的任务（如文档摘要、标签提取），用批量 API 异步处理。
模型端开启 speculative decoding。如果你自部署模型，可以用小模型预先猜测输出，大模型验证，加速 2-3 倍。

三、生产级代码实现

检索链延迟虽然不如 LLM 生成大，但优化后用户感知明显——因为它是"等待开始"的阶段。

减少 Embedding 维度。从 1536 维降到 768 维，编码和搜索速度都能翻倍。用 OpenAItext-embedding-3-small替代text-embedding-3-large。
使用向量索引加速。Milvus/Redis 的 HNSW 索引比 FLAT 搜索快 10-100 倍。EF_SEARCH参数从默认值调低到 40-80，平衡精度和速度。
检索结果裁剪。top_k=5和top_k=20的召回差距通常小于延迟差距。在 95% 的场景里，5 个结果足够。
Embedding 缓存。高频查询的 Embedding 结果缓存起来。用户问"怎么退款"和同事问"退款流程"，向量几乎一样，不需要重新编码。
并行检索。如果用了混合检索（向量 + BM25 + 知识图谱），三个检索器应该并发执行，不要串行。
粗排+精排两阶段。先用简单方法（向量相似度）召回 50 个候选，再用重排模型精排 top 5。两阶段比一次性精排快。
使用更快的 Embedding 模型。BGE-small-en 的编码速度是 BGE-large-en 的 3 倍，精度损失小于 3%。
文档分块优化。chunk_size 从 2000 减到 500，检索的每个 chunk 更短，向量检索更快，且给 LLM 的上下文更精炼。
提前终止搜索。设定相似度阈值（如 0.7），当已召回的结果相似度足够高时，停止搜索。
使用近似近邻替代精确搜索。ANNS 在百万级数据上比精确搜索快 1000 倍，精度损失 1-2%，完全可接受。

四、边界分析与架构权衡

连接池复用。HTTP 连接复用可以将 Embedding 和 LLM API 调用的连接建立时间从 50ms 降到 1ms。
使用 HTTP/2 或 gRPC。多路复用减少连接数，减少 TCP 握手开销。
Pre-warm 向量索引。启动时预加载索引到内存。冷启动第一次检索可能慢 10 倍，因为索引从磁盘加载到内存。
CDN 就近部署向量服务。如果用户在中国、向量服务在美国，网络延迟就 200ms 起步。
请求压缩。大 Prompt（>10KB）使用 gzip 压缩传输，减少网络传输时间。
异步非阻塞架构。用 asyncio 替代同步调用。同步等待一个 API 的时间可以用来发起另一个请求。
使用消息队列削峰。突发流量时，请求先入队列，由 Worker 池并行处理，避免 API 限流导致的排队。
超时和降级。设置合理的超时时间（总延迟 8 秒），超时后返回一个快速但可能不完美的答案。
预热模型推理引擎。自部署时，推理引擎（vLLM/TGI）的第一次推理需要加载模型到 GPU，延迟可能是正常推理的 10 倍。系统启动时做一次预热推理。
延迟分段监控。在 Embedding、检索、LLM 生成各阶段打点。不知道延迟在哪，优化就是盲人摸象。

结论

import asyncio import time from dataclasses import dataclass, field from typing import Any, Optional import logging logger = logging.getLogger(__name__) @dataclass class LatencyTracker: embed_ms: float = 0 search_ms: float = 0 rerank_ms: float = 0 llm_ms: float = 0 total_ms: float = 0 def summary(self) -> str: parts = [] if self.total_ms > 0: parts.append(f"Total: {self.total_ms:.0f}ms") parts.append(f"Embed: {self.embed_ms:.0f}ms ({self.embed_ms/self.total_ms*100:.0f}%)") parts.append(f"Search: {self.search_ms:.0f}ms ({self.search_ms/self.total_ms*100:.0f}%)") parts.append(f"LLM: {self.llm_ms:.0f}ms ({self.llm_ms/self.total_ms*100:.0f}%)") return " | ".join(parts) class OptimizedRAGPipeline: def __init__( self, embed_model, vector_store, llm_client, embedding_cache_size: int = 1000, search_top_k: int = 5, enable_streaming: bool = True, max_tokens: int = 512, ): self.embed_model = embed_model self.vector_store = vector_store self.llm_client = llm_client self.embedding_cache: dict[str, list[float]] = {} self.search_top_k = search_top_k self.enable_streaming = enable_streaming self.max_tokens = max_tokens async def query(self, question: str, timeout: float = 8.0) -> dict: tracker = LatencyTracker() t_start = time.perf_counter() try: # Phase 1: Embedding (with cache) — 措施 14 t1 = time.perf_counter() if question in self.embedding_cache: query_vector = self.embedding_cache[question] else: query_vector = await asyncio.wait_for( self.embed_model.encode(question), timeout=1.0 ) self.embedding_cache[question] = query_vector if len(self.embedding_cache) > 1000: self.embedding_cache.pop(next(iter(self.embedding_cache))) tracker.embed_ms = (time.perf_counter() - t1) * 1000 # Phase 2: Vector Search — 措施 13 (top_k=5) t2 = time.perf_counter() search_results = await asyncio.wait_for( self.vector_store.search(query_vector, top_k=self.search_top_k), timeout=0.5, ) tracker.search_ms = (time.perf_counter() - t2) * 1000 # Phase 3: Construct prompt — 措施 4 (精简 prompt) context = "\n".join([r.content for r in search_results]) prompt = f"基于以下信息回答问题。\n{context}\n\n问题：{question}" # Phase 4: LLM Generation with streaming — 措施 2 t3 = time.perf_counter() response = await asyncio.wait_for( self.llm_client.generate( prompt=prompt, max_tokens=self.max_tokens, stream=self.enable_streaming, ), timeout=6.0, ) tracker.llm_ms = (time.perf_counter() - t3) * 1000 tracker.total_ms = (time.perf_counter() - t_start) * 1000 logger.info(f"RAG query completed: {tracker.summary()}") return { "answer": response, "latency": tracker, "sources": [r.metadata for r in search_results], } except asyncio.TimeoutError: elapsed = (time.perf_counter() - t_start) * 1000 logger.warning(f"RAG query timeout after {elapsed:.0f}ms") return { "answer": "抱歉，查询超时，请稍后重试。", "latency": tracker, "error": "timeout", } except Exception as e: elapsed = (time.perf_counter() - t_start) * 1000 logger.error(f"RAG query failed after {elapsed:.0f}ms: {e}") return { "answer": "系统暂时无法处理您的请求。", "latency": tracker, "error": str(e), }