Graph vs. Relational for Card Fraud - Enterprise Multi-Agent LangGraph RAG for Internet & Mobile Banking

Part I: Why Graph Databases Outperform Relational Models in Card Fraud

Before building the next system, it is critical to articulate why we abandon the relational model for fraud investigation. This isn’t technology hype—it’s mathematical necessity.

The Join Explosion Problem

In a relational database, finding a fraud ring requires recursive self-joins:

-- Find cardholders sharing devices with our suspect (3 hops)
WITH RECURSIVE device_chain AS (
    SELECT c.customer_id, t.device_fp, 1 AS depth
    FROM transactions t JOIN cards c ON t.card_token = c.token
    WHERE t.txn_id IN (:trigger_txns)
    
    UNION ALL
    
    SELECT c2.customer_id, t2.device_fp, dc.depth + 1
    FROM device_chain dc
    JOIN transactions t2 ON t2.device_fp = dc.device_fp
    JOIN cards c2 ON t2.card_token = c2.token
    WHERE dc.depth < 3 AND t2.timestamp > NOW() - INTERVAL '30 days'
)
SELECT * FROM device_chain WHERE depth > 1;

Problems at scale:

Metric	Relational (PostgreSQL)	Graph (Neo4j)
3-hop query on 1B txns	45-120 seconds	15-80 milliseconds
Query plan stability	Degrades with data growth	Consistent O(k) traversal
Concurrent read performance	Lock contention on CTEs	Lock-free native traversal
Schema evolution for new fraud patterns	ALTER TABLE + migration	Add relationship type instantly
Path explanation for auditors	Reconstruct from join logic	Native path objects returned

Three Structural Advantages for Cards

Index-Free Adjacency: Each node stores direct pointers to its neighbors. Traversing from Device → Transaction → Cardholder is a pointer chase, not an index lookup + hash join. At 3+ hops, this difference is 100-1000x.
Polymorphic Relationships: Fraud rings evolve. Today it’s shared devices; tomorrow it’s shared shipping addresses, then shared WiFi fingerprints, then behavioral biometric clusters. In RDBMS, each new signal requires schema changes. In graph, you add a new relationship type with zero downtime.
Natural Pattern Matching: Cypher’s MATCH path = (a)-[*1..3]->(b) directly expresses "find connections." SQL expresses the mechanism of finding connections. When your LLM agents generate queries, generating correct Cypher is dramatically more reliable than generating correct recursive CTEs.

When to Stay Relational

Graph is not better for:

Transaction ledger storage (use columnar/OLAP)
Real-time authorization decisions (use Redis + ML)
Regulatory reporting aggregations (use data warehouse)
Simple account CRUD (use PostgreSQL)

Graph wins specifically for relationship-intensive investigative queries at depth ≥ 2. For card fraud, that’s exactly the workload.

Part II: Internet & Mobile Banking — End-to-End Multi-Agent LangGraph RAG

Now we apply these principles to a different but equally graph-critical domain: Internet & Mobile Banking (IMB). Here the use case shifts from fraud rings to Account Takeover (ATO) Investigation & Digital Channel Risk.

The IMB Challenge

Digital banking generates rich behavioral telemetry: login events, device changes, credential resets, session patterns, and cross-channel activity. ATO investigations require correlating:

Login anomalies across web/mobile/API channels
Device fingerprint evolution over time
Credential change sequences
Linked accounts and authorized users
Unstructured incident reports and security policies

This is a temporal identity graph problem—fundamentally different from card fraud’s transaction graph.

Graph Ontology for Digital Banking

// Identity & Access Fabric
(:Customer)-[:HAS_PROFILE]->(:DigitalProfile)
(:DigitalProfile)-[:AUTHENTICATED_VIA]->(:Credential)
(:Credential)-[:RESET_HISTORY]->(:CredentialEvent)
(:DigitalProfile)-[:ACTIVE_SESSION]->(:Session)
(:Session)-[:ORIGINATED_FROM]->(:Device)
(:Session)-[:ACCESSED_CHANNEL]->(:Channel)  // WEB, MOBILE_IOS, MOBILE_ANDROID, API

// Behavioral Signals
(:Device)-[:GEOLOCATION_HISTORY]->(:GeoPoint)
(:Device)-[:BEHAVIORAL_BASELINE]->(:BehaviorProfile)
(:Session)-[:TRIGGERED_ALERT]->(:SecurityAlert)
(:Customer)-[:LINKED_ACCOUNT]->(:DepositAccount)
(:Customer)-[:AUTHORIZED_USER]->(:AuthorizedUser)

// Investigative Knowledge
(:IncidentCase)-[:INVESTIGATES]->(:SecurityAlert)
(:IncidentCase)-[:REFERENCES_POLICY]->(:SecurityPolicy)
(:IncidentCase)-[:CORRELATES_WITH]->(:IncidentCase)
(:SecurityPolicy)-[:SUPERSEDES]->(:SecurityPolicy)  // Version chain

Key differences from card fraud graph:

Centered on identity sessions, not transactions
Temporal credential events replace transaction timestamps
Channel nodes enable cross-platform attack detection
Behavioral baselines as first-class entities for anomaly context

Real-Time Use Case: Suspicious Cross-Channel Account Takeover

Scenario: Customer "Jane Doe" reports unauthorized transfers. Security ops sees:

Password reset via mobile app at 2:00 AM
New device enrollment 5 minutes later
Web login from different geo 20 minutes after
Two outbound transfers totaling $48,000

Analyst needs to determine: Is this ATO? What was the attack vector? What policy governs remediation? Are other customers affected?

Implementation

Step 1: Typed State with Session Memory

from typing import Annotated, List, Dict, Any, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class IMBInvestigationState(TypedDict):
    """State for digital banking ATO investigation."""
    messages: Annotated[List[Any], add_messages]
    
    # Case context
    incident_id: str
    customer_id: str
    alert_ids: List[str]
    analyst_id: str
    
    # Graph-derived identity intelligence
    session_timeline: List[Dict[str, Any]]
    device_history: List[Dict[str, Any]]
    credential_events: List[Dict[str, Any]]
    linked_accounts_at_risk: List[str]
    correlated_incidents: List[Dict[str, Any]]
    attack_vector_assessment: str
    
    # RAG-derived policy context
    applicable_security_policies: List[Dict[str, str]]
    regulatory_requirements: List[str]
    remediation_steps: List[str]
    
    # Workflow control
    phase: str  # "triage", "identity_analysis", "policy_retrieval", "synthesis", "complete"
    escalation_required: bool
    sar_filing_needed: bool
    error_log: List[str]

Step 2: Graph Tools for Identity Traversal

from langchain_neo4j import Neo4jGraph
from langchain_core.tools import tool
from datetime import datetime, timedelta

graph = Neo4jGraph(
    url="bolt://imb-graph-prod.databases.neo4j.io:7687",
    username="neo4j",
    password="${NEO4J_PASSWORD}",
    database="digital-banking",
    read_only=True
)

@tool
def reconstruct_session_timeline(customer_id: str, lookback_days: int = 7) -> list:
    """
    Rebuilds chronological session + credential + device timeline.
    Critical for establishing attack sequence in ATO cases.
    """
    query = """
    MATCH (c:Customer {customer_id: $cid})-[:HAS_PROFILE]->(dp:DigitalProfile)
    OPTIONAL MATCH (dp)-[:ACTIVE_SESSION]->(s:Session)-[:ORIGINATED_FROM]->(d:Device)
    OPTIONAL MATCH (s)-[:ACCESSED_CHANNEL]->(ch:Channel)
    OPTIONAL MATCH (s)-[:TRIGGERED_ALERT]->(a:SecurityAlert)
    OPTIONAL MATCH (dp)-[:AUTHENTICATED_VIA]->(cred:Credential)-[:RESET_HISTORY]->(ce:CredentialEvent)
    WHERE s.started_at > datetime() - duration($lookback)
       OR ce.event_time > datetime() - duration($lookback)
    WITH s, d, ch, a, ce,
         coalesce(s.started_at, ce.event_time) AS event_time,
         CASE 
           WHEN s IS NOT NULL THEN 'SESSION'
           WHEN ce IS NOT NULL THEN 'CREDENTIAL_EVENT'
         END AS event_type
    ORDER BY event_time ASC
    RETURN 
        event_time,
        event_type,
        ch.name AS channel,
        d.device_fingerprint AS device,
        d.is_trusted AS device_trusted,
        a.alert_type AS alert,
        ce.event_subtype AS cred_event,
        s.geo_country AS country,
        s.ip_risk_score AS ip_risk
    """
    return graph.query(query, {
        "cid": customer_id,
        "lookback": f"P{lookback_days}D"
    })

@tool
def find_correlated_atto_incidents(customer_id: str, device_fps: List[str]) -> list:
    """
    Finds other customers who experienced similar attack patterns.
    Uses device + behavioral similarity to detect campaign-level ATO.
    """
    query = """
    UNWIND $fps AS fp
    MATCH (d:Device {device_fingerprint: fp})<-[:ORIGINATED_FROM]-(s:Session)<-[:ACTIVE_SESSION]-
          (dp:DigitalProfile)<-[:HAS_PROFILE]-(other:Customer)
    WHERE other.customer_id <> $cid
      AND s.started_at > datetime() - duration('P30D')
    OPTIONAL MATCH (other_inc:IncidentCase)-[:INVESTIGATES]->(alert:SecurityAlert)
    WHERE alert.customer_id = other.customer_id
      AND alert.created_at > datetime() - duration('P30D')
    RETURN DISTINCT
        other.customer_id AS affected_customer,
        fp AS shared_device,
        count(DISTINCT s) AS shared_sessions,
        collect(DISTINCT other_inc.incident_id)[0..5] AS related_incidents
    ORDER BY shared_sessions DESC
    LIMIT 20
    """
    return graph.query(query, {"cid": customer_id, "fps": device_fps})

@tool
def retrieve_security_policy(policy_context: List[str]) -> list:
    """
    Hybrid RAG for security policies. Combines vector similarity
    with graph-based version resolution to ensure current policy.
    """
    query = """
    CALL db.index.fulltext.queryNodes('securityPolicySearch', $query_text) YIELD node, score
    WHERE node.effective_date <= datetime()
      AND NOT EXISTS {
        MATCH (node)-[:SUPERSEDES]->(newer:SecurityPolicy)
        WHERE newer.effective_date <= datetime()
      }
      AND score > 0.65
    RETURN 
        node.policy_id,
        node.title,
        node.category,
        substring(node.content, 0, 600) AS excerpt,
        score AS relevance
    ORDER BY score DESC
    LIMIT 5
    """
    return graph.query(query, {"query_text": " ".join(policy_context)})

Step 3: LangGraph Multi-Agent Orchestration

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Agent specializations
identity_llm = llm.bind_tools([reconstruct_session_timeline, find_correlated_atto_incidents])
policy_llm = llm.bind_tools([retrieve_security_policy])

def identity_analysis_agent(state: IMBInvestigationState) -> IMBInvestigationState:
    """Reconstructs attack timeline and identifies correlated victims."""
    prompt = f"""You are a Digital Identity Forensics Analyst.
    Customer {state['customer_id']} has alerts: {state['alert_ids']}.
    Reconstruct the session/credential timeline and check for correlated ATO incidents.
    Determine the likely attack vector (credential stuffing, SIM swap, phishing, etc.)."""
    response = identity_llm.invoke(
        [{"role": "user", "content": prompt}] + state["messages"]
    )
    return {"messages": [response], "phase": "identity_analysis"}

def policy_retrieval_agent(state: IMBInvestigationState) -> IMBInvestigationState:
    """Fetches applicable security policies based on attack vector and findings."""
    context_keywords = ["account takeover", state.get("attack_vector_assessment", "unknown")]
    if state.get("linked_accounts_at_risk"):
        context_keywords.append("linked account remediation")
    if state.get("correlated_incidents"):
        context_keywords.append("campaign response")
    
    prompt = f"""Retrieve current security policies relevant to: {context_keywords}.
    Only return active, non-superseded policies."""
    response = policy_llm.invoke([{"role": "user", "content": prompt}])
    return {"messages": [response], "phase": "policy_retrieval"}

def synthesis_agent(state: IMBInvestigationState) -> IMBInvestigationState:
    """Produces final ATO determination with policy-backed remediation plan."""
    prompt = f"""
    You are a Senior Digital Banking Security Officer. Synthesize this ATO investigation:
    
    CUSTOMER: {state['customer_id']}
    INCIDENT: {state['incident_id']}
    SESSION TIMELINE EVENTS: {len(state.get('session_timeline', []))}
    ATTACK VECTOR: {state.get('attack_vector_assessment', 'Undetermined')}
    CORRELATED VICTIMS: {len(state.get('correlated_incidents', []))}
    ACCOUNTS AT RISK: {state.get('linked_accounts_at_risk', [])}
    
    APPLICABLE POLICIES:
    {chr(10).join(f"- [{p['category']}] {p['title']}: {p['excerpt'][:200]}" 
                  for p in state.get('applicable_security_policies', []))}
    
    Produce:
    1. ATO Determination (Confirmed / Suspected / False Positive)
    2. Attack Vector Narrative (chronological)
    3. Customer Remediation Steps (per policy)
    4. Campaign Assessment (isolated vs. coordinated)
    5. SAR Filing Recommendation (Yes/No with justification)
    6. Escalation Recommendation
    """
    response = llm.invoke(prompt)
    
    exposure = len(state.get("linked_accounts_at_risk", []))
    correlated = len(state.get("correlated_incidents", []))
    
    return {
        "messages": [response],
        "phase": "complete",
        "escalation_required": correlated > 2 or exposure > 3,
        "sar_filing_needed": state.get("attack_vector_assessment") == "confirmed_ato"
    }

# Build graph
workflow = StateGraph(IMBInvestigationState)

workflow.add_node("identity_analysis", identity_analysis_agent)
workflow.add_node("identity_tools", ToolNode([
    reconstruct_session_timeline, 
    find_correlated_atto_incidents
]))
workflow.add_node("policy_retrieval", policy_retrieval_agent)
workflow.add_node("policy_tools", ToolNode([retrieve_security_policy]))
workflow.add_node("synthesis", synthesis_agent)

workflow.set_entry_point("identity_analysis")
workflow.add_edge("identity_analysis", "identity_tools")
workflow.add_edge("identity_tools", "policy_retrieval")
workflow.add_edge("policy_retrieval", "policy_tools")
workflow.add_edge("policy_tools", "synthesis")
workflow.add_conditional_edges(
    "synthesis",
    lambda s: END if not s.get("escalation_required") else "escalation_queue"
)

# Persistent state for audit trail and case resumption
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://${PG_CONN}")
app = workflow.compile(checkpointer=checkpointer)93

Step 4: Real-Time Streaming Execution

async def investigate_atto(customer_id: str, alert_ids: List[str], analyst_id: str):
    incident_id = f"ATO-{datetime.utcnow().strftime('%Y%m%d')}-{customer_id[-6:]}"
    config = {"configurable": {"thread_id": incident_id}}
    
    initial_state: IMBInvestigationState = {
        "messages": [],
        "incident_id": incident_id,
        "customer_id": customer_id,
        "alert_ids": alert_ids,
        "analyst_id": analyst_id,
        "phase": "triage",
        "escalation_required": False,
        "sar_filing_needed": False,
        "error_log": []
    }
    
    async for event in app.astream_events(initial_state, config=config, version="v2"):
        kind = event["event"]
        
        if kind == "on_tool_end":
            yield {
                "type": "identity_signal",
                "tool": event["name"],
                "data": event["data"]["output"],
                "timestamp": datetime.utcnow().isoformat()
            }
        elif kind == "on_chat_model_stream":
            yield {
                "type": "synthesis_token",
                "content": event["data"]["chunk"].content
            }
        elif kind == "on_chain_end" and event["name"] == "synthesis":
            yield {
                "type": "investigation_complete",
                "incident_id": incident_id,
                "escalation": event["data"]["output"].get("escalation_required"),
                "sar_needed": event["data"]["output"].get("sar_filing_needed")
            }

Performance Optimization for IMB Scale

Challenge	Solution	Impact
Session volume (millions/day)	Time-partitioned `:Session_YYYYMM` labels + 90-day hot store	Active set < 50M nodes
Device fingerprint cardinality	Bloom filter pre-check before graph traversal	Eliminates 80% of empty traversals
Policy version resolution	Graph-native `SUPERSEDES` chain vs. metadata filtering	Guarantees current policy without external logic
Timeline reconstruction latency	Pre-materialized `:SessionTimeline` view updated via CDC	Read latency < 20ms
Correlation query fan-out	Degree-bounded traversal (`LIMIT` per hop)	Prevents supernode explosion

// Critical indexes for IMB
CREATE INDEX customer_profile IF NOT EXISTS 
FOR (c:Customer) ON (c.customer_id);

CREATE INDEX session_time_channel IF NOT EXISTS 
FOR (s:Session) ON (s.started_at, s.channel);

CREATE INDEX device_fp_trusted IF NOT EXISTS 
FOR (d:Device) ON (d.device_fingerprint, d.is_trusted);

CREATE FULLTEXT INDEX securityPolicySearch FOR (p:SecurityPolicy) 
ON EACH [p.title, p.content, p.category, p.keywords];

Compliance Considerations Specific to Digital Banking

FFIEC Authentication Guidance: Graph-traced session timelines provide evidence of multi-factor authentication compliance and step-up verification triggers.
Reg E / Liability Determination: The chronological attack narrative directly supports Regulation E error resolution timelines and liability assignment.
GLBA Safeguards Rule: Policy retrieval agent ensures every remediation recommendation cites current, active security policy—demonstrating reasonable safeguards.
BSA/AML SAR: Correlated incident detection provides the "pattern of suspicious activity" narrative required for SAR filing.
Data Minimization: Customer PII stored only in vaulted profile nodes; graph relationships use internal IDs. Session logs auto-purge per retention policy.

Conclusion

Internet & Mobile Banking demands a temporal identity graph fundamentally different from both lending and card fraud graphs. The key insights:

Sessions are first-class nodes, not attributes. This enables timeline reconstruction that SQL cannot match.
Credential events form causal chains that reveal attack vectors through traversal, not correlation.
Policy versioning belongs in the graph, eliminating the most common RAG failure mode in regulated domains.
LangGraph’s persistent state turns investigations into resumable, auditable workflows—not ephemeral chat sessions.

The relational model fails here because digital banking risk is inherently topological and temporal. No amount of indexing compensates for the fundamental mismatch between tabular storage and identity-graph queries. This implementation is for architectural reference. Production IMB security systems require FFIEC compliance validation, penetration testing of the graph layer, model risk management review for any automated decisions, and coordination with your institution’s BSA officer and CISO. Never expose graph APIs directly to client applications; always mediate through authenticated service layers with field-level access control.