RAG

引言 Haystack是deepset开发的企业级NLP框架，在RAG领域有着深厚积累。2026年的Haystack已经发展成为一个完整的RAG解决方案框架。本文将分享Haystack在RAG实践中的经验。 Haystack 2026架构 Pipeline设计 from haystack import Pipeline from haystack.components.embedders import OllamaEmbedder from haystack.components.retrievers import ChromaRetriever from haystack.components.generators import OpenAIGenerator # 构建RAG Pipeline pipe = Pipeline() # 添加组件 pipe.add_component("embedder", OllamaEmbedder(model="bge-large-zh")) pipe.add_component("retriever", ChromaRetriever(top_k=5)) pipe.add_component("generator", OpenAIGenerator(model="gpt-5")) # 连接组件 pipe.connect("embedder.embedding", "retriever.query_embedding") pipe.connect("retriever.documents", "generator.documents") 文档处理 from haystack.components.converters import PDFToDocument, MarkdownToDocument from haystack.components.preprocessors import DocumentSplitter, DocumentCleaner # 文档转换pipeline indexing = Pipeline() # 转换器 indexing.add_component("pdf_converter", PDFToDocument()) indexing.add_component("md_converter", MarkdownToDocument()) # 预处理 indexing.add_component("cleaner", DocumentCleaner()) indexing.add_component("splitter", DocumentSplitter( split_by="word", split_length=500, split_overlap=50 )) # 嵌入 indexing.add_component("embedder", OllamaEmbedder(model="bge-large-zh")) indexing.add_component("writer", ChromaDocumentWriter()) # 连接 indexing.connect("pdf_converter.documents", "cleaner.documents") indexing.connect("cleaner.documents", "splitter.documents") indexing.connect("splitter.documents", "embedder.documents") indexing.connect("embedder.documents", "writer.documents") RAG优化实践 实践一：混合检索 from haystack.components.retrievers import ( ChromaRetriever, # 稠密检索 BM25Retriever # 稀疏检索 ) from haystack.components.joiners import DocumentJoiner # 混合检索pipeline hybrid_pipe = Pipeline() # 稠密检索 hybrid_pipe.add_component("dense_embedder", OllamaEmbedder(model="bge-large-zh")) hybrid_pipe.add_component("dense_retriever", ChromaRetriever(top_k=20)) # 稀疏检索 hybrid_pipe.add_component("sparse_retriever", BM25Retriever(top_k=20)) # 融合 hybrid_pipe.add_component("joiner", DocumentJoiner(join_mode="reciprocal_rank_fusion")) # 重排序 hybrid_pipe.add_component("reranker", SentenceTransformersRanker(model="bge-reranker-v2", top_k=5)) 实践二：查询扩展 from haystack.components.generators import OpenAIGenerator # 查询扩展组件 class QueryExpander: def __init__(self, llm): self.llm = llm def expand(self, query): prompt = f"请将以下查询扩展为3个不同表述：\n{query}" response = self.llm.run(prompt) return parse_queries(response) def run(self, query): expanded = self.expand(query) return {"queries": expanded} # 在pipeline中使用 pipe.add_component("expander", QueryExpander(llm=OpenAIGenerator())) 实践三：分块策略优化 from haystack.components.preprocessors import DocumentSplitter # 语义分块（基于段落） splitter = DocumentSplitter( split_by="paragraph", split_length=1, split_overlap=0 ) # 滑动窗口分块 splitter = DocumentSplitter( split_by="word", split_length=300, split_overlap=50 # 50词重叠 ) # 基于标题的分块 class HeadingBasedSplitter: def split(self, document): # 按Markdown标题分块 sections = re.split(r'^#+\s', document.content, flags=re.MULTILINE) return [Document(content=s.strip()) for s in sections if s.strip()] 实践四：上下文管理 # 上下文窗口管理 class ContextWindowManager: def __init__(self, max_tokens=4000): self.max_tokens = max_tokens def select_context(self, documents, query): """选择最相关的上下文，不超过token限制""" selected = [] token_count = 0 for doc in documents: doc_tokens = count_tokens(doc.content) if token_count + doc_tokens > self.max_tokens: # 截断最后一个文档 remaining = self.max_tokens - token_count if remaining > 100: # 至少100 token才包含 doc.content = doc.content[:remaining] selected.append(doc) break selected.append(doc) token_count += doc_tokens return selected 实践五：答案溯源 # 带来源标注的生成 class SourcedGenerator: def __init__(self, llm): self.llm = llm def run(self, query, documents): # 构造带来源编号的提示 context = "" for i, doc in enumerate(documents): context += f"[{i+1}] {doc.content}\n\n" prompt = f""" 基于以下参考信息回答问题。在回答中标注信息来源。 参考信息： {context} 问题：{query} 回答格式：答案内容[来源编号] """ response = self.llm.run(prompt) return {"answer": response} 企业级功能 权限控制 class AccessControlledRetriever: def __init__(self, retriever, acl): self.retriever = retriever self.acl = acl # 访问控制列表 def run(self, query, user_id): # 检索 documents = self.retriever.run(query) # 过滤：只返回用户有权限的文档 accessible = [ doc for doc in documents if self.acl.has_access(user_id, doc.metadata.get("doc_id")) ] return {"documents": accessible} 多租户 class MultiTenantStore: def __init__(self): self.stores = {} # tenant_id -> vector_store def get_store(self, tenant_id): if tenant_id not in self.stores: self.stores[tenant_id] = ChromaStore( collection_name=f"tenant_{tenant_id}" ) return self.stores[tenant_id] 缓存 from haystack.components.cachers import CacheChecker pipe.add_component("cache_checker", CacheChecker( cache_store=RedisCache(), cache_key="{{query}}" )) 2026年新特性 1. 多模态RAG from haystack.components.embedders import CLIPEmbedder # 图文混合RAG pipe.add_component("image_embedder", CLIPEmbedder()) pipe.add_component("text_embedder", OllamaEmbedder(model="bge-large-zh")) 2. 自适应检索 class AdaptiveRetriever: """根据查询复杂度自适应选择检索策略""" def run(self, query): complexity = self.assess_complexity(query) if complexity == "simple": return self.simple_retrieve(query) elif complexity == "medium": return self.hybrid_retrieve(query) else: return self.multi_hop_retrieve(query) 3. 评估集成 from haystack.components.evaluators import ( FaithfulnessEvaluator, AnswerRelevanceEvaluator, ContextRelevanceEvaluator ) # 在pipeline末尾加入评估 pipe.add_component("faithfulness", FaithfulnessEvaluator()) pipe.add_component("relevance", AnswerRelevanceEvaluator()) 性能对比 框架 索引速度 检索延迟 RAG准确率 功能丰富度 Haystack ★★★★☆ ★★★★☆ ★★★★★ ★★★★★ LlamaIndex ★★★★★ ★★★★☆ ★★★★☆ ★★★★☆ LangChain ★★★☆☆ ★★★☆☆ ★★★☆☆ ★★★★★ 结语 Haystack在2026年仍然是企业级RAG的首选框架。其Pipeline架构清晰、组件丰富、可扩展性强，特别适合需要精细控制RAG流程的企业应用。 ...

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论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 嵌入模型是RAG系统、语义搜索和推荐系统的基石。虽然不如生成模型那样引人注目，但嵌入模型的质量直接决定了上游应用的效果上限。2026年，嵌入模型在多语言、长文本和领域特化方面取得了显著进步。本文将为你提供全面的选型指南。 为什么嵌入模型很重要 嵌入模型将文本转化为高维向量，使得语义相似的文本在向量空间中距离更近。这个看似简单的操作是以下应用的核心： RAG系统：通过向量检索找到相关文档 语义搜索：理解用户查询的真实意图 推荐系统：基于内容相似度推荐 聚类分析：发现文本集合中的主题结构 去重与匹配：识别语义重复内容 嵌入模型的质量提升10%，可能带来RAG系统端到端效果提升20-30%。 主流嵌入模型 商业API OpenAI text-embedding-4-large — 3072维，综合最强 Voyage AI voyage-3 — 2048维，检索特化 Cohere Embed v4 — 1536维，多语言强 Google text-gecko-3 — 768维，速度快 开源模型 BGE-large-zh-v2 (智源) — 1024维，中文最强 E5-mistral-7b (微软) — 4096维，英文最强 GTE-large-zh (阿里) — 1024维，中文优秀 Jina embeddings v3 (Jina AI) — 1024维，长文本强 Nomic embed v2 (Nomic) — 768维，开源可复现 BGE-M3 (智源) — 1024维，多语言+多功能 核心基准对比 MTEB 2026 MTEB（Massive Text Embedding Benchmark）是最权威的嵌入模型评估基准： 英文检索 模型 维度 得分 特点 voyage-3 2048 72.8 检索之王 text-embedding-4-large 3072 71.5 综合最佳 E5-mistral-7b 4096 70.3 开源英文最佳 Cohere Embed v4 1536 69.7 多语言均衡 Jina embeddings v3 1024 68.5 长文本强 Nomic embed v2 768 66.2 轻量高效 中文检索 模型 维度 得分 特点 BGE-large-zh-v2 1024 74.5 中文之王 GTE-large-zh 1024 72.8 中文优秀 BGE-M3 1024 71.3 多语言兼中文 text-embedding-4-large 3072 70.2 商业中文最佳 Cohere Embed v4 1536 68.7 长文本检索 在处理长文档（>8K tokens）的检索任务上： ...

长尾问题：AI知识的冰山模型 AI模型的知识像一座冰山——水面之上是常见知识（高频、大量训练数据覆盖），水面之下是长尾知识（低频、训练数据稀少）。 GPT-6能回答关于Python编程、美国历史、量子力学的常见问题——这是冰山之上。但当被问到"15世纪蒙古萨满教的具体仪式"或"某种罕见遗传病的最新治疗方案"时，AI的表现急剧下降——这是冰山之下。 长尾知识的重要性 为什么长尾很重要 专业性：专家工作中最需要的就是长尾知识 创新性：突破性创新往往来自长尾领域的交叉 公平性：罕见疾病患者、少数族裔语言、小众领域——AI在这些领域的无能会造成实际伤害 可靠性：如果AI在长尾问题上"幻觉"，用户可能无法识别（因为用户也不懂） 长尾知识的类型 类型1：罕见但存在 罕见病（发病率<1/10000的疾病） 小众语言（使用人数<10万的语言） 冷门历史事件 类型2：新产生 2026年最新发表的论文 刚刚发生的新闻 新发布的软件版本 类型3：专业深度 某个特定法律条文的具体适用 某种材料的特定合成条件 某个算法在特定硬件上的优化参数 类型4：隐性知识 无法用文字充分描述的技能 行业内部的"潜规则" 需要实践经验才能理解的知识 当前AI在长尾上的表现 量化评估 知识频率 AI准确率 人类专家准确率 前1%（常见） 94.2% 92.1% 前1-5% 87.3% 88.5% 前5-20% 72.1% 79.3% 后20%（长尾） 41.6% 65.2% 在长尾知识上，AI的准确率急剧下降，而人类专家虽然也有下降，但幅度小得多。 典型问题 1. 幻觉 在长尾问题上，AI倾向于"编造"看似合理但实际错误的答案。因为训练数据稀少，模型无法区分"知道"和"不知道"。 2. 过度泛化 AI将常见知识错误地泛化到罕见场景： 罕见病的症状被误诊为常见病 小众语言的语法被按主流语言处理 冷门法律被按主流法律解释 3. 置信度校准失败 AI在长尾问题上的置信度通常过高——它"不知道自己不知道"。这比"不知道"更危险，因为用户可能信任高置信度的错误答案。 解决方案 1. 检索增强生成（RAG） 最直接的方案——在生成答案前检索外部知识库： 用户问题 → 判断是否需要外部知识 → 检索相关文档 → 结合检索结果生成答案 2026年RAG的改进： 多跳检索：支持多步骤的信息检索 混合检索：向量检索+关键词检索+语义重排序 自适应检索：根据问题难度调整检索深度 实时检索：连接实时数据源获取最新信息 局限：RAG的效果取决于外部知识库的覆盖度和质量。对于非常冷门的知识，可能根本没有可检索的文档。 ...

引言 基础RAG（检索-增强-生成）已经成为了AI应用的标配：用户提问→检索相关文档→LLM基于文档生成回答。但这种简单模式在面对复杂问题时力不从心。 2026年，RAG架构已经演进出多种高级模式：多跳检索、自适应检索、推理增强检索、迭代检索等。这些模式让RAG系统能够处理更复杂的问题、提供更准确的答案。本文将系统介绍这些高级模式。 一、基础RAG的局限 1.1 单次检索不够 复杂问题需要多步推理和多轮检索： 问题: "对比GPT-4和Claude在代码生成任务上的表现" 基础RAG: 一次检索"GPT-4 Claude 代码生成对比" → 可能找不到直接的对比文档 高级RAG: 检索1: "GPT-4 代码生成性能" 检索2: "Claude 代码生成性能" 检索3: "GPT-4 vs Claude benchmark" → 综合多个来源生成对比 1.2 检索质量不稳定 简单向量检索可能返回语义相关但信息不足的文档。 1.3 缺乏推理能力 基础RAG是"检索+拼接"，缺乏对检索结果的推理和整合。 二、多跳检索 2.1 原理 多跳检索（Multi-hop Retrieval）通过多轮检索逐步收集信息，每轮检索基于前一轮的结果。 问题: "那位发明了Transformer的作者后来创立了哪家公司？" 跳1: 检索"Transformer论文作者" → 得到作者列表（包括Ashish Vaswani等） 跳2: 检索"Ashish Vaswani 创立公司" → 得到Adept AI 答案: Adept AI 2.2 实现 class MultiHopRetriever: def __init__(self, retriever, llm, max_hops=3): self.retriever = retriever self.llm = llm self.max_hops = max_hops async def retrieve(self, question): accumulated_context = [] for hop in range(self.max_hops): # 1. 根据当前上下文生成下一步检索查询 if hop == 0: query = question else: query = await self.generate_next_query( question, accumulated_context ) if query is None: # 不需要更多检索 break # 2. 执行检索 results = await self.retriever.search(query, top_k=5) # 3. 过滤和评估结果 relevant = await self.filter_relevant(results, question, accumulated_context) accumulated_context.extend(relevant) # 4. 检查是否已有足够信息回答 if await self.has_enough_info(question, accumulated_context): break return accumulated_context async def generate_next_query(self, question, context): """基于当前上下文生成下一步查询""" prompt = f""" 原始问题: {question} 已知信息: {context} 还需要什么信息才能回答问题？请生成一个搜索查询。 如果已经有足够信息，返回"DONE"。 """ response = await self.llm.call(prompt) if "DONE" in response: return None return response.strip() 三、自适应检索 3.1 原理 不是所有问题都需要检索，也不是所有问题都需要相同类型的检索。自适应检索根据问题特征选择最优的检索策略。 ...

引言 检索增强生成（RAG）是2026年最实用的大模型应用架构之一。从基础的"检索-生成"流水线，到Agentic RAG的智能检索代理，RAG技术正在经历深刻的架构演进。本文系统梳理RAG的发展脉络和技术选型。 RAG 的演进阶段 第一阶段：基础 RAG（Naive RAG） 最原始的RAG架构，包含三个步骤： 将文档切分为片段 对片段进行向量化并存储 检索最相关的片段，拼接后输入模型 优点： 简单、快速、易于实现 缺点： 检索质量依赖单一向量相似度，无法处理复杂查询 第二阶段：增强 RAG（Enhanced RAG） 在基础RAG之上增加优化： 查询优化： 查询重写：将用户问题转化为更适合检索的形式 查询分解：将复杂问题拆分为多个子查询 多路检索：同时使用向量检索和关键词检索 后处理： 重排序（Reranking）：对检索结果进行精排 上下文压缩：去除冗余片段，保留关键信息 自适应窗口：根据问题类型调整上下文长度 第三阶段：Agentic RAG Agentic RAG的核心思想：让Agent自主决定如何检索、何时检索、检索什么。 ┌─────────────────────────────────────────────┐ │ Agentic RAG 架构 │ │ │ │ ┌──────────┐ ┌──────────┐ │ │ │ Agent │───▶│ 检索 │ │ │ │ 规划器 │ │ 策略 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ │ │ │ 反思 │◀───│ 多路 │ │ │ │ 与调整 │ │ 检索 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────────────────┐ │ │ │ 最终生成与验证 │ │ │ └──────────────────────────┘ │ └─────────────────────────────────────────────┘ 主流 RAG 框架对比 LangChain / LangGraph 特点： 生态最丰富，组件最全 适合： 快速原型开发、复杂RAG管线 ...

引言 幻觉（Hallucination）是大模型最顽固的问题之一。即使是最先进的模型，在回答事实性问题时仍会生成看似合理但完全错误的信息。2026年，随着大模型在更多关键场景中的应用，理解幻觉的根因并有效缓解成为刚需。 幻觉的五大根因 1. 训练数据局限 大模型的知识截止于训练数据，对于训练数据中不存在或过时的信息，模型会通过"推理"生成看似合理的答案。 典型表现： 引用不存在的论文或研究 编造公司或产品 给出过时的法律条款 2. 概率采样本质 语言模型基于概率分布生成文本，其本质是"最可能的下一个词"而非"最正确的事实"。当多个答案在概率上接近时，模型可能选择错误的那个。 3. 过度泛化 模型在训练中学到的一般规律被错误地应用到特定场景，导致"以偏概全"的错误推断。 4. 指令误解 当用户指令模糊或存在歧义时，模型可能以错误的方式理解意图，生成偏离事实的回答。 5. 自洽性缺失 大模型缺乏对自身输出的内在验证机制，无法区分"我知道的"和"我认为可能的"。 2026年主流缓解技术 技术一：RAG增强检索 通过检索外部知识库补充模型的知识，是当前最实用的方案。 # RAG 幻觉缓解流程 def rag_mitigate(query): # 1. 检索相关文档 docs = retriever.search(query, top_k=5) # 2. 相关性过滤 relevant_docs = filter_relevance(docs, query) # 3. 带检索的生成 response = model.generate( prompt=f"基于以下信息回答：{relevant_docs}\n\n问题：{query}" ) # 4. 事实一致性检查 if check_factuality(response, relevant_docs) < threshold: return "我无法确认该信息的准确性" return response 效果： 在事实性问题上的幻觉率降低60-80% 技术二：自我反思（Self-Reflection） 让模型对自己的输出进行二次检查，发现并修正可能的错误。 方法： Chain-of-Verification：先生成答案，再验证每个事实点 Self-Correction：让模型识别自己回答中的矛盾 Multi-Agent Debate：多个Agent相互质疑 技术三：不确定性校准 让模型学会表达"不知道"，而非强行编造答案。 实现方式： 置信度评分：对每个预测输出置信度 拒绝回答：当置信度低于阈值时，明确告知"不确定" 证据标注：要求模型标注每个事实点的来源 技术四：微调与对齐 通过偏好数据微调，让模型更倾向于给出准确而非"有趣"的答案。 ...

引言 记忆系统是Agent智能的核心支柱之一。一个没有记忆的Agent就像没有长期记忆的人类——每次对话都是第一次见面。2026年，Agent记忆系统从简单的向量检索演进为多层次、多模态的持久记忆架构。 记忆系统的三层架构 第一层：工作记忆（Working Memory） 工作记忆对应Agent的当前上下文窗口，负责处理当前任务所需的即时信息。 技术要点： 上下文窗口扩展：200K+ tokens成为主流 滑动窗口机制：保留最相关的历史片段 注意力蒸馏：从长上下文中提取关键信息 第二层：短期记忆（Short-term Memory） 短期记忆存储最近几天到几周内与当前任务相关的数据。 实现方案： 向量数据库（Milvus、Qdrant、Weaviate） 增量索引与自动清理 基于任务相似度的检索优化 第三层：长期记忆（Long-term Memory） 长期记忆是Agent的"知识库"，存储经过整理、抽象后的持久化信息。 关键技术： 知识图谱（Neo4j、NebulaGraph） 结构化摘要生成 记忆融合与冲突解决 2026年记忆系统的前沿进展 1. 神经符号记忆 将神经网络的感知能力与符号推理的逻辑能力结合，实现更可靠的长期记忆。 # 伪代码：神经符号记忆融合 class NeuroSymbolicMemory: def store(self, experience): # 神经网络提取特征 embedding = self.encoder(experience) # 符号系统提取关系 schema = self.knowledge_extractor(experience) # 联合存储 self.vector_db.upsert(embedding) self.graph_db.merge(schema) def recall(self, query): # 双通道检索 vector_hits = self.vector_db.search(query) graph_hits = self.graph_db.traverse(query) # 重排序融合 return self.reranker(vector_hits + graph_hits) 2. 记忆压缩与摘要 面对无限增长的记忆数据，如何高效压缩和摘要成为关键问题。 主流方法： 基于重要性的记忆保留 周期性记忆摘要生成 基于任务的记忆激活 3. 多模态记忆 2026年的Agent记忆不再局限于文本，而是支持图像、音频、视频等多模态数据的存储与检索。 工程实践建议 选型指南 场景 推荐方案 快速原型 Chroma + 本地存储 生产环境 Milvus / Qdrant 集群 复杂关系 Neo4j + 向量混合 多模态 Milvus + CLIP编码 常见陷阱 记忆污染：错误信息被永久存储 检索延迟：大规模向量检索的性能瓶颈 记忆冲突：不同时期信息之间的矛盾 隐私泄露：敏感信息未做脱敏处理 结语 Agent记忆系统是通向真正智能的关键一步。随着技术的持续演进，未来的Agent将拥有越来越接近人类的记忆能力——不仅记得住，还能想得深。 ...

从Naive RAG到Agentic RAG的演进 RAG技术的发展经历了三个阶段： Naive RAG (2023) → Advanced RAG (2024) → Agentic RAG (2025-2026) 朴素检索 增强检索 智能体检索 单次查询 多次优化 自主决策 固定管道 模块化 动态规划 传统RAG是一个"固定管道"：查询→检索→拼接→生成，一锤子买卖。Agentic RAG将LLM作为"大脑"，让它自主决定是否需要检索、检索什么、检索几次、何时停止。这就像从"查字典"升级为"咨询一个会查资料的研究员"。 Agentic RAG的核心架构 ┌─────────────────────────────────────────────────────┐ │ Agentic RAG Architecture │ ├─────────────────────────────────────────────────────┤ │ │ │ ┌──────────┐ ┌──────────────────────────────┐ │ │ │ User │───▶│ Agent Controller │ │ │ │ Query │ │ (LLM + Planning + Memory) │ │ │ └──────────┘ └──────────┬───────────────────┘ │ │ │ │ │ ┌─────────┼─────────┐ │ │ ▼ ▼ ▼ │ │ ┌─────────┐ ┌───────┐ ┌─────────┐ │ │ │ Router │ │ Query │ │ Evaluator│ │ │ │ (路由) │ │Rewrite│ │ (评估器) │ │ │ └────┬────┘ └───┬───┘ └────┬────┘ │ │ │ │ │ │ │ ┌─────────┼──────────┼──────────┘ │ │ ▼ ▼ ▼ │ │ ┌──────────┐ ┌────────┐ ┌──────────┐ ┌─────────┐ │ │ │ Vector │ │ Web │ │ SQL │ │ Graph │ │ │ │ Store │ │ Search │ │ DB │ │ Store │ │ │ └──────────┘ └────────┘ └──────────┘ └─────────┘ │ │ │ └─────────────────────────────────────────────────────┘ 四大核心组件 组件 职责 关键能力 Agent Controller 任务规划与决策 分解问题、决定工具调用顺序 Router 工具路由 选择合适的数据源（向量库/搜索引擎/数据库） Query Rewriter 查询改写 将用户问题改写为更有效的检索查询 Evaluator 结果评估 判断检索结果是否充分，决定是否需要追加检索 核心模式解析 模式1：迭代检索（Iterative Retrieval） from langgraph.graph import StateGraph, END from typing import TypedDict, List, Annotated import operator class AgentState(TypedDict): question: str retrieved_docs: Annotated[List[str], operator.add] answer: str iteration: int sufficient: bool def retrieve_node(state: AgentState) -> dict: """基于当前问题+已有信息生成查询并检索""" query = query_rewriter(state["question"], state["retrieved_docs"]) docs = vector_store.similarity_search(query, k=5) return {"retrieved_docs": docs, "iteration": state["iteration"] + 1} def evaluate_node(state: AgentState) -> dict: """评估检索结果是否足以回答问题""" if state["iteration"] >= 3: return {"sufficient": True} prompt = f"""基于以下检索到的文档，能否完整回答用户问题？ 问题: {state['question']} 文档: {state['retrieved_docs']} 如果信息充分回答"SUFFICIENT"，否则回答"INSUFFICIENT"并说明缺什么。""" response = llm.invoke(prompt) sufficient = "SUFFICIENT" in response return {"sufficient": sufficient} def answer_node(state: AgentState) -> dict: """基于所有检索文档生成最终回答""" prompt = f"问题: {state['question']}\n\n参考资料:\n{state['retrieved_docs']}" return {"answer": llm.invoke(prompt)} # 构建工作流 workflow = StateGraph(AgentState) workflow.add_node("retrieve", retrieve_node) workflow.add_node("evaluate", evaluate_node) workflow.add_node("answer", answer_node) workflow.set_entry_point("retrieve") workflow.add_edge("retrieve", "evaluate") workflow.add_conditional_edges( "evaluate", lambda state: "answer" if state["sufficient"] else "retrieve" ) workflow.add_edge("answer", END) agent = workflow.compile() 模式2：多路检索（Multi-Route Retrieval） def route_query(state: AgentState) -> str: """根据问题类型选择检索路径""" routing_prompt = f""" 分析以下问题，选择最合适的数据源： 问题: {state['question']} 选项: - "vector": 适合从内部文档库检索事实性信息 - "web": 适合需要最新信息的时效性问题 - "sql": 适合结构化数据查询 - "graph": 适合关系推理和多方关联 返回JSON: {{"route": "...", "reason": "..."}} """ decision = llm.invoke(routing_prompt) return json.loads(decision)["route"] 模式3：自我纠正检索（Self-Corrective RAG） def grade_documents(state: AgentState) -> dict: """对检索文档打分，过滤低质量结果""" graded_docs = [] for doc in state["retrieved_docs"]: score_prompt = f"""评估文档与问题的相关性(0-10): 问题: {state['question']} 文档: {doc.page_content[:500]} 只返回数字。""" score = int(llm.invoke(score_prompt).strip()) if score >= 6: graded_docs.append(doc) # 如果所有文档都不相关，触发查询改写 if not graded_docs: new_query = llm.invoke(f"改写这个查询以获得更好的检索结果: {state['question']}") return {"retrieved_docs": [], "question": new_query} return {"retrieved_docs": graded_docs} 工程实现：基于LangGraph的完整Agentic RAG from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolNode from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage # 定义工具 @tool def vector_search(query: str) -> str: """从内部知识库检索相关文档""" docs = vector_store.similarity_search(query, k=5) return "\n\n".join([d.page_content for d in docs]) @tool def web_search(query: str) -> str: """从网络搜索最新信息""" results = search_api.search(query, num_results=5) return "\n\n".join([r["content"] for r in results]) @tool def sql_query(question: str) -> str: """查询业务数据库""" sql = text2sql(question) return db.execute(sql) # Agent决策循环 tools = [vector_search, web_search, sql_query] llm = ChatOpenAI(model="gpt-4o", temperature=0) llm_with_tools = llm.bind_tools(tools) def agent_decision(state): messages = state["messages"] response = llm_with_tools.invoke(messages) return {"messages": [response]} def should_continue(state): last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return END # 构建Graph graph = StateGraph(dict) graph.add_node("agent", agent_decision) graph.add_node("tools", ToolNode(tools)) graph.set_entry_point("agent") graph.add_conditional_edges("agent", should_continue) graph.add_edge("tools", "agent") # 工具执行后回到agent做下一轮决策 app = graph.compile() # 执行 result = app.invoke({ "messages": [HumanMessage(content="对比2026年Q1三大云厂商的AI服务收入")] }) 性能基准对比 我们在三个数据集上对比了不同RAG架构的表现： ...

核心矛盾：检索 vs 全量上下文 随着GPT-5（2M token）、Claude 4（200K token）、Gemini 2.5 Pro（1M token）的发布，“长上下文模型能否替代RAG"成为2026年最热门的技术争论之一。 两种思路的对比： RAG思路：只给LLM看相关的片段 ┌─────────┐ ┌──────────┐ ┌─────────┐ │ 用户查询 │───▶│ 检索Top-K │───▶│ LLM生成 │ └─────────┘ │ 相关文档 │ └─────────┘ └──────────┘ Long Context思路：把所有文档都塞给LLM ┌─────────┐ ┌────────────────────┐ ┌─────────┐ │ 用户查询 │───▶│ 全部文档(长上下文) │───▶│ LLM生成 │ └─────────┘ └────────────────────┘ └─────────┘ 表面上看，长上下文模型让RAG变得多余——既然能塞下整个知识库，为什么还要检索？但实际情况远没有这么简单。 六大维度全面对比 1. 上下文利用率（Context Utilization） LLM对长上下文的中间部分利用率低，这是经过大量研究验证的"Lost in the Middle"现象。 文档位置对注意力的影响（GPT-5实测）： 文档开头 ──► 注意力权重高 ▼ ──► 文档结尾 ▲ ▲ 用户问题 系统提示 容易引用 容易引用 文档中间 ──► 注意力权重低 ◄── 关键信息 ▼ 容易被忽略 实测数据（1000份文档，每篇500字，共500K token上下文）： ...