如何优化 RAG 系统架构：解决检索相关性与幻觉控制
RAG 检索增强生成的"幻觉"问题，本文通过多路召回解决了
前言
RAG 系统最大的问题是什么？不是检索不到，而是检索到了不相关的内容，导致大模型"胡诌"。
本文做了一个 RAG 问答系统，用户问 A，检索到文档 B，大模型就基于 B 回答。结果答非所问。后来用了多路召回 + 重排序，解决了这个问题。今天聊聊。
一、 底层原理
1.1 RAG 的检索"幻觉"
RAG 系统的问题不是大模型在幻觉，而是"上下文污染"：
graph TD A["用户提问"] --> B["单路检索"] B --> C["检索到文档"] C --> D{"相关？"} D -->|相关| E["正确回答"] D -->|不相关| F["大模型被误导"] F --> G["输出幻觉"] H["多路召回"] --> I["向量检索"] H --> J["关键词检索"] H --> K["语义检索"] I --> L["融合排序"] J --> L K --> L L --> M["高质量上下文"] M --> N["准确回答"]核心问题：
- 单路检索不够全面
- 检索噪音影响大模型
- 上下文质量决定回答质量
1.2 检索方案对比
| 方案 | 召回率 | 精确率 | 实现难度 |
|---|---|---|---|
| 单路向量 | 中 | 中 | 低 |
| 关键词 | 低 | 高 | 低 |
| 多路召回 | 高 | 中 | 中 |
| 多路+重排 | 高 | 高 | 中 |
二、 快速上手
2.1 基础 RAG
from typing import List, Dict class SimpleRAG: def __init__(self, vector_store, llm): self.vector_store = vector_store self.llm = llm def query(self, question: str) -> str: docs = self.vector_store.similarity_search(question, k=5) context = "\n\n".join(docs) prompt = f"基于以下内容回答问题：\n{context}\n\n问题: {question}" return self.llm(prompt)2.2 多路召回版
class MultiRouteRAG: def __init__(self, vector_store, keyword_store, llm): self.vector_store = vector_store self.keyword_store = keyword_store self.llm = llm def retrieve(self, query: str, k=5) -> List[str]: vector_results = self.vector_store.similarity_search(query, k=k) keyword_results = self.keyword_store.search(query, k=k) combined = self._fusion(vector_results, keyword_results) return combined[:k] def _fusion(self, *results): seen = set() merged = [] for docs in results: for doc in docs: if doc not in seen: seen.add(doc) merged.append(doc) return merged def query(self, question: str) -> str: docs = self.retrieve(question) context = "\n\n".join(docs[:3]) prompt = f"基于以下内容回答问题：\n{context}\n\n问题: {question}" return self.llm(prompt)三、 核心 API / 深水区
3.1 RAG 优化技巧速查
| 技术 | 解决的问题 | 效果 |
|---|---|---|
| 多路召回 | 召回率不够 | 高 |
| 重排序 | 精确率低 | 高 |
| 查询改写 | 表达不清晰 | 中 |
| 切片策略 | 文档太大 | 中 |
3.2 重排序
class ReRanker: def __init__(self, llm): self.llm = llm def rerank(self, query: str, docs: List[str], k=3) -> List[str]: scored = [] for doc in docs: prompt = f"评估文档与问题的相关度（0-10）：\n问题: {query}\n文档: {doc[:200]}" score_text = self.llm(prompt) try: score = float(score_text.strip()) except: score = 5 scored.append((score, doc)) scored.sort(key=lambda x: x[0], reverse=True) return [doc for _, doc in scored[:k]]3.3 查询改写
class QueryRewriter: def __init__(self, llm): self.llm = llm def rewrite(self, query: str) -> str: prompt = f"""把用户问题改写成更易检索的格式： 原问题: {query} 改写后: """ return self.llm(prompt) def expand(self, query: str, n=3) -> List[str]: prompt = f"生成 {n} 个不同表达的检索查询：{query}" result = self.llm(prompt) return [q.strip() for q in result.split("\n") if q.strip()][:n]四、 实战演练
4.1 完整的 RAG 问答系统
from typing import List, Dict, Any from dataclasses import dataclass @dataclass class RAGConfig: multi_route: bool = True rerank: bool = True query_rewrite: bool = True top_k: int = 5 class OptimizedRAG: def __init__(self, vector_store, keyword_store, llm, config: RAGConfig): self.vector_store = vector_store self.keyword_store = keyword_store self.llm = llm self.config = config self.reranker = ReRanker(llm) self.rewriter = QueryRewriter(llm) def query(self, question: str) -> Dict[str, Any]: queries = [question] if self.config.query_rewrite: rewritten = self.rewriter.rewrite(question) if rewritten: queries.append(rewritten) all_docs = [] for q in queries: if self.config.multi_route: docs = self._multi_retrieve(q) else: docs = self.vector_store.similarity_search(q, k=self.config.top_k) all_docs.extend(docs) unique_docs = list(dict.fromkeys(all_docs)) if self.config.rerank: unique_docs = self.reranker.rerank(question, unique_docs, k=3) else: unique_docs = unique_docs[:3] context = "\n\n".join(unique_docs) answer = self.llm(f"基于资料回答：\n{context}\n\n问题: {question}") return {"answer": answer, "sources": unique_docs[:2]} def _multi_retrieve(self, query: str) -> List[str]: vector_results = self.vector_store.similarity_search(query, k=self.config.top_k) keyword_results = self.keyword_store.search(query, k=self.config.top_k) seen = set() merged = [] for d in vector_results + keyword_results: if d not in seen: seen.add(d) merged.append(d) return merged config = RAGConfig() rag = OptimizedRAG(vector_store, keyword_store, llm, config) result = rag.query("今天天气怎么样") print(result)五、 避坑指南与最佳实践
💡技巧：多路召回 + 重排序
单一检索方式不够，混合才靠谱。
⚠️警告：不要喂太多文档
Top-3 就够了，多了反而是噪音。
✅推荐：查询改写提升召回
用户表述不清时，改写能提升命中率。
六、 综合实战演示
6.1 生产级 RAG 管道
from typing import List, Dict, Any import time import hashlib class ProductionRAG: def __init__(self, vector_store, keyword_store, llm, cache_size=1000): self.vector_store = vector_store self.keyword_store = keyword_store self.llm = llm self.cache = {} self.cache_size = cache_size def query(self, question: str) -> Dict: cache_key = hashlib.md5(question.encode()).hexdigest() if cache_key in self.cache: return self.cache[cache_key] start = time.time() vector_docs = self.vector_store.similarity_search(question, k=5) keyword_docs = self.keyword_store.search(question, k=5) seen = set() all_docs = [] for d in vector_docs + keyword_docs: if d not in seen: seen.add(d) all_docs.append(d) context = "\n\n".join(all_docs[:3]) answer = self.llm(f"基于资料回答：\n{context[:2000]}\n\n问题: {question}") result = { "answer": answer, "sources": all_docs[:2], "latency": time.time() - start } if len(self.cache) < self.cache_size: self.cache[cache_key] = result return result rag = ProductionRAG(vector_store, keyword_store, llm) result = rag.query("最近有什么新闻") print(result["answer"])总结
RAG 检索优化要点：
- 多路召回提升召回率
- 重排序提升精确率
- 查询改写提升覆盖率
- 缓存加速响应
