RAG的核心承诺是有据可查的答案但很多RAG系统的实际表现是检索到了文档还是捏造了答案。这不是矛盾这是工程问题。RAG幻觉的来源分析RAG系统的幻觉与纯LLM的幻觉不同它有特定的发生机制。理解来源才能对症下药。类型1检索失败型幻觉根本原因检索器没找到相关文档或者找到的文档相关性太低模型在没有证据的情况下凭训练数据推测答案。类型2检索噪声型幻觉根本原因检索到了大量文档但其中有很多不相关的噪声。模型无法区分哪些是有效信息导致混淆。类型3语义漂移型幻觉根本原因文档里有相关主题的内容但不完全回答用户的问题。模型把相近的信息当作正确答案返回。类型4引用虚构型幻觉根本原因模型在总结多个文档时会发明文档中没有但逻辑上合理的细节。类型5过时信息型幻觉根本原因知识库中的文档已经过时但模型没有意识到把旧信息当作当前事实输出。## 检索质量优化从源头减少幻觉### 1. 分块策略对检索质量的影响pythonfrom langchain.text_splitter import ( RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter)from typing import Listclass SmartChunker: 智能文档分块器根据内容类型选择最优策略 def chunk_document(self, content: str, doc_type: str) - List[str]: if doc_type technical_doc: # 技术文档按标题层级分块保留结构 return self._markdown_split(content) elif doc_type legal_document: # 法律文档按条款分块保持完整性 return self._clause_split(content) elif doc_type conversation: # 对话记录按会话轮次分块 return self._conversation_split(content) else: # 通用文档语义感知分块 return self._semantic_split(content) def _semantic_split(self, content: str) - List[str]: 语义感知分块通过计算相邻句子的语义相似度 在语义转折点分块而不是机械地按字符数截断 from sentence_transformers import SentenceTransformer import numpy as np model SentenceTransformer(BAAI/bge-m3) # 先按句号分句 sentences content.split(。) sentences [s.strip() for s in sentences if s.strip()] if len(sentences) 1: return [content] # 计算相邻句子的语义相似度 embeddings model.encode(sentences) similarities [] for i in range(len(embeddings) - 1): sim float(np.dot(embeddings[i], embeddings[i1])) similarities.append(sim) # 在相似度骤降的地方分块 threshold np.mean(similarities) - np.std(similarities) chunks [] current_chunk [sentences[0]] for i, sim in enumerate(similarities): if sim threshold and len(current_chunk) 3: chunks.append(。.join(current_chunk) 。) current_chunk [sentences[i1]] else: current_chunk.append(sentences[i1]) if current_chunk: chunks.append(。.join(current_chunk)) return chunks def _markdown_split(self, content: str) - List[str]: 按Markdown标题层级分块 splitter MarkdownHeaderTextSplitter( headers_to_split_on[ (#, h1), (##, h2), (###, h3), ] ) docs splitter.split_text(content) # 如果某个段落太长再做二次分块 final_chunks [] for doc in docs: if len(doc.page_content) 1500: sub_splitter RecursiveCharacterTextSplitter( chunk_size800, chunk_overlap100 ) sub_chunks sub_splitter.split_text(doc.page_content) final_chunks.extend(sub_chunks) else: final_chunks.append(doc.page_content) return final_chunks### 2. 混合检索减少检索空洞pythonfrom rank_bm25 import BM25Okapiimport numpy as npclass HybridRetriever: BM25 向量检索的混合检索器 def __init__(self, vector_store, chunks: List[str]): self.vector_store vector_store self.chunks chunks # 建立BM25索引 tokenized_corpus [doc.split() for doc in chunks] self.bm25 BM25Okapi(tokenized_corpus) def retrieve( self, query: str, top_k: int 10, alpha: float 0.5 # 向量检索权重1-alpha为BM25权重 ) - List[tuple]: 混合检索向量检索擅长语义匹配 BM25擅长关键词精确匹配两者互补 # 向量检索 vector_results self.vector_store.similarity_search_with_score( query, ktop_k * 2 ) vector_scores {doc.page_content: score for doc, score in vector_results} # BM25检索 bm25_scores_raw self.bm25.get_scores(query.split()) bm25_scores {} for i, score in enumerate(bm25_scores_raw): bm25_scores[self.chunks[i]] score # 归一化两种分数到[0,1] def normalize(scores: dict) - dict: if not scores: return scores max_score max(scores.values()) min_score min(scores.values()) if max_score min_score: return {k: 1.0 for k in scores} return { k: (v - min_score) / (max_score - min_score) for k, v in scores.items() } norm_vector normalize(vector_scores) norm_bm25 normalize(bm25_scores) # 合并分数RRF或加权平均 all_docs set(norm_vector.keys()) | set(norm_bm25.keys()) combined_scores {} for doc in all_docs: v_score norm_vector.get(doc, 0) b_score norm_bm25.get(doc, 0) combined_scores[doc] alpha * v_score (1 - alpha) * b_score # 排序并返回top-k sorted_docs sorted( combined_scores.items(), keylambda x: x[1], reverseTrue ) return sorted_docs[:top_k]### 3. Reranker二次排序pythonfrom sentence_transformers import CrossEncoderclass RerankerFilter: 使用CrossEncoder重排序 CrossEncoder对查询和文档对做精细评分 比向量相似度更准确但更慢适合在粗检索后精排 def __init__(self): # 专门用于中文的重排序模型 self.reranker CrossEncoder(BAAI/bge-reranker-v2-m3) def rerank( self, query: str, documents: List[str], top_k: int 5, min_score: float 0.5 ) - List[tuple]: 重排序并过滤低质量结果 # 构建(query, doc)对CrossEncoder做精细评分 pairs [(query, doc) for doc in documents] scores self.reranker.predict(pairs) # 排序并过滤 ranked sorted( zip(documents, scores), keylambda x: x[1], reverseTrue ) # 过滤掉相关性过低的文档 filtered [ (doc, score) for doc, score in ranked if score min_score ] return filtered[:top_k]## 生成质量控制让模型知道自己不知道### 1. 有据可查提示设计pythonGROUNDED_GENERATION_PROMPT 你是一个严格基于提供文档回答问题的助手。核心原则1. **只使用文档中的信息**不要使用训练数据中的知识2. **标注引用来源**每个关键论断都要说明来自哪个文档3. **明确表达不确定性**如果文档没有明确回答直接说文档中没有找到相关信息4. **不要推断**文档中没有写的内容不要推测---参考文档{documents}---用户问题{question}回答格式要求- 首先判断文档是否包含回答问题的充分信息- 如果信息充分直接回答并在相关论断后标注[来源文档X]- 如果信息不足明确说明根据提供的文档无法完整回答这个问题。文档涉及的相关信息是...async def grounded_generate( question: str, retrieved_docs: List[str], llm_client) - dict: 基于文档的扎实生成返回答案和置信度 # 格式化文档 formatted_docs \n\n.join([ f文档{i1}\n{doc} for i, doc in enumerate(retrieved_docs) ]) prompt GROUNDED_GENERATION_PROMPT.format( documentsformatted_docs, questionquestion ) answer await llm_client.complete(prompt) # 后处理检测是否真的有文档支撑 has_citation [来源 in answer claims_no_info 无法完整回答 in answer or 没有找到相关信息 in answer return { answer: answer, is_grounded: has_citation or claims_no_info, confidence: high if has_citation else (low if not claims_no_info else medium) }### 2. 自动幻觉检测pythonFAITHFULNESS_CHECK_PROMPT 判断以下答案是否完全基于提供的上下文还是包含了上下文中没有的信息。上下文{context}答案{answer}请检查答案中的每个关键论断1. 该论断在上下文中是否有明确依据2. 还是这是模型的推断或虚构以JSON格式输出{{ is_faithful: true/false, unfaithful_claims: [虚构或无据论断1, 虚构或无据论断2], faithfulness_score: 0-11表示完全基于上下文}}class HallucinationDetector: 幻觉自动检测器 def __init__(self, judge_llm): self.judge judge_llm # 用更强的模型做判断 async def check_faithfulness( self, context: str, answer: str ) - dict: 检查答案是否忠实于检索到的文档 result_json await self.judge.complete( FAITHFULNESS_CHECK_PROMPT.format( contextcontext, answeranswer ), response_format{type: json_object} ) result json.loads(result_json) return { is_faithful: result.get(is_faithful, False), unfaithful_claims: result.get(unfaithful_claims, []), score: result.get(faithfulness_score, 0.0) } async def check_with_threshold( self, context: str, answer: str, threshold: float 0.8 ) - tuple[str, dict]: 检测幻觉低于阈值时触发重新生成 返回(最终答案, 检测结果) check_result await self.check_faithfulness(context, answer) if check_result[score] threshold: return answer, check_result # 幻觉检测失败返回保守答案 conservative_answer ( f根据文档可以提供以下有依据的信息\n\n f{self._extract_grounded_parts(answer, check_result[unfaithful_claims])}\n\n f注部分内容{, .join(check_result[unfaithful_claims][:2])}等 f无法在文档中得到确认已移除。 ) return conservative_answer, check_result def _extract_grounded_parts( self, answer: str, unfaithful_claims: List[str] ) - str: 从答案中移除无据论断保留有据部分 result answer for claim in unfaithful_claims: if claim in result: result result.replace(claim, [已移除无文档依据]) return result## 完整的反幻觉RAG流水线pythonclass AntiHallucinationRAGPipeline: 完整的反幻觉RAG流水线 def __init__(self): self.retriever HybridRetriever(vector_store, chunks) self.reranker RerankerFilter() self.generator GroundedGenerator(llm_client) self.detector HallucinationDetector(judge_llm) self.metrics MetricsCollector() async def answer(self, question: str) - dict: 完整的问答流程最大化准确性 # 第1步混合检索 raw_results self.retriever.retrieve(question, top_k20) docs [doc for doc, _ in raw_results] if not docs: return { answer: 抱歉知识库中没有找到与您问题相关的信息。, is_reliable: False, confidence: none } # 第2步Reranker精排 reranked self.reranker.rerank(question, docs, top_k5) # 如果最高分太低主动告知 if reranked and reranked[0][1] 0.3: return { answer: 找到了一些相关内容但相关性较低以下信息仅供参考\n\n self._format_docs([d for d, _ in reranked]), is_reliable: False, confidence: low } # 第3步有据生成 context self._format_docs([d for d, _ in reranked]) gen_result await self.generator.generate(question, context) # 第4步幻觉检测 final_answer, faithfulness await self.detector.check_with_threshold( contextcontext, answergen_result[answer], threshold0.75 ) # 记录指标 await self.metrics.record({ question: question, retrieval_count: len(reranked), faithfulness_score: faithfulness[score], unfaithful_claims: faithfulness[unfaithful_claims] }) return { answer: final_answer, is_reliable: faithfulness[score] 0.75, confidence: high if faithfulness[score] 0.9 else medium, source_docs: [d for d, _ in reranked], faithfulness_score: faithfulness[score] } def _format_docs(self, docs: List[str]) - str: return \n\n---\n\n.join([f来源{i1}{doc} for i, doc in enumerate(docs)])## 效果评估使用RAGASpythonfrom ragas import evaluatefrom ragas.metrics import faithfulness, answer_relevancy, context_precisionasync def evaluate_rag_quality(test_dataset: list) - dict: 使用RAGAS评估RAG系统质量 results evaluate( test_dataset, metrics[ faithfulness, # 答案是否忠实于检索文档 answer_relevancy, # 答案是否回答了问题 context_precision # 检索文档是否相关 ] ) print(fFaithfulness忠实度: {results[faithfulness]:.2%}) print(fAnswer Relevancy相关性: {results[answer_relevancy]:.2%}) print(fContext Precision检索精度: {results[context_precision]:.2%}) return results优化后一个生产级RAG系统的幻觉率Faithfulness低于0.7的比例从优化前的23%降到了6%以下。关键是要接受有时候承认不知道比给出错误答案更好这个原则——这才是可信AI系统的基础。