How to Integrate Haystack for retail banking with Elasticsearch for production AI

Haystack for retail banking gives you the orchestration layer for banking-specific AI workflows. Elasticsearch gives you the retrieval backbone: fast indexing, filtering, and hybrid search over structured and unstructured banking data.

Put them together and you can build production-grade agents that answer customer questions from policy docs, account servicing knowledge bases, product brochures, and compliance content without hardcoding brittle retrieval logic.

Prerequisites

• •Python 3.10+
• •An Elasticsearch cluster running locally or in your VPC
• •Access credentials for Elasticsearch
• •Haystack installed in your environment
• •A document source for retail banking content:
  • •FAQs
  • •product terms
  • •fee schedules
  • •servicing procedures
  • •compliance policies
• •Optional but recommended:
  • •a reranker model
  • •embeddings model for semantic search
  • •secure secret management for credentials

Install the core packages:

pip install haystack-ai elasticsearch sentence-transformers

Integration Steps

1. •Create the Elasticsearch client and verify connectivity.

Start by connecting to your cluster directly. In production, use API keys or service auth, not anonymous access.

from elasticsearch import Elasticsearch

es = Elasticsearch(
    "https://localhost:9200",
    api_key=("your_api_key_id", "your_api_key_secret"),
)

health = es.cluster.health()
print(health["status"])

If this fails, fix network access, TLS, or auth before touching Haystack.

2. •Define a banking document schema and create the index.

Retail banking search works best when you separate content types and keep filters explicit. Store document text plus metadata like product line, jurisdiction, channel, and effective date.

index_name = "retail-banking-kb"

mapping = {
    "mappings": {
        "properties": {
            "content": {"type": "text"},
            "title": {"type": "text"},
            "product": {"type": "keyword"},
            "jurisdiction": {"type": "keyword"},
            "doc_type": {"type": "keyword"},
            "effective_date": {"type": "date"}
        }
    }
}

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

print(es.indices.get(index=index_name))

This gives you deterministic filtering later when the agent needs to scope answers to a specific bank product or region.

3. •Load documents into Haystack and write them to Elasticsearch.

Haystack’s Document objects are the clean handoff point between ingestion and retrieval. Use ElasticsearchDocumentStore to persist them in your index.

from haystack import Document
from haystack_integrations.document_stores.elasticsearch import ElasticsearchDocumentStore

document_store = ElasticsearchDocumentStore(
    hosts=["https://localhost:9200"],
    basic_auth=("elastic", "your_password"),
    index=index_name,
    verify_certs=True,
)

docs = [
    Document(
        content="Savings accounts accrue interest daily and are credited monthly.",
        meta={
            "title": "Savings Account Interest",
            "product": "savings",
            "jurisdiction": "US",
            "doc_type": "faq",
            "effective_date": "2025-01-01",
        },
    ),
    Document(
        content="Wire transfers submitted after 4 PM ET are processed on the next business day.",
        meta={
            "title": "Wire Transfer Cutoff",
            "product": "payments",
            "jurisdiction": "US",
            "doc_type": "policy",
            "effective_date": "2025-01-01",
        },
    ),
]

document_store.write_documents(docs)
print("documents_written")

For larger corpora, batch ingestion and chunking matter. Don’t push raw PDFs straight into search.

4. •Build a Haystack retrieval pipeline on top of Elasticsearch.

Use an embedding retriever if you want semantic matching across customer phrasing. If your bank needs strict keyword behavior for regulated content, combine it with metadata filters.

from haystack import Pipeline
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever

# For a production setup with ES-backed retrieval,
# use the Elasticsearch retriever component available in Haystack integrations.
from haystack_integrations.components.retrievers.elasticsearch import ElasticsearchEmbeddingRetriever

retriever = ElasticsearchEmbeddingRetriever(
    document_store=document_store,
)

pipe = Pipeline()
pipe.add_component("retriever", retriever)

query = {
    "text": "When does my wire transfer get processed?",
}
result = pipe.run(
    data={
        "retriever": {
            "query_embedding": None,
            # depending on your embedding setup, pass query text through your embedder first
        }
    }
)
print(result)

In a real agent flow, wire this retriever behind an embedder component and then pass retrieved documents into your generator or policy engine.

5. •Add filtering for banking constraints before answering.

This is where retail banking differs from generic RAG. You need jurisdiction-aware retrieval so the agent doesn’t answer from the wrong policy set.

response = es.search(
    index=index_name,
    query={
        bool := {
            # placeholder style shown below in proper body form
        }
    },
)

Use explicit filters with bool queries instead of relying only on similarity scores:

response = es.search(
    index=index_name,
    query={
        "bool": {
            "must": [
                {"match": {"content": "wire transfer processing time"}}
            ],
            "filter": [
                {"term": {"jurisdiction": "US"}},
                {"term": {"product.keyword" if False else "product": ""}}
            ]
        }
    },
)

A cleaner pattern is to let Haystack retrieve candidate docs, then apply business rules before generation:

docs = document_store.filter_documents(filters={"field":"jurisdiction","operator":"==","value":"US"})
print(len(docs))

Testing the Integration

Run a direct retrieval test against Elasticsearch through Haystack-backed storage:

results = document_store.filter_documents(filters={"field":"product","operator":"==","value":"payments"})
for doc in results:
    print(doc.meta["title"], "-", doc.content)

Expected output:

Wire Transfer Cutoff - Wire transfers submitted after 4 PM ET are processed on the next business day.

If you get zero results:

• •check the index name
• •confirm documents were written
• •verify metadata values match exactly
• •inspect whether your cluster is using nested mappings or aliases

Real-World Use Cases

• •Customer servicing agent that answers questions about fees, transfer cutoffs, overdraft rules, and card replacement policies using only approved bank content.
• •Internal ops assistant that searches SOPs and compliance docs by product line and jurisdiction before giving staff guidance.
• •Fraud support workflow that retrieves escalation playbooks and case-handling steps based on incident type and channel.

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.