NLP 进阶：RAG 检索增强生成——从幻觉困境到知识锚定的工程实践 NLP 进阶RAG 检索增强生成——从幻觉困境到知识锚定的工程实践一、LLM 的知识盲区与幻觉困境当模型自信地胡说大语言模型在开放域问答中表现惊艳但其知识来源于训练数据存在两个根本性缺陷知识时效性滞后训练截止日期之后的事件一无所知和长尾知识缺失低频领域的专业知识覆盖不足。更危险的是LLM 面对知识盲区时不会坦承我不知道而是自信地生成看似合理但完全错误的内容——这就是幻觉Hallucination问题。以企业内部知识库问答为例当用户询问公司 2025 年 Q3 的差旅报销政策变更了哪些条款通用 LLM 只能编造答案因为这类信息从未出现在其训练数据中。即便使用最新的模型也无法保证回答的准确性。RAGRetrieval-Augmented Generation的核心思路是不让模型凭记忆回答而是先从知识库中检索相关文档再将文档作为上下文提供给模型让模型基于事实生成回答。代码是人与机器的对话而 RAG 更像是给这段对话配备了一座图书馆——不再凭空想象而是先翻书再回答。如同古人解卦需先查阅典籍RAG 让生成过程有了知识的锚点。二、RAG 的双阶段架构检索与生成的协同机制RAG 系统分为检索Retrieval和生成Generation两个阶段。检索阶段将用户查询转换为向量在知识库中找到语义最相关的文档片段生成阶段将检索到的文档与用户问题拼接为 Prompt由 LLM 基于上下文生成回答。graph TB Q[用户查询] -- QR[查询预处理与改写] QR -- QE[查询向量化 Embedding] QE -- VS[向量相似度检索] VS -- RR[检索结果重排序 Rerank] RR -- CT[上下文组装与截断] subgraph 知识库构建离线 D[原始文档] -- DC[文档切分 Chunking] DC -- DE[文档向量化 Embedding] DE -- VI[向量索引构建br/FAISS/Milvus] end VI -.- VS CT -- LLM[大语言模型生成] LLM -- A[最终回答 来源引用] subgraph 质量保障 H[幻觉检测] S[来源追溯] end A -- H A -- S检索质量是 RAG 系统的生命线。向量检索基于语义相似度但语义相似不等于信息相关——如何减肥和如何增重语义相近但信息相反。Rerank 阶段使用交叉编码器Cross-Encoder对检索结果进行精排将查询和文档拼接后计算相关性分数比双塔向量的点积更准确但计算成本更高。文档切分策略直接影响检索粒度。过大的 Chunk 包含过多无关信息稀释了关键内容的相似度过小的 Chunk 丢失上下文检索到的片段可能无法独立理解。实践中通常按段落或语义边界切分Chunk 大小在 256-512 Token 之间相邻 Chunk 之间保留 50-100 Token 的重叠。三、生产级 RAG 系统实现以下代码实现了一个完整的 RAG 系统包含文档处理、向量检索、Rerank 和幻觉检测import hashlib import logging from typing import Any, Dict, List, Optional, Tuple from dataclasses import dataclass, field from abc import ABC, abstractmethod import numpy as np logger logging.getLogger(__name__) dataclass class Document: 文档数据结构 content: str metadata: Dict[str, Any] field(default_factorydict) doc_id: str def __post_init__(self): if not self.doc_id: self.doc_id hashlib.md5( self.content.encode() ).hexdigest()[:12] dataclass class Chunk: 文档片段 content: str doc_id: str chunk_index: int embedding: Optional[np.ndarray] None metadata: Dict[str, Any] field(default_factorydict) class DocumentChunker: 文档切分器支持多种切分策略 def __init__( self, chunk_size: int 400, chunk_overlap: int 80, separator: str \n, ): if chunk_overlap chunk_size: raise ValueError( f重叠大小 ({chunk_overlap}) 必须小于 f块大小 ({chunk_size}) ) self._chunk_size chunk_size self._chunk_overlap chunk_overlap self._separator separator def chunk(self, doc: Document) - List[Chunk]: 将文档切分为片段 # 按分隔符拆分 segments doc.content.split(self._separator) segments [s.strip() for s in segments if s.strip()] chunks [] current_text chunk_index 0 for segment in segments: # 简化的 Token 估算中文约 1.5 字/Token estimated_tokens len(current_text) / 1.5 if estimated_tokens len(segment) / 1.5 self._chunk_size: if current_text: chunks.append(Chunk( contentcurrent_text.strip(), doc_iddoc.doc_id, chunk_indexchunk_index, metadata{**doc.metadata}, )) chunk_index 1 # 保留重叠部分 overlap_text current_text[-self._chunk_overlap * 2:] current_text overlap_text self._separator segment else: current_text segment else: current_text self._separator segment # 处理最后一块 if current_text.strip(): chunks.append(Chunk( contentcurrent_text.strip(), doc_iddoc.doc_id, chunk_indexchunk_index, metadata{**doc.metadata}, )) return chunks class EmbeddingService(ABC): 向量化服务抽象接口 abstractmethod async def embed_texts(self, texts: List[str]) - np.ndarray: 批量文本向量化 pass abstractmethod async def embed_query(self, query: str) - np.ndarray: 查询向量化 pass class VectorStore(ABC): 向量存储抽象接口 abstractmethod async def add(self, chunks: List[Chunk]) - None: 添加文档片段 pass abstractmethod async def search( self, query_embedding: np.ndarray, top_k: int 5 ) - List[Tuple[Chunk, float]]: 向量相似度检索 pass class InMemoryVectorStore(VectorStore): 基于内存的向量存储适用于中小规模知识库 def __init__(self): self._chunks: List[Chunk] [] self._embeddings: Optional[np.ndarray] None async def add(self, chunks: List[Chunk]) - None: embeddings [ c.embedding for c in chunks if c.embedding is not None ] if not embeddings: raise ValueError(所有 Chunk 都缺少 embedding) new_embeddings np.stack(embeddings) if self._embeddings is None: self._embeddings new_embeddings else: self._embeddings np.vstack( [self._embeddings, new_embeddings] ) self._chunks.extend(chunks) async def search( self, query_embedding: np.ndarray, top_k: int 5 ) - List[Tuple[Chunk, float]]: if self._embeddings is None or len(self._chunks) 0: return [] # 余弦相似度 query_norm query_embedding / ( np.linalg.norm(query_embedding) 1e-10 ) doc_norms self._embeddings / ( np.linalg.norm(self._embeddings, axis1, keepdimsTrue) 1e-10 ) similarities doc_norms query_norm top_indices np.argsort(similarities)[::-1][:top_k] results [] for idx in top_indices: score float(similarities[idx]) if score 0.3: # 最低相似度阈值 results.append((self._chunks[idx], score)) return results class Reranker(ABC): 重排序器抽象接口 abstractmethod async def rerank( self, query: str, chunks: List[Chunk], top_k: int 3 ) - List[Tuple[Chunk, float]]: pass class LLMReranker(Reranker): 基于 LLM 的重排序器利用 LLM 判断文档与查询的相关性 def __init__(self, llm_client: Any): self._llm llm_client async def rerank( self, query: str, chunks: List[Chunk], top_k: int 3 ) - List[Tuple[Chunk, float]]: scored_chunks [] for chunk in chunks: prompt ( f请判断以下文档片段与查询的相关性 f给出 0-10 的评分。\n\n f查询: {query}\n\n f文档: {chunk.content[:500]}\n\n f相关性评分仅输出数字: ) try: response await self._llm.generate( prompt, temperature0.0, max_tokens10 ) score float(response.strip()) scored_chunks.append((chunk, score / 10.0)) except (ValueError, Exception) as e: logger.warning(f重排序评分失败: {e}) scored_chunks.append((chunk, 0.5)) scored_chunks.sort(keylambda x: x[1], reverseTrue) return scored_chunks[:top_k] class HallucinationDetector: 幻觉检测器验证生成内容是否被检索文档支撑 def __init__(self, llm_client: Any): self._llm llm_client async def check( self, answer: str, context_chunks: List[Chunk] ) - Dict[str, Any]: 检测回答中的幻觉内容 context_text \n\n.join( f[文档{i1}]: {c.content[:300]} for i, c in enumerate(context_chunks) ) prompt ( 请检查以下回答是否被提供的文档内容所支撑。\n 对于回答中的每个关键声明判断是否有文档依据。\n\n f文档:\n{context_text}\n\n f回答:\n{answer}\n\n 请以 JSON 格式输出:\n {supported_claims: 数量, unsupported_claims: 数量, hallucination_risk: low/medium/high, details: 具体说明} ) try: import json response await self._llm.generate( prompt, temperature0.0, max_tokens300 ) result json.loads(response.strip()) return result except Exception as e: logger.warning(f幻觉检测失败: {e}) return { hallucination_risk: unknown, details: f检测失败: {str(e)}, } class RAGPipeline: RAG 完整管道 def __init__( self, embedding_service: EmbeddingService, vector_store: VectorStore, reranker: Optional[Reranker] None, hallucination_detector: Optional[HallucinationDetector] None, max_context_tokens: int 3000, ): self._embedding embedding_service self._store vector_store self._reranker reranker self._hallucination_detector hallucination_detector self._max_context_tokens max_context_tokens async def index_documents(self, docs: List[Document]) - int: 索引文档到知识库 chunker DocumentChunker() all_chunks [] for doc in docs: chunks chunker.chunk(doc) all_chunks.extend(chunks) # 批量向量化 texts [c.content for c in all_chunks] embeddings await self._embedding.embed_texts(texts) for i, chunk in enumerate(all_chunks): chunk.embedding embeddings[i] await self._store.add(all_chunks) logger.info( f已索引 {len(docs)} 篇文档 f共 {len(all_chunks)} 个片段 ) return len(all_chunks) async def query( self, question: str, llm_client: Any, top_k: int 5, ) - Dict[str, Any]: 执行 RAG 查询 # 1. 查询向量化 query_embedding await self._embedding.embed_query(question) # 2. 向量检索 search_results await self._store.search( query_embedding, top_ktop_k ) if not search_results: return { answer: 未找到相关文档无法回答该问题, sources: [], hallucination_risk: none, } # 3. 重排序 candidate_chunks [chunk for chunk, _ in search_results] if self._reranker: ranked_results await self._reranker.rerank( question, candidate_chunks, top_k3 ) final_chunks [c for c, _ in ranked_results] else: final_chunks candidate_chunks[:3] # 4. 组装上下文控制 Token 预算 context_parts [] total_tokens 0 for chunk in final_chunks: estimated_tokens len(chunk.content) / 1.5 if total_tokens estimated_tokens self._max_context_tokens: break context_parts.append(chunk.content) total_tokens estimated_tokens context_text \n\n---\n\n.join(context_parts) # 5. LLM 生成 prompt ( 请基于以下文档内容回答问题。 如果文档中没有相关信息请明确说明。\n\n f文档:\n{context_text}\n\n f问题: {question}\n\n 回答请引用文档来源: ) answer await llm_client.generate( prompt, temperature0.1, max_tokens500 ) # 6. 幻觉检测 hallucination_result {hallucination_risk: unknown} if self._hallucination_detector: hallucination_result ( await self._hallucination_detector.check( answer, final_chunks ) ) return { answer: answer, sources: [ {doc_id: c.doc_id, chunk_index: c.chunk_index} for c in final_chunks ], hallucination_risk: hallucination_result.get( hallucination_risk, unknown ), hallucination_details: hallucination_result.get( details, ), }关键工程实践文档切分保留重叠区域避免语义断裂向量检索设置最低相似度阈值过滤无关结果上下文组装控制 Token 预算防止超出 LLM 窗口幻觉检测验证生成内容是否被文档支撑。四、RAG 系统的权衡检索质量的天花板效应检索质量决定生成上限RAG 的回答质量受限于检索结果的相关性。如果关键文档未被检索到LLM 无法凭空生成正确答案。在知识库规模增大时向量检索的召回率会下降需要引入混合检索向量 关键词 BM25提升召回。Chunk 粒度的两难大 Chunk 保留完整上下文但引入噪声降低检索精度小 Chunk 检索精准但可能丢失关键上下文。一种折中方案是检索小 Chunk返回大 Chunk——用小 Chunk 做相似度匹配但返回包含该 Chunk 的完整段落作为上下文。延迟与质量的取舍Rerank 阶段显著提升检索质量但每个候选文档都需要一次 LLM 调用或 Cross-Encoder 推理延迟从毫秒级上升到秒级。在实时对话场景中可能需要跳过 Rerank 或使用轻量级模型。知识库更新的时效性向量索引构建是离线批处理新文档入库后需要重新计算 Embedding 并更新索引。在知识频繁更新的场景中索引更新的延迟成为瓶颈。禁用场景答案需要复杂推理而非简单检索的场景如数学证明RAG 的检索结果可能反而干扰推理知识库质量极低大量错误或过时信息时RAG 会放大错误对延迟要求极高的实时系统检索生成的两阶段延迟无法接受。五、总结RAG 通过先检索后生成的架构将 LLM 的回答锚定在知识库的事实上有效缓解了幻觉问题。核心组件包括文档切分、向量化、向量检索、重排序和幻觉检测。生产实践中需关注Chunk 大小与重叠的平衡、最低相似度阈值过滤、Token 预算控制、混合检索提升召回率。RAG 适用于知识密集型问答场景但在需要复杂推理或对延迟极度敏感的场景中需谨慎评估必要时应结合其他技术方案。