RAG的核心挑战 RAG（检索增强生成）让LLM能够基于私有知识回答问题。但一个简单的RAG原型——文档切块→向量化→相似度检索→拼接到prompt——在实际使用中往往效果不佳。2026年的RAG优化需要从文档处理、检索质量、上下文管理和生成控制四个维度系统提升。
文档处理优化 智能分块 from langchain.text_splitter import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter class SmartChunker: def __init__(self, chunk_size=512, chunk_overlap=64): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap # 通用分割器 self.general_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""] ) # Markdown分割器 self.md_splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] ) def chunk(self, text, doc_type="auto"): if doc_type == "auto": doc_type = self.detect_type(text) if doc_type == "markdown": # 先按标题分块，再按大小细分 md_chunks = self.md_splitter.split_text(text) final_chunks = [] for chunk in md_chunks: if len(chunk.page_content) > self.chunk_size * 2: sub_chunks = self.general_splitter.split_text(chunk.page_content) for sc in sub_chunks: sc.metadata = chunk.metadata final_chunks.append(sc) else: final_chunks.append(chunk) return final_chunks else: return self.general_splitter.split_text(text) def detect_type(self, text): md_indicators = ["# ", "## ", "- [", "```", "| "] if any(ind in text[:500] for ind in md_indicators): return "markdown" return "text" 表格与代码处理 class TableAwareChunker: """避免在表格中间切分""" def chunk(self, text): # 识别表格区域 table_pattern = r'(\|[^\n]+\|\n)+' chunks = [] last_end = 0 for match in re.finditer(table_pattern, text): # 表格前的文本正常切分 pre_text = text[last_end:match.start()] if pre_text.strip(): chunks.extend(self.split_text(pre_text)) # 表格作为完整块 chunks.append(text[match.start():match.end()]) last_end = match.end() # 剩余文本 if last_end < len(text): chunks.extend(self.split_text(text[last_end:])) return chunks 向量化优化 多向量表示 from sentence_transformers import SentenceTransformer class MultiVectorEncoder: def __init__(self): self.dense_model = SentenceTransformer('BAAI/bge-large-zh-v1.5') # 稀疏向量用于关键词匹配 self.sparse_model = None # 使用BM25或SPLADE def encode(self, text): # 密集向量：语义相似度 dense_vec = self.dense_model.encode(text, normalize_embeddings=True) # 稀疏向量：精确匹配 sparse_vec = self.sparse_model.encode(text) if self.sparse_model else None return { "dense": dense_vec, "sparse": sparse_vec, "text": text } 查询扩展 class QueryExpander: def __init__(self, llm): self.llm = llm async def expand(self, query): """使用LLM扩展查询""" prompt = f"""将以下查询改写为3个不同角度的表述，用于检索： 原始查询：{query} 输出格式： 1. [改写1] 2. [改写2] 3. [改写3]""" response = await self.llm.ainvoke(prompt) expansions = self.parse_expansions(response) expansions.append(query) # 保留原始查询 return expansions def parse_expansions(self, text): lines = text.strip().split('\n') return [line.split('.', 1)[1].strip() for line in lines if '.' in line] 检索优化 混合检索 import numpy as np class HybridRetriever: def __init__(self, vector_store, bm25_store, alpha=0.7): self.vector_store = vector_store # 密集向量检索 self.bm25_store = bm25_store # BM25稀疏检索 self.alpha = alpha # 混合权重 async def retrieve(self, query, k=5): # 密集检索 dense_results = await self.vector_store.asimilarity_search_with_score( query, k=k*2 ) # 稀疏检索 sparse_results = self.bm25_store.search(query, k=k*2) # 分数归一化 dense_scores = self.normalize_scores([s for _, s in dense_results]) sparse_scores = self.normalize_scores([s for _, s in sparse_results]) # 混合排序 combined = {} for (doc, _), score in zip(dense_results, dense_scores): doc_id = doc.metadata["id"] combined[doc_id] = combined.get(doc_id, 0) + self.alpha * score combined.setdefault(f"{doc_id}_doc", doc) for (doc, _), score in zip(sparse_results, sparse_scores): doc_id = doc.metadata["id"] combined[doc_id] = combined.get(doc_id, 0) + (1 - self.alpha) * score combined.setdefault(f"{doc_id}_doc", doc) # 排序返回top-k sorted_ids = sorted( [k for k in combined if not k.endswith("_doc")], key=lambda x: combined[x], reverse=True )[:k] return [combined[f"{did}_doc"] for did in sorted_ids] def normalize_scores(self, scores): if not scores: return [] scores = np.array(scores) if scores.max() > scores.min(): return (scores - scores.min()) / (scores.max() - scores.min()) return np.ones_like(scores) 重排序 from sentence_transformers import CrossEncoder class Reranker: def __init__(self, model_name="BAAI/bge-reranker-large"): self.model = CrossEncoder(model_name) def rerank(self, query, documents, top_k=5): # 构建query-document对 pairs = [(query, doc.page_content) for doc in documents] # 计算相关性分数 scores = self.model.predict(pairs) # 排序 ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True) return [doc for doc, _ in ranked[:top_k]] 上下文管理 动态上下文窗口 class ContextManager: def __init__(self, max_tokens=4096): self.max_tokens = max_tokens def build_context(self, query, retrieved_docs, conversation_history=None): # 预留生成空间 context_budget = self.max_tokens - 512 # 分配：历史对话30%，检索文档70% history_budget = int(context_budget * 0.3) docs_budget = context_budget - history_budget # 构建历史上下文 history_text = self.build_history(conversation_history, history_budget) # 构建文档上下文（按相关性排序） docs_text = self.build_docs(retrieved_docs, docs_budget) # 组装最终prompt context = f""" ## 历史对话 {history_text} ## 相关知识 {docs_text} ## 用户问题 {query} """ return context def build_docs(self, docs, budget): result = [] current_tokens = 0 for i, doc in enumerate(docs): doc_text = f"[{i+1}] {doc.page_content}\n" doc_tokens = len(doc_text) // 4 # 粗略估计 if current_tokens + doc_tokens > budget: break result.append(doc_text) current_tokens += doc_tokens return "\n".join(result) 引用标注 class CitationGenerator: def __init__(self, llm): self.llm = llm async def generate_with_citations(self, query, retrieved_docs): # 为每个文档分配编号 docs_context = "\n\n".join([ f"[{i+1}] {doc.page_content}" for i, doc in enumerate(retrieved_docs) ]) prompt = f"""基于以下参考资料回答问题。在回答中使用 [编号] 标注信息来源。 参考资料： {docs_context} 问题：{query} 要求： 1. 只使用参考资料中的信息 2. 用 [编号] 标注每条信息的来源 3. 如果资料中没有相关信息，说明"根据现有资料无法回答" """ response = await self.llm.ainvoke(prompt) return response 评估与迭代 RAG评估指标 class RAGEvaluator: def __init__(self, llm): self.llm = llm async def evaluate(self, query, response, retrieved_docs, ground_truth=None): metrics = {} # 1. 检索相关性 metrics["retrieval_relevance"] = await self.eval_retrieval( query, retrieved_docs ) # 2. 回答忠实度（是否基于检索内容） metrics["faithfulness"] = await self.eval_faithfulness( response, retrieved_docs ) # 3. 回答完整性 if ground_truth: metrics["completeness"] = await self.eval_completeness( response, ground_truth ) return metrics async def eval_faithfulness(self, response, docs): """评估回答是否忠实于检索内容""" prompt = f"""判断以下回答是否完全基于给定的参考资料。 参考资料：{' '.join(d.page_content[:200] for d in docs)} 回答：{response} 请输出： 1. 忠实/不忠实 2. 不忠实的部分（如有） """ result = await self.llm.ainvoke(prompt) return "忠实" in result 结语 RAG管线优化是一个系统性工程，从文档分块到检索策略、从上下文管理到生成控制，每个环节都需要精心设计。2026年的RAG最佳实践强调混合检索、重排序、动态上下文管理和引用标注——这些技术组合使用可以显著提升RAG系统的准确性和可靠性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
