Haystack 2026: 面向生产级 RAG 与 Agent 流水线的端到端 NLP 框架 —— 配置指南

python -m venv haystack-env
source haystack-env/bin/activate

# 安装 Haystack 核心
pip install haystack-ai

# 验证安装
python -c "import haystack; print(haystack.__version__)"
# 预期输出: 2.12.x

# 安装所有常用额外依赖
pip install "haystack-ai[all]"

# 或按需安装特定额外依赖
pip install haystack-ai opensearch-py  # 用于 OpenSearch
pip install haystack-ai qdrant-client   # 用于 Qdrant
pip install haystack-ai weaviate-client # 用于 Weaviate

# 设置 OpenAI API 密钥
export OPENAI_API_KEY="sk-your-key-here"

# 本地模型支持，安装 HuggingFace
pip install transformers torch sentence-transformers

# verify_setup.py
from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersDocumentEmbedder
from haystack.document_stores import InMemoryDocumentStore

print("Haystack imported successfully")
print(f"Components available: embedders, retrievers, generators, routers")

store = InMemoryDocumentStore()
print(f"Document store initialized: {store.count_documents()} docs")

# basic_rag.py
from haystack import Pipeline, Document
from haystack.document_stores import InMemoryDocumentStore
from haystack.components.embedders import (
    SentenceTransformersDocumentEmbedder,
    SentenceTransformersTextEmbedder,
)
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder

# 创建文档存储并添加文档
doc_store = InMemoryDocumentStore()
documents = [
    Document(content="Haystack is an open-source NLP framework for building search systems."),
    Document(content="RAG combines retrieval with generation for more accurate answers."),
    Document(content="Document stores in Haystack support multiple backends including OpenSearch and Qdrant."),
    Document(content="Embeddings convert text into dense vectors for semantic search."),
    Document(content="Haystack pipelines are directed acyclic graphs of components."),
]

# 嵌入并写入文档
doc_embedder = SentenceTransformersDocumentEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
)
doc_embedder.warm_up()
embeddings = doc_embedder.run(documents=documents)
doc_store.write_documents(embeddings["documents"])

# 构建 RAG 流水线
rag = Pipeline()
rag.add_component("embedder", SentenceTransformersTextEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
))
rag.add_component("retriever", InMemoryEmbeddingRetriever(
    document_store=doc_store, top_k=3
))
rag.add_component("prompt_builder", PromptBuilder(
    template="""Answer based on context.
Context: {% for doc in documents %}
- {{ doc.content }}{% endfor %}
Question: {{ query }}
Answer:"""
))
rag.add_component("generator", OpenAIGenerator(model="gpt-4o-mini"))

# 连接组件
rag.connect("embedder", "retriever")
rag.connect("retriever", "prompt_builder.documents")
rag.connect("prompt_builder", "generator")

# 运行流水线
result = rag.run({
    "embedder": {"text": "What is Haystack?"},
    "prompt_builder": {"query": "What is Haystack?"},
})
print(result["generator"]["replies"][0])

python basic_rag.py

# rag_with_reranker.py
from haystack import Pipeline
from haystack.document_stores import InMemoryDocumentStore
from haystack.components.embedders import (
    SentenceTransformersDocumentEmbedder,
    SentenceTransformersTextEmbedder,
)
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack.components.rankers import TransformersSimilarityRanker
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder

doc_store = InMemoryDocumentStore()
# ... (与上面相同的文档设置)

pipeline = Pipeline()
pipeline.add_component("embedder", SentenceTransformersTextEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
))
pipeline.add_component("retriever", InMemoryEmbeddingRetriever(
    document_store=doc_store, top_k=10
))
pipeline.add_component("ranker", TransformersSimilarityRanker(
    model="cross-encoder/ms-marco-MiniLM-L-6-v2", top_k=3
))
pipeline.add_component("prompt_builder", PromptBuilder(
    template="""Answer based on context.
Context: {% for doc in documents %}
- {{ doc.content }}{% endfor %}
Question: {{ query }}
Answer:"""
))
pipeline.add_component("generator", OpenAIGenerator(model="gpt-4o-mini"))

# 通过重排序器连接
pipeline.connect("embedder", "retriever")
pipeline.connect("retriever", "ranker")
pipeline.connect("ranker", "prompt_builder.documents")
pipeline.connect("prompt_builder", "generator")

result = pipeline.run({
    "embedder": {"text": "How does Haystack handle document storage?"},
    "prompt_builder": {"query": "How does Haystack handle document storage?"},
})
print(result["generator"]["replies"][0])

# branching_pipeline.py
from haystack import Pipeline
from haystack.components.routers import ConditionalRouter
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

pipeline = Pipeline()

# 路由器根据查询类型决定路径
pipeline.add_component("router", ConditionalRouter(routes={
    "condition": "{{ 'technical' in query.lower() }}",
    "output": "{{ query }}",
    "output_type": str,
}))

# 技术分支，提供详细上下文
tech_prompt = """You are a technical assistant. Provide detailed, accurate answers.
Question: {{ query }}
Answer:"""
pipeline.add_component("tech_builder", PromptBuilder(template=tech_prompt))
pipeline.add_component("tech_generator", OpenAIGenerator(model="gpt-4o"))

# 简单分支，用于一般查询
general_prompt = """Provide a concise answer.
Question: {{ query }}
Answer:"""
pipeline.add_component("general_builder", PromptBuilder(template=general_prompt))
pipeline.add_component("general_generator", OpenAIGenerator(model="gpt-4o-mini"))

# 连接路由器输出
pipeline.connect("router.output", "tech_builder")
pipeline.connect("router.fallback_output", "general_builder")

result = pipeline.run({"router": {"query": "What is vector similarity search?"}})

# opensearch_store.py
from haystack.document_stores import OpenSearchDocumentStore

store = OpenSearchDocumentStore(
    host="localhost",
    port=9200,
    index="documents",
    embedding_dim=384,
    use_ssl=True,
    verify_certs=True,
)

# 与嵌入检索器一起使用
from haystack.components.retrievers import OpenSearchEmbeddingRetriever

retriever = OpenSearchEmbeddingRetriever(document_store=store, top_k=5)

# qdrant_store.py
from haystack_integrations.document_stores.qdrant import QdrantDocumentStore

store = QdrantDocumentStore(
    host="localhost",
    port=6333,
    index="haystack_docs",
    embedding_dim=384,
    recreate_index=True,
)

from haystack_integrations.components.retrievers.qdrant import QdrantEmbeddingRetriever

retriever = QdrantEmbeddingRetriever(document_store=store, top_k=5)

# local_llm.py
from haystack.components.generators import HuggingFaceLocalGenerator

generator = HuggingFaceLocalGenerator(
    model="meta-llama/Llama-3.2-3B-Instruct",
    task="text-generation",
    generation_kwargs={"max_new_tokens": 256, "temperature": 0.7},
)
generator.warm_up()

result = generator.run("Explain RAG pipelines in one paragraph.")
print(result["replies"][0])

# custom_component.py
from haystack import component
from typing import Any, Dict, List

@component
class TokenCounter:
    """统计输入文本中 token 数量的自定义组件。"""

    @component.output_types(token_count=int, text=str)
    def run(self, text: str) -> Dict[str, Any]:
        # 简单空格分词（生产环境使用 tiktoken）
        token_count = len(text.split())
        return {"token_count": token_count, "text": text}

# 在流水线中使用
from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator

pipe = Pipeline()
pipe.add_component("counter", TokenCounter())
pipe.add_component("generator", OpenAIGenerator(model="gpt-4o-mini"))
pipe.connect("counter.text", "generator.prompt")

result = pipe.run({"counter": {"text": "Summarize quantum computing."}})
print(f"Tokens: {result['counter']['token_count']}")
print(f"Response: {result['generator']['replies'][0]}")

# web_search_tool.py
from haystack import Pipeline
from haystack.components.websearch import SerperDevWebSearch
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

web_search = SerperDevWebSearch(api_key="your-serper-key")

pipeline = Pipeline()
pipeline.add_component("search", web_search)
pipeline.add_component("builder", PromptBuilder(
    template="""Use search results to answer.
Results: {% for doc in documents %}
- {{ doc.content }}{% endfor %}
Question: {{ query }}
Answer:"""
))
pipeline.add_component("generator", OpenAIGenerator(model="gpt-4o-mini"))

pipeline.connect("search.documents", "builder.documents")
pipeline.connect("builder", "generator")

result = pipeline.run({
    "search": {"query": "latest AI models 2026"},
    "builder": {"query": "What are the latest AI models released in 2026?"},
})
print(result["generator"]["replies"][0])

# async_pipeline.py
import asyncio
from haystack import Pipeline
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder

async def run_queries(queries: list):
    pipeline = Pipeline()
    pipeline.add_component("builder", PromptBuilder(
        template="Answer concisely: {{ query }}"
    ))
    pipeline.add_component("generator", OpenAIGenerator(model="gpt-4o-mini"))
    pipeline.connect("builder", "generator")

    tasks = [
        pipeline.run_async({"builder": {"query": q}})
        for q in queries
    ]
    return await asyncio.gather(*tasks)

results = asyncio.run(run_queries([
    "What is Haystack?",
    "Explain vector search.",
    "How does RAG work?",
]))

# serialize_pipeline.py
from haystack import Pipeline

# 将流水线保存为 YAML（版本控制友好）
rag_pipeline.dump("rag_pipeline.yaml")

# 从 YAML 加载流水线
loaded = Pipeline.loads(open("rag_pipeline.yaml").read())
result = loaded.run({
    "embedder": {"text": "What is Haystack?"},
    "prompt_builder": {"query": "What is Haystack?"},
})

# evaluate_pipeline.py
from haystack import Pipeline, Document
from haystack.components.evaluators import SASEvaluator, FaithfulnessEvaluator

# 真实标签数据
ground_truth = [
    {"query": "What is Haystack?", "expected": "An NLP framework"},
    {"query": "What is RAG?", "expected": "Retrieval-Augmented Generation"},
]

# 运行流水线并收集预测
predictions = []
for item in ground_truth:
    result = rag_pipeline.run({
        "embedder": {"text": item["query"]},
        "prompt_builder": {"query": item["query"]},
    })
    predictions.append(result["generator"]["replies"][0])

# 使用语义相似度评估
sas_evaluator = SASEvaluator()
sas_result = sas_evaluator.run(
    ground_truth_answers=[g["expected"] for g in ground_truth],
    predicted_answers=predictions,
)
print(f"SAS Score: {sas_result['score']:.3f}")

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8000
CMD ["python", "serve.py"]

# serve.py
from fastapi import FastAPI
from haystack import Pipeline
import yaml

app = FastAPI()

# 启动时加载流水线
with open("rag_pipeline.yaml") as f:
    pipeline = Pipeline.loads(f.read())

@app.post("/query")
async def query(question: str):
    result = pipeline.run({
        "embedder": {"text": question},
        "prompt_builder": {"query": question},
    })
    return {
        "answer": result["generator"]["replies"][0],
        "documents": [d.content for d in result.get("retriever", {}).get("documents", [])],
    }

# docker-compose.yml
version: "3.8"
services:
  haystack-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      - opensearch

  opensearch:
    image: opensearchproject/opensearch:2.14.0
    environment:
      - discovery.type=single-node
      - DISABLE_SECURITY_PLUGIN=true
    ports:
      - "9200:9200"
    volumes:
      - osdata:/usr/share/opensearch/data

volumes:
  osdata: