LangGraph is the go-to framework for building stateful, graph-based AI agents that need to loop, branch, and retry based on real-world conditions. If you are building a loan underwriting bot, a KYC pipeline, or a credit assessment agent for Indian users, you almost certainly need to parse financial documents along the way.

This guide shows you exactly how to connect LangGraph's StateGraph to Lekha's financial document API so your agents can read bank statements, salary slips, and ITR documents natively — and act on the extracted data in subsequent nodes.

Why LangGraph + Lekha?

LangGraph models agent logic as a directed graph where each node is a function and edges encode conditional routing. Lekha turns unstructured Indian financial PDFs into structured JSON your agent can reason over.

Together they unlock workflows that were previously painful to build:

Conditional routing: parse a document → if income is below threshold, route to rejection node; otherwise route to approval node

Retry loops: if extraction confidence is low, ask the user to re-upload a cleaner scan

Parallel fan-out: fetch salary slip, bank statement, and CIBIL report simultaneously in parallel nodes

Human-in-the-loop: pause the graph when a document fails validation and wait for a human reviewer

Prerequisites

pip install langgraph langchain-anthropic httpx python-dotenv

You'll also need:

A Lekha API key from lekhadev.com (free tier available)

An Anthropic API key for the LLM nodes

export LEKHA_API_KEY=lk_live_...
export ANTHROPIC_API_KEY=sk-ant-...

Step 1: Define the Agent State

LangGraph passes a shared TypedDict state through every node. Define it to hold your document payloads and derived fields.

from typing import TypedDict, Optional, Literal
from langgraph.graph import StateGraph, END
class LoanState(TypedDict):
    # Inputs
    bank_statement_pdf: bytes
    salary_slip_pdf: bytes
    applicant_name: str
# Extracted
    bank_data: Optional[dict]
    salary_data: Optional[dict]
# Derived
    monthly_income: Optional[float]
    average_balance: Optional[float]
    foir: Optional[float]
# Decision
    decision: Optional[Literal["approved", "rejected", "review"]]
    reason: Optional[str]

Step 2: Build the Lekha Extraction Nodes

Each node is a plain Python function that receives the state dict, performs work, and returns a partial update.

import httpx
import base64
import os
LEKHA_BASE = "https://api.lekhadev.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['LEKHA_API_KEY']}"}
def encode_pdf(pdf_bytes: bytes) -> str:
    return base64.b64encode(pdf_bytes).decode()
async def extract_bank_statement(state: LoanState) -> dict:
    """Node: send bank statement to Lekha, store structured JSON."""
    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(
            f"{LEKHA_BASE}/extract",
            headers=HEADERS,
            json={
                "document": encode_pdf(state["bank_statement_pdf"]),
                "document_type": "bank_statement",
            },
        )
        resp.raise_for_status()
        data = resp.json()["data"]
# Derive average balance from the last 3 months of transactions
    transactions = data.get("transactions", [])
    balances = [t["balance"] for t in transactions if t.get("balance") is not None]
    avg_balance = sum(balances[-90:]) / len(balances[-90:]) if balances else 0.0
return {
        "bank_data": data,
        "average_balance": round(avg_balance, 2),
    }
async def extract_salary_slip(state: LoanState) -> dict:
    """Node: send salary slip to Lekha, store structured JSON."""
    async with httpx.AsyncClient(timeout=60) as client:
        resp = await client.post(
            f"{LEKHA_BASE}/extract",
            headers=HEADERS,
            json={
                "document": encode_pdf(state["salary_slip_pdf"]),
                "document_type": "salary_slip",
            },
        )
        resp.raise_for_status()
        data = resp.json()["data"]
return {
        "salary_data": data,
        "monthly_income": data.get("net_pay", 0.0),
    }

> Tip: Lekha's response for salary_slip includes gross_pay, net_pay, deductions, employer_name, and pay_period. For bank_statement you get account_number, transactions[], opening_balance, and closing_balance. See the full schema at lekhadev.com/docs.

Step 3: Add a Calculation Node

Once both extraction nodes have run, calculate the Fixed Obligation to Income Ratio (FOIR) — the standard Indian lending metric.

def calculate_foir(state: LoanState) -> dict:
    """Node: compute FOIR from extracted data."""
    income = state.get("monthly_income", 0.0) or 0.0
# Sum recurring debits that look like EMI payments
    transactions = (state.get("bank_data") or {}).get("transactions", [])
    suspected_emis = [
        abs(t["amount"])
        for t in transactions
        if t.get("transaction_type") == "debit"
        and any(kw in (t.get("description") or "").upper()
                for kw in ["EMI", "LOAN", "NACH", "ECS"])
    ]
    # Annualise to monthly average (last 90 days → ÷ 3)
    monthly_obligations = sum(suspected_emis) / 3 if suspected_emis else 0.0
foir = (monthly_obligations / income * 100) if income else 100.0
return {"foir": round(foir, 2)}

Step 4: Add a Decision Node

The decision node applies your lending policy rules and produces a final verdict.

def make_decision(state: LoanState) -> dict:
    """Node: apply lending policy and emit a decision."""
    income = state.get("monthly_income", 0.0) or 0.0
    foir = state.get("foir", 100.0) or 100.0
    avg_balance = state.get("average_balance", 0.0) or 0.0
# Example policy thresholds
    MIN_INCOME = 30_000          # ₹30,000/month
    MAX_FOIR = 50                # 50%
    MIN_BALANCE = 10_000         # ₹10,000 avg balance
if income < MIN_INCOME:
        return {"decision": "rejected", "reason": f"Monthly income ₹{income:,.0f} below minimum ₹{MIN_INCOME:,.0f}"}
if foir > MAX_FOIR:
        return {"decision": "rejected", "reason": f"FOIR {foir:.1f}% exceeds maximum {MAX_FOIR}%"}
if avg_balance < MIN_BALANCE:
        return {"decision": "review", "reason": f"Average balance ₹{avg_balance:,.0f} is low — requires manual check"}
return {"decision": "approved", "reason": f"All checks passed. FOIR: {foir:.1f}%, Income: ₹{income:,.0f}/mo"}

Step 5: Wire the Graph

Now connect all nodes into a LangGraph StateGraph. Both extraction nodes run in the same step (parallel fan-out) and the rest follow sequentially.

from langgraph.graph import StateGraph, END
def build_loan_graph() -> StateGraph:
    graph = StateGraph(LoanState)
# Add nodes
    graph.add_node("extract_bank", extract_bank_statement)
    graph.add_node("extract_salary", extract_salary_slip)
    graph.add_node("calculate_foir", calculate_foir)
    graph.add_node("decide", make_decision)
# Fan-out: both extractions start from the entry point
    graph.set_entry_point("extract_bank")
    graph.add_edge("extract_bank", "extract_salary")
    graph.add_edge("extract_salary", "calculate_foir")
    graph.add_edge("calculate_foir", "decide")
# Conditional exit
    graph.add_conditional_edges(
        "decide",
        lambda s: s["decision"],
        {
            "approved": END,
            "rejected": END,
            "review": END,
        },
    )
return graph.compile()

> For true parallel fan-out (both PDFs uploaded at the same time), use LangGraph's send API or split into separate fan_out / fan_in nodes. The sequential version above is simpler to debug.

Step 6: Run the Agent

import asyncio
async def main():
    app = build_loan_graph()
with open("hdfc_statement.pdf", "rb") as f:
        bank_pdf = f.read()
    with open("salary_slip_june.pdf", "rb") as f:
        salary_pdf = f.read()
result = await app.ainvoke({
        "bank_statement_pdf": bank_pdf,
        "salary_slip_pdf": salary_pdf,
        "applicant_name": "Priya Sharma",
        "bank_data": None,
        "salary_data": None,
        "monthly_income": None,
        "average_balance": None,
        "foir": None,
        "decision": None,
        "reason": None,
    })
print(f"Decision: {result['decision'].upper()}")
    print(f"Reason:   {result['reason']}")
    print(f"FOIR:     {result['foir']}%")
    print(f"Income:   ₹{result['monthly_income']:,.0f}/month")
asyncio.run(main())

Sample output:

Decision: APPROVED
Reason:   All checks passed. FOIR: 32.4%, Income: ₹85,000/month
FOIR:     32.4%
Income:   ₹85,000/month

Adding Human-in-the-Loop for "Review" Cases

LangGraph's interrupt mechanism lets you pause the graph for a human decision and resume it later — perfect for the review branch:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt
def human_review_node(state: LoanState) -> dict:
    """Pause and wait for a human reviewer."""
    verdict = interrupt({
        "applicant": state["applicant_name"],
        "foir": state["foir"],
        "avg_balance": state["average_balance"],
        "reason": state["reason"],
    })
    # verdict is whatever the human sends when they resume the graph
    return {"decision": verdict["decision"], "reason": verdict["reason"]}
Rebuild with checkpointing for persistence
memory = MemorySaver()
app = build_loan_graph_with_review(checkpointer=memory)
Resume a paused graph after human input:
app.invoke(Command(resume={"decision": "approved", "reason": "Verified manually"}),
           config={"configurable": {"thread_id": "app-123"}})

Handling Document Errors Gracefully

Lekha returns a consistent error structure when a document can't be parsed. Catch it inside your node and route accordingly:

async def extract_bank_statement_safe(state: LoanState) -> dict:
    try:
        async with httpx.AsyncClient(timeout=60) as client:
            resp = await client.post(
                f"{LEKHA_BASE}/extract",
                headers=HEADERS,
                json={
                    "document": encode_pdf(state["bank_statement_pdf"]),
                    "document_type": "bank_statement",
                },
            )
            if not resp.json().get("success"):
                error = resp.json().get("error", {})
                return {
                    "decision": "rejected",
                    "reason": f"Document error: {error.get('message', 'Unknown')}",
                }
            data = resp.json()["data"]
    except httpx.TimeoutException:
        return {"decision": "review", "reason": "Extraction timed out — retry or manual review"}
# ... rest of extraction
    return {"bank_data": data}

Testing the Graph Locally

LangGraph ships with a built-in Studio UI for visualising your graph and stepping through state changes. Run it locally:

pip install langgraph-cli
langgraph dev

Open http://localhost:8123 to see your node graph, inspect state at each step, and replay runs. Try the Lekha playground alongside it to generate sample extractions for your test fixtures.

Deploying to LangGraph Platform

When you're ready for production, deploy your graph to LangGraph Platform for managed persistence, scaling, and observability:

# langgraph.json
{
  "graphs": {
    "loan_agent": "./loan_agent.py:app"
  },
  "env": ".env"
}
langgraph deploy

Your graph gets a REST API with /runs, /threads, and /assistants endpoints — making it straightforward to integrate with a Next.js frontend or a webhook handler.

---

FAQ

What Indian document types does Lekha support?

Lekha supports bank statements (all major Indian banks including HDFC, SBI, ICICI, Axis, Kotak, and 20+ others), salary slips, ITR documents, Form 16, CIBIL credit reports, CAS mutual fund statements, GST invoices, and balance sheets. Full list at lekhadev.com/docs.

Can I run extraction nodes in parallel in LangGraph?

Yes — use LangGraph's fan_out pattern with Send events. Both Lekha API calls can fire simultaneously to cut end-to-end latency roughly in half. For most loan workflows the sequential pattern is simpler to start with and fast enough.

How does Lekha handle password-protected PDFs?

Pass the password field in the request body: {"document": "...", "document_type": "bank_statement", "password": "PRIYA1990"}. Lekha will decrypt the PDF before extraction. Passwords are never stored.

Is Lekha DPDP-compliant for use in production lending pipelines?

Yes. Lekha processes documents in-memory and never persists document content to disk. It is designed for DPDP (Digital Personal Data Protection Act) compliance. See the DPDP compliance guide for details.

---

Ready to build your own LangGraph financial agent? Sign up for a free Lekha API key at lekhadev.com — no credit card required. Explore the live Lekha playground to test extraction on any Indian financial document in under 30 seconds.