Haystack Integration for NLP Pipelines

Haystack Integration for NLP Pipelines

Haystack (deepset) — production-ready framework for building NLP pipelines with native support for RAG, question answering, and document processing. Unlike LangChain's imperative style, Haystack uses a declarative pipeline model—components connect in a graph, data flows between them in a typed manner. This simplifies testing, versioning, and component replacement.

When Haystack Is the Right Choice

Haystack outperforms competitors in several scenarios:

• Document-centric tasks — when work primarily involves searching and processing document corpora
• Production-grade RAG — need a reliable, testable system, not a prototype
• Team prefers explicit configuration — YAML pipelines are easier to audit than LangChain Python code
• Multi-hop question answering — Haystack has built-in components for complex search

LangChain / LlamaIndex are preferable for rapid prototyping and agent scenarios with multiple tools.

Key Haystack 2.x Abstractions

Haystack 2.x (current version) changed architecture compared to 1.x:

• Component — base unit, has typed @component.input and @component.output
• Pipeline — graph of components, connected by data types
• Document — unified object with content, meta, embedding, id
• DocumentStore — storage abstraction (InMemory, Elasticsearch, OpenSearch, pgvector, Weaviate, Qdrant, Milvus)

from haystack import Pipeline, Document
from haystack.components.retrievers import InMemoryBM25Retriever
from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import RAGPromptBuilder

# Build RAG pipeline
pipeline = Pipeline()
pipeline.add_component("retriever", InMemoryBM25Retriever(document_store=store))
pipeline.add_component("prompt_builder", RAGPromptBuilder())
pipeline.add_component("generator", OpenAIGenerator(model="gpt-4o-mini"))

pipeline.connect("retriever.documents", "prompt_builder.documents")
pipeline.connect("prompt_builder.prompt", "generator.prompt")

DocumentStore Connection

DocumentStore choice depends on scale and requirements:

Qdrant setup:

from haystack_integrations.document_stores.qdrant import QdrantDocumentStore

document_store = QdrantDocumentStore(
    url="http://localhost:6333",
    index="documents",
    embedding_dim=1536,
    recreate_index=False,
)

Document Indexing

Indexing pipeline is separate from search pipeline:

from haystack.components.converters import PyPDFToDocument
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter
from haystack.components.embedders import OpenAIDocumentEmbedder
from haystack.components.writers import DocumentWriter

indexing = Pipeline()
indexing.add_component("converter", PyPDFToDocument())
indexing.add_component("cleaner", DocumentCleaner())
indexing.add_component("splitter", DocumentSplitter(
    split_by="sentence", split_length=5, split_overlap=2
))
indexing.add_component("embedder", OpenAIDocumentEmbedder(
    model="text-embedding-3-small"
))
indexing.add_component("writer", DocumentWriter(document_store=document_store))

indexing.connect("converter.documents", "cleaner.documents")
indexing.connect("cleaner.documents", "splitter.documents")
indexing.connect("splitter.documents", "embedder.documents")
indexing.connect("embedder.documents", "writer.documents")

Hybrid Search

Haystack supports hybrid search (BM25 + semantic) via DocumentJoiner:

from haystack.components.retrievers import InMemoryBM25Retriever, InMemoryEmbeddingRetriever
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import MetaFieldRanker

pipeline.add_component("bm25", InMemoryBM25Retriever(document_store=store, top_k=10))
pipeline.add_component("semantic", InMemoryEmbeddingRetriever(document_store=store, top_k=10))
pipeline.add_component("joiner", DocumentJoiner(join_mode="reciprocal_rank_fusion"))

pipeline.connect("bm25.documents", "joiner.documents")
pipeline.connect("semantic.documents", "joiner.documents")

RRF (Reciprocal Rank Fusion) combines results from both retrievers without score normalization.

Custom Components

Haystack easily extends with custom components:

from haystack import component
from haystack.dataclasses import Document
from typing import List

@component
class CustomReranker:
    @component.output_types(documents=List[Document])
    def run(self, documents: List[Document], query: str) -> dict:
        # Your reranking logic
        scored = [(doc, self.score(doc, query)) for doc in documents]
        ranked = sorted(scored, key=lambda x: x[1], reverse=True)
        return {"documents": [doc for doc, _ in ranked[:5]]}

Pipeline Serialization and Deployment

Pipelines serialize to YAML—a key advantage for DevOps:

# Export
pipeline.to_yaml("rag_pipeline.yaml")

# Import
pipeline = Pipeline.from_yaml("rag_pipeline.yaml")

YAML pipeline files can be stored in Git, code reviewed, deployed via CI/CD. Haystack Hayhooks provides REST API for serving pipelines:

pip install hayhooks
hayhooks run --pipelines-dir ./pipelines

After startup, pipeline is available at /pipeline/rag/run.

RAG Quality Assessment

Haystack has built-in evaluation tools:

from haystack.evaluation import EvaluationResult
from haystack.components.evaluators import (
    FaithfulnessEvaluator,
    ContextRelevanceEvaluator,
    SASEvaluator
)

Metrics: Faithfulness (answer matches context), Context Relevance (context relevant to question), SAS (semantic similarity of answer to benchmark).

Performance and Scalability

• Async mode via pipeline.run_async() for concurrent request processing
• Batching for embedder components (up to 10x speedup during indexing)
• Caching via CachingChecker + Redis — cache search results for identical queries
• Prometheus metrics via Hayhooks middleware

Typical RAG pipeline performance: 1–3 seconds per query using gpt-4o-mini and Qdrant.

Integration Timeline

• Basic RAG pipeline (1 DocumentStore, 1 LLM): 1–2 weeks
• Hybrid search + custom reranker: 3–4 weeks
• Production deployment + monitoring + evaluation: 6–8 weeks