RAG 技术现状与避坑指南：检索增强不是银弹

一、从狂热到理性：RAG 技术的一年实践反思

2025 年到 2026 年，RAG（Retrieval-Augmented Generation，检索增强生成）技术经历了从"银弹"到"鸡肋"的评价转变。某知识管理平台的实践数据揭示了真相：

• 上线初期：号称准确率 95%（基于 100 条测试集）
• 上线三月后：用户满意度 62%，主要投诉"答非所问"
• 深入分析：测试集过于简单，真实场景的查询复杂度远超测试

这不是 RAG 本身的问题，而是工程落地的偏差。本文将客观分析 RAG 技术的现状、局限性和避坑指南。

二、RAG 的技术原理与核心挑战

RAG 的工作流程

RAG 的核心思想是：在生成答案之前，先从知识库中检索相关信息，然后将检索结果作为上下文提供给大模型。

标准流程：

1. 索引阶段：将文档分块 → 向量化 → 存入向量数据库
2. 检索阶段：用户查询向量化 → 相似度检索 → 返回 Top-K 结果
3. 生成阶段：将检索结果拼接到 prompt → 调用大模型生成答案

核心挑战一：文本分块（Chunking）

问题：如何将一个 50 页的 PDF 文档切分成合适的块？

反模式：

# 错误示例：固定长度切分（忽略语义） def bad_chunking(text: str, chunk_size: int = 500) -> List[str]: chunks = [] for i in range(0, len(text), chunk_size): chunks.append(text[i:i+chunk_size]) return chunks # 问题：可能把一个完整的句子切成两半 # 例如："RAG 技术的核心是检索增强。生成模型通过..." # 被切成 ["RAG 技术的核心是检索增强。", "生成模型通过..."] # 第二块失去了上下文

正确做法：语义感知的分块

from typing import List import re class SemanticChunker: """语义感知的文本分块器""" def __init__(self, max_chunk_size: int = 500, overlap: int = 50): self.max_chunk_size = max_chunk_size self.overlap = overlap def chunk(self, text: str) -> List[str]: """按语义边界分块""" # 1. 先按段落分割 paragraphs = text.split('\n\n') chunks = [] current_chunk = [] current_size = 0 for para in paragraphs: para = para.strip() if not para: continue # 如果单个段落就超过限制，按句子分割 if len(para) > self.max_chunk_size: sentences = self._split_sentences(para) for sent in sentences: if current_size + len(sent) > self.max_chunk_size: # 保存当前块 if current_chunk: chunks.append(' '.join(current_chunk)) # 开始新块（带重叠） current_chunk = current_chunk[-self.overlap:] if self.overlap > 0 else [] current_size = sum(len(s) for s in current_chunk) current_chunk.append(sent) current_size += len(sent) else: # 普通段落 if current_size + len(para) > self.max_chunk_size: chunks.append(' '.join(current_chunk)) current_chunk = current_chunk[-self.overlap:] if self.overlap > 0 else [] current_size = sum(len(s) for s in current_chunk) current_chunk.append(para) current_size += len(para) # 保存最后一个块 if current_chunk: chunks.append(' '.join(current_chunk)) return chunks def _split_sentences(self, text: str) -> List[str]: """按句子分割""" sentences = re.split(r'(?<=[。！？\.!?])\s+', text) return [s.strip() for s in sentences if s.strip()]

核心挑战二：检索精度

问题：用户问"如何退款"，检索返回了"退款政策"的文档，但具体内容是关于"不可退款的商品类型"，导致生成的答案错误。

解决方案：混合检索 + Rerank

class HybridRetriever: """混合检索器：向量检索 + 关键词检索""" def __init__(self, vector_db, keyword_index): self.vector_db = vector_db self.keyword_index = keyword_index def retrieve(self, query: str, top_k: int = 10) -> List[Document]: # 1. 向量检索 query_embedding = self.embed(query) vector_results = self.vector_db.search(query_embedding, top_k=top_k) # 2. 关键词检索（BM25） keyword_results = self.keyword_index.search(query, top_k=top_k) # 3. 结果融合（Reciprocal Rank Fusion） merged = self._rrf_merge(vector_results, keyword_results) # 4. Rerank（使用交叉编码器） reranked = self._rerank(query, merged[:top_k*2]) return reranked[:top_k] def _rrf_merge(self, list1: List, list2: List, k: int = 60) -> List: """Reciprocal Rank Fusion 融合""" scores = {} for rank, doc in enumerate(list1, 1): doc_id = doc.id if doc_id not in scores: scores[doc_id] = {'doc': doc, 'score': 0} scores[doc_id]['score'] += 1 / (k + rank) for rank, doc in enumerate(list2, 1): doc_id = doc.id if doc_id not in scores: scores[doc_id] = {'doc': doc, 'score': 0} scores[doc_id]['score'] += 1 / (k + rank) # 按分数排序 sorted_docs = sorted(scores.values(), key=lambda x: x['score'], reverse=True) return [item['doc'] for item in sorted_docs] def _rerank(self, query: str, docs: List[Document]) -> List[Document]: """使用交叉编码器重排序""" # 这里使用 Cohere Rerank 或 BGE Reranker from sentence_transformers import CrossEncoder model = CrossEncoder('BAAI/bge-reranker-v1.5') pairs = [[query, doc.content] for doc in docs] scores = model.predict(pairs) # 按重排序分数排序 doc_score_pairs = list(zip(docs, scores)) doc_score_pairs.sort(key=lambda x: x[1], reverse=True) return [doc for doc, _ in doc_score_pairs]

三、生产级 RAG 系统实现

完整的 RAG Pipeline

from dataclasses import dataclass from typing import List, Optional import logging @dataclass class RetrievalResult: """检索结果""" document: Document score: float source: str # 'vector', 'keyword', 'hybrid' @dataclass class GenerationResult: """生成结果""" answer: str sources: List[RetrievalResult] # 引用的来源 confidence: float # 置信度 class RAGPipeline: """生产级 RAG Pipeline""" def __init__( self, retriever: HybridRetriever, llm_client, prompt_template: str, config: dict ): self.retriever = retriever self.llm = llm_client self.prompt_template = prompt_template self.config = config def query(self, user_query: str) -> GenerationResult: # 1. Query 理解（可选：改写、扩展） processed_query = self._process_query(user_query) # 2. 检索 retrieved_docs = self.retriever.retrieve( processed_query, top_k=self.config.get('top_k', 5) ) if not retrieved_docs: return GenerationResult( answer="抱歉，我没有找到相关信息。", sources=[], confidence=0.0 ) # 3. 上下文压缩（去除冗余） compressed_context = self._compress_context( retrieved_docs, max_tokens=self.config.get('max_context_tokens', 2000) ) # 4. 组装 Prompt prompt = self.prompt_template.format( context=compressed_context, question=user_query ) # 5. 调用大模型 answer = self.llm.generate( prompt, temperature=self.config.get('temperature', 0.1), max_tokens=self.config.get('max_tokens', 500) ) # 6. 后处理（提取来源、计算置信度） result = self._post_process(answer, retrieved_docs) return result def _process_query(self, query: str) -> str: """Query 改写（解决口语化问题）""" # 示例：将"咋退款"改写为"如何申请退款" rewrite_prompt = f"将以下用户提问改写为更适合文档检索的query：\n{query}" # 可选：只用大模型改写复杂查询 if len(query) < 10 or query.isascii(): return query rewritten = self.llm.generate(rewrite_prompt, temperature=0.0, max_tokens=100) return rewritten.strip() def _compress_context(self, docs: List[Document], max_tokens: int) -> str: """上下文压缩：去除冗余，控制 token 数""" # 简单策略：按相关性排序，截取前 N 个 total_tokens = 0 selected_docs = [] for doc in docs: doc_tokens = len(doc.content) // 2 # 粗略估计：1 个中文字符 ≈ 2 tokens if total_tokens + doc_tokens > max_tokens: break selected_docs.append(doc) total_tokens += doc_tokens # 组装上下文 context_parts = [] for i, doc in enumerate(selected_docs, 1): context_parts.append(f"[文档{i}] {doc.content}") return '\n\n'.join(context_parts)

四、边界分析与 Trade-offs

RAG 不是银弹：适用场景分析

适合 RAG 的场景：

1. 知识密集但变化不频繁：产品文档、API 文档、政策法规
2. 需要引用来源：医疗咨询、法律咨询（必须给出依据）
3. 个性化问答：企业内部知识库（每个部门有不同的文档）

不适合 RAG 的场景：

1. 实时数据：股票价格、天气（应该用 API 调用）
2. 推理密集型：数学题、逻辑推理（直接让模型推理更好）
3. 创意生成：写诗、写小说（不需要检索）

成本与性能的 Trade-off

实测数据（某客服场景）：

结论：Top-5 是性价比最高的选择。

常见问题与解决方案

问题 1：检索到了无关内容

• 原因：向量检索的相似度分数不可靠
• 解决：设置相似度阈值（如 cosine_sim > 0.7），低于阈值的直接返回"不知道"

问题 2：上下文太长，超出模型窗口

• 原因：检索了太多文档
• 解决：上下文压缩 + 动态选择 Top-K

问题 3：幻觉问题

• 原因：模型忽略检索结果，自己"编造"答案
• 解决：在 prompt 中强调"只基于给定上下文回答，不要编造"

五、总结

RAG 技术现状总结：

能做好的场景：

• 文档问答（FAQ、产品手册）
• 技术文档检索（API 文档、代码注释）
• 法规合规查询（法律条文、政策文件）

做不好的场景：

• 实时信息查询
• 复杂推理
• 多轮对话（需要记忆管理）

避坑清单：

1. 文本分块必须考虑语义边界，不能简单按字数切分
2. 检索必须混合（向量 + 关键词），单靠向量检索不够
3. Rerank 是提升精度的关键，不能省略
4. 必须设置相似度阈值，避免检索无关内容
5. 成本优化：Top-K 选 3-5，上下文压缩必须做

未来方向：

• Agentic RAG：让模型自主决定检索几次、检索什么
• 多模态 RAG：支持图片、表格的检索
• 实时 RAG：支持流式更新知识库

下一篇文章，我们将深入探讨后端可观测性建设清单。