Autonomous Multi-Agent Fraud Triage & SAR Generation Architecture

import os
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage

class FraudState(TypedDict):
    transaction_id: str
    account_id: str
    amount_usd: float
    graph_risk_score: float
    sanctions_hit: bool
    agent_verdicts: List[dict]
    sar_narrative: str
    requires_human_review: bool

def graph_topology_agent(state: FraudState):
    # Air-gapped Neo4j Cypher query & vector similarity
    # Evaluates velocity anomalies and cyclical entity patterns
    return {"graph_risk_score": 0.89, "agent_verdicts": [{"agent": "GraphTopology", "status": "FLAGGED"}]}

def sanctions_screening_agent(state: FraudState):
    # Air-gapped OFAC / PEP fuzzy matching
    return {"sanctions_hit": False, "agent_verdicts": [{"agent": "OFACScreening", "status": "CLEARED"}]}

def sar_generator_agent(state: FraudState):
    # Generates structured FinCEN SAR narrative following regulatory standards
    narrative = f"SAR-2026: Account {state['account_id']} exhibited rapid velocity anomalies (Score: {state['graph_risk_score']})."
    return {"sar_narrative": narrative, "requires_human_review": True}

workflow = StateGraph(FraudState)
workflow.add_node("graph_agent", graph_topology_agent)
workflow.add_node("sanctions_agent", sanctions_screening_agent)
workflow.add_node("sar_agent", sar_generator_agent)

workflow.set_entry_point("graph_agent")
workflow.add_edge("graph_agent", "sanctions_agent")
workflow.add_edge("sanctions_agent", "sar_agent")
workflow.add_edge("sar_agent", END)

fraud_triage_app = workflow.compile()