RAG 检索结果总是不相关我重构了检索架构后命中率提了 40%前言我们团队搞 RAG 半年了最大的痛就是检索不相关。用户问 A搜出文档 B大模型就胡诌。后来我们重构了检索架构做了分层检索和相关性重排序命中率从 60% 提到了 85%。今天聊聊工程落地的经验。一、底层原理1.1 RAG 检索为什么不准RAG 的检索环节是决定输出质量的关键graph TD A[用户提问] -- B[向量检索] B -- C[Top-K 文档] C -- D{相关性够吗} D --|不够| E[大模型缺乏依据] E -- F[输出幻觉内容] D --|够| G[准确回答] H[优化点] -- I[多路召回] H -- J[重排序] H -- K[查询改写]主要问题单一路径检索覆盖不够向量相似度不一定等于语义相关检索到的文档噪音多大模型被噪音误导1.2 常见检索方案对比方案召回率精确率实现难度纯向量检索中中低纯关键词检索低高低混合检索高高中分层检索高高中二、快速上手先看基础版 RAGfrom typing import List, Dict class SimpleRAG: def __init__(self, retriever, llm): self.retriever retriever self.llm llm def query(self, question): # 单一路径检索 docs self.retriever.retrieve(question, k5) context \n\n.join(docs) prompt f基于以下内容回答问题\n{context}\n\n问题{question} return self.llm(prompt)再看改进版混合检索class HybridRetriever: def __init__(self, vector_store, keyword_store): self.vector_store vector_store self.keyword_store keyword_store def retrieve(self, query, k5): # 向量检索 vector_results self.vector_store.similarity_search(query, kk) # 关键词检索 keyword_results self.keyword_store.search(query, kk) # 融合去重 combined self.fusion(vector_results, keyword_results) return combined[:k] def fusion(self, *results): # 简单融合按分数加权 doc_scores {} for i, docs in enumerate(results): weight 1.0 / (i 1) for doc in docs: doc_id doc.get(id) score doc.get(score, 0) * weight doc_scores[doc_id] doc_scores.get(doc_id, 0) score sorted_docs sorted(doc_scores.items(), keylambda x: x[1], reverseTrue) return [{id: k, score: v} for k, v in sorted_docs]三、核心 API / 深水区3.1 RAG 优化策略速查策略解决的问题效果查询改写用户表述不清晰中混合检索召回率不够高重排序精确率低高切片策略文档太大中元数据过滤噪音多高3.2 查询改写的实现class QueryRewriter: def __init__(self, llm): self.llm llm def rewrite(self, query): prompt f将用户的问题改写成更适合检索的格式 原始问题{query} 改写规则 1. 提取核心关键词 2. 补充同义词 3. 组织成多个子查询 只返回改写后的查询不要其他内容 return self.llm(prompt) def rewrite_multi(self, query, n3): prompt f将用户问题改写成{n}个不同的检索查询 原始问题{query} 返回格式每行一个查询不要序号 result self.llm(prompt) return [q.strip() for q in result.split(\n) if q.strip()][:n]3.3 相关性重排序class ReRanker: def __init__(self, llm): self.llm llm def rerank(self, query, docs, k3): scored_results [] for doc in docs: prompt f判断文档与问题的相关性0-10分 问题{query} 文档内容{doc[:200]} 只返回分数 score_text self.llm(prompt) try: score float(score_text.strip()) except: score 0 scored_results.append((score, doc)) scored_results.sort(keylambda x: x[0], reverseTrue) return [doc for _, doc in scored_results[:k]]四、实战演练完整的优化版 RAG 系统from typing import List, Dict, Any, Optional import json class OptimizedRAG: def __init__(self, vector_store, keyword_store, llm): self.retriever HybridRetriever(vector_store, keyword_store) self.rewriter QueryRewriter(llm) self.reranker ReRanker(llm) self.llm llm def query(self, question: str) - Dict[str, Any]: # 1. 查询改写 queries self.rewriter.rewrite_multi(question) all_queries [question] queries # 2. 多路检索 all_docs [] for q in all_queries: docs self.retriever.retrieve(q, k10) all_docs.extend(docs) # 3. 去重 seen set() unique_docs [] for doc in all_docs: if doc[id] not in seen: seen.add(doc[id]) unique_docs.append(doc) # 4. 重排序 doc_contents [d.get(content, ) for d in unique_docs] top_docs self.reranker.rerank(question, doc_contents, k3) # 5. 构建上下文 context \n\n.join(top_docs) # 6. 生成回答 prompt f基于以下参考资料回答问题。 如果参考资料不足以回答问题请直接说不知道。 参考资料 {context} 问题{question} 回答 answer self.llm(prompt) # 7. 回答后验证 verification self._verify_answer(answer, context, question) return { answer: answer, sources: top_docs, verified: verification } def _verify_answer(self, answer, context, question): prompt f验证回答是否基于提供的资料 资料{context[:500]} 回答{answer} 如果回答基于资料返回 true否则返回 false result self.llm(prompt) return true in result.lower()五、避坑指南与最佳实践 **技巧多路召回 重排序单一检索方式不够混合才靠谱。⚠️ **警告不要喂太多文档Top-5 就够了多了反而是噪音。✅ **推荐做回答后验证用 LLM 自检看回答是否基于检索结果。六、综合实战演示生产级 RAG 管道import time import hashlib from typing import List, Dict, Any from dataclasses import dataclass dataclass class RAGResult: query: str answer: str sources: List[str] latency: float confidence: float class ProductionRAG: def __init__(self, retriever, llm, cache_size1000): self.retriever retriever self.llm llm self.cache {} self.cache_size cache_size def query(self, question: str) - RAGResult: start time.time() # 检查缓存 cache_key hashlib.md5(question.encode()).hexdigest() if cache_key in self.cache: cached self.cache[cache_key] return RAGResult( queryquestion, answercached[answer], sourcescached[sources], latency0, confidencecached.get(confidence, 0.9) ) # 检索 docs self.retriever.retrieve(question, k5) doc_contents [d.get(content, ) for d in docs] context \n\n.join(doc_contents) # 生成 prompt f基于资料回答\n{context}\n\n问题{question} answer self.llm(prompt) # 缓存 if len(self.cache) self.cache_size: self.cache[cache_key] { answer: answer, sources: doc_contents[:2], confidence: 0.8 } return RAGResult( queryquestion, answeranswer, sourcesdoc_contents[:2], latencytime.time() - start, confidence0.8 ) rag ProductionRAG(HybridRetriever(...), llm) result rag.query(今天天气怎么样) print(f回答: {result.answer})七、总结RAG 检索优化不是单一手段能搞定的多路检索 重排序查询改写提高命中回答后验证防幻觉缓存加速搞好了这些RAG 系统就能在真实场景中落地了。
RAG 检索结果总是不相关？我重构了检索架构后命中率提了 40%
RAG 检索结果总是不相关我重构了检索架构后命中率提了 40%前言我们团队搞 RAG 半年了最大的痛就是检索不相关。用户问 A搜出文档 B大模型就胡诌。后来我们重构了检索架构做了分层检索和相关性重排序命中率从 60% 提到了 85%。今天聊聊工程落地的经验。一、底层原理1.1 RAG 检索为什么不准RAG 的检索环节是决定输出质量的关键graph TD A[用户提问] -- B[向量检索] B -- C[Top-K 文档] C -- D{相关性够吗} D --|不够| E[大模型缺乏依据] E -- F[输出幻觉内容] D --|够| G[准确回答] H[优化点] -- I[多路召回] H -- J[重排序] H -- K[查询改写]主要问题单一路径检索覆盖不够向量相似度不一定等于语义相关检索到的文档噪音多大模型被噪音误导1.2 常见检索方案对比方案召回率精确率实现难度纯向量检索中中低纯关键词检索低高低混合检索高高中分层检索高高中二、快速上手先看基础版 RAGfrom typing import List, Dict class SimpleRAG: def __init__(self, retriever, llm): self.retriever retriever self.llm llm def query(self, question): # 单一路径检索 docs self.retriever.retrieve(question, k5) context \n\n.join(docs) prompt f基于以下内容回答问题\n{context}\n\n问题{question} return self.llm(prompt)再看改进版混合检索class HybridRetriever: def __init__(self, vector_store, keyword_store): self.vector_store vector_store self.keyword_store keyword_store def retrieve(self, query, k5): # 向量检索 vector_results self.vector_store.similarity_search(query, kk) # 关键词检索 keyword_results self.keyword_store.search(query, kk) # 融合去重 combined self.fusion(vector_results, keyword_results) return combined[:k] def fusion(self, *results): # 简单融合按分数加权 doc_scores {} for i, docs in enumerate(results): weight 1.0 / (i 1) for doc in docs: doc_id doc.get(id) score doc.get(score, 0) * weight doc_scores[doc_id] doc_scores.get(doc_id, 0) score sorted_docs sorted(doc_scores.items(), keylambda x: x[1], reverseTrue) return [{id: k, score: v} for k, v in sorted_docs]三、核心 API / 深水区3.1 RAG 优化策略速查策略解决的问题效果查询改写用户表述不清晰中混合检索召回率不够高重排序精确率低高切片策略文档太大中元数据过滤噪音多高3.2 查询改写的实现class QueryRewriter: def __init__(self, llm): self.llm llm def rewrite(self, query): prompt f将用户的问题改写成更适合检索的格式 原始问题{query} 改写规则 1. 提取核心关键词 2. 补充同义词 3. 组织成多个子查询 只返回改写后的查询不要其他内容 return self.llm(prompt) def rewrite_multi(self, query, n3): prompt f将用户问题改写成{n}个不同的检索查询 原始问题{query} 返回格式每行一个查询不要序号 result self.llm(prompt) return [q.strip() for q in result.split(\n) if q.strip()][:n]3.3 相关性重排序class ReRanker: def __init__(self, llm): self.llm llm def rerank(self, query, docs, k3): scored_results [] for doc in docs: prompt f判断文档与问题的相关性0-10分 问题{query} 文档内容{doc[:200]} 只返回分数 score_text self.llm(prompt) try: score float(score_text.strip()) except: score 0 scored_results.append((score, doc)) scored_results.sort(keylambda x: x[0], reverseTrue) return [doc for _, doc in scored_results[:k]]四、实战演练完整的优化版 RAG 系统from typing import List, Dict, Any, Optional import json class OptimizedRAG: def __init__(self, vector_store, keyword_store, llm): self.retriever HybridRetriever(vector_store, keyword_store) self.rewriter QueryRewriter(llm) self.reranker ReRanker(llm) self.llm llm def query(self, question: str) - Dict[str, Any]: # 1. 查询改写 queries self.rewriter.rewrite_multi(question) all_queries [question] queries # 2. 多路检索 all_docs [] for q in all_queries: docs self.retriever.retrieve(q, k10) all_docs.extend(docs) # 3. 去重 seen set() unique_docs [] for doc in all_docs: if doc[id] not in seen: seen.add(doc[id]) unique_docs.append(doc) # 4. 重排序 doc_contents [d.get(content, ) for d in unique_docs] top_docs self.reranker.rerank(question, doc_contents, k3) # 5. 构建上下文 context \n\n.join(top_docs) # 6. 生成回答 prompt f基于以下参考资料回答问题。 如果参考资料不足以回答问题请直接说不知道。 参考资料 {context} 问题{question} 回答 answer self.llm(prompt) # 7. 回答后验证 verification self._verify_answer(answer, context, question) return { answer: answer, sources: top_docs, verified: verification } def _verify_answer(self, answer, context, question): prompt f验证回答是否基于提供的资料 资料{context[:500]} 回答{answer} 如果回答基于资料返回 true否则返回 false result self.llm(prompt) return true in result.lower()五、避坑指南与最佳实践 **技巧多路召回 重排序单一检索方式不够混合才靠谱。⚠️ **警告不要喂太多文档Top-5 就够了多了反而是噪音。✅ **推荐做回答后验证用 LLM 自检看回答是否基于检索结果。六、综合实战演示生产级 RAG 管道import time import hashlib from typing import List, Dict, Any from dataclasses import dataclass dataclass class RAGResult: query: str answer: str sources: List[str] latency: float confidence: float class ProductionRAG: def __init__(self, retriever, llm, cache_size1000): self.retriever retriever self.llm llm self.cache {} self.cache_size cache_size def query(self, question: str) - RAGResult: start time.time() # 检查缓存 cache_key hashlib.md5(question.encode()).hexdigest() if cache_key in self.cache: cached self.cache[cache_key] return RAGResult( queryquestion, answercached[answer], sourcescached[sources], latency0, confidencecached.get(confidence, 0.9) ) # 检索 docs self.retriever.retrieve(question, k5) doc_contents [d.get(content, ) for d in docs] context \n\n.join(doc_contents) # 生成 prompt f基于资料回答\n{context}\n\n问题{question} answer self.llm(prompt) # 缓存 if len(self.cache) self.cache_size: self.cache[cache_key] { answer: answer, sources: doc_contents[:2], confidence: 0.8 } return RAGResult( queryquestion, answeranswer, sourcesdoc_contents[:2], latencytime.time() - start, confidence0.8 ) rag ProductionRAG(HybridRetriever(...), llm) result rag.query(今天天气怎么样) print(f回答: {result.answer})七、总结RAG 检索优化不是单一手段能搞定的多路检索 重排序查询改写提高命中回答后验证防幻觉缓存加速搞好了这些RAG 系统就能在真实场景中落地了。
