How to Integrate Haystack for pension funds with Elasticsearch for RAG

If you’re building an AI agent for a pension fund, you need retrieval that is both fast and auditable. Haystack gives you the RAG orchestration layer, while Elasticsearch gives you indexed search over policy docs, investment memos, actuarial reports, and member communications.

The useful part of this combo is simple: your agent can answer questions like “What’s our glide path policy for default funds?” or “Show the latest contribution escalation rules” by pulling grounded evidence from a controlled document store.

Prerequisites

• •Python 3.10+
• •An Elasticsearch cluster running locally or in your environment
• •An Elasticsearch user with index permissions
• •A Haystack installation with RAG components available
• •Access to an embedding model, either:
  • •a local model
  • •OpenAI-compatible embeddings
  • •another Haystack-supported embedding backend
• •Pension fund documents ready to ingest:
  • •policy PDFs
  • •trustee meeting notes
  • •compliance memos
  • •member FAQ content

Install the Python packages:

pip install haystack-ai elasticsearch sentence-transformers pypdf

Integration Steps

1) Connect to Elasticsearch and create a target index

Start by creating a dedicated index for pension fund knowledge. Keep the schema simple: text content, metadata, and vector fields if you plan to do semantic retrieval.

from elasticsearch import Elasticsearch

es = Elasticsearch(
    "http://localhost:9200",
    basic_auth=("elastic", "changeme")
)

index_name = "pension-fund-rag"

mapping = {
    "mappings": {
        "properties": {
            "content": {"type": "text"},
            "source": {"type": "keyword"},
            "doc_type": {"type": "keyword"},
            "page": {"type": "integer"},
            "embedding": {
                "type": "dense_vector",
                "dims": 384,
                "index": True,
                "similarity": "cosine"
            }
        }
    }
}

if not es.indices.exists(index=index_name):
    es.indices.create(index=index_name, **mapping)

For production, lock this down with role-based access control and separate indexes per business domain.

2) Load pension fund documents into Haystack documents

Haystack works best when you normalize your data into Document objects before indexing. This keeps metadata attached all the way through retrieval.

from haystack import Document

documents = [
    Document(
        content="Default fund members are enrolled in the lifecycle strategy until age 55.",
        meta={"source": "trustee_policy_2024.pdf", "doc_type": "policy", "page": 12}
    ),
    Document(
        content="Contribution escalation is reviewed annually by the board and applied each April.",
        meta={"source": "member_faq.docx", "doc_type": "faq", "page": 3}
    ),
]

If your source files are PDFs, extract text first and chunk them before creating Document instances. For pension data, keep chunks small enough to preserve citation precision.

3) Embed documents and write them into Elasticsearch

Use a Haystack embedding component to turn each chunk into a vector, then push both text and embedding into Elasticsearch. This is the core of semantic retrieval.

from haystack.components.embedders import SentenceTransformersDocumentEmbedder

embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
embedder.warm_up()

embedded_docs = embedder.run(documents=documents)["documents"]

bulk_actions = []
for doc in embedded_docs:
    bulk_actions.append({
        "_index": index_name,
        "_source": {
            "content": doc.content,
            **doc.meta,
            "embedding": doc.embedding.tolist()
        }
    })

from elasticsearch.helpers import bulk
bulk(es, bulk_actions)

This pattern keeps indexing explicit. In regulated environments, that matters because you can log every document that enters the retrieval corpus.

4) Build a Haystack retrieval pipeline backed by Elasticsearch

Now wire Haystack to query Elasticsearch using vector search. Use a retriever that can query the dense vector field you created earlier.

from haystack import Pipeline
from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.components.retrievers import InMemoryEmbeddingRetriever

query_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
query_embedder.warm_up()

# If you're using Elastic's native vector search through your own query layer,
# keep Haystack responsible for orchestration and pass retrieved docs back in.
pipeline = Pipeline()
pipeline.add_component("query_embedder", query_embedder)

question = "When are contribution escalations reviewed?"
result = pipeline.run({"query_embedder": {"text": question}})
query_embedding = result["query_embedder"]["embedding"]

At this point, send query_embedding to Elasticsearch with a script_score or kNN query. That’s the cleanest production pattern when you want full control over ranking logic and filters.

search_response = es.search(
    index=index_name,
    size=3,
    query={
        "script_score": {
            "query": {"match_all": {}},
            "script": {
                "source": """
                    cosineSimilarity(params.query_vector, 'embedding') + 1.0
                """,
                "params": {"query_vector": query_embedding.tolist()}
            }
        }
    }
)

5) Feed retrieved context into your generation step

Once Elasticsearch returns top matches, pass those chunks into your LLM prompt. Keep citations attached so your agent can explain where answers came from.

top_hits = search_response["hits"]["hits"]

context = "\n\n".join(
    f"[{hit['_source']['source']} p.{hit['_source'].get('page', '?')}] {hit['_source']['content']}"
    for hit in top_hits
)

prompt = f"""
You are answering questions about pension fund operations.
Use only the context below.

Context:
{context}

Question: {question}
Answer:
"""

print(prompt)

In a real agent system, this prompt goes into your generator component, then back to the user with references attached.

Testing the Integration

Run a direct retrieval test against a known pension policy question. You want to confirm two things: Elasticsearch returns relevant chunks, and Haystack produces embeddings consistently.

test_question = "What month is contribution escalation applied?"
q_vec = query_embedder.run(text=test_question)["embedding"]

resp = es.search(
    index=index_name,
    size=1,
    query={
        "script_score": {
            "query": {"match_all": {}},
            "script": {
                "source": """
                    cosineSimilarity(params.query_vector, 'embedding') + 1.0
                """,
                "params": {"query_vector": q_vec.tolist()}
            }
        }
    }
)

hit = resp["hits"]["hits"][0]["_source"]
print(hit["source"])
print(hit["content"])

Expected output:

member_faq.docx
Contribution escalation is reviewed annually by the board and applied each April.

If that comes back cleanly, your ingestion, embedding, indexing, and retrieval path are connected.

Real-World Use Cases

• •

  Trustee Q&A assistant

  • •Answer governance questions from board packs, policy docs, and investment committee minutes with citations.
• •

  Member support agent

  • •Retrieve accurate responses about contributions, retirement dates, default funds, and benefit options from approved knowledge sources.
• •

  Compliance evidence lookup

  • •Let internal teams search policy history and operational procedures during audits without manually digging through shared drives.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.