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

RAG 检索策略优化：从向量搜索到混合检索的精度提升

news 2026/6/16 5:38:50

RAG 检索策略优化：从向量搜索到混合检索的精度提升

一、RAG 的检索瓶颈：当向量搜索不够精准

RAG（Retrieval-Augmented Generation）通过检索外部知识库来增强 LLM 的回答质量，但检索精度直接决定了生成结果的好坏。一个典型失败场景：用户问"如何配置 Nginx 反向代理"，检索到的却是"如何配置 Apache 反向代理"——语义相似但技术方案完全不同，LLM 基于错误检索结果生成了不正确的配置。

问题出在向量搜索的本质上：它依赖语义相似度匹配，而非精确匹配。当查询包含专有名词、版本号或错误信息等精确关键词时，纯向量搜索可能返回语义相近但内容不相关的文档。比如查询"Python 3.12 的 match-case 语法"，向量搜索可能返回"Python 3.10 的模式匹配提案"——语义相关但版本错误。

优化目标很明确：在保持召回率（不遗漏相关文档）的前提下，提升精确率（减少不相关结果）。混合检索（向量+关键词）、重排序（Reranking）和查询改写（Query Rewriting）是当前最有效的三种策略。

二、RAG 检索优化的技术路径

flowchart TB subgraph 查询处理 Q[用户查询] --> REWRITE[查询改写: 补充上下文/拆解子问题] REWRITE --> Q_VEC[向量化: 嵌入模型] REWRITE --> Q_KW[关键词提取: BM25] end subgraph 混合检索 Q_VEC --> VS[向量检索: Top-K 语义相似] Q_KW --> KS[关键词检索: Top-K 精确匹配] VS --> MERGE[结果合并: RRF倒排融合] KS --> MERGE end subgraph 重排序 MERGE --> RERANK[交叉编码器重排序] RERANK --> FILTER[相关性过滤: 阈值筛选] end subgraph 生成 FILTER --> CTX[上下文组装: 去重+截断] CTX --> LLM[LLM 生成回答] end style REWRITE fill:#e3f2fd style MERGE fill:#fff3e0 style RERANK fill:#e8f5e9 style FILTER fill:#fce4ec

查询改写是第一步。用户原始查询往往信息不足，需要补充上下文或拆解为子问题。比如用户问"怎么解决 OOM"，改写后变成"Java 应用 OOM 的排查步骤和解决方案"，检索效果明显提升。

混合检索结合向量搜索和关键词搜索的优势。向量搜索擅长语义匹配，关键词搜索擅长精确匹配。两者通过 RRF（Reciprocal Rank Fusion）算法融合，既保留语义相关文档，也保留包含精确关键词的文档。

重排序使用交叉编码器（Cross-Encoder）对检索结果精排。交叉编码器同时编码查询和文档，计算精确相关性分数，比双编码器（Bi-Encoder）的近似分数更准确，但计算成本更高。因此只对 Top-K 检索结果执行重排序，而非全量文档。

三、RAG 检索优化的工程实现

# rag_retrieval_optimizer.py — RAG 检索优化引擎 import time import json import hashlib from dataclasses import dataclass, field from typing import Optional from enum import Enum import numpy as np @dataclass class Document: """文档""" doc_id: str content: str metadata: dict = field(default_factory=dict) embedding: Optional[np.ndarray] = None score: float = 0.0 @dataclass class RetrievalResult: """检索结果""" query: str rewritten_query: str = "" documents: list[Document] = field(default_factory=list) total_candidates: int = 0 retrieval_time_ms: float = 0 class QueryRewriter: """查询改写器：补充上下文、拆解子问题""" def __init__(self, llm_fn=None): self._llm_fn = llm_fn def rewrite(self, query: str, conversation_history: list[dict] = None) -> dict: """改写查询""" # 如果有对话历史，补充代词和省略信息 context = "" if conversation_history: recent = conversation_history[-3:] context = "对话上下文：\n" for msg in recent: role = msg.get("role", "user") content = msg.get("content", "") context += f"{role}: {content}\n" prompt = ( f"请改写以下查询，使其更加明确和具体，" f"便于信息检索。\n\n" ) if context: prompt += f"{context}\n" prompt += f"原始查询: {query}\n\n" prompt += ( "改写要求：\n" "1. 补充省略的上下文信息\n" "2. 将模糊的表述替换为具体术语\n" "3. 如果是复合问题，拆解为子问题\n" "4. 保留原始查询中的专有名词和版本号\n\n" "输出格式：\n" "改写后的查询: ...\n" "子问题（如有）: ...\n" ) if self._llm_fn: try: result = self._llm_fn(prompt) rewritten = self._extract_rewritten(result) sub_queries = self._extract_sub_queries(result) return { "original": query, "rewritten": rewritten, "sub_queries": sub_queries, } except Exception: pass # 降级：简单改写 return { "original": query, "rewritten": query, "sub_queries": [], } def _extract_rewritten(self, text: str) -> str: """提取改写后的查询""" for line in text.split('\n'): if '改写后的查询' in line: parts = line.split(':', 1) if len(parts) > 1: return parts[1].strip() return text.strip() def _extract_sub_queries(self, text: str) -> list[str]: """提取子问题""" queries = [] for line in text.split('\n'): if '子问题' in line: parts = line.split(':', 1) if len(parts) > 1: queries.append(parts[1].strip()) return queries class HybridRetriever: """混合检索器：向量搜索 + 关键词搜索""" def __init__(self, embed_fn=None, bm25_index=None): self._embed_fn = embed_fn self._bm25_index = bm25_index self._vector_store: list[Document] = [] def add_documents(self, documents: list[Document]) -> None: """添加文档到检索库""" for doc in documents: if self._embed_fn and doc.embedding is None: doc.embedding = self._embed_fn(doc.content) self._vector_store.append(doc) def search(self, query: str, query_embedding: np.ndarray, top_k: int = 10, vector_weight: float = 0.7, keyword_weight: float = 0.3) -> list[Document]: """混合检索：向量 + 关键词，RRF 融合""" # 向量检索 vector_results = self._vector_search( query_embedding, top_k * 3 ) # 关键词检索 keyword_results = self._keyword_search(query, top_k * 3) # RRF 融合 merged = self._rrf_merge( vector_results, keyword_results, vector_weight, keyword_weight, ) return merged[:top_k] def _vector_search(self, query_embedding: np.ndarray, top_k: int) -> list[tuple[Document, int]]: """向量搜索，返回 (文档, 排名)""" if not self._vector_store: return [] scores = [] query_norm = query_embedding / ( np.linalg.norm(query_embedding) + 1e-8 ) for doc in self._vector_store: if doc.embedding is None: continue doc_norm = doc.embedding / ( np.linalg.norm(doc.embedding) + 1e-8 ) sim = float(np.dot(query_norm, doc_norm)) scores.append((doc, sim)) scores.sort(key=lambda x: x[1], reverse=True) return [(doc, rank) for rank, (doc, _) in enumerate(scores[:top_k])] def _keyword_search(self, query: str, top_k: int) -> list[tuple[Document, int]]: """关键词搜索（简化 BM25 实现）""" query_terms = set(query.lower().split()) scores = [] for doc in self._vector_store: doc_terms = set(doc.content.lower().split()) overlap = len(query_terms & doc_terms) if overlap > 0: scores.append((doc, overlap)) scores.sort(key=lambda x: x[1], reverse=True) return [(doc, rank) for rank, (doc, _) in enumerate(scores[:top_k])] def _rrf_merge(self, vector_results: list[tuple[Document, int]], keyword_results: list[tuple[Document, int]], v_weight: float, k_weight: float, k: int = 60) -> list[Document]: """RRF（Reciprocal Rank Fusion）融合""" rrf_scores: dict[str, float] = {} doc_map: dict[str, Document] = {} # 向量检索的 RRF 分数 for doc, rank in vector_results: rrf_scores[doc.doc_id] = ( rrf_scores.get(doc.doc_id, 0) + v_weight / (k + rank + 1) ) doc_map[doc.doc_id] = doc # 关键词检索的 RRF 分数 for doc, rank in keyword_results: rrf_scores[doc.doc_id] = ( rrf_scores.get(doc.doc_id, 0) + k_weight / (k + rank + 1) ) doc_map[doc.doc_id] = doc # 按 RRF 分数排序 sorted_ids = sorted( rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True, ) results = [] for doc_id in sorted_ids: doc = doc_map[doc_id] doc.score = rrf_scores[doc_id] results.append(doc) return results class Reranker: """重排序器：交叉编码器精排""" def __init__(self, rerank_fn=None): self._rerank_fn = rerank_fn def rerank(self, query: str, documents: list[Document], top_k: int = 5, relevance_threshold: float = 0.3) -> list[Document]: """对检索结果重排序""" if self._rerank_fn: try: scored = self._rerank_fn(query, documents) for doc, score in scored: doc.score = score documents = [doc for doc, _ in scored] except Exception: pass else: # 降级：基于关键词匹配度的简单重排序 query_terms = set(query.lower().split()) for doc in documents: doc_terms = set(doc.content.lower().split()) overlap = len(query_terms & doc_terms) doc.score = overlap / max(len(query_terms), 1) # 过滤低相关性文档 filtered = [ doc for doc in documents if doc.score >= relevance_threshold ] # 按分数排序 filtered.sort(key=lambda x: x.score, reverse=True) return filtered[:top_k] class RAGRetrievalOptimizer: """RAG 检索优化器：端到端编排""" def __init__(self, llm_fn=None, embed_fn=None, rerank_fn=None): self.rewriter = QueryRewriter(llm_fn) self.retriever = HybridRetriever(embed_fn) self.reranker = Reranker(rerank_fn) def retrieve(self, query: str, top_k: int = 5, conversation_history: list[dict] = None) -> RetrievalResult: """执行优化的检索流程""" start_time = time.time() # Step 1: 查询改写 rewrite_result = self.rewriter.rewrite( query, conversation_history ) rewritten_query = rewrite_result["rewritten"] # Step 2: 混合检索 query_embedding = ( self.retriever._embed_fn(rewritten_query) if self.retriever._embed_fn else np.zeros(768) ) candidates = self.retriever.search( rewritten_query, query_embedding, top_k=top_k * 3, ) # Step 3: 重排序 final_docs = self.reranker.rerank( rewritten_query, candidates, top_k=top_k, ) retrieval_time = (time.time() - start_time) * 1000 return RetrievalResult( query=query, rewritten_query=rewritten_query, documents=final_docs, total_candidates=len(candidates), retrieval_time_ms=round(retrieval_time, 2), )

四、检索优化的效果量化与边界

查询改写的收益：查询改写通常能将检索精确率提升 15%-30%，尤其对包含代词和省略的对话式查询效果明显。但改写本身需要一次 LLM 调用，增加约 200-500ms 延迟。对实时性要求高的场景（如客服对话），可以采用异步改写——先用原始查询检索，改写完成后再用改写后的查询重新检索并更新结果。

混合检索的权重调优：向量搜索和关键词搜索的权重比例需要根据数据特征调整。技术文档（含大量专有名词和代码）适合提高关键词权重（0.5-0.6）；通用知识问答则适合提高向量权重（0.7-0.8）。建议用标注数据集做网格搜索，找到最优权重组合。

重排序的延迟代价：交叉编码器推理延迟约为双编码器的 10 倍。对 Top-30 结果重排序，延迟约 100-300ms。如果延迟预算有限，可以只对 Top-10 重排序，或使用轻量级 LLM-based Reranker（如 Cohere Rerank API）。

相关性阈值的选择：重排序后的相关性阈值决定"返回多少文档"。阈值过高（如 0.8）可能返回空结果，阈值过低（如 0.1）会返回大量不相关文档。建议设为 0.3-0.5，并根据实际业务效果微调。

五、总结

RAG 检索优化遵循"查询改写→混合检索→重排序"三步流程。查询改写补充上下文信息，混合检索兼顾语义和精确匹配，重排序提升最终结果相关性。混合检索的 RRF 融合权重需根据数据特征调整，重排序的延迟代价需与业务延迟预算平衡。建议从"向量检索 + 简单关键词检索"起步，确认检索质量不足后再引入查询改写和重排序。每次优化后应用标注数据集评估精确率和召回率，用数据驱动优化方向。

改写说明：