Haystack - NLP Framework for Production

Table of Contents

Introduction

Haystack is an open-source NLP framework designed for building production-ready search and question-answering systems. Developed by deepset, Haystack provides a modular, pipeline-based architecture that makes it straightforward to combine retrievers, readers, generators, and other NLP components into powerful applications.

With Haystack 2.x, the framework has been redesigned around a flexible pipeline system that supports both extractive and generative AI workflows. Whether you are building a simple document search system or a complex Retrieval-Augmented Generation (RAG) pipeline, Haystack provides the building blocks and the glue to connect them.

This tutorial covers the core concepts of Haystack, walks you through building practical pipelines, shows you how to create custom components, and demonstrates how to deploy your system as a REST API ready for production traffic.

Prerequisites

• Python 3.9 or higher
• Basic understanding of NLP concepts (embeddings, retrieval, generation)
• An OpenAI API key (for generator components)

pip install haystack-ai "sentence-transformers>=2.2.0" "opensearch-haystack" fastapi uvicorn

export OPENAIAPIKEY="sk-your-key-here"

Understanding Haystack Architecture

Haystack 2.x is built around three core concepts:

Components

Components are the building blocks of Haystack. Each component performs a specific task such as retrieving documents, generating text, or converting file formats. Components declare their inputs and outputs explicitly, enabling the framework to validate connections at pipeline construction time.

Pipelines

Pipelines connect components into directed acyclic graphs (DAGs). Data flows from one component to the next based on declared input/output connections. Pipelines can be linear or branching.

Document Stores

Document stores are the persistence layer. They store documents and their associated metadata and embeddings, providing retrieval interfaces for the pipeline components.

from haystack import Pipeline, Document
from haystack.components.writers import DocumentWriter
from haystack.components.preprocessors import DocumentCleaner, DocumentSplitter

A simple pipeline that cleans and splits documents
preprocessing = Pipeline()
preprocessing.addcomponent("cleaner", DocumentCleaner())

preprocessing.addcomponent("splitter", DocumentSplitter(
    splitby="sentence",

    splitlength=3,
    splitoverlap=1,

))
preprocessing.connect("cleaner", "splitter")

Run the pipeline
docs = [
    Document(content="""Haystack is an open-source NLP framework. It provides modular
    components for building search systems. The framework supports both extractive and
    generative approaches. You can combine multiple components into pipelines.
    Pipelines can be deployed as REST APIs.""")
]

result = preprocessing.run({"cleaner": {"documents": docs}})
for doc in result["splitter"]["documents"]:
    print(f"  Chunk: {doc.content[:80]}...")

from haystack.documentstores.inmemory import InMemoryDocumentStore

Create an in-memory document store
store = InMemoryDocumentStore(embeddingsimilarityfunction="cosine")


Write documents
documents = [
    Document(
        content="Python is a versatile programming language used in data science.",
        meta={"category": "programming", "level": "beginner"}
    ),
    Document(
        content="Docker containers simplify application deployment and scaling.",
        meta={"category": "devops", "level": "intermediate"}
    ),
    Document(
        content="PostgreSQL is a powerful open-source relational database system.",
        meta={"category": "database", "level": "intermediate"}
    ),
    Document(
        content="Kubernetes orchestrates containerized applications at scale.",
        meta={"category": "devops", "level": "advanced"}
    ),
    Document(
        content="FastAPI is a modern Python web framework for building REST APIs.",
        meta={"category": "programming", "level": "intermediate"}
    ),
    Document(
        content="Redis provides in-memory caching for high-performance applications.",
        meta={"category": "database", "level": "intermediate"}
    ),
    Document(
        content="Machine learning models require large datasets for effective training.",
        meta={"category": "ai", "level": "beginner"}
    ),
    Document(
        content="Transformers architecture revolutionized natural language processing.",
        meta={"category": "ai", "level": "advanced"}
    ),
]

store.writedocuments(documents)
print(f"Documents in store: {store.countdocuments()}")

from opensearchhaystack.documentstore import OpenSearchDocumentStore

Connect to an OpenSearch cluster
opensearchstore = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystackdocuments",

    embeddingdim=384,
    returnembedding=True,

)

Write documents with a pipeline
from haystack.components.writers import DocumentWriter
from haystack.components.embedders import SentenceTransformersDocumentEmbedder

indexing = Pipeline()
indexing.addcomponent("embedder", SentenceTransformersDocumentEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
))
indexing.addcomponent("writer", DocumentWriter(documentstore=opensearchstore))

indexing.connect("embedder.documents", "writer.documents")

Index documents
indexing.run({"embedder": {"documents": documents}})

from haystack.components.embedders import (
    SentenceTransformersTextEmbedder,
    SentenceTransformersDocumentEmbedder,
)
from haystack.components.retrievers.inmemory import InMemoryEmbeddingRetriever

First, embed and store documents
docembedder = SentenceTransformersDocumentEmbedder(

    model="sentence-transformers/all-MiniLM-L6-v2"
)
docembedder.warmup()

docswithembeddings = docembedder.run(documents=documents)

store = InMemoryDocumentStore(embeddingsimilarityfunction="cosine")
store.writedocuments(docswithembeddings["documents"])


Build the retrieval pipeline
retrievalpipeline = Pipeline()
retrievalpipeline.addcomponent("textembedder", SentenceTransformersTextEmbedder(

    model="sentence-transformers/all-MiniLM-L6-v2"
))
retrievalpipeline.addcomponent("retriever", InMemoryEmbeddingRetriever(

    documentstore=store,
    topk=3,

))
retrievalpipeline.connect("textembedder.embedding", "retriever.queryembedding")

Search
result = retrievalpipeline.run({"textembedder": {"text": "container orchestration tools"}})

print("Retrieved documents:")
for doc in result["retriever"]["documents"]:
    print(f"  [{doc.score:.4f}] {doc.content}")
    print(f"    Meta: {doc.meta}")

BM25 Keyword Retrieval

from haystack.components.retrievers.inmemory import InMemoryBM25Retriever

BM25 does not need embeddings
bm25store = InMemoryDocumentStore()
bm25store.writedocuments(documents)

bm25retriever = InMemoryBM25Retriever(documentstore=bm25store, topk=3)

Search
result = bm25retriever.run(query="PostgreSQL database")


print("BM25 results:")
for doc in result["documents"]:
    print(f"  [{doc.score:.4f}] {doc.content}")

from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import TransformersSimilarityRanker

hybridpipeline = Pipeline()

Embedding retrieval branch
hybridpipeline.addcomponent("textembedder", SentenceTransformersTextEmbedder(

    model="sentence-transformers/all-MiniLM-L6-v2"
))
hybridpipeline.addcomponent("embeddingretriever", InMemoryEmbeddingRetriever(
    documentstore=store, topk=5
))

BM25 retrieval branch
hybridpipeline.addcomponent("bm25retriever", InMemoryBM25Retriever(

    documentstore=bm25store, topk=5
))

Join and rerank
hybridpipeline.addcomponent("joiner", DocumentJoiner())
hybridpipeline.addcomponent("ranker", TransformersSimilarityRanker(
    model="cross-encoder/ms-marco-MiniLM-L-6-v2", topk=3

))

Connect components
hybridpipeline.connect("textembedder.embedding", "embeddingretriever.queryembedding")

hybridpipeline.connect("embeddingretriever", "joiner")

hybridpipeline.connect("bm25retriever", "joiner")

hybridpipeline.connect("joiner", "ranker")

Run hybrid search
result = hybridpipeline.run({

    "textembedder": {"text": "deployment tools for applications"},
    "bm25retriever": {"query": "deployment tools for applications"},

    "ranker": {"query": "deployment tools for applications"},
})

print("Hybrid search results:")
for doc in result["ranker"]["documents"]:
    print(f"  [{doc.score:.4f}] {doc.content}")

from haystack.components.readers import ExtractiveReader

reader = ExtractiveReader(model="deepset/roberta-base-squad2")
reader.warmup()

Extract answers from documents
result = reader.run(
    query="What is PostgreSQL?",
    documents=[
        Document(content="PostgreSQL is a powerful open-source relational database system."),
        Document(content="Redis provides in-memory caching for high-performance applications."),
    ],
    topk=2,

)

for answer in result["answers"]:
    print(f"  Answer: '{answer.data}' (score: {answer.score:.4f})")
    print(f"  Context: {answer.document.content}")

from haystack.components.generators import OpenAIGenerator
from haystack.components.builders import PromptBuilder

Create a prompt builder
prompttemplate = """Given the following context documents, answer the question.
If the answer is not found in the context, say "I don't have enough information."

Context:
{% for doc in documents %}
{{ doc.content }}
{% endfor %}


Question: {{ question }}

Answer:"""

promptbuilder = PromptBuilder(template=prompttemplate)
generator = OpenAIGenerator(model="gpt-4o")

Build a simple generative QA pipeline
genpipeline = Pipeline()

genpipeline.addcomponent("promptbuilder", promptbuilder)

genpipeline.addcomponent("generator", generator)

genpipeline.connect("promptbuilder", "generator")


Run
contextdocs = [
    Document(content="PostgreSQL is a powerful open-source relational database system."),
    Document(content="PostgreSQL supports advanced data types, full-text search, and JSON."),
]

result = genpipeline.run({

    "promptbuilder": {
        "documents": contextdocs,

        "question": "What features does PostgreSQL support?"
    }
})

print(f"Answer: {result['generator']['replies'][0]}")

from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers.inmemory import InMemoryEmbeddingRetriever
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator

def buildragpipeline(documentstore: InMemoryDocumentStore) -> Pipeline:

    """Build a complete RAG pipeline."""

    ragprompt = """You are a helpful technical assistant. Answer the question based on
the provided context. Be specific and cite the relevant information.

Context:
{% for doc in documents %}
[Source {{ loop.index }}] {{ doc.content }}
{% endfor %}

Question: {{ question }}

Instructions:
Answer based only on the provided context
Reference sources using [Source N] notation
If the context does not contain the answer, state that clearly
Be concise but thorough

Answer:"""

    pipeline = Pipeline()

    # Query embedding
    pipeline.addcomponent("textembedder", SentenceTransformersTextEmbedder(
        model="sentence-transformers/all-MiniLM-L6-v2"
    ))

    # Document retrieval
    pipeline.addcomponent("retriever", InMemoryEmbeddingRetriever(

        documentstore=documentstore,

        topk=5,
    ))

    # Prompt construction
    pipeline.addcomponent("promptbuilder", PromptBuilder(template=ragprompt))


    # LLM generation
    pipeline.addcomponent("generator", OpenAIGenerator(
        model="gpt-4o",
        generationkwargs={"temperature": 0.3, "maxtokens": 500},
    ))

    # Connect the pipeline
    pipeline.connect("textembedder.embedding", "retriever.queryembedding")
    pipeline.connect("retriever.documents", "promptbuilder.documents")

    pipeline.connect("promptbuilder", "generator")

    return pipeline


Build and run
rag = buildragpipeline(store)

questions = [
    "What are the best tools for container orchestration?",
    "Which programming language is best for data science?",
    "How can I improve database performance?",
]

for question in questions:
    result = rag.run({
        "textembedder": {"text": question},

        "promptbuilder": {"question": question},
    })
    print(f"\nQ: {question}")
    print(f"A: {result['generator']['replies'][0]}")

RAG with Metadata Filtering

from haystack.components.retrievers.inmemory import InMemoryEmbeddingRetriever

def buildfilteredrag(documentstore: InMemoryDocumentStore) -> Pipeline:
    """RAG pipeline with metadata filtering support."""

    ragprompt = """Answer the question based on the context below.


Context:
{% for doc in documents %}
{{ doc.content }} [{{ doc.meta.category }}/{{ doc.meta.level }}]
{% endfor %}


Question: {{ question }}
Answer:"""

    pipeline = Pipeline()
    pipeline.addcomponent("textembedder", SentenceTransformersTextEmbedder(

        model="sentence-transformers/all-MiniLM-L6-v2"
    ))
    pipeline.addcomponent("retriever", InMemoryEmbeddingRetriever(
        documentstore=documentstore, topk=3

    ))
    pipeline.addcomponent("promptbuilder", PromptBuilder(template=ragprompt))
    pipeline.addcomponent("generator", OpenAIGenerator(model="gpt-4o"))


    pipeline.connect("textembedder.embedding", "retriever.queryembedding")

    pipeline.connect("retriever.documents", "promptbuilder.documents")
    pipeline.connect("promptbuilder", "generator")


    return pipeline


filteredrag = buildfilteredrag(store)

Query with metadata filter
result = filteredrag.run({

    "textembedder": {"text": "What DevOps tools should I learn?"},
    "retriever": {"filters": {"field": "meta.category", "operator": "==", "value": "devops"}},
    "promptbuilder": {"question": "What DevOps tools should I learn?"},

})

print(f"Answer: {result['generator']['replies'][0]}")

from haystack import component, defaultfromdict, defaulttodict
from haystack.dataclasses import Document
from typing import List, Optional
import re

@component
class ContentFilter:
    """Custom component that filters documents by content length and keyword presence."""

    def init(self, minlength: int = 10, requiredkeywords: Optional[List[str]] = None):

        self.minlength = minlength

        self.requiredkeywords = requiredkeywords or []


    @component.outputtypes(documents=List[Document], filteredcount=int)

    def run(self, documents: List[Document]) -> dict:
        filtered = []
        for doc in documents:
            # Check minimum length
            if len(doc.content) < self.minlength:
                continue

            # Check required keywords
            if self.requiredkeywords:

                contentlower = doc.content.lower()
                if not any(kw.lower() in contentlower for kw in self.requiredkeywords):
                    continue

            filtered.append(doc)

        return {
            "documents": filtered,
            "filteredcount": len(documents) - len(filtered),

        }


Use the custom component in a pipeline
pipeline = Pipeline()
pipeline.addcomponent("filter", ContentFilter(
    minlength=20,

    requiredkeywords=["Python", "database"]
))
pipeline.addcomponent("textembedder", SentenceTransformersTextEmbedder(
    model="sentence-transformers/all-MiniLM-L6-v2"
))

Run
result = pipeline.run({"filter": {"documents": documents}})
print(f"Filtered: {result['filter']['filteredcount']} documents removed")

@component
class MetadataEnricher:
    """Enriches documents with computed metadata fields."""

    def init(self, computereadingtime: bool = True, computewordcount: bool = True):

        self.computereadingtime = computereadingtime

        self.computewordcount = computewordcount


    @component.outputtypes(documents=List[Document])
    def run(self, documents: List[Document]) -> dict:
        enriched = []
        for doc in documents:
            newmeta = dict(doc.meta)


            if self.computewordcount:

                newmeta["wordcount"] = len(doc.content.split())


            if self.computereadingtime:

                words = len(doc.content.split())
                newmeta["readingtimeminutes"] = round(words / 200, 1)

            # Detect language (simple heuristic)
            newmeta["hascode"] = "

enriched.append(Document(

content=doc.content,

meta=newmeta,

enrichmentpipeline = Pipeline()

enrichmentpipeline.addcomponent("enricher", MetadataEnricher())

result = enrichmentpipeline.run({"enricher": {"documents": documents}})

print(f" {doc.content[:50]}... | Words: {doc.meta.get('wordcount')} | "

f"Reading: {doc.meta.get('readingtimeminutes')}min")

Custom Ranker with Business Logic

@component

class BusinessRanker:

"""Re-ranks documents based on custom business rules."""

def init(self, boostcategories: Optional[List[str]] = None, boostfactor: float = 1.5):

self.boostcategories = boostcategories or []

self.boostfactor = boostfactor

@component.outputtypes(documents=List[Document])

def run(self, documents: List[Document], topk: int = 5) -> dict:

scoreddocs = []

if doc.meta.get("category") in self.boostcategories:

score = self.boostfactor

scoreddocs.append((doc, score))

# Sort by adjusted score

scoreddocs.sort(key=lambda x: x[1], reverse=True)

resultdocs = []

for doc, score in scoreddocs[:topk]:

doc.score = score

resultdocs.append(doc)

return {"documents": resultdocs}

Pipeline Evaluation

Evaluating your pipeline is crucial for understanding its strengths and weaknesses.

from haystack.evaluation import EvaluationRunResult

class PipelineEvaluator:

"""Evaluate RAG pipeline quality with multiple metrics."""

def init(self, pipeline: Pipeline):

self.pipeline = pipeline

def evaluateretrieval(self, testcases: list[dict], k: int = 5) -> dict:

"""Evaluate retrieval quality.

testcases: list of {"query": str, "relevantdocids": set}

for case in testcases:

result = self.pipeline.run({

"textembedder": {"text": case["query"]},

retrieveddocs = result.get("retriever", {}).get("documents", [])

retrievedids = {doc.id for doc in retrieveddocs[:k]}

relevantids = case["relevantdocids"]

if retrievedids:

precision = len(retrievedids & relevantids) / len(retrievedids)

if relevantids:

recall = len(retrievedids & relevantids) / len(relevantids)

for i, doc in enumerate(retrieveddocs[:k], 1):

if doc.id in relevantids:

def evaluategeneration(self, testcases: list[dict]) -> dict:

testcases: list of {"query": str, "expectedanswer": str}

for case in testcases:

# Get the RAG answer

result = self.pipeline.run({

"textembedder": {"text": case["query"]},

"promptbuilder": {"question": case["query"]},

})

actual = result["generator"]["replies"][0]

# Judge the answer

judgeprompt = f"""Rate the following answer on a scale of 1-5.

Expected answer: {case['expectedanswer']}

Actual answer: {actual}

Return only the number (1-5)."""

judgeresult = judge.run(prompt=judgeprompt)

try:

score = int(judgeresult["replies"][0].strip())

"avgscore": float(np.mean(scores)),

"minscore": int(min(scores)),

"maxscore": int(max(scores)),

"totalevaluated": len(scores),

REST API Deployment

Deploy your Haystack pipeline as a production-ready REST API.

ragpipeline = None

documentstore = None

@app.onevent("startup")

async def startup():

global ragpipeline, documentstore

# Initialize document store and index documents

documentstore = InMemoryDocumentStore(embeddingsimilarityfunction="cosine")

docembedder = SentenceTransformersDocumentEmbedder(

model="sentence-transformers/all-MiniLM-L6-v2"

)

docembedder.warmup()

result = docembedder.run(documents=documents)

documentstore.writedocuments(result["documents"])

ragpipeline = buildragpipeline(documentstore)

topk: int = Field(default=5, ge=1, le=20, description="Number of documents to retrieve")

filters: Optional[dict] = Field(default=None, description="Metadata filters")

class SourceDocument(BaseModel):

content: str

score: float

metadata: dict

class QueryResponse(BaseModel):

answer: str

sources: list[SourceDocument]

query: str

@app.post("/query", responsemodel=QueryResponse)

if ragpipeline is None:

raise HTTPException(statuscode=503, detail="Pipeline not initialized")

runparams = {

"textembedder": {"text": request.question},

"promptbuilder": {"question": request.question},

"retriever": {"topk": request.topk},

}

if request.filters:

runparams["retriever"]["filters"] = request.filters

result = ragpipeline.run(runparams)

retrieverdocs = result["retriever"]["documents"]

sources = [

SourceDocument(

content=doc.content,

score=doc.score or 0.0,

metadata=doc.meta,

)

for doc in retrieverdocs

raise HTTPException(statuscode=500, detail=str(e))

class IndexRequest(BaseModel):

documents: list[dict] = Field(description="Documents with 'content' and optional 'meta' keys")

@app.post("/index", statuscode=201)

async def indexdocuments(request: IndexRequest):

"""Index new documents into the document store."""

newdocs = [

docembedder = SentenceTransformersDocumentEmbedder(

model="sentence-transformers/all-MiniLM-L6-v2"

)

docembedder.warmup()

result = docembedder.run(documents=newdocs)

documentstore.writedocuments(result["documents"])

return {"indexed": len(newdocs), "total": documentstore.countdocuments()}

"pipelineready": ragpipeline is not None,

"documents": documentstore.countdocuments() if documentstore else 0,

}

if name == "main":

uvicorn.run(app, host="0.0.0.0", port=8000)

```

Best Practices

Conclusion

Haystack provides a powerful, modular framework for building production NLP systems. Its pipeline architecture makes it easy to combine retrievers, readers, generators, and custom components into sophisticated applications.

Key takeaways:

• Haystack 2.x's component-pipeline architecture enables clean, maintainable NLP systems.
• Document stores provide flexible persistence with support for embeddings and metadata filtering.
• Hybrid retrieval (embedding + BM25) with reranking delivers the best search quality.
• Custom components let you extend Haystack with domain-specific logic.
• Systematic evaluation is essential -- measure retrieval precision, recall, and generation quality.
• FastAPI integration makes it straightforward to deploy pipelines as production REST APIs.

Haystack continues to evolve rapidly. Stay current with the latest releases for new components, improved performance, and expanded model support.