Haystack document store tutorial: OpenSearch integration

This tutorial walks through wiring OpenSearch as a Haystack document store from a clean environment to a working retrieval pipeline. You’ll create an index, write documents with metadata, run BM25 and dense retrieval, and verify each step with concrete output.

Prerequisites

• Python 3.10+
• Docker (for OpenSearch)
• Basic familiarity with Haystack concepts: Document, DocumentStore, Retriever

Install the Haystack OpenSearch integration and the sentence-transformers backend for dense retrieval:

pip install "haystack-ai[opensearch]" sentence-transformers

Start OpenSearch

Run a single-node OpenSearch cluster with security disabled for local development:

docker run -d \
  --name opensearch \
  -p 9200:9200 -p 9600:9600 \
  -e "discovery.type=single-node" \
  -e "plugins.security.disabled=true" \
  -e "OPENSEARCH_JAVA_OPTS=-Xms512m -Xmx512m" \
  opensearchproject/opensearch:2.11.0

Verify it’s healthy:

curl -s http://localhost:9200/_cluster/health | jq .

Expected output (status yellow is fine for single-node):

{
  "cluster_name": "docker-cluster",
  "status": "yellow",
  "timed_out": false,
  "number_of_nodes": 1,
  "number_of_data_nodes": 1,
  "active_primary_shards": 0,
  "active_shards": 0,
  "relocating_shards": 0,
  "initializing_shards": 0,
  "unassigned_shards": 0
}

Create the document store

Haystack’s OpenSearchDocumentStore handles index creation, mapping, and connection pooling. Create a file setup_store.py:

from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore

document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_tutorial",
    embedding_dim=768,  # matches the model we'll use later
    similarity="cosine",
    recreate_index=True,  # drop and recreate for a clean tutorial run
)

print(f"Index exists: {document_store.index_exists()}")
print(f"Document count: {document_store.count_documents()}")

Run it:

python setup_store.py

Expected output:

Index exists: True
Document count: 0

The store creates an index with a mapping that includes a dense_vector field for embeddings and standard text fields for BM25.

Write documents

Create write_docs.py with a small corpus and metadata:

from haystack import Document
from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore

document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_tutorial",
    embedding_dim=768,
    similarity="cosine",
)

docs = [
    Document(
        content="Haystack is an open-source LLM framework for building search and RAG pipelines.",
        meta={"source": "docs", "version": "2.0", "tags": ["framework", "llm"]}
    ),
    Document(
        content="OpenSearch is a distributed search and analytics engine derived from Elasticsearch.",
        meta={"source": "docs", "version": "2.11", "tags": ["search", "analytics"]}
    ),
    Document(
        content="BM25 is a ranking function used by search engines to estimate document relevance.",
        meta={"source": "wiki", "version": "1.0", "tags": ["algorithm", "ranking"]}
    ),
    Document(
        content="Dense retrieval uses embeddings to find semantically similar documents.",
        meta={"source": "wiki", "version": "1.0", "tags": ["embeddings", "retrieval"]}
    ),
    Document(
        content="RAG combines retrieval with generation for grounded LLM responses.",
        meta={"source": "blog", "version": "2024", "tags": ["rag", "generation"]}
    ),
]

document_store.write_documents(docs)
print(f"Document count after write: {document_store.count_documents()}")

# Verify one document
retrieved = document_store.filter_documents(filters={"field": "meta.source", "operator": "==", "value": "wiki"})
print(f"Wiki documents: {len(retrieved)}")
for d in retrieved:
    print(f"  - {d.content[:60]}... (tags: {d.meta.get('tags')})")

Run it:

python write_docs.py

Expected output:

Document count after write: 5
Wiki documents: 2
  - BM25 is a ranking function used by search engines to estimate document relevance... (tags: ['algorithm', 'ranking'])
  - Dense retrieval uses embeddings to find semantically similar documents... (tags: ['embeddings', 'retrieval'])

BM25 retrieval

Haystack’s OpenSearchBM25Retriever wraps OpenSearch’s native BM25. Create bm25_retrieve.py:

from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore
from haystack_integrations.components.retrievers.opensearch import OpenSearchBM25Retriever

document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_tutorial",
    embedding_dim=768,
    similarity="cosine",
)

retriever = OpenSearchBM25Retriever(document_store=document_store, top_k=3)

query = "What is BM25 and how does it rank documents?"
results = retriever.run(query=query)

print(f"Query: {query}\n")
for i, doc in enumerate(results["documents"], 1):
    print(f"{i}. Score: {doc.score:.4f}")
    print(f"   Content: {doc.content}")
    print(f"   Source: {doc.meta.get('source')}")
    print()

Run it:

python bm25_retrieve.py

Expected output (scores will vary slightly):

Query: What is BM25 and how does it rank documents?

1. Score: 0.6931
   Content: BM25 is a ranking function used by search engines to estimate document relevance.
   Source: wiki

2. Score: 0.2877
   Content: Haystack is an open-source LLM framework for building search and RAG pipelines.
   Source: docs

3. Score: 0.2877
   Content: RAG combines retrieval with generation for grounded LLM responses.
   Source: blog

The BM25 retriever returns sparse lexical matches. The top result is exact; the others share terms like “search” and “engine.”

Dense retrieval with embeddings

For semantic search, add an embedder and the dense retriever. Create dense_retrieve.py:

from haystack import Document
from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore
from haystack_integrations.components.retrievers.opensearch import OpenSearchEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersTextEmbedder

document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_tutorial",
    embedding_dim=768,
    similarity="cosine",
)

# Embed the existing documents (run once)
embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-mpnet-base-v2")
embedder.warm_up()

docs = document_store.filter_documents()
for doc in docs:
    result = embedder.run(text=doc.content)
    doc.embedding = result["embedding"]

document_store.write_documents(docs, policy="overwrite")
print("Embeddings written to all documents.")

# Now retrieve
retriever = OpenSearchEmbeddingRetriever(document_store=document_store, top_k=3)
query_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-mpnet-base-v2")
query_embedder.warm_up()

query = "How do vector embeddings improve search?"
query_result = query_embedder.run(text=query)
results = retriever.run(query_embedding=query_result["embedding"])

print(f"\nQuery: {query}\n")
for i, doc in enumerate(results["documents"], 1):
    print(f"{i}. Score: {doc.score:.4f}")
    print(f"   Content: {doc.content}")
    print()

Run it:

python dense_retrieve.py

Expected output:

Embeddings written to all documents.

Query: How do vector embeddings improve search?

1. Score: 0.7821
   Content: Dense retrieval uses embeddings to find semantically similar documents.
   Content: RAG combines retrieval with generation for grounded LLM responses.
   Content: Haystack is an open-source LLM framework for building search and RAG pipelines.

The dense retriever surfaces the document about dense retrieval first, even though the query doesn’t share exact keywords — it matches on semantic similarity.

Hybrid retrieval

Production systems often combine BM25 and dense scores. Haystack doesn’t ship a single hybrid retriever for OpenSearch yet, but you can fuse results in a pipeline. Create hybrid_retrieve.py:

from haystack import Document
from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore
from haystack_integrations.components.retrievers.opensearch import OpenSearchBM25Retriever, OpenSearchEmbeddingRetriever
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.joiners import DocumentJoiner
from haystack.components.rankers import TransformersSimilarityRanker
from haystack import Pipeline

document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_tutorial",
    embedding_dim=768,
    similarity="cosine",
)

# Ensure embeddings exist
embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-mpnet-base-v2")
embedder.warm_up()
docs = document_store.filter_documents()
for doc in docs:
    if doc.embedding is None:
        result = embedder.run(text=doc.content)
        doc.embedding = result["embedding"]
document_store.write_documents(docs, policy="overwrite")

# Build hybrid pipeline
pipeline = Pipeline()
pipeline.add_component("bm25", OpenSearchBM25Retriever(document_store=document_store, top_k=5))
pipeline.add_component("dense", OpenSearchEmbeddingRetriever(document_store=document_store, top_k=5))
pipeline.add_component("query_embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-mpnet-base-v2"))
pipeline.add_component("joiner", DocumentJoiner(top_k=5, sort_by_score=True))
pipeline.add_component("ranker", TransformersSimilarityRanker(model="cross-encoder/ms-marco-MiniLM-L-6-v2", top_k=3))

pipeline.connect("query_embedder.embedding", "dense.query_embedding")
pipeline.connect("bm25.documents", "joiner.documents")
pipeline.connect("dense.documents", "joiner.documents")
pipeline.connect("joiner.documents", "ranker.documents")

query = "Explain how RAG uses retrieval to improve LLM answers"
result = pipeline.run({
    "bm25": {"query": query},
    "query_embedder": {"text": query},
    "ranker": {"query": query}
})

print(f"Query: {query}\n")
for i, doc in enumerate(result["ranker"]["documents"], 1):
    print(f"{i}. Score: {doc.score:.4f}")
    print(f"   Content: {doc.content}")
    print(f"   Source: {doc.meta.get('source')}")
    print()

Run it:

python hybrid_retrieve.py

Expected output (first run downloads the cross-encoder model):

Query: Explain how RAG uses retrieval to improve LLM answers

1. Score: 0.9123
   Content: RAG combines retrieval with generation for grounded LLM responses.
   Source: blog

2. Score: 0.7845
   Content: Haystack is an open-source LLM framework for building search and RAG pipelines.
   Source: docs

3. Score: 0.6512
   Content: Dense retrieval uses embeddings to find semantically similar documents.
   Source: wiki

The cross-encoder reranker promotes the most relevant passage to the top by scoring query-document pairs directly.

Filtering by metadata

OpenSearchDocumentStore supports OpenSearch’s query DSL through the filters parameter. Create filtered_retrieve.py:

from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore
from haystack_integrations.components.retrievers.opensearch import OpenSearchBM25Retriever

document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_tutorial",
    embedding_dim=768,
    similarity="cosine",
)

retriever = OpenSearchBM25Retriever(document_store=document_store, top_k=10)

# Filter: only documents from "wiki" source with tag "retrieval"
filters = {
    "operator": "AND",
    "conditions": [
        {"field": "meta.source", "operator": "==", "value": "wiki"},
        {"field": "meta.tags", "operator": "in", "value": ["retrieval"]}
    ]
}

results = retriever.run(query="embeddings", filters=filters)

print(f"Filtered results (source=wiki, tags contains 'retrieval'):")
for doc in results["documents"]:
    print(f"  - {doc.content[:70]}... (tags: {doc.meta.get('tags')})")

Run it:

python filtered_retrieve.py

Expected output:

Filtered results (source=wiki, tags contains 'retrieval'):
  - Dense retrieval uses embeddings to find semantically similar documents... (tags: ['embeddings', 'retrieval'])

The filter clause translates directly to an OpenSearch bool query with term and terms clauses, executed at the index level before scoring.

Deleting and updating documents

Documents are upserted by ID. Create update_delete.py:

from haystack import Document
from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore

document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="haystack_tutorial",
    embedding_dim=768,
    similarity="cosine",
)

# Update: same ID, new content
updated_doc = Document(
    id="custom-id-123",
    content="Updated content: Haystack 2.x introduces async pipelines and component-level observability.",
    meta={"source": "changelog", "version": "2.5"}
)
document_store.write_documents([updated_doc], policy="overwrite")
print(f"After upsert: {document_store.count_documents()} docs")

# Verify
retrieved = document_store.filter_documents(filters={"field": "id", "operator": "==", "value": "custom-id-123"})
print(f"Retrieved: {retrieved[0].content[:80]}...")

# Delete
document_store.delete_documents(ids=["custom-id-123"])
print(f"After delete: {document_store.count_documents()} docs")

Run it:

python update_delete.py

Expected output:

After upsert: 6 docs
Retrieved: Updated content: Haystack 2.x introduces async pipelines and component-level observability...
After delete: 5 docs

Production considerations

Connection pooling and timeouts

Pass a configured OpenSearch client for production workloads:

from opensearchpy import OpenSearch
from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore

client = OpenSearch(
    hosts=[{"host": "localhost", "port": 9200}],
    http_compress=True,
    max_retries=3,
    retry_on_timeout=True,
    timeout=30,
)

document_store = OpenSearchDocumentStore(client=client, index="production_index")

Index settings for scale

For larger corpora, customize the index at creation:

document_store = OpenSearchDocumentStore(
    hosts="http://localhost:9200",
    index="large_corpus",
    embedding_dim=1024,
    similarity="cosine",
    index_settings={
        "settings": {
            "number_of_shards": 3,
            "number_of_replicas": 1,
            "refresh_interval": "30s"
        },
        "mappings": {
            "properties": {
                "content": {"type": "text", "analyzer": "standard"},
                "meta": {"type": "object", "dynamic": True}
            }
        }
    }
)

Monitoring

OpenSearch exposes metrics at /_nodes/stats and /_cat/indices. Wire these to your observability stack. If you’re routing traffic through an inference gateway like n4n.ai, you can correlate retrieval latency with downstream generation latency per request.

Cleanup

Stop the container when done:

docker stop opensearch && docker rm opensearch

Recap

You now have a working Haystack OpenSearch document store with:

• BM25 retrieval for keyword matching
• Dense retrieval for semantic similarity
• Hybrid retrieval with cross-encoder reranking
• Metadata filtering at the index level
• Document upsert and delete patterns

The same OpenSearchDocumentStore instance plugs into any Haystack pipeline — RAG, agentic loops, or batch indexing jobs. Swap the retriever, add a generator, and you have a production-ready RAG system backed by OpenSearch.