How to Build a claims processing Agent Using LangGraph in Python for retail banking

By Cyprian AaronsUpdated 2026-04-21
claims-processinglanggraphpythonretail-banking

A claims processing agent in retail banking takes an incoming customer claim, classifies it, pulls the right account and transaction context, checks policy and regulatory rules, and routes the case to auto-approval, request-for-more-info, or human review. It matters because claims are high-volume, time-sensitive, and audit-sensitive: if you get the workflow wrong, you create customer friction, compliance risk, and a messy paper trail.

Architecture

A production claims agent for retail banking usually needs these components:

  • Ingress layer

    • Accepts claim payloads from API, CRM, or back-office queues.
    • Normalizes fields like customer ID, product type, claim amount, channel, and evidence links.
  • State model

    • Stores the working case context across steps.
    • Includes extracted facts, rule decisions, confidence scores, and audit metadata.
  • Decision graph

    • Uses LangGraph to route between extraction, validation, policy checks, enrichment, and final disposition.
    • Keeps branching logic explicit instead of hiding it inside one big LLM prompt.
  • Policy and compliance tools

    • Checks KYC status, account ownership, dispute windows, AML flags, and product-specific claim rules.
    • Enforces explainability by storing why a path was chosen.
  • Human review handoff

    • Escalates suspicious or incomplete claims to an analyst queue.
    • Preserves all intermediate state so a reviewer can see exactly what happened.
  • Audit storage

    • Writes immutable logs for every node execution.
    • Supports retention requirements and data residency constraints.

Implementation

  1. Define the state and the graph nodes

Use TypedDict for the graph state so every node reads and writes predictable fields. In retail banking, keep PII minimal in state; store references to records instead of copying full documents around.

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class ClaimState(TypedDict):
    claim_id: str
    customer_id: str
    product_type: str
    amount: float
    evidence_count: int
    kyc_passed: bool
    policy_passed: bool
    needs_human_review: bool
    decision: Literal["approve", "reject", "review"]
    reason: str

def classify_claim(state: ClaimState) -> ClaimState:
    # Replace with real classifier / LLM extraction
    return {
        **state,
        "evidence_count": state.get("evidence_count", 0),
        "needs_human_review": False,
        "reason": "Claim classified and normalized"
    }

def check_policy(state: ClaimState) -> ClaimState:
    kyc_ok = state["kyc_passed"]
    policy_ok = state["amount"] <= 5000.0 and state["product_type"] in {"debit_card", "cashback"}
    return {
        **state,
        "policy_passed": policy_ok,
        "needs_human_review": not (kyc_ok and policy_ok),
        "reason": "Policy check complete"
    }

def decide(state: ClaimState) -> ClaimState:
    if not state["kyc_passed"]:
        return {**state, "decision": "review", "reason": "KYC failed"}
    if not state["policy_passed"]:
        return {**state, "decision": "review", "reason": "Policy failed"}
    return {**state, "decision": "approve", "reason": "Auto-approved"}

graph = StateGraph(ClaimState)
graph.add_node("classify_claim", classify_claim)
graph.add_node("check_policy", check_policy)
graph.add_node("decide", decide)

graph.add_edge(START, "classify_claim")
graph.add_edge("classify_claim", "check_policy")
graph.add_edge("check_policy", "decide")
graph.add_edge("decide", END)

app = graph.compile()
  1. Add conditional routing for exceptions

Real claims workflows need branching. LangGraph’s add_conditional_edges is the right tool when the next step depends on current state. For example, incomplete claims should request more information while suspicious ones go straight to review.

from langgraph.graph import StateGraph

def route_after_policy(state: ClaimState) -> str:
    if not state["kyc_passed"]:
        return "human_review"
    if not state["policy_passed"]:
        return "human_review"
    return "decide"

def human_review(state: ClaimState) -> ClaimState:
    return {
        **state,
        "decision": "review",
        "reason": f"Sent to analyst queue for claim {state['claim_id']}"
    }

workflow = StateGraph(ClaimState)
workflow.add_node("classify_claim", classify_claim)
workflow.add_node("check_policy", check_policy)
workflow.add_node("human_review", human_review)
workflow.add_node("decide", decide)

workflow.add_edge(START, "classify_claim")
workflow.add_edge("classify_claim", "check_policy")
workflow.add_conditional_edges(
    "check_policy",
    route_after_policy,
    {
        "human_review": "human_review",
        "decide": "decide",
    },
)
workflow.add_edge("human_review", END)
workflow.add_edge("decide", END)

claims_app = workflow.compile()
  1. Invoke the graph with a real claim payload

At runtime you pass a plain dictionary matching the state schema. In production you would enrich this from your core banking system before invoking the graph.

input_state = {
    "claim_id": "CLM-100245",
    "customer_id": "CUST-7781",
    "product_type": "debit_card",
    "amount": 250.00,
    "evidence_count": 2,
    "kyc_passed": True,
}

result = claims_app.invoke(input_state)
print(result["decision"])
print(result["reason"])
  1. Persist checkpoints for auditability

For banking workflows you want resumability and traceability. LangGraph supports checkpointing through its compile-time configuration; wire it to your persistence layer so an analyst can resume a partially processed case without losing prior decisions.

Typical pattern:

  • Use a checkpointer backed by Postgres or another durable store.
  • Store thread_id per claim so each case has a stable execution history.
  • Keep node-level inputs/outputs for audit review.

Production Considerations

  • Compliance logging

    • Log every node transition with timestamp, claim ID, operator ID if human review occurred, and decision reason.
    • Make sure logs are tamper-evident and retained according to your bank’s policy.
  • Data residency

    • Keep customer data in-region.
    • If your LangGraph app calls external models or tools, ensure those services are approved for the same jurisdiction as the claim data.
  • Guardrails on automation

    • Hard-code thresholds for auto-approval; do not let an LLM override policy limits.
    • Separate extraction from decisioning so model hallucinations cannot directly approve payouts.
  • Monitoring

    • Track approval rate, manual review rate, false positive escalations, average handling time, and rollback frequency.
    • Alert when one product line suddenly shifts behavior; that usually means upstream data quality or rule drift.

Common Pitfalls

  1. Putting business rules inside one prompt

    • This makes audits painful and failures opaque.
    • Keep deterministic checks in Python nodes and use models only where they add value.
  2. Over-sharing PII across nodes

    • Developers often pass full statements or identity documents through every step.
    • Pass references or minimal fields only; fetch sensitive records only inside trusted tools.
  3. Skipping human-review paths

    • Retail banking claims always have edge cases: chargebacks outside window limits, disputed card-not-present transactions, duplicate submissions.
    • Build explicit escalation paths with add_conditional_edges, not ad hoc exception handling.
  4. No replayable audit trail

    • If you cannot reconstruct why a claim was approved six months later, you will fail internal controls reviews.
    • Persist graph state transitions and final outputs with immutable storage from day one.

Keep learning

By Cyprian Aarons, AI Consultant at Topiax.

Want the complete 8-step roadmap?

Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.

Get the Starter Kit

Related Guides