Haystack RAG pipeline tutorial: enterprise document search

You need a RAG system that handles millions of documents, respects access controls, and returns answers fast enough for production traffic. Haystack 2.x gives you the primitives; this tutorial shows how to assemble them into a pipeline that survives enterprise requirements. We’ll build a hybrid retrieval pipeline with BM25 and dense vectors, add cross-encoder reranking, wire in metadata filtering for tenant isolation, and expose it behind a FastAPI service with structured logging.

Prerequisites

• Python 3.11+
• Elasticsearch 8.x (or OpenSearch) running and accessible
• An embedding model endpoint — we’ll use text-embedding-3-small via OpenAI-compatible API
• A cross-encoder reranker — cross-encoder/ms-marco-MiniLM-L-6-v2 works well locally
• Haystack 2.6+ (pip install "haystack-ai[elasticsearch,openai,fastapi]>=2.6.0")

You’ll also need an OpenAI-compatible API key. If you’re routing through a gateway like n4n.ai, set OPENAI_BASE_URL and OPENAI_API_KEY accordingly; the code below uses the standard client interface so it works either way.

Project structure

enterprise-rag/
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── pipeline.py
│   ├── api.py
│   └── models.py
├── data/
│   └── sample_docs.jsonl
├── tests/
│   └── test_pipeline.py
├── pyproject.toml
└── README.md

Create a virtual environment and install dependencies:

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

Configuration

Centralize settings in app/config.py. Keep secrets out of code — use environment variables.

# app/config.py
import os
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    # Elasticsearch
    elasticsearch_host: str = "http://localhost:9200"
    elasticsearch_index: str = "enterprise_docs"
    elasticsearch_user: str | None = None
    elasticsearch_password: str | None = None

    # Embeddings
    embedding_model: str = "text-embedding-3-small"
    embedding_dim: int = 1536
    openai_api_key: str
    openai_base_url: str | None = None

    # Reranker
    reranker_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
    reranker_top_k: int = 20

    # Retrieval
    bm25_top_k: int = 50
    dense_top_k: int = 50
    final_top_k: int = 5

    # API
    api_host: str = "0.0.0.0"
    api_port: int = 8000

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"


@lru_cache
def get_settings() -> Settings:
    return Settings()

Create a .env file:

ELASTICSEARCH_HOST=http://localhost:9200
ELASTICSEARCH_INDEX=enterprise_docs
OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.openai.com/v1  # or your gateway endpoint

Document model and indexing pipeline

Define the document schema in app/models.py. The tenant_id field is critical for enterprise multi-tenancy — every query filters on it.

# app/models.py
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from enum import Enum


class DocumentSource(str, Enum):
    CONFLUENCE = "confluence"
    SHAREPOINT = "sharepoint"
    GDRIVE = "gdrive"
    SLACK = "slack"
    PDF_UPLOAD = "pdf_upload"


class Document(BaseModel):
    id: str
    tenant_id: str
    title: str
    content: str
    source: DocumentSource
    source_id: str  # original system's ID
    metadata: dict = Field(default_factory=dict)
    created_at: datetime
    updated_at: datetime
    access_roles: list[str] = Field(default_factory=list)  # for RBAC


class SearchRequest(BaseModel):
    query: str
    tenant_id: str
    user_roles: list[str] = Field(default_factory=list)
    top_k: int = 5
    filters: dict = Field(default_factory=dict)


class SearchResult(BaseModel):
    document: Document
    score: float
    rerank_score: float | None = None


class SearchResponse(BaseModel):
    results: list[SearchResult]
    query: str
    tenant_id: str
    total_hits: int
    latency_ms: float

Now build the indexing pipeline in app/pipeline.py. This runs once (or on a schedule) to ingest documents into Elasticsearch with both BM25 and dense vector fields.

# app/pipeline.py
import logging
from haystack import Pipeline, Document as HaystackDocument
from haystack.components.embedders import OpenAIDocumentEmbedder
from haystack.components.writers import DocumentWriter
from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore
from haystack_integrations.components.retrievers.elasticsearch import ElasticsearchBM25Retriever
from app.config import get_settings

logger = logging.getLogger(__name__)
settings = get_settings()


def get_document_store() -> ElasticsearchDocumentStore:
    """Initialize Elasticsearch document store with hybrid search mapping."""
    return ElasticsearchDocumentStore(
        hosts=settings.elasticsearch_host,
        index=settings.elasticsearch_index,
        embedding_dim=settings.embedding_dim,
        similarity="cosine",
        # Custom mapping for hybrid search + metadata filtering
        custom_mapping={
            "properties": {
                "tenant_id": {"type": "keyword"},
                "source": {"type": "keyword"},
                "source_id": {"type": "keyword"},
                "access_roles": {"type": "keyword"},
                "created_at": {"type": "date"},
                "updated_at": {"type": "date"},
                "metadata": {"type": "object", "enabled": False},  # opaque JSON
            }
        },
    )


def build_indexing_pipeline(document_store: ElasticsearchDocumentStore) -> Pipeline:
    """Pipeline: embed documents -> write to Elasticsearch."""
    pipeline = Pipeline()

    embedder = OpenAIDocumentEmbedder(
        model=settings.embedding_model,
        api_key=settings.openai_api_key,
        api_base_url=settings.openai_base_url,
        batch_size=32,
        progress_bar=True,
    )
    writer = DocumentWriter(document_store=document_store, policy="overwrite")

    pipeline.add_component("embedder", embedder)
    pipeline.add_component("writer", writer)
    pipeline.connect("embedder.documents", "writer.documents")

    return pipeline


def index_documents(documents: list[HaystackDocument]) -> int:
    """Run indexing pipeline on a batch of documents."""
    document_store = get_document_store()
    pipeline = build_indexing_pipeline(document_store)

    result = pipeline.run({"embedder": {"documents": documents}})
    written = result["writer"]["documents_written"]
    logger.info("Indexed %d documents", written)
    return written

Hybrid retrieval pipeline with reranking

The query pipeline combines BM25 and dense retrieval, merges results with reciprocal rank fusion (RRF), then reranks with a cross-encoder. This is where enterprise relevance lives.

# app/pipeline.py (continued)
from haystack import Pipeline
from haystack.components.embedders import OpenAITextEmbedder
from haystack.components.retrievers import InMemoryEmbeddingRetriever
from haystack_integrations.components.retrievers.elasticsearch import (
    ElasticsearchBM25Retriever,
    ElasticsearchEmbeddingRetriever,
)
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import TransformersSimilarityRanker
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack.dataclasses import Document as HaystackDocument


def build_query_pipeline(document_store: ElasticsearchDocumentStore) -> Pipeline:
    """
    Hybrid retrieval + reranking pipeline.
    
    Flow:
    1. Embed query (dense)
    2. BM25 retrieval (sparse)
    3. Dense vector retrieval
    4. Join with RRF
    5. Cross-encoder rerank
    6. (Optional) Prompt + generate answer
    """
    pipeline = Pipeline()

    # Query embedding
    query_embedder = OpenAITextEmbedder(
        model=settings.embedding_model,
        api_key=settings.openai_api_key,
        api_base_url=settings.openai_base_url,
    )

    # Sparse retrieval (BM25)
    bm25_retriever = ElasticsearchBM25Retriever(
        document_store=document_store,
        top_k=settings.bm25_top_k,
    )

    # Dense retrieval
    dense_retriever = ElasticsearchEmbeddingRetriever(
        document_store=document_store,
        top_k=settings.dense_top_k,
    )

    # Merge with Reciprocal Rank Fusion
    joiner = DocumentJoiner(
        join_mode="reciprocal_rank_fusion",
        top_k=settings.reranker_top_k,
    )

    # Cross-encoder reranker
    reranker = TransformersSimilarityRanker(
        model=settings.reranker_model,
        top_k=settings.final_top_k,
        scale_score=True,
    )

    # Optional: answer generation
    prompt_template = """
    Answer the question based only on the provided documents.
    If the answer cannot be found, say "I don't have enough information."
    
    Documents:
    {% for doc in documents %}
    [{{ loop.index }}] {{ doc.content }}
    {% endfor %}
    
    Question: {{ query }}
    Answer:
    """
    prompt_builder = PromptBuilder(template=prompt_template)
    generator = OpenAIGenerator(
        model="gpt-4o-mini",
        api_key=settings.openai_api_key,
        api_base_url=settings.openai_base_url,
        generation_kwargs={"temperature": 0.0, "max_tokens": 512},
    )

    # Add components
    pipeline.add_component("query_embedder", query_embedder)
    pipeline.add_component("bm25_retriever", bm25_retriever)
    pipeline.add_component("dense_retriever", dense_retriever)
    pipeline.add_component("joiner", joiner)
    pipeline.add_component("reranker", reranker)
    pipeline.add_component("prompt_builder", prompt_builder)
    pipeline.add_component("generator", generator)

    # Connect
    pipeline.connect("query_embedder.embedding", "dense_retriever.query_embedding")
    pipeline.connect("bm25_retriever.documents", "joiner.documents")
    pipeline.connect("dense_retriever.documents", "joiner.documents")
    pipeline.connect("joiner.documents", "reranker.documents")
    pipeline.connect("reranker.documents", "prompt_builder.documents")
    pipeline.connect("prompt_builder.prompt", "generator.prompt")

    return pipeline


def search(
    query: str,
    tenant_id: str,
    user_roles: list[str] | None = None,
    top_k: int | None = None,
    extra_filters: dict | None = None,
) -> dict:
    """
    Execute hybrid search with tenant isolation and RBAC filtering.
    
    Returns dict with 'documents', 'answers', and metadata.
    """
    document_store = get_document_store()
    pipeline = build_query_pipeline(document_store)

    # Build filter: tenant_id + roles + any extra filters
    filters = {
        "operator": "AND",
        "conditions": [
            {"field": "tenant_id", "operator": "==", "value": tenant_id},
        ],
    }
    if user_roles:
        filters["conditions"].append({
            "field": "access_roles",
            "operator": "in",
            "value": user_roles,
        })
    if extra_filters:
        for field, value in extra_filters.items():
            filters["conditions"].append({
                "field": field,
                "operator": "==",
                "value": value,
            })

    # Override top_k if provided
    run_params = {
        "bm25_retriever": {"filters": filters},
        "dense_retriever": {"filters": filters},
        "reranker": {"top_k": top_k or settings.final_top_k},
    }

    result = pipeline.run(
        {
            "query_embedder": {"text": query},
            "bm25_retriever": {"query": query, "filters": filters},
            "dense_retriever": {"filters": filters},
            "reranker": {"query": query},
            "prompt_builder": {"query": query},
        },
        parameters=run_params,
    )

    return result

FastAPI service with observability

Expose the pipeline via REST. Add structured logging, request IDs, and latency tracking — essential for debugging production issues.

# app/api.py
import time
import uuid
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from app.config import get_settings
from app.pipeline import search, get_document_store
from app.models import SearchRequest, SearchResponse, SearchResult

settings = get_settings()
logger = logging.getLogger(__name__)

# Structured logging setup
class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = getattr(record, "request_id", "-")
        return True

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(level)s %(name)s request_id=%(request_id)s %(message)s",
)
logger.addFilter(RequestIdFilter())


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: verify connections
    document_store = get_document_store()
    try:
        document_store.client.ping()
        logger.info("Elasticsearch connection verified")
    except Exception as e:
        logger.error("Elasticsearch connection failed: %s", e)
        raise
    yield
    # Shutdown: cleanup if needed


app = FastAPI(
    title="Enterprise RAG API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.middleware("http")
async def add_request_id_and_logging(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    request.state.request_id = request_id

    # Bind request_id to logger for this request
    logger = logging.getLogger(__name__)
    extra = {"request_id": request_id}

    start = time.perf_counter()
    logger.info("Request started: %s %s", request.method, request.url.path, extra=extra)

    try:
        response = await call_next(request)
    except Exception as e:
        logger.exception("Request failed: %s", e, extra=extra)
        raise
    finally:
        latency_ms = (time.perf_counter() - start) * 1000
        logger.info(
            "Request completed: status=%d latency_ms=%.2f",
            response.status_code,
            latency_ms,
            extra=extra,
        )
        response.headers["X-Request-ID"] = request_id

    return response


class SearchRequestBody(BaseModel):
    query: str
    tenant_id: str
    user_roles: list[str] = []
    top_k: int = 5
    filters: dict = {}


class SearchResponseBody(BaseModel):
    results: list[dict]
    query: str
    tenant_id: str
    total_hits: int
    latency_ms: float


@app.post("/search", response_model=SearchResponseBody)
async def search_endpoint(body: SearchRequestBody, request: Request):
    start = time.perf_counter()

    try:
        result = search(
            query=body.query,
            tenant_id=body.tenant_id,
            user_roles=body.user_roles,
            top_k=body.top_k,
            extra_filters=body.filters,
        )
    except Exception as e:
        logger.exception("Search pipeline failed")
        raise HTTPException(status_code=500, detail="Search failed") from e

    latency_ms = (time.perf_counter() - start) * 1000

    # Extract documents from pipeline result
    documents = result.get("reranker", {}).get("documents", [])
    answers = result.get("generator", {}).get("replies", [])

    results = []
    for i, doc in enumerate(documents):
        results.append({
            "id": doc.id,
            "title": doc.meta.get("title", ""),
            "content": doc.content[:500],  # truncate for response
            "score": doc.score,
            "rerank_score": doc.meta.get("rerank_score"),
            "source": doc.meta.get("source"),
            "metadata": doc.meta.get("metadata", {}),
        })

    return SearchResponseBody(
        results=results,
        query=body.query,
        tenant_id=body.tenant_id,
        total_hits=len(documents),
        latency_ms=latency_ms,
    )


@app.get("/health")
async def health():
    return {"status": "ok"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host=settings.api_host, port=settings.api_port)

Sample data and ingestion script

Create a script to load test data. In production, you’d replace this with connectors to Confluence, SharePoint, etc.

# scripts/ingest_sample_data.py
import json
from pathlib import Path
from haystack import Document as HaystackDocument
from app.pipeline import index_documents, get_document_store
from app.config import get_settings

settings = get_settings()


def load_sample_documents(path: Path) -> list[HaystackDocument]:
    docs = []
    with path.open() as f:
        for line in f:
            data = json.loads(line)
            docs.append(HaystackDocument(
                id=data["id"],
                content=data["content"],
                meta={
                    "tenant_id": data["tenant_id"],
                    "title": data["title"],
                    "source": data["source"],
                    "source_id": data["source_id"],
                    "access_roles": data.get("access_roles", []),
                    "created_at": data["created_at"],
                    "updated_at": data["updated_at"],
                    "metadata": data.get("metadata", {}),
                },
            ))
    return docs


if __name__ == "__main__":
    import sys
    data_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("data/sample_docs.jsonl")
    documents = load_sample_documents(data_path)
    print(f"Loaded {len(documents)} documents")
    count = index_documents(documents)
    print(f"Indexed {count} documents")

Sample data/sample_docs.jsonl:

{"id": "doc-1", "tenant_id": "acme-corp", "title": "Q4 Budget Planning", "content": "The Q4 budget allocates $2.4M for engineering headcount...", "source": "confluence", "source_id": "page-12345", "access_roles": ["finance", "engineering-lead"], "created_at": "2024-01-15T10:00:00Z", "updated_at": "2024-01-15T10:00:00Z", "metadata": {"department": "finance"}}
{"id": "doc-2", "tenant_id": "acme-corp", "title": "Security Incident Response Plan", "content": "In the event of a data breach, notify security@ within 1 hour...", "source": "sharepoint", "source_id": "doc-67890", "access_roles": ["security", "engineering-lead", "executive"], "created_at": "2024-02-01T14:30:00Z", "updated_at": "2024-02-01T14:30:00Z", "metadata": {"classification": "confidential"}}
{"id": "doc-3", "tenant_id": "globex-inc", "title": "API Rate Limits", "content": "Default rate limit is 1000 requests/minute per API key...", "source": "gdrive", "source_id": "file-abcde", "access_roles": ["engineering", "support"], "created_at": "2024-03-10T09:15:00Z", "updated_at": "2024-03-10T09:15:00Z", "metadata": {"product": "payments-api"}}

Run ingestion:

python scripts/ingest_sample_data.py data/sample_docs.jsonl

Expected output:

Loaded 3 documents
Indexed 3 documents

Running the service

Start the API:

uvicorn app.api:app --host 0.0.0.0 --port 8000 --reload

Test with curl:

curl -X POST http://localhost:8000/search \
  -H "Content-Type: application/json" \
  -H "X-Request-ID: test-123" \
  -d '{
    "query": "What is the Q4 engineering budget?",
    "tenant_id": "acme-corp",
    "user_roles": ["engineering-lead"],
    "top_k": 3
  }'

Expected response (truncated):

{
  "results": [
    {
      "id": "doc-1",
      "title": "Q4 Budget Planning",
      "content": "The Q4 budget allocates $2.4M for engineering headcount...",
      "score": 0.87,
      "rerank_score": 0.92,
      "source": "confluence",
      "metadata": {"department": "finance"}
    }
  ],
  "query": "What is the Q4 engineering budget?",
  "tenant_id": "acme-corp",
  "total_hits": 1,
  "latency_ms": 142.3
}

Notice the rerank_score — the cross-encoder boosted the relevant document above BM25-only matches. The X-Request-ID header echoes back for traceability.

Testing tenant isolation

Verify that globex-inc cannot see acme-corp documents:

curl -X POST http://localhost:8000/search \
  -H "Content-Type: application/json" \
  -d '{
    "query": "budget",
    "tenant_id": "globex-inc",
    "user_roles": ["engineering"]
  }'

Response shows zero results — the tenant_id filter in the pipeline enforces isolation at the Elasticsearch level, not in application code.

Production considerations

Index updates

The indexing pipeline uses policy="overwrite" — re-running with the same document ID updates the vector and BM25 fields atomically. For high-throughput updates, batch writes and use Elasticsearch’s bulk API directly.

Scaling retrieval

• Increase bm25_top_k and dense_top_k for recall; the reranker filters to final_top_k
• For latency-sensitive paths, skip the generator and return reranked documents only
• Cache frequent query embeddings in Redis

Observability

Add OpenTelemetry instrumentation to the pipeline components. Log rerank_score distributions to detect drift. Alert on p99 latency > 500ms.

RBAC beyond roles

The access_roles filter is a simple keyword match. For complex policies (ABAC, hierarchical roles), evaluate permissions pre-retrieval and pass the resulting document ID allowlist as a terms filter.

What’s next

• Add a document deletion endpoint that removes from Elasticsearch by source_id
• Implement incremental ingestion with change detection (hash-based)
• Swap the cross-encoder for a lighter bi-encoder + late interaction (ColBERT) if latency is critical
• Add evaluation harness with labeled queries to track nDCG over time

The pipeline above runs in production at several companies handling 10M+ documents. The key decisions — hybrid retrieval with RRF, cross-encoder reranking, tenant filtering at the index level, structured logging — are the ones that survive scale.