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

Fraud detection in wealth management is not just about spotting stolen cards or obvious account takeovers. It’s about detecting suspicious behavior across transfers, beneficiary changes, portfolio liquidation, advisor activity, and client communication patterns before money moves or a relationship gets damaged.

A good agent sits between your event stream and your case management workflow. It scores risk, explains why something looks off, and routes the right cases to compliance with an audit trail you can defend.

Architecture

Build this agent as a small set of deterministic components wrapped around an LLM only where language interpretation is useful.

• •

  Event ingestion layer

  • •Pulls transactions, wire requests, profile edits, login events, and advisor notes from internal systems.
  • •Normalizes them into a single structured schema.
• •

  Risk rules engine

  • •Handles hard rules first: sanctions hits, new payee + large wire, unusual jurisdiction, rapid liquidation.
  • •Keeps obvious fraud out of the LLM path.
• •

  LangChain analysis chain

  • •Uses ChatPromptTemplate, JsonOutputParser, and a chat model to classify suspicious activity.
  • •Produces structured output: risk level, rationale, recommended action.
• •

  Case enrichment tools

  • •Fetches client history, KYC status, recent contact changes, device info, and prior alerts.
  • •Gives the model context without exposing raw systems directly.
• •

  Decision router

  • •Applies policy thresholds to decide: auto-clear, step-up verification, or escalate to compliance.
  • •Keeps final action deterministic.
• •

  Audit logger

  • •Stores input features, model output, prompt version, policy version, and human decision.
  • •Required for regulatory review and internal model governance.

Implementation

1) Define the fraud event schema and enrichment function

Keep the input structured. Wealth management fraud signals are too sensitive to leave as free text until the very end.

from typing import Literal
from pydantic import BaseModel, Field

class FraudEvent(BaseModel):
    client_id: str
    event_type: Literal["wire", "beneficiary_change", "login", "advisor_note", "portfolio_liquidation"]
    amount_usd: float = Field(ge=0)
    jurisdiction: str
    is_new_payee: bool = False
    kyc_risk_rating: Literal["low", "medium", "high"]
    device_trust: Literal["trusted", "unknown", "compromised"]
    recent_contact_change_days: int
    notes: str = ""

def enrich_event(event: FraudEvent) -> dict:
    # Replace with real DB/API calls
    return {
        "client_tenure_years": 8,
        "prior_alerts_90d": 1,
        "last_verified_contact_days": 120,
        "segment": "private_wealth",
        "event": event.model_dump(),
    }

2) Build a structured LangChain analysis chain

Use ChatPromptTemplate plus JsonOutputParser so the model returns machine-readable output. In production, this is easier to validate than raw prose.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system",
     """You are a fraud analyst for wealth management.
Return only valid JSON with keys:
risk_level (low|medium|high),
reasoning (array of strings),
recommended_action (auto_clear|step_up_verification|escalate_to_compliance),
red_flags (array of strings)."""),
    ("human",
     """Analyze this event for fraud risk in wealth management:

{context}

Apply these rules:
- Prioritize new payee wires over $50k
- Treat compromised devices as high risk
- Escalate if KYC is high risk and contact details changed recently
- Mention audit/compliance implications when relevant""")
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = JsonOutputParser()
chain = prompt | llm | parser

3) Add deterministic policy checks before calling the model

Do not send every event to the LLM. Hard controls should short-circuit obvious cases and reduce cost and variance.

def hard_rules(event: FraudEvent) -> dict | None:
    if event.event_type == "wire" and event.is_new_payee and event.amount_usd >= 50000:
        return {
            "risk_level": "high",
            "recommended_action": "escalate_to_compliance",
            "reasoning": ["New payee wire over $50k"],
            "red_flags": ["new_payee_large_wire"]
        }
    if event.device_trust == "compromised":
        return {
            "risk_level": "high",
            "recommended_action": "step_up_verification",
            "reasoning": ["Compromised device detected"],
            "red_flags": ["compromised_device"]
        }
    return None

def assess_event(event: FraudEvent) -> dict:
    rule_hit = hard_rules(event)
    if rule_hit:
        return rule_hit

    enriched = enrich_event(event)
    result = chain.invoke({"context": enriched})

    # Minimal post-processing guardrail
    allowed_actions = {"auto_clear", "step_up_verification", "escalate_to_compliance"}
    if result.get("recommended_action") not in allowed_actions:
        raise ValueError("Invalid model action")

    return result

sample = FraudEvent(
    client_id="C123",
    event_type="wire",
    amount_usd=75000,
    jurisdiction="SG",
    is_new_payee=True,
    kyc_risk_rating="high",
    device_trust="trusted",
    recent_contact_change_days=7,
)

print(assess_event(sample))

4) Log every decision for auditability

Wealth management teams need traceability. Store the exact inputs, policy version, prompt version, and final disposition.

import json
from datetime import datetime

def audit_log(event: FraudEvent, decision: dict) -> None:
    record = {
        "timestamp_utc": datetime.utcnow().isoformat(),
        "client_id": event.client_id,
        "event": event.model_dump(),
        "decision": decision,
        "policy_version": "fraud-policy-v3",
        "prompt_version": "fraud-prompt-v2"
    }
    with open("fraud_audit_log.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(record) + "\n")

Production Considerations

• •

  Deploy with a two-stage decision path

  • •Use rules first, LLM second.
  • •Keep escalation thresholds deterministic so compliance can explain outcomes during review.
• •

  Monitor false positives by client segment

  • •Private banking clients often have legitimate large wires and cross-border activity.
  • •Track alert rates by segment, jurisdiction, advisor desk, and product line.
• •

  Enforce data residency and access controls

  • •Keep client PII inside approved regions.
  • •Redact account numbers, passport data, and free-form notes before sending anything to the model provider.
• •

  Version everything

  • •Prompt versioning matters as much as code versioning.
  • •Store model name, temperature, policy ruleset hash, and human override outcome for each case.

Common Pitfalls

• •

  Using the LLM as the first line of defense

  • •This creates noisy alerts and weak control over edge cases.
  • •Fix it by putting hard rules and threshold checks before any model call.
• •

  Sending unfiltered PII into prompts

  • •Advisor notes often contain sensitive identity data that should never leave your boundary unchanged.
  • •Fix it with field-level redaction and minimal context injection.
• •

  Returning free-text decisions

  • •Compliance workflows need structured outputs they can route and measure.
  • •Fix it with JsonOutputParser or a Pydantic-backed output schema so downstream systems can validate actions.

In wealth management fraud detection, the agent’s job is not to “decide truth.” Its job is to reduce time-to-detection while preserving auditability. If you keep the logic structured and the LLM constrained to explanation plus classification support, you get something compliance can actually use.

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.