How to Integrate Haystack for payments with Elasticsearch for RAG

Combining Haystack for payments with Elasticsearch gives you a clean pattern for payment-aware RAG: retrieve policy, invoice, and transaction context from search, then let the agent answer or act with the payment layer attached. In practice, this is what you want when an assistant needs to answer “why was this charge declined?”, “can I retry this invoice?”, or “what’s the refund status for order 1842?” without bouncing between systems.

Prerequisites

• •Python 3.10+
• •An Elasticsearch cluster running locally or in your environment
• •A Haystack for payments project configured with:
  • •API key or service credentials
  • •Payment processor connection details
• •pip access to install dependencies
• •Documents indexed in Elasticsearch:
  • •invoices
  • •payment events
  • •refund policies
  • •transaction notes
• •A vector-capable Elasticsearch index if you plan to do semantic retrieval

Install the core packages:

pip install haystack-ai elasticsearch python-dotenv

Integration Steps

1. •Connect to Elasticsearch and create a retriever-friendly index

You want Elasticsearch to store both metadata and text chunks that your agent can retrieve during RAG. For production, keep payment records normalized and index only the fields needed for retrieval.

from elasticsearch import Elasticsearch

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

index_name = "payments-rag"

if not es.indices.exists(index=index_name):
    es.indices.create(
        index=index_name,
        mappings={
            "properties": {
                "doc_id": {"type": "keyword"},
                "content": {"type": "text"},
                "customer_id": {"type": "keyword"},
                "payment_id": {"type": "keyword"},
                "doc_type": {"type": "keyword"},
            }
        },
    )

2. •Write payment documents into Elasticsearch

This is where you load receipts, dispute notes, and payment status updates. Keep content human-readable so Haystack can retrieve useful context later.

docs = [
    {
        "doc_id": "inv-1001",
        "content": "Invoice 1001 was paid on 2026-04-18. Card ending 4242. Status: settled.",
        "customer_id": "cust-77",
        "payment_id": "pay-9001",
        "doc_type": "invoice",
    },
    {
        "doc_id": "refund-2002",
        "content": "Refund request for payment pay-9001 was approved and sent to processor.",
        "customer_id": "cust-77",
        "payment_id": "pay-9001",
        "doc_type": "refund",
    },
]

for doc in docs:
    es.index(index=index_name, id=doc["doc_id"], document=doc)

es.indices.refresh(index=index_name)

3. •Build a Haystack RAG pipeline that queries Elasticsearch

Haystack’s ElasticsearchDocumentStore is the bridge here. It lets your agent retrieve relevant payment context before generating an answer.

from haystack import Document, Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIChatGenerator
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.document_stores.elasticsearch import ElasticsearchDocumentStore

document_store = ElasticsearchDocumentStore(
    hosts=["http://localhost:9200"],
    index=index_name,
    basic_auth=("elastic", "changeme"),
)

# If you're syncing from ES directly into Haystack docs:
haystack_docs = [
    Document(content=d["content"], meta={k: v for k, v in d.items() if k != "content"})
    for d in docs
]
document_store.write_documents(haystack_docs)

template = """
Use the retrieved payment context to answer the question.

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

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

pipe = Pipeline()
pipe.add_component("retriever", InMemoryBM25Retriever(document_store=document_store))
pipe.add_component("prompt_builder", PromptBuilder(template=template))
pipe.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))

pipe.connect("retriever.documents", "prompt_builder.documents")
pipe.connect("prompt_builder.prompt", "llm.messages")

4. •Attach your payments SDK call after retrieval

This is the part most teams miss. Retrieval answers the “what happened?” question; the payments SDK handles “do it now” actions like capture, refund, or retry.

If your Haystack for payments SDK exposes an action client, wire it after retrieval so the agent has context before calling the payment API.

from haystack_payments import PaymentsClient

payments = PaymentsClient(api_key="hp_test_123")

result = pipe.run({
    "retriever": {
        "query": "What is the status of refund for pay-9001?"
    },
    "prompt_builder": {
        "question": "What is the status of refund for pay-9001?"
    }
})

answer = result["llm"]["replies"][0].text
print(answer)

# Example action based on retrieved context:
if "approved" in answer.lower():
    refund_response = payments.refunds.get(payment_id="pay-9001")
    print(refund_response)

5. •Add filtering so agents only retrieve customer-scoped records

For banking and insurance workloads, this matters more than fancy prompting. Scope every query by customer or account identifier before you generate a response.

query = {
    "_source": ["doc_id", "content", "customer_id", "payment_id", "doc_type"],
    "query": {
        "bool": {
            "must": [
                {"match": {"content": {"query": "refund status"}}},
                {"term": {"customer_id": {"value": "cust-77"}}}
            ]
        }
    }
}

search_results = es.search(index=index_name, body=query)
for hit in search_results["hits"]["hits"]:
    print(hit["_source"]["content"])

Testing the Integration

Run a retrieval-plus-payment check with one known record and verify both systems return consistent data.

test_query = {
    "_source": ["doc_id", "content"],
    "query": {
        "match_phrase": {
            "content": {
                "query": "\"refund request\""
            }
        }
    }
}

resp = es.search(index=index_name, body=test_query)
print(f"Hits: {resp['hits']['total']['value']}")

haystack_result = pipe.run({
    "retriever": {"query": "refund status for pay-9001"},
    "prompt_builder": {"question": "refund status for pay-9001"}
})

print(haystack_result["llm"]["replies"][0].text)

Expected output:

Hits: 1
Refund request for payment pay-9001 was approved and sent to processor.

Real-World Use Cases

• •

  Payment support agent

  • •Retrieve invoice history from Elasticsearch.
  • •Use Haystack to answer chargeback and refund questions with exact transaction context.
• •

  Ops copilot for finance teams

  • •Search settlement logs, failed captures, and retry attempts.
  • •Trigger payment actions only after retrieving policy-compliant evidence.
• •

  Claims-to-payment workflows

  • •Index claim notes and payout records.
  • •Let an agent explain why a payout was delayed and whether a retry should be issued.

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.