引言 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流程的企业应用。
...
