How to Build a claims processing Agent Using LangGraph in Python for pension funds

A claims processing agent for pension funds takes a member’s claim, extracts the required details, checks policy and eligibility rules, routes missing information back to the claimant, and prepares a decision package for human review or downstream systems. It matters because pension operations are document-heavy, compliance-sensitive, and slow when handled manually; the agent reduces turnaround time without removing auditability or control.

Architecture

• •

  Ingress layer

  • •Accepts claim forms, scanned documents, emails, or portal submissions.
  • •Normalizes everything into a single ClaimState object.
• •

  Document extraction node

  • •Pulls structured fields from PDFs or OCR text.
  • •Extracts member ID, plan type, benefit type, dates, employer history, and supporting evidence.
• •

  Rules and eligibility node

  • •Checks pension-specific rules:
    • •vesting status
    • •retirement age
    • •disability criteria
    • •beneficiary validation
    • •jurisdiction-specific requirements
• •

  Compliance and risk node

  • •Flags missing consent, suspicious changes in bank details, mismatched identities, or residency issues.
  • •Produces an audit trail for every decision.
• •

  Human review node

  • •Sends edge cases to an operations analyst.
  • •Keeps the final decision with a human when confidence is low or policy requires it.
• •

  Persistence and audit layer

  • •Stores state transitions, extracted evidence, rule outputs, and reviewer actions.
  • •Supports regulator requests and internal audits.

Implementation

1. Define the state model

Use TypedDict so every node reads and writes a predictable shape. Keep raw inputs separate from derived fields so you can trace where each value came from.

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

class ClaimState(TypedDict):
    claim_id: str
    raw_text: str
    member_id: str
    claim_type: str
    extracted_facts: dict
    eligibility_status: Literal["pending", "eligible", "ineligible", "needs_review"]
    compliance_flags: list[str]
    decision: str
    audit_log: Annotated[list[str], operator.add]

2. Build the core nodes

Each node should do one job. In production, replace the placeholder logic with OCR output, database lookups, and policy services.

def extract_claim_data(state: ClaimState) -> dict:
    text = state["raw_text"].lower()

    facts = {
        "has_id": "member id" in text,
        "has_bank_details": "bank" in text,
        "has_medical_docs": "medical" in text,
        "retirement_claim": "retirement" in text,
    }

    return {
        "member_id": state.get("member_id", ""),
        "claim_type": "retirement" if facts["retirement_claim"] else "unknown",
        "extracted_facts": facts,
        "audit_log": [f"Extracted facts for {state['claim_id']}"],
    }

def check_eligibility(state: ClaimState) -> dict:
    facts = state["extracted_facts"]
    if not facts.get("has_id"):
        return {
            "eligibility_status": "needs_review",
            "decision": "missing_member_id",
            "audit_log": ["Eligibility blocked: member ID missing"],
        }

    if state["claim_type"] == "retirement":
        return {
            "eligibility_status": "eligible",
            "decision": "pre_approved_for_review",
            "audit_log": ["Retirement claim passed basic eligibility checks"],
        }

    return {
        "eligibility_status": "ineligible",
        "decision": "unsupported_claim_type",
        "audit_log": ["Claim type not recognized"],
    }

def compliance_check(state: ClaimState) -> dict:
    flags = []
    facts = state["extracted_facts"]

    if not facts.get("has_bank_details"):
        flags.append("missing_bank_details")

    if state["eligibility_status"] == "eligible" and not facts.get("has_medical_docs"):
        flags.append("documentation_gap")

    status = state["eligibility_status"]
    if flags:
        status = "needs_review"

    return {
        "compliance_flags": flags,
        "eligibility_status": status,
        "audit_log": [f"Compliance flags: {flags}" if flags else "Compliance clear"],
    }

def route_decision(state: ClaimState) -> str:
    if state["eligibility_status"] == "eligible" and not state["compliance_flags"]:
        return END
    return END

3. Wire the graph with StateGraph

This is the actual LangGraph pattern: create a graph from a typed state, add nodes, connect edges, compile it, then invoke it with an initial state.

graph = StateGraph(ClaimState)

graph.add_node("extract_claim_data", extract_claim_data)
graph.add_node("check_eligibility", check_eligibility)
graph.add_node("compliance_check", compliance_check)

graph.add_edge(START, "extract_claim_data")
graph.add_edge("extract_claim_data", "check_eligibility")
graph.add_edge("check_eligibility", "compliance_check")
graph.add_conditional_edges(
    "compliance_check",
    route_decision,
)

app = graph.compile()

result = app.invoke({
    "claim_id": "CLM-10001",
    "raw_text": (
        "Member ID provided. Retirement claim submitted with bank details "
        "and supporting medical documents."
    ),
    "member_id": "",
    "claim_type": "",
    "extracted_facts": {},
    "eligibility_status": "pending",
    # Annotated[list[str], operator.add] means LangGraph can merge lists across nodes
     # but you still need an initial list here.
     # Keep this field non-empty-safe in production code paths.
     #
     # Audit logs should be immutable downstream once persisted.
     #
     # This example keeps it simple.
     #
     # The graph will append entries returned by nodes.
     #
     # Don't store secrets here.
    
})
print(result)

4. Add a human-review branch for real operations

Pension claims often need manual approval when evidence is incomplete or jurisdictional rules apply. Use conditional routing to send those cases to a review queue instead of auto-closing them.

def needs_human_review(state: ClaimState) -> bool:
    return (
        state["eligibility_status"] == "needs_review"
        or len(state["compliance_flags"]) > 0
        or state["decision"] == ""
    )

In practice you would route to a human_review node that writes to your case management system and waits for analyst input before continuing.

Production Considerations

• •

  Auditability first

  • •Persist every node input/output with timestamps and versioned policy references.
  • •Regulators will ask why a claim was approved or delayed; your logs need to answer that without reconstructing guesses from prompts.
• •

  Data residency

  • •Keep member data inside approved regions.
  • •If you use external model APIs, isolate them behind a residency-aware gateway or use self-hosted models for sensitive jurisdictions.
• •

  Guardrails for pension rules

  • •Hard-code non-negotiable checks outside the LLM path:
    • •identity verification
    • •beneficiary consent
    • •retirement age thresholds
    • •payment destination validation
  • •The model can summarize evidence; it should not override statutory rules.
• •

  Operational monitoring

  • •Track:
    • •percentage of claims auto-resolved
    • •human-review rate
    • •missing-document frequency
    • •time-to-decision by claim type
  • •Alert on spikes in bank-detail changes or repeated submissions from the same identity cluster.

Common Pitfalls

• •

  Letting the LLM make final eligibility decisions

  • •Don’t do this for pension funds.
  • •Use deterministic rule services for eligibility and reserve the model for extraction and summarization.
• •

  Mixing raw documents with derived decisions in one blob

  • •This kills traceability.
  • •Keep raw_text, extracted facts, compliance flags, and final decisions as separate fields in state.
• •

  Skipping human review on ambiguous claims

  • •Pension claims often fail because documents are incomplete rather than invalid.
  • •Route ambiguous cases to analysts instead of forcing an automated yes/no outcome.

If you build this as a small LangGraph workflow first and keep policy enforcement outside the model path, you get something pension operations can actually trust: fast enough to reduce backlog, strict enough to survive audit.

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.