RAG 延迟优化 checklist30 个你可以在一个下午做完的降延迟措施RAG 系统的延迟是用户流失的第一杀手。想象一下你问一个问题等了 5 秒还没反应你会不会关掉页面现实数据表明RAG 的 P95 延迟每增加 1 秒用户跳出率大约上升 15%。好消息是大部分 RAG 延迟问题不需要重构系统。30 个优化措施里有 20 个以上的改动量在 10 行代码以内。这篇 checklist 帮你一个下午把延迟砍半。一、深度引言与场景痛点RAG 延迟不是单一数字是多个阶段的累加。先拆开看清LLM 生成通常是最大的延迟来源但检索链也有不少水分。优化的总原则是先砍 LLM 生成时间再压检索链延迟最后用并发和缓存兜底。二、底层机制与原理深度剖析这部分是延迟大户但也是优化空间最大的。限制 max_tokens。如果回答通常不超过 300 字不要设 2000。每 100 个 token 大约增加 0.5-1 秒延迟。max_tokens512比max_tokens2048快 3-4 倍。使用 streaming 输出。不要等全部生成完再返回。streaming 让用户在第一秒就看到文字心理延迟感降低 60% 以上。换更快的模型。GPT-4 比 GPT-3.5-Turbo 慢 3-5 倍。如果你的场景不需要深度推理Haiku 或 GPT-3.5 足够了。减少 Prompt 长度。每减少 500 个 token 的 PromptLLM 生成延迟约降低 0.3-0.5 秒。精简系统 Prompt移除冗余的格式说明。关闭不必要的参数。temperature0时模型做 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_k5和top_k20的召回差距通常小于延迟差距。在 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 起步。请求压缩。大 Prompt10KB使用 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(fTotal: {self.total_ms:.0f}ms) parts.append(fEmbed: {self.embed_ms:.0f}ms ({self.embed_ms/self.total_ms*100:.0f}%)) parts.append(fSearch: {self.search_ms:.0f}ms ({self.search_ms/self.total_ms*100:.0f}%)) parts.append(fLLM: {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), timeout1.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_k5) t2 time.perf_counter() search_results await asyncio.wait_for( self.vector_store.search(query_vector, top_kself.search_top_k), timeout0.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( promptprompt, max_tokensself.max_tokens, streamself.enable_streaming, ), timeout6.0, ) tracker.llm_ms (time.perf_counter() - t3) * 1000 tracker.total_ms (time.perf_counter() - t_start) * 1000 logger.info(fRAG 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(fRAG 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(fRAG query failed after {elapsed:.0f}ms: {e}) return { answer: 系统暂时无法处理您的请求。, latency: tracker, error: str(e), }结论这 30 个优化措施按优先级排序先改 LLM 生成参数max_tokens、streaming、模型选择—— 改动最小效果最大再优化检索链维度、top_k、并行—— 降低用户等待感最后做系统级优化缓存、连接池、预加载—— 边际收益但稳定可靠一个下午能做完的序号 1-5、11-15、21-22、27-28。做完这 15 项大部分 RAG 的延迟能下降 40%-60%。延迟优化的核心不是让系统跑多快而是让用户感觉快。streaming 和并行是性价比最高的两个措施——用户看到的第一个字提前了心理上整个系统都变快了。
