How to Build a fraud detection Agent Using LangGraph in Python for wealth management

A fraud detection agent for wealth management watches client activity, flags suspicious patterns, and decides what to do next: block, step up verification, or route to a human reviewer. That matters because the cost of a missed event is not just financial loss; it also includes compliance exposure, reputational damage, and broken trust with high-net-worth clients.

Architecture

• •

  Event intake layer

  • •Receives portfolio trades, wire requests, beneficiary changes, login events, and profile updates.
  • •Normalizes them into a single schema before they hit the graph.
• •

  Risk scoring node

  • •Applies deterministic rules first: unusual transfer size, new device, offshore destination, rapid beneficiary change.
  • •Produces a structured risk score and reason codes.
• •

  Policy and compliance node

  • •Checks against wealth-management controls: KYC status, AML thresholds, jurisdiction restrictions, sanctions screening.
  • •Decides whether the case can auto-clear or must be escalated.
• •

  Investigation node

  • •Pulls account history, recent behavior, advisor notes, and prior alerts.
  • •Summarizes evidence for an analyst without exposing unnecessary PII.
• •

  Decision node

  • •Chooses one of three actions: allow, step_up, or escalate.
  • •Writes an audit-friendly decision record.
• •

  Audit sink

  • •Persists inputs, outputs, timestamps, model version, rule version, and final action.
  • •Required for internal review and regulator-facing traceability.

Implementation

1) Define the state and decision model

Use a typed state so every node reads and writes predictable fields. For wealth management workflows, keep the state small and explicit; don’t pass raw client records around if you only need risk signals and a few identifiers.

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

Action = Literal["allow", "step_up", "escalate"]

class FraudState(TypedDict):
    client_id: str
    event_type: str
    amount: float
    jurisdiction: str
    kyc_status: str
    risk_score: int
    reasons: List[str]
    action: Action
    audit_log: List[str]

2) Build the nodes

Keep rule logic deterministic. In regulated environments, this makes it easier to explain why a transfer was stopped or why a review was triggered.

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

    if state["amount"] > 100000:
        score += 40
        reasons.append("large_transfer")
    if state["jurisdiction"] in {"high_risk_country", "sanctioned_region"}:
        score += 50
        reasons.append("jurisdiction_risk")
    if state["event_type"] == "beneficiary_change":
        score += 25
        reasons.append("beneficiary_change")
    if state["kyc_status"] != "verified":
        score += 30
        reasons.append("kyc_not_verified")

    state["risk_score"] = min(score, 100)
    state["reasons"] = reasons
    return state

def decide_action(state: FraudState) -> FraudState:
    if state["risk_score"] >= 80:
        state["action"] = "escalate"
    elif state["risk_score"] >= 40:
        state["action"] = "step_up"
    else:
        state["action"] = "allow"

    state["audit_log"].append(
        f"decision={state['action']} risk_score={state['risk_score']} reasons={','.join(state['reasons'])}"
    )
    return state

def audit_event(state: FraudState) -> FraudState:
    # Replace with DB write / SIEM / immutable log sink in production.
    print({"client_id": state["client_id"], "audit_log": state["audit_log"]})
    return state

3) Wire the graph with LangGraph

This is the core pattern. StateGraph gives you a clear control flow for scoring, deciding, and auditing. You can add conditional routing later if you want different paths for wire transfers versus profile changes.

workflow = StateGraph(FraudState)

workflow.add_node("score_risk", score_risk)
workflow.add_node("decide_action", decide_action)
workflow.add_node("audit_event", audit_event)

workflow.add_edge(START, "score_risk")
workflow.add_edge("score_risk", "decide_action")
workflow.add_edge("decide_action", "audit_event")
workflow.add_edge("audit_event", END)

app = workflow.compile()

initial_state: FraudState = {
    "client_id": "C12345",
    "event_type": "wire_transfer",
    "amount": 250000.0,
    "jurisdiction": "high_risk_country",
    "kyc_status": "verified",
    "risk_score": 0,
    "reasons": [],
    "action": "allow",
    "audit_log": []
}

result = app.invoke(initial_state)
print(result["action"])

4) Add conditional escalation for analyst review

In production you usually want branching logic. LangGraph supports this with add_conditional_edges, which is cleaner than stuffing every decision into one function.

def route_by_risk(state: FraudState) -> Action:
    return state["action"]

workflow2 = StateGraph(FraudState)
workflow2.add_node("score_risk", score_risk)
workflow2.add_node("decide_action", decide_action)
workflow2.add_node("audit_event", audit_event)

workflow2.add_edge(START, "score_risk")
workflow2.add_edge("score_risk", "decide_action")

workflow2.add_conditional_edges(
    "decide_action",
    route_by_risk,
    {
        "allow": END,
        "step_up": END,
        "escalate": "audit_event",
    },
)

workflow2.add_edge("audit_event", END)
app2 = workflow2.compile()

For wealth management teams that need human oversight on high-value movements or account changes, route "escalate" to an analyst queue instead of auto-blocking everything. That reduces false positives while preserving control.

Production Considerations

• •

  Auditability

  • •Store every input signal, rule hit, final action, and graph version.
  • •Keep immutable logs for regulator review and internal model governance.
• •

  Data residency

  • •Keep client data in-region if your firm has jurisdictional constraints.
  • •If you use external model calls later in the graph, ensure the provider supports your residency requirements or redact sensitive fields before sending them out.
• •

  Compliance guardrails

  • •Separate fraud detection from final enforcement when policy requires human approval.
  • •Encode thresholds for AML/KYC exceptions explicitly so analysts can see why a case escalated.
• •

  Monitoring

  • •Track false positive rate by event type: wire transfers often behave differently from login anomalies.
  • •Alert on drift in risk scores after product launches, market volatility spikes, or advisor workflow changes.

Common Pitfalls

1. •

   Using an LLM for deterministic fraud rules

   • •Don’t ask a model to decide whether a $500k wire to an offshore account is suspicious when a rule can do it better.
   • •Use deterministic logic first; reserve LLMs for summarization or analyst notes.
2. •

   Leaking too much PII through the graph

   • •Wealth management data includes tax IDs, account balances, beneficiary details, and advisor notes.
   • •Pass only the fields each node needs; redact before any external call or logging sink.
3. •

   Skipping explainability

   • •If your output is just "escalate", operations teams will hate it.
   • •Always attach reason codes like large_transfer, kyc_not_verified, or jurisdiction_risk so compliance can review decisions quickly.
4. •

   Treating all high-risk events as auto-blocks

   • •That creates unnecessary friction for legitimate clients moving capital between managed accounts.
   • •Use step-up authentication or analyst review for borderline cases instead of hard denial every time.

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.