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

A claims processing agent in investment banking takes a claim, classifies it, pulls the right internal records, checks policy and compliance rules, and routes the case to approval, rejection, or human review. It matters because claims workflows sit at the intersection of client trust, regulatory exposure, and operational cost; a bad decision can create audit findings, delayed settlements, or control breaches.

Architecture

• •

  Ingress layer

  • •Accepts claim payloads from case management systems, email ingestion, or API gateways.
  • •Normalizes fields like claimant ID, trade reference, product type, jurisdiction, and timestamps.
• •

  Classification node

  • •Determines claim type: failed settlement, fee dispute, corporate action error, misbooking, or documentation issue.
  • •Drives downstream routing and determines which controls apply.
• •

  Evidence retrieval node

  • •Pulls data from internal systems: OMS/EMS logs, trade blotter, confirmations, client agreements, and prior cases.
  • •Must be read-only and fully auditable.
• •

  Policy and compliance check node

  • •Applies business rules and regulatory constraints.
  • •Checks for data residency restrictions, KYC/AML flags, sanctions exposure, and escalation thresholds.
• •

  Decision node

  • •Produces one of three outcomes: approve, reject with reason codes, or escalate to human review.
  • •Emits structured output for downstream case management.
• •

  Audit trail store

  • •Persists every state transition, tool call, retrieved record ID, and final rationale.
  • •Required for model risk review and internal audit.

Implementation

1. Define the state and decision schema

Use a typed state object so every node reads and writes predictable fields. In banking workflows, this prevents loose dicts from turning into untraceable control failures.

from typing import TypedDict, Annotated
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END
from operator import add

class ClaimDecision(BaseModel):
    outcome: str  # approve | reject | escalate
    reason_code: str
    summary: str

class ClaimState(TypedDict):
    claim_id: str
    claimant_id: str
    product_type: str
    jurisdiction: str
    details: str
    classification: str
    evidence_ids: list[str]
    compliance_flags: list[str]
    decision: dict

def merge_lists(left: list[str], right: list[str]) -> list[str]:
    return left + right

class GraphState(TypedDict):
    claim_id: str

The important part is not the model choice. It is making the workflow deterministic enough that audit can replay it later.

2. Build the core nodes

Each node should do one job. Keep LLM usage constrained to classification and summarization; use code for policy checks.

from langgraph.graph import StateGraph

def classify_claim(state: ClaimState) -> dict:
    text = state["details"].lower()
    if "failed settlement" in text:
        cls = "settlement_failure"
    elif "fee" in text or "charge" in text:
        cls = "fee_dispute"
    else:
        cls = "other"
    return {"classification": cls}

def retrieve_evidence(state: ClaimState) -> dict:
    # Replace with real read-only integrations to OMS / case store / document vault.
    evidence_map = {
        "settlement_failure": ["trade_blotter_2024_11_12", "confirm_77821"],
        "fee_dispute": ["fee_schedule_q3", "client_agreement_991"],
        "other": ["case_notes_001"]
    }
    return {"evidence_ids": evidence_map.get(state["classification"], [])}

def compliance_check(state: ClaimState) -> dict:
    flags = []
    if state["jurisdiction"] in {"US", "UK"}:
        flags.append("residency_review_required")
    if state["product_type"] == "structured_product":
        flags.append("product_risk_escalation")
    return {"compliance_flags": flags}

def decide(state: ClaimState) -> dict:
    if "product_risk_escalation" in state["compliance_flags"]:
        result = ClaimDecision(
            outcome="escalate",
            reason_code="CONTROL_102",
            summary="Structured product claims require manual review."
        )
    elif state["classification"] == "fee_dispute":
        result = ClaimDecision(
            outcome="reject",
            reason_code="POLICY_014",
            summary="Fee disputes require contract-level validation."
        )
    else:
        result = ClaimDecision(
            outcome="approve",
            reason_code="AUTO_001",
            summary="Claim meets automated processing criteria."
        )
    return {"decision": result.model_dump()}

3. Wire the graph with LangGraph’s actual API

StateGraph, add_node, add_edge, add_conditional_edges, compile are the core primitives you want here. This is the pattern that scales when you need more branches later.

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(ClaimState)

workflow.add_node("classify_claim", classify_claim)
workflow.add_node("retrieve_evidence", retrieve_evidence)
workflow.add_node("compliance_check", compliance_check)
workflow.add_node("decide", decide)

workflow.add_edge(START, "classify_claim")
workflow.add_edge("classify_claim", "retrieve_evidence")
workflow.add_edge("retrieve_evidence", "compliance_check")
workflow.add_edge("compliance_check", "decide")
workflow.add_edge("decide", END)

app = workflow.compile()

result = app.invoke({
    "claim_id": "CLM-10021",
    "claimant_id": "CUST-8841",
    "product_type": "equity_swap",
    "jurisdiction": "UK",
    "details": "Client reports failed settlement on equity swap leg.",
    "classification": "",
    "evidence_ids": [],
    "compliance_flags": [],
    "decision": {}
})

print(result["decision"])

If you need escalation branching later, swap the linear edge after compliance_check for add_conditional_edges. That keeps the graph readable instead of burying routing logic inside a single monolithic function.

4. Add human review for risky cases

Investment banking claims cannot be fully automated when controls are unclear. Use a conditional branch to route flagged cases into an analyst queue.

def route_after_compliance(state: ClaimState) -> str:
    if state["compliance_flags"]:
        return "human_review"
    return "decide"

def human_review(state: ClaimState) -> dict:
    return {
        "decision": {
            "outcome": "escalate",
            "reason_code": ",".join(state["compliance_flags"]),
            "summary": f"Manual review required for {state['claim_id']}."
        }
    }

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

workflow.add_edge(START, "classify_claim")
workflow.add_edge("classify_claim", "retrieve_evidence")
workflow.add_edge("retrieve_evidence", "compliance_check")
workflow.add_conditional_edges(
    "compliance_check",
    route_after_compliance,
    {"human_review": "human_review", "decide": "decide"}
)
workflow.add_edge("human_review", END)
workflow.add_edge("decide", END)

app = workflow.compile()

Production Considerations

• •

  Auditability

  • •Persist input state, node outputs, tool calls, and final decision payloads.
  • •Store immutable logs with correlation IDs so compliance can reconstruct every claim path.
• •

  Data residency

  • •Keep EU client data in-region if your bank has residency obligations.
  • •Do not send sensitive claim content to external model endpoints unless legal approves that transfer path.
• •

  Guardrails

  • •Enforce allowlisted tools only; no free-form system access from the agent.
  • •Validate all outputs against schemas before writing back to case systems.
  • •Require human approval for high-risk products like structured notes or cross-border disputes.
• •

  Monitoring

  • •Track auto-approval rate by product type and jurisdiction.
  • •Alert on spikes in escalations, repeated rejections by reason code, or missing evidence retrievals.
  • •Feed model drift checks into your control dashboard if you use an LLM for classification.

Common Pitfalls

1. •

   Using the LLM as the decision engine

   • •Don’t let the model decide payout eligibility directly.
   • •Use deterministic policy code for approvals and reserve LLMs for classification or summarization.
2. •

   Skipping structured state

   • •Passing raw strings between nodes makes debugging painful and breaks auditability.
   • •Use typed state with explicit fields so every transition is inspectable.
3. •

   Ignoring jurisdiction-specific controls

   • •A claim that is valid in one region may violate another region’s handling rules.
   • •Encode residency checks, retention policies, and escalation thresholds per jurisdiction before go-live.

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.