How to Build a fraud detection Agent Using LangGraph in Python for pension funds

A fraud detection agent for pension funds reviews member requests, contribution changes, benefit withdrawals, and account activity to flag suspicious patterns before money leaves the system. It matters because pension operations sit under strict compliance, auditability, and data residency requirements, and a missed fraud case can become both a financial loss and a regulatory incident.

Architecture

• •

  Ingress layer

  • •Receives events from member portals, admin back offices, claims systems, and batch files.
  • •Normalizes each event into a common fraud-review schema.
• •

  Policy and compliance router

  • •Checks jurisdiction, fund rules, KYC status, and whether the request can be processed automatically.
  • •Forces human review for restricted cases like early withdrawals or address changes tied to payout events.
• •

  Risk scoring node

  • •Runs deterministic checks first: velocity, device mismatch, bank-account change proximity, unusual withdrawal amount.
  • •Optionally calls an LLM for narrative summarization or case classification, not for final approval.
• •

  Evidence collector

  • •Pulls supporting context: member history, prior claims, contribution patterns, beneficiary changes, and previous alerts.
  • •Produces an audit-friendly evidence bundle for downstream reviewers.
• •

  Decision node

  • •Emits one of three actions: approve, hold_for_review, or escalate.
  • •Stores the reason codes used to reach that decision.
• •

  Audit sink

  • •Persists every state transition, score, and tool call.
  • •Required for regulator review and internal model governance.

Implementation

1) Define the state and nodes

Use a typed state object so every step in the graph is explicit. For pension funds, keep the state small enough to audit but rich enough to explain decisions later.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class FraudState(TypedDict):
    event_id: str
    member_id: str
    event_type: str
    amount: float
    jurisdiction: str
    kyc_status: str
    risk_score: int
    reasons: list[str]
    decision: str
    audit_log: list[str]

def enrich_event(state: FraudState) -> FraudState:
    audit = state.get("audit_log", [])
    audit.append(f"Received {state['event_type']} for member {state['member_id']}")
    return {**state, "audit_log": audit}

def score_risk(state: FraudState) -> FraudState:
    score = 0
    reasons = []

    if state["event_type"] == "benefit_withdrawal" and state["amount"] > 50000:
        score += 40
        reasons.append("large_withdrawal")

    if state["kyc_status"] != "verified":
        score += 30
        reasons.append("kyc_not_verified")

    if state["jurisdiction"] in {"ZA", "EU"} and state["event_type"] == "bank_change":
        score += 20
        reasons.append("sensitive_change_in_regulated_jurisdiction")

    return {
        **state,
        "risk_score": score,
        "reasons": reasons,
        "audit_log": state.get("audit_log", []) + [f"Risk scored at {score}"],
    }

def decide(state: FraudState) -> FraudState:
    score = state["risk_score"]
    if score >= 60:
        decision = "escalate"
    elif score >= 30:
        decision = "hold_for_review"
    else:
        decision = "approve"

    return {
        **state,
        "decision": decision,
        "audit_log": state.get("audit_log", []) + [f"Decision={decision}"],
    }

This is plain LangGraph code using TypedDict plus node functions that return partial state updates. The important part is that every update is deterministic and explainable.

2) Build the graph with conditional routing

Use StateGraph for orchestration and add_conditional_edges when you need different paths based on risk. That gives you explicit control over which cases go straight through and which ones get reviewed.

def route(state: FraudState) -> str:
    if state["decision"] == "escalate":
        return "human_review"
    if state["decision"] == "hold_for_review":
        return "case_queue"
    return END

def send_to_human_review(state: FraudState) -> FraudState:
    return {
        **state,
        "audit_log": state.get("audit_log", []) + ["Sent to human investigator"],
    }

def send_to_case_queue(state: FraudState) -> FraudState:
    return {
        **state,
        "audit_log": state.get("audit_log", []) + ["Queued for secondary review"],
    }

graph = StateGraph(FraudState)
graph.add_node("enrich_event", enrich_event)
graph.add_node("score_risk", score_risk)
graph.add_node("decide", decide)
graph.add_node("human_review", send_to_human_review)
graph.add_node("case_queue", send_to_case_queue)

graph.add_edge(START, "enrich_event")
graph.add_edge("enrich_event", "score_risk")
graph.add_edge("score_risk", "decide")
graph.add_conditional_edges("decide", route)

app = graph.compile()

This pattern keeps policy logic outside the LLM. For pension funds, that matters because approval rules must be stable enough to defend during audits.

3) Run the agent on a real event

At runtime you pass in a single event payload. The output includes the final decision plus an audit trail you can persist in your case management system.

input_state = {
    "event_id": "evt_10021",
    "member_id": "M-88421",
    "event_type": "benefit_withdrawal",
    "amount": 78000.0,
    "jurisdiction": "EU",
    "kyc_status": "pending",
    "risk_score": 0,
    "reasons": [],
    "decision": "",
    "audit_log": [],
}

result = app.invoke(input_state)

print(result["decision"])
print(result["risk_score"])
print(result["reasons"])
print("\n".join(result["audit_log"]))

If you want streaming visibility in production workflows, use app.stream(...) instead of invoke(...). That lets your ops team see intermediate states without waiting for the full graph to finish.

4) Add an LLM only where it helps

For pension fraud detection, the LLM should summarize evidence or classify notes from investigators. Do not let it make final payout decisions.

A clean pattern is to add a node that calls a model after deterministic scoring has already happened. Keep prompts short, structured, and logged with versioning so compliance can reproduce results later.

Production Considerations

• •

  Deployment isolation

  • •Run the graph inside your regulated environment or private cloud region.
  • •Keep member PII inside approved data residency boundaries; do not ship raw records to external model APIs unless your legal team has signed off.
• •

  Monitoring

  • •Track false positives by event type: withdrawals, beneficiary updates, bank detail changes.
  • •Log node-level latency and decision distributions so you can catch drift when fraud patterns change.
• •

  Guardrails

  • •Hard-block auto-approval for high-risk pension events like first-time lump-sum withdrawals or bank-account changes before payout.
  • •Require human sign-off when KYC is incomplete or when jurisdiction-specific rules trigger enhanced due diligence.
• •

  Auditability

  • •Persist the full graph trace with reason codes.
  • •Store model version, prompt version, rule version, and timestamp for each decision so investigations can reconstruct the exact path taken.

Common Pitfalls

• •

  Using the LLM as the final decider

  • •This is the fastest way to create an un-auditable workflow.
  • •Keep deterministic rules in control flow and use the model only for summarization or triage support.
• •

  Ignoring pension-specific policy gates

  • •A generic fraud model will miss things like early-access restrictions, retirement age rules, or beneficiary change sensitivity.
  • •Encode those as explicit checks in graph nodes before any automated action is taken.
• •

  Skipping replayable audit logs

  • •If you only store the final label, you cannot defend decisions during regulator review.
  • •Save intermediate scores, routed branches, node outputs, and version metadata for every case.

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.