Haystack Financial Document Pipelines with Lekha
Build production-ready Indian financial document pipelines using Haystack and Lekha API. Parse bank statements, salary slips, and ITRs with TypeScript.
Haystack by deepset is one of the most popular Python frameworks for building production AI pipelines. When you need to process Indian financial documents — bank statements, salary slips, Form 16s, CAS statements — combining Haystack's orchestration power with Lekha's specialized extraction API gives you a pipeline that handles real-world Indian document formats reliably.
This guide shows you how to build a Haystack pipeline that extracts structured data from Indian financial documents, routes them by type, and produces clean JSON ready for downstream decision logic.
Why Haystack + Lekha?
Haystack excels at building composable, maintainable AI pipelines with branching logic, custom components, and observability hooks. Lekha handles the hard part of Indian financial documents — complex table layouts, multiple bank formats, vernacular number formats like "₹1,45,000", and multi-page PDFs.
Together they let you:
Prerequisites
You'll need:
haystack-ai>=2.0 installedpip install haystack-ai requests
Get your API key at lekhadev.com — the free tier covers 50 documents/month to get started.
Building Your First Lekha Component
Haystack's component system lets you wrap any API as a pipeline node. Here's a custom LekhaExtractor component:
from haystack import component, default_from_dict, default_to_dict
from haystack.dataclasses import ByteStream
from typing import Optional
import requests
import base64
@component
class LekhaExtractor:
"""Extracts structured data from Indian financial documents via Lekha API."""
def __init__(self, api_key: str, document_type: Optional[str] = None):
self.api_key = api_key
self.document_type = document_type # None = auto-classify
self.base_url = "https://api.lekhadev.com/v1"
@component.output_types(
document_type=str,
structured_data=dict,
confidence=float,
error=Optional[str],
)
def run(self, file: ByteStream, filename: str = "document.pdf"):
"""
Extract structured data from a financial document.
Args:
file: Document as ByteStream
filename: Original filename (helps with type detection)
Returns:
document_type, structured_data, confidence, error
"""
encoded = base64.b64encode(file.data).decode("utf-8")
mime_type = file.mime_type or "application/pdf"
payload = {
"file": encoded,
"filename": filename,
"mime_type": mime_type,
}
if self.document_type:
payload["document_type"] = self.document_type
try:
response = requests.post(
f"{self.base_url}/extract",
json=payload,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
timeout=60,
)
response.raise_for_status()
result = response.json()
if result.get("success"):
data = result["data"]
return {
"document_type": data.get("document_type", "unknown"),
"structured_data": data,
"confidence": data.get("confidence", 0.0),
"error": None,
}
else:
return {
"document_type": "unknown",
"structured_data": {},
"confidence": 0.0,
"error": result.get("error", {}).get("message", "Extraction failed"),
}
except requests.RequestException as e:
return {
"document_type": "unknown",
"structured_data": {},
"confidence": 0.0,
"error": str(e),
}
def to_dict(self):
return default_to_dict(self, api_key=self.api_key, document_type=self.document_type)
@classmethod
def from_dict(cls, data):
return default_from_dict(cls, data)
A Complete Loan Eligibility Pipeline
Here's a real-world pipeline that processes a loan application package — bank statement, salary slip, and Form 16 — and synthesizes an eligibility assessment:
from haystack import Pipeline
from haystack.components.routers import MetadataRouter
from haystack.dataclasses import ByteStream
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
import os
def build_loan_eligibility_pipeline(lekha_api_key: str) -> Pipeline:
pipeline = Pipeline()
# Add Lekha extractor — auto-classifies any Indian financial document
pipeline.add_component("extractor", LekhaExtractor(api_key=lekha_api_key))
# Prompt builder synthesizes the extracted data for LLM analysis
template = """
You are a loan underwriter analyzing Indian financial documents.
Document Type: {{ document_type }}
Extracted Data:
{{ structured_data }}
Based on this data, provide:
1. Monthly income (if determinable)
2. Average monthly balance (for bank statements)
3. Any red flags (bounced EMIs, irregular income, excessive withdrawals)
4. Preliminary eligibility: ELIGIBLE / NEEDS_REVIEW / INELIGIBLE
Be concise. Output JSON only.
"""
pipeline.add_component(
"prompt_builder",
PromptBuilder(template=template)
)
pipeline.add_component(
"llm",
OpenAIGenerator(
api_key=os.environ["OPENAI_API_KEY"],
model="gpt-4o-mini",
)
)
# Wire up the pipeline
pipeline.connect("extractor.document_type", "prompt_builder.document_type")
pipeline.connect("extractor.structured_data", "prompt_builder.structured_data")
pipeline.connect("prompt_builder.prompt", "llm.prompt")
return pipeline
Run it
pipeline = build_loan_eligibility_pipeline(os.environ["LEKHA_API_KEY"])
with open("applicant_salary_slip.pdf", "rb") as f:
file_bytes = f.read()
result = pipeline.run({
"extractor": {
"file": ByteStream(data=file_bytes, mime_type="application/pdf"),
"filename": "applicant_salary_slip.pdf",
}
})
print(result["llm"]["replies"][0])
Branching by Document Type
For applications that receive mixed documents — some users upload salary slips, others upload ITRs — you can branch the pipeline using Haystack's ConditionalRouter:
from haystack.components.routers import ConditionalRouter
routes = [
{
"condition": "{{ document_type == 'bank_statement' }}",
"output": "{{ structured_data }}",
"output_name": "bank_data",
"output_type": dict,
},
{
"condition": "{{ document_type == 'salary_slip' }}",
"output": "{{ structured_data }}",
"output_name": "salary_data",
"output_type": dict,
},
{
"condition": "{{ document_type not in ['bank_statement', 'salary_slip'] }}",
"output": "{{ structured_data }}",
"output_name": "other_data",
"output_type": dict,
},
]
router = ConditionalRouter(routes=routes)
pipeline.add_component("router", router)
pipeline.connect("extractor.document_type", "router.document_type")
pipeline.connect("extractor.structured_data", "router.structured_data")
Now connect each branch to specialized analyzers...
This routing pattern keeps your pipeline clean — the bank statement branch can compute running balances and flag EMI bounces, while the salary slip branch focuses on gross pay, deductions, and employer PF contributions.
Batch Processing with Async Pipelines
Processing a folder of documents in parallel:
import asyncio
from pathlib import Path
from haystack import AsyncPipeline
async def process_application_folder(folder: str, lekha_api_key: str):
"""Process all documents in a loan application folder concurrently."""
pipeline = AsyncPipeline()
pipeline.add_component("extractor", LekhaExtractor(api_key=lekha_api_key))
docs = list(Path(folder).glob("*.pdf"))
async def process_one(doc_path: Path):
with open(doc_path, "rb") as f:
file_bytes = f.read()
result = await pipeline.run_async({
"extractor": {
"file": ByteStream(data=file_bytes, mime_type="application/pdf"),
"filename": doc_path.name,
}
})
return {
"file": doc_path.name,
"type": result["extractor"]["document_type"],
"data": result["extractor"]["structured_data"],
"confidence": result["extractor"]["confidence"],
}
results = await asyncio.gather(*[process_one(d) for d in docs])
return results
Run for a batch
results = asyncio.run(
process_application_folder("./applications/user_12345/", os.environ["LEKHA_API_KEY"])
)
for r in results:
print(f"{r['file']}: {r['type']} (confidence: {r['confidence']:.0%})")
What Lekha Extracts
When you send a bank statement, you get back a structured payload like:
{
"document_type": "bank_statement",
"bank_name": "HDFC Bank",
"account_number": "XXXX1234",
"period": { "from": "2026-04-01", "to": "2026-09-30" },
"opening_balance": 45230.5,
"closing_balance": 67890.25,
"transactions": [
{
"date": "2026-04-03",
"description": "NEFT-EMPLOYER PAYROLL",
"debit": null,
"credit": 85000.0,
"balance": 130230.5,
"category": "salary"
},
{
"date": "2026-04-05",
"description": "EMI HDFC HOME LOAN",
"debit": 32450.0,
"credit": null,
"balance": 97780.5,
"category": "emi"
}
],
"summary": {
"total_credits": 510000.0,
"total_debits": 487340.25,
"average_monthly_balance": 58234.67,
"emi_obligations": 32450.0
}
}
All amounts are numbers (not strings), all dates are ISO 8601, and categories are pre-tagged — ready to feed directly into your underwriting logic. Try it live at lekhadev.com/playground.
Connecting to Haystack Tracing
Haystack 2.x includes built-in OpenTelemetry tracing. Add the Lekha span metadata to make extraction quality visible in your dashboards:
from haystack.tracing import tracer
@component
class LekhaExtractorWithTracing(LekhaExtractor):
@component.output_types(
document_type=str,
structured_data=dict,
confidence=float,
error=Optional[str],
)
def run(self, file: ByteStream, filename: str = "document.pdf"):
with tracer.trace("lekha_extraction") as span:
span.set_content_tag("lekha.filename", filename)
result = super().run(file, filename)
span.set_content_tag("lekha.document_type", result["document_type"])
span.set_content_tag("lekha.confidence", result["confidence"])
return result
This integrates cleanly with Langfuse, Jaeger, or any OpenTelemetry-compatible backend — giving you per-document extraction latency, type distribution, and confidence trends over time.
Production Checklist
Before you ship a Haystack + Lekha pipeline to production:
tenacity retries with exponential backoff.requests timeout accordingly and handle gracefully.FAQ
Does Lekha work with all Indian bank formats? Yes. Lekha supports 30+ Indian banks including SBI, HDFC, ICICI, Axis, Kotak, PNB, Canara, Union Bank, Bank of Baroda, and more. The API auto-detects the bank format from the document. See the full list at lekhadev.com/docs. Can I use Haystack with Lekha without an LLM? Absolutely. TheLekhaExtractor component works standalone — you don't need an LLM if you just want structured JSON from your documents. The LLM step in the examples above is optional post-processing.
How does Lekha handle multi-page statements?
Lekha processes all pages automatically and returns a unified transaction list with totals. You don't need to split or pre-process multi-page PDFs.
Is Lekha DPDP compliant?
Lekha processes documents in memory and does not store them on disk. Extracted structured data (amounts, dates, categories) does not contain Personally Identifiable Information by default. See our compliance docs for details on data handling.
Ready to add Lekha to your Haystack pipeline? Sign up at lekhadev.com and get your API key in under a minute. The free tier includes 50 extractions/month — enough to build and validate your pipeline before scaling up.