RAG系统的7个反模式从Naive RAG到Advanced RAG的避坑指南作者钟伊人 | 日期2026-07-29 | Week5 总结与避坑模块一Naive RAG的七大反模式反模式1直接分块无语义感知问题描述将文档按固定长度如500字符直接切分。一个句子被拦腰截断语义完全丢失。这会导致检索到的块内容不完整回答质量差。 反模式1固定长度分块错误示例 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 # 问题示例 # text 深度学习是机器学习的一个重要分支。它使用多层神经网络来学习数据的表示。 # 可能在分支和它使用之间切断导致语义不完整正确模式语义感知分块 正确模式基于句子/段落的语义分块 import re from typing import List, Dict import tiktoken class SemanticChunker: 语义感知的分块器 保持句子和段落的完整性 def __init__(self, max_chunk_tokens: int 512, overlap_tokens: int 50, tokenizer_name: str cl100k_base): Args: max_chunk_tokens: 每个块的最大token数 overlap_tokens: 相邻块的重叠token数保持上下文连贯 tokenizer_name: 分词器名称 self.max_tokens max_chunk_tokens self.overlap_tokens overlap_tokens self.tokenizer tiktoken.get_encoding(tokenizer_name) def chunk_markdown(self, markdown_text: str) - List[Dict]: 对Markdown文档进行语义分块 策略 1. 按标题# ## ###分段 2. 每段内按句子分割 3. 组合句子直到达到token限制 chunks [] # 步骤1按标题分节 sections self._split_by_headers(markdown_text) current_chunk_sentences [] current_tokens 0 for section_title, section_content in sections: sentences self._split_into_sentences(section_content) for sentence in sentences: sentence_tokens len(self.tokenizer.encode(sentence)) # 如果单个句子超过限制强制分块尽量在标点处切分 if sentence_tokens self.max_tokens: # 将当前块保存 if current_chunk_sentences: chunks.append({ content: .join(current_chunk_sentences), metadata: {section: section_title} }) # 保留最后几句作为重叠 overlap self._get_overlap_sentences( current_chunk_sentences, self.overlap_tokens ) current_chunk_sentences overlap current_tokens sum( len(self.tokenizer.encode(s)) for s in current_chunk_sentences ) # 对超长句子进行强制分块 forced_chunks self._force_chunk_sentence( sentence, self.max_tokens ) for fc in forced_chunks: chunks.append({ content: fc, metadata: {section: section_title, forced: True} }) continue # 如果加入这个句子会超限先保存当前块 if current_tokens sentence_tokens self.max_tokens: if current_chunk_sentences: chunks.append({ content: .join(current_chunk_sentences), metadata: {section: section_title} }) # 保留最后几句作为重叠 overlap self._get_overlap_sentences( current_chunk_sentences, self.overlap_tokens ) current_chunk_sentences overlap current_tokens sum( len(self.tokenizer.encode(s)) for s in current_chunk_sentences ) current_chunk_sentences.append(sentence) current_tokens sentence_tokens # 保存最后一个块 if current_chunk_sentences: chunks.append({ content: .join(current_chunk_sentences), metadata: {section: sections[-1][0] if sections else unknown} }) return chunks def _split_by_headers(self, text: str) - List[tuple]: 按Markdown标题分节 sections [] current_title 引言 current_content [] for line in text.split(\n): if re.match(r^#{1,6}\s, line): # 保存上一节 if current_content: sections.append((current_title, \n.join(current_content))) current_title line.lstrip(#).strip() current_content [] else: current_content.append(line) # 保存最后一节 if current_content: sections.append((current_title, \n.join(current_content))) return sections def _split_into_sentences(self, text: str) - List[str]: 按句子分割中文英文 # 中文句子分割。 # 英文句子分割. ! ? ; sentence_endings r(?[。!?])\s* sentences re.split(sentence_endings, text) return [s.strip() for s in sentences if s.strip()] def _get_overlap_sentences(self, sentences: List[str], overlap_tokens: int) - List[str]: 获取最后N个句子作为重叠 overlap [] token_count 0 for sentence in reversed(sentences): s_tokens len(self.tokenizer.encode(sentence)) if token_count s_tokens overlap_tokens: break overlap.insert(0, sentence) token_count s_tokens return overlap def _force_chunk_sentence(self, sentence: str, max_tokens: int) - List[str]: 对超长句子进行强制分块 words sentence.split() chunks [] current_chunk [] current_tokens 0 for word in words: word_tokens len(self.tokenizer.encode(word)) if current_tokens word_tokens max_tokens: chunks.append( .join(current_chunk)) current_chunk [word] current_tokens word_tokens else: current_chunk.append(word) current_tokens word_tokens if current_chunk: chunks.append( .join(current_chunk)) return chunks反模式2检索Top-K后直接使用无重排序问题描述Naive RAG直接取向量检索的Top-K结果。向量相似度高不等于相关性高。缺乏重排序Reranking导致不相关块被送入LLM。正确模式两阶段检索检索重排序 正确模式两阶段检索 第一阶段向量检索召回候选集召回率高 第二阶段精排模型重新排序精确度高 from typing import List, Tuple import numpy as np class TwoStageRetriever: 两阶段检索器 生产级实现向量检索 精排模型 def __init__(self, vector_store, reranker_modelNone, initial_top_k: int 20, final_top_k: int 5): Args: vector_store: 向量数据库客户端 reranker_model: 精排模型如Cohere Rerank、BGE Reranker initial_top_k: 第一阶段召回数量 final_top_k: 第二阶段返回数量 self.vector_store vector_store self.reranker reranker_model self.initial_k initial_top_k self.final_k final_top_k def retrieve(self, query: str) - List[Dict]: 两阶段检索 阶段1向量检索召回候选集宽召回 阶段2精排模型重新排序精确排 # 阶段1向量检索 candidates self.vector_store.search( queryquery, top_kself.initial_k ) if not self.reranker: # 如果没有精排模型返回向量检索结果 return candidates[:self.final_k] # 阶段2精排 reranked self.reranker.rerank( queryquery, documents[c[content] for c in candidates], top_kself.final_k ) # 根据精排结果重新组织返回数据 result [] for r in reranked: idx r[index] result.append({ **candidates[idx], rerank_score: r[score] }) return result class SimpleReranker: 简易精排模型基于交叉编码器 生产环境应使用专门的精排模型如BGE Reranker def __init__(self, model_name: str BAAI/bge-reranker-v2-m3): 初始化精排模型 需要使用支持中文的精排模型 # 实际实现中应加载CrossEncoder模型 # from sentence_transformers import CrossEncoder # self.model CrossEncoder(model_name) self.model_name model_name print(f精排模型 {model_name} 加载完成模拟) def rerank(self, query: str, documents: List[str], top_k: int 5) - List[Dict]: 精排文档 精排模型会同时考虑query和document给出相关性得分 这比向量检索的双向编码器精确得多 # 实际实现 # scores self.model.predict([[query, doc] for doc in documents]) # 这里用模拟数据 scores np.random.rand(len(documents)) # 模拟得分 # 按得分排序 indexed_scores [(i, float(s)) for i, s in enumerate(scores)] indexed_scores.sort(keylambda x: x[1], reverseTrue) return [ {index: idx, score: score} for idx, score in indexed_scores[:top_k] ]反模式3无查询改写用户问题直扔检索器问题描述用户的问题可能含糊、简写、有歧义。直接将原始问题用于检索召回率低。例如怎么用 → 缺少上下文检索效果差。正确模式查询改写Query Rewriting 正确模式查询改写 在检索前用LLM将用户问题改写为更适合检索的形式 class QueryRewriter: 查询改写器 将用户原始问题改写为更利于检索的查询 def __init__(self, llm_client): self.llm llm_client def rewrite(self, original_query: str, chat_history: List[Dict] None) - str: 改写查询 改写策略 1. 扩展简写怎么用 → 如何使用[产品名]的功能 2. 消除歧义根据上下文 3. 添加同义词提高召回率 4. 生成结构化查询用于混合检索 # 构建改写提示词 prompt self._build_rewrite_prompt(original_query, chat_history) # 调用LLM进行改写 rewritten self.llm.generate(prompt) return rewritten.strip() def _build_rewrite_prompt(self, query: str, history: List[Dict] None) - str: 构建查询改写的提示词 history_text if history: history_text 对话历史\n for h in history[-3:]: # 只取最近3轮 history_text f用户{h.get(user, )}\n history_text f助手{h.get(assistant, )}\n return f{history_text} 请将以下用户问题改写为3个更适合向量数据库检索的查询。 要求 1. 消除歧义补充上下文 2. 使用完整、规范的表达 3. 保留关键信息 原始问题{query} 输出格式JSON {{ rewritten_queries: [查询1, 查询2, 查询3], expanded_query: 扩展后的查询包含同义词 }} def multi_query_retrieve(self, original_query: str, retriever, chat_history: List[Dict] None) - List[Dict]: 多查询检索 将改写后的多个查询分别检索然后合并去重 # 改写查询 rewrite_result self.rewrite(original_query, chat_history) # 解析改写结果简化处理 import json try: result json.loads(rewrite_result) queries result.get(rewritten_queries, [original_query]) except json.JSONDecodeError: queries [original_query] # 对每个查询进行检索 all_results [] seen_contents set() for query in queries: results retriever.retrieve(query) for r in results: content_hash hash(r[content]) if content_hash not in seen_contents: seen_contents.add(content_hash) all_results.append(r) # 按相关性排序用第一个查询的得分 all_results.sort(keylambda x: x.get(score, 0), reverseTrue) return all_results[:retriever.final_k]MermaidNaive RAG vs Advanced RAG模块二Advanced RAG的核心技术技术1混合检索Hybrid Search向量检索擅长语义匹配关键词检索擅长精确匹配。混合检索结合两者优势显著提升召回率。 混合检索实现 结合向量检索语义匹配和BM25关键词匹配 import numpy as np from typing import List, Dict, Tuple class HybridRetriever: 混合检索器 结合向量检索和稀疏检索BM25 def __init__(self, vector_store, bm25_index, vector_weight: float 0.7, bm25_weight: float 0.3): Args: vector_store: 向量数据库 bm25_index: BM25索引如rank_bm25 vector_weight: 向量检索得分权重 bm25_weight: BM25得分权重 self.vector_store vector_store self.bm25 bm25_index self.vector_weight vector_weight self.bm25_weight bm25_weight def retrieve(self, query: str, top_k: int 5) - List[Dict]: 混合检索 步骤 1. 向量检索语义相似度 2. BM25检索关键词匹配 3. 归一化得分 4. 加权融合 5. 重新排序 # 1. 向量检索 vector_results self.vector_store.search(query, top_ktop_k * 2) vector_scores self._normalize_scores( [r.get(score, 0) for r in vector_results] ) # 2. BM25检索 bm25_results self.bm25.search(query, top_ktop_k * 2) bm25_scores self._normalize_scores( [r.get(score, 0) for r in bm25_results] ) # 3. 合并得分Reciprocal Rank Fusion或加权求和 merged self._merge_results( vector_results, vector_scores, bm25_results, bm25_scores, top_k ) return merged def _normalize_scores(self, scores: List[float]) - List[float]: Min-Max归一化 if not scores: return scores min_s min(scores) max_s max(scores) if max_s min_s: return [0.5] * len(scores) return [(s - min_s) / (max_s - min_s) for s in scores] def _merge_results(self, vector_results: List[Dict], vector_scores: List[float], bm25_results: List[Dict], bm25_scores: List[float], top_k: int) - List[Dict]: 合并两种检索结果 # 使用RRFReciprocal Rank Fusion rrf_scores {} # 向量检索结果的RRF得分 for rank, (result, score) in enumerate(zip(vector_results, vector_scores), 1): doc_id result.get(id, hash(result[content])) rrf self.vector_weight / (rank 60) # 60是RRF常数 rrf_scores[doc_id] rrf_scores.get(doc_id, 0) rrf # BM25检索结果的RRF得分 for rank, (result, score) in enumerate(zip(bm25_results, bm25_scores), 1): doc_id result.get(id, hash(result[content])) rrf self.bm25_weight / (rank 60) rrf_scores[doc_id] rrf_scores.get(doc_id, 0) rrf # 按RRF得分排序 sorted_docs sorted(rrf_scores.items(), keylambda x: x[1], reverseTrue) # 构建返回结果 result_map {} for r in vector_results bm25_results: doc_id r.get(id, hash(r[content])) result_map[doc_id] r return [ {**result_map[doc_id], hybrid_score: score} for doc_id, score in sorted_docs[:top_k] if doc_id in result_map ]技术2上下文压缩Context Compression检索到的块可能包含大量无关信息。上下文压缩用LLM提取相关内容减少噪音和token消耗。 上下文压缩器 在将检索结果送入LLM之前提取相关内容 class ContextCompressor: 上下文压缩器 使用LLM从检索结果中提取与问题相关的内容 def __init__(self, llm_client, max_compressed_tokens: int 1000): self.llm llm_client self.max_tokens max_compressed_tokens def compress(self, query: str, retrieved_docs: List[Dict]) - str: 压缩检索结果 策略 1. 用LLM从每个文档中提取相关句子 2. 合并所有相关句子 3. 如果超过token限制进一步压缩 compressed_parts [] for i, doc in enumerate(retrieved_docs): # 用LLM提取相关内容 extraction_prompt f根据以下问题从文档中提取所有相关内容。 只输出相关内容不要输出无关内容。 如果文档中没有相关内容输出无相关内容。 问题{query} 文档 {doc[content]} 相关内容 extracted self.llm.generate(extraction_prompt) if 无相关内容 not in extracted and len(extracted.strip()) 10: compressed_parts.append(f[文档{i1}] {extracted.strip()}) # 合并所有相关部分 compressed \n\n.join(compressed_parts) # 如果超过token限制进一步摘要 if len(compressed) self.max_tokens * 4: # 粗略估计 compressed self._summarize(compressed, query) return compressed def _summarize(self, text: str, query: str) - str: 进一步摘要如果压缩后仍然太长 prompt f根据问题{query}对以下内容进行摘要只保留回答该问题所需的信息。 内容 {text} 摘要 return self.llm.generate(prompt)技术3HyDEHypothetical Document EmbeddingsHyDE先用LLM生成一个假设性回答再用这个回答去检索。实验表明这种方法能显著提升检索准确率。 HyDE检索器 先生成假设性回答再进行检索 class HyDERetriever: HyDEHypothetical Document Embeddings检索器 核心思想 1. 用LLM根据问题生成一个假设性回答即使不完全正确 2. 用这个假设性回答的向量去检索 3. 假设性回答的向量比原始问题更接近目标文档 def __init__(self, llm_client, vector_store): self.llm llm_client self.vector_store vector_store def retrieve(self, query: str, top_k: int 5) - List[Dict]: HyDE检索 为什么有效 - 原始问题的向量是查询向量 - 假设性回答的向量更接近文档向量 - 文档向量之间的相似度比查询-文档相似度更高 # 步骤1生成假设性回答 hypothetical_doc self._generate_hypothetical_document(query) # 步骤2用假设性回答进行向量检索 results self.vector_store.search_by_text( texthypothetical_doc, top_ktop_k ) return results def _generate_hypothetical_document(self, query: str) - str: 生成假设性回答 prompt f请写一个可能回答以下问题的文档片段。 不需要完全正确但要包含可能相关的信息。 问题{query} 假设性文档片段 return self.llm.generate(prompt, max_tokens200) def retrieve_with_fusion(self, query: str, top_k: int 5) - List[Dict]: HyDE 原始查询融合检索 结合两种检索结果进一步提升召回率 # HyDE检索 hyde_results self.retrieve(query, top_ktop_k) # 原始查询检索 original_results self.vector_store.search(query, top_ktop_k) # 融合结果去重 merged {} for r in hyde_results original_results: doc_id r.get(id, hash(r[content])) if doc_id not in merged: merged[doc_id] r return list(merged.values())[:top_k]模块三RAG系统的评估与迭代如何评估RAG系统的质量RAG系统的评估需要关注两个核心指标检索质量召回率、精确率生成质量答案准确性、完整性 RAG系统评估框架 from dataclasses import dataclass from typing import List, Dict, Optional dataclass class RAGEvaluationResult: RAG评估结果 retrieval_recall: float # 检索召回率 retrieval_precision: float # 检索精确率 answer_relevance: float # 回答相关性 answer_correctness: float # 回答正确性 faithfulness: float # 忠实度是否幻觉 overall_score: float # 综合得分 class RAGEvaluator: RAG系统评估器 评估维度 1. 检索质量召回率、精确率需要有标注数据 2. 生成质量用LLM-as-Judge评估 3. 端到端用标准数据集评估 def __init__(self, llm_client): self.llm llm_client def evaluate_retrieval(self, queries: List[str], ground_truth_docs: List[List[str]], retrieved_docs: List[List[str]]) - Dict: 评估检索质量 Args: queries: 查询列表 ground_truth_docs: 每个查询的相关文档列表标注 retrieved_docs: 每个查询检索到的文档列表 recalls [] precisions [] for i, query in enumerate(queries): gt_set set(ground_truth_docs[i]) retrieved_set set(retrieved_docs[i]) if not gt_set: continue # 召回率 检索到的相关文档 / 所有相关文档 recall len(gt_set retrieved_set) / len(gt_set) recalls.append(recall) # 精确率 检索到的相关文档 / 所有检索到的文档 if retrieved_set: precision len(gt_set retrieved_set) / len(retrieved_set) else: precision 0 precisions.append(precision) return { mean_recall: round(sum(recalls) / len(recalls), 3) if recalls else 0, mean_precision: round(sum(precisions) / len(precisions), 3) if precisions else 0, f1_score: self._calc_f1(recalls, precisions) } def evaluate_generation(self, queries: List[str], generated_answers: List[str], reference_answers: List[str]) - Dict: 评估生成质量 使用LLM-as-Judge让另一个LLM评估答案质量 scores [] for query, gen_answer, ref_answer in zip(queries, generated_answers, reference_answers): # 用LLM评估 eval_prompt f请对以下回答进行评分1-5分 问题{query} 生成的回答{gen_answer} 参考答案{ref_answer} 评分标准 - 相关性1-5分回答是否相关 - 正确性1-5分回答是否正确 - 完整性1-5分回答是否完整 输出JSON格式 {{relevance: 分数, correctness: 分数, completeness: 分数, reason: 理由}} eval_result self.llm.generate(eval_prompt) # 解析评分简化 import json try: scores_dict json.loads(eval_result) overall (scores_dict.get(relevance, 3) scores_dict.get(correctness, 3) scores_dict.get(completeness, 3)) / 3 scores.append(overall) except json.JSONDecodeError: scores.append(3.0) # 默认中等分数 return { mean_generation_score: round(sum(scores) / len(scores), 3) if scores else 0, individual_scores: scores } def _calc_f1(self, recalls: List[float], precisions: List[float]) - float: 计算F1得分 avg_recall sum(recalls) / len(recalls) if recalls else 0 avg_precision sum(precisions) / len(precisions) if precisions else 0 if avg_recall avg_precision 0: return 0 return round(2 * avg_recall * avg_precision / (avg_recall avg_precision), 3)MermaidRAG评估闭环模块四生产环境中的RAG工程实践工程实践1增量更新与版本管理文档会不断更新RAG系统需要支持增量更新。否则每次文档更新都要重新索引全部内容。 RAG系统的增量更新管理 import hashlib from dataclasses import dataclass from typing import List, Optional from datetime import datetime dataclass class DocumentVersion: 文档版本 doc_id: str content_hash: str created_at: datetime is_deleted: bool False class RAGDocumentManager: RAG文档管理器 支持增量更新和版本管理 def __init__(self, vector_store): self.vector_store vector_store self.doc_versions: Dict[str, DocumentVersion] {} def update_document(self, doc_id: str, new_content: str, chunk_strategy: str semantic) - bool: 更新文档增量更新 策略 1. 计算新内容的hash 2. 与旧版本比较 3. 只更新有变化的部分 new_hash self._calc_hash(new_content) # 检查是否有变化 if doc_id in self.doc_versions: old_version self.doc_versions[doc_id] if old_version.content_hash new_hash: print(f文档 {doc_id} 无变化跳过更新) return False # 删除旧版本的向量 self._delete_old_vectors(doc_id, old_version) # 分块并嵌入新内容 chunks self._chunk_document(new_content, chunk_strategy) embeddings self._embed_chunks(chunks) # 存入向量数据库 vector_ids [] for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)): vector_id self.vector_store.add( embeddingembedding, metadata{ doc_id: doc_id, chunk_index: i, content: chunk, version_hash: new_hash } ) vector_ids.append(vector_id) # 更新版本记录 self.doc_versions[doc_id] DocumentVersion( doc_iddoc_id, content_hashnew_hash, created_atdatetime.now() ) print(f文档 {doc_id} 更新完成新增 {len(chunks)} 个块) return True def delete_document(self, doc_id: str): 删除文档软删除 if doc_id in self.doc_versions: self.doc_versions[doc_id].is_deleted True # 从向量数据库中删除 self.vector_store.delete_by_metadata(doc_id, doc_id) print(f文档 {doc_id} 已删除) def _calc_hash(self, content: str) - str: 计算内容的SHA256哈希 return hashlib.sha256(content.encode()).hexdigest() def _chunk_document(self, content: str, strategy: str) - List[str]: 分块文档简化实现 # 实际应使用模块一中的SemanticChunker return [content[i:i500] for i in range(0, len(content), 500)] def _embed_chunks(self, chunks: List[str]) - List[List[float]]: 将分块转换为向量简化实现 # 实际应调用嵌入模型 import numpy as np return [np.random.rand(768).tolist() for _ in chunks] def _delete_old_vectors(self, doc_id: str, old_version: DocumentVersion): 删除旧版本的向量 self.vector_store.delete_by_metadata(doc_id, doc_id) print(f已删除文档 {doc_id} 的旧版本向量)工程实践2防止幻觉Hallucination的策略RAG的核心价值是减少幻觉。但如果检索质量差LLM仍可能产生幻觉。 防止RAG系统幻觉的策略 class HallucinationGuard: 幻觉防护器 多层策略防止RAG系统产生幻觉 def __init__(self, llm_client): self.llm llm_client def generate_with_guardrails(self, query: str, context: str, max_hallucination_risk: float 0.3) - str: 生成回答并检测幻觉风险 策略 1. 在提示词中明确要求不知道就说不知道 2. 生成后检测幻觉用LLM评估 3. 如果幻觉风险高重新生成或返回无法确定 # 步骤1带防护栏的生成 guarded_prompt f根据以下参考文档回答问题。 规则 - 只根据参考文档回答 - 如果参考文档中没有相关信息明确说根据现有资料无法回答 - 不要编造信息 参考文档 {context} 问题{query} 回答 answer self.llm.generate(guarded_prompt) # 步骤2检测幻觉 hallucination_risk self._detect_hallucination(query, context, answer) # 步骤3如果风险高尝试重新生成或返回保守回答 if hallucination_risk max_hallucination_risk: print(f检测到高幻觉风险{hallucination_risk:.2f}使用保守策略) return 根据现有资料我无法给出确切答案。建议您咨询相关专业人士。 return answer def _detect_hallucination(self, query: str, context: str, answer: str) - float: 检测幻觉风险 返回0-1之间的风险分数 detection_prompt f评估以下回答的幻觉风险。 参考文档 {context} 问题{query} 回答{answer} 评估标准 - 回答中的所有信息是否都能在参考文档中找到依据 - 回答是否添加了参考文档中没有的信息 输出JSON格式 {{hallucination_risk: 0.0-1.0的分数, reason: 理由}} 只输出JSON不要输出其他内容。 result self.llm.generate(detection_prompt) import json try: result_dict json.loads(result) return float(result_dict.get(hallucination_risk, 0.5)) except (json.JSONDecodeError, ValueError): return 0.5 # 默认中等风险模块五从Naive RAG到Advanced RAG的迁移路径渐进式升级策略不需要一次性实现所有Advanced RAG技术。可以逐步升级每次升级都测量效果提升。 RAG系统升级路径规划 UPGRADE_ROADMAP [ { stage: 1, name: Naive RAG, features: [基础向量检索, Top-K直接拼接], expected_improvement: 基准 }, { stage: 2, name: 更好的分块, features: [语义感知分块, 重叠窗口], expected_improvement: 召回率15% }, { stage: 3, name: 查询改写, features: [查询扩展, 多查询检索], expected_improvement: 召回率20% }, { stage: 4, name: 混合检索, features: [向量BM25, 得分融合], expected_improvement: 召回率25% }, { stage: 5, name: 精排, features: [Reranker模型, 两阶段检索], expected_improvement: 精确率30% }, { stage: 6, name: 上下文压缩, features: [相关内容提取, 动态token管理], expected_improvement: 生成质量20% }, { stage: 7, name: Advanced RAG, features: [HyDE, 自我评估, 迭代优化], expected_improvement: 端到端质量40% }, ] def print_upgrade_roadmap(): 打印升级路线图 print( * 70) print(RAG系统升级路线图) print( * 70) for step in UPGRADE_ROADMAP: print(f\n阶段{step[stage]}{step[name]}) print(f 功能{, .join(step[features])}) print(f 预期提升{step[expected_improvement]}) print(\n建议每次升级后都用评估框架测量效果确保升级有效) def should_upgrade_to_next_stage(current_metrics: Dict) - bool: 判断是否应该升级到下一阶段 决策规则 1. 当前阶段的指标已经稳定 2. 进一步优化当前阶段收益递减 3. 用户反馈显示有明确痛点 # 示例决策逻辑 if current_metrics.get(retrieval_recall, 0) 0.7: print(建议先提升检索召回率升级到阶段3-4) return True if current_metrics.get(answer_correctness, 0) 0.7: print(建议先提升回答质量升级到阶段5-6) return True return FalseMermaid7个反模式与解决方案对照纯技术总结7大反模式固定分块破坏语义、无重排序精确率低、无查询改写召回率低、无混合检索、无上下文压缩、无幻觉防护、无评估迭代Advanced RAG核心技术语义分块句子级重叠、两阶段检索向量召回精排、查询改写LLM扩展、混合检索向量BM25RRF融合、HyDE假设性回答检索、上下文压缩LLM提取相关句评估框架检索质量召回率/精确率/F1、生成质量LLM-as-Judge评分、端到端标准数据集工程实践增量更新按hash检测变化、幻觉防护防护栏提示词事后检测、版本管理软删除向量删除升级路径7阶段渐进升级每阶段后用评估框架验证效果