引言

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系统不只是"检索+生成"，而是文档处理、检索策略、上下文管理和答案溯源的完整体系。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。