Graph-Native Fraud & Credit Risk for Cards

The Debit/Credit Cards domain presents a unique challenge distinct from mortgages or commercial lending: velocity. While a mortgage underwriter has days to investigate, a card transaction authorization decision must be made in <200ms. Post-transaction fraud investigations, however, require deep relational analysis that traditional rule engines cannot provide. This article details an enterprise-grade Graph-RAG Multi-Agent System for the Cards module. We model the graph specifically for high-cardinality transactional data and low-latency fraud ring detection. Using LangGraph with persistent state, we orchestrate specialized agents that combine real-time transaction signals with historical graph topology and unstructured fraud policy documents.

1. The Card-Specific Graph Ontology

Unlike lending graphs centered on static entities (properties, companies), card graphs are temporal and behavioral. The schema must support both real-time inference and forensic investigation.

Core Entities

Entity	Description	Key Properties
`Cardholder`	Primary account holder	`customer_id`, `risk_tier`, `kyc_verified_at`
`Card`	Physical/virtual payment instrument	`pan_token`, `product_code`, `status`, `issued_at`
`Merchant`	Point-of-sale entity	`mcc_code`, `merchant_id`, `geo_location`, `risk_score`
`Transaction`	Individual authorization/event	`txn_id`, `amount`, `timestamp`, `auth_code`, `response_code`
`Device`	Hardware used for transaction	`device_fingerprint`, `os`, `ip_hash`, `is_emulator`
`Address`	Shipping/billing/AVS location	`address_hash`, `geo_coords`, `type`
`FraudCase`	Investigation container	`case_id`, `status`, `analyst_id`, `created_at`
`PolicyDocument`	Unstructured fraud rules/SOPs	`doc_id`, `version`, `effective_date`

Critical Relationships

// Transactional Fabric (High Volume - Time Partitioned)
(:Cardholder)-[:HOLDS]->(:Card)
(:Card)-[:INITIATED]->(:Transaction)
(:Transaction)-[:PROCESSED_AT]->(:Merchant)
(:Transaction)-[:USED_DEVICE]->(:Device)
(:Transaction)-[:SHIPPED_TO]->(:Address)
(:Transaction)-[:BILLED_TO]->(:Address)

// Behavioral & Risk Signals (Derived/Computed)
(:Device)-[:PREVIOUSLY_USED_BY]->(:Cardholder)  // Device sharing signal
(:Merchant)-[:FLAGGED_FOR]->(:FraudType)         // Merchant risk categorization
(:Address)-[:LINKED_TO_FRAUD_RING]->(:FraudRing) // Spatial clustering result
(:Cardholder)-[:REPORTED_STOLEN]->(:Card)        // Temporal status change

// Investigative & Knowledge
(:FraudCase)-[:INVESTIGATES]->(:Transaction)
(:FraudCase)-[:REFERENCES_POLICY]->(:PolicyDocument)
(:FraudCase)-[:ASSIGNED_TO]->(:Analyst)
(:FraudCase)-[:RELATED_TO]->(:FraudCase)         // Case merging/linking

Why This Schema Matters for Cards

Device Fingerprint as Hub: Fraud rings reuse devices across stolen cards. Device is the highest-value node for contagion detection.
Temporal Edges: Transactions are time-series. We use Neo4j’s temporal indexing or time-bucketed labels (:Transaction_2026_08) to avoid scanning billions of nodes.
Tokenized PANs: Never store raw card numbers. Use vault tokens as identifiers.
MCC Hierarchy: Merchant Category Codes form a taxonomy. Graph traversal up/down MCC hierarchies enables category-level risk scoring.

2. Real-Time Use Case: Post-Authorization Fraud Ring Investigation

Scenario: A batch of 47 transactions from different cardholders was flagged by the ML model as "potential coordinated fraud." All transactions occurred within a 3-hour window at merchants in the same metro area, but no single rule triggered a decline. A fraud analyst needs to determine: Is this a fraud ring? What is the total exposure? Which policy applies?

This requires traversing device sharing, address proximity, merchant connections, AND retrieving relevant SOPs—all within a single investigative session with memory.

3. End-to-End Implementation

Prerequisites

pip install langgraph langchain-neo4j neo4j redis qdrant-client pydantic langchain-openai

Step 1: Typed State with Memory & Checkpointing

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

class CardFraudState(TypedDict):
    """Persistent state for card fraud investigation."""
    messages: Annotated[List[Any], add_messages]
    
    # Investigation context
    case_id: str
    trigger_txn_ids: List[str]
    analyst_id: str
    
    # Graph-derived intelligence
    linked_cardholders: List[Dict[str, Any]]
    shared_devices: List[Dict[str, Any]]
    merchant_cluster: Dict[str, Any]
    estimated_exposure: float
    fraud_ring_confidence: float  # 0.0 - 1.0
    
    # RAG-derived policy context
    applicable_policies: List[Dict[str, str]]  # {title, excerpt, relevance_score}
    recommended_actions: List[str]
    
    # Workflow control
    investigation_phase: str  # "triage", "deep_dive", "synthesis", "complete"
    human_approval_required: bool
    error_log: List[str]

Step 2: Graph Tools Optimized for Card Velocity

from langchain_neo4j import Neo4jGraph
from langchain_core.tools import tool

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

@tool
def find_device_contagion(txn_ids: List[str], max_hops: int = 3) -> dict:
    """
    Finds all cardholders connected through shared devices.
    Uses bidirectional traversal optimized for device-centric fraud rings.
    """
    query = """
    UNWIND $txn_ids AS txn_id
    MATCH (t:Transaction {txn_id: txn_id})-[:USED_DEVICE]->(d:Device)
    MATCH (d)<-[:USED_DEVICE]-(other_txn:Transaction)<-[:INITIATED]-(ch:Cardholder)
    WHERE other_txn.txn_id <> txn_id
      AND other_txn.timestamp > t.timestamp - duration('P30D')
    WITH DISTINCT ch, d, count(other_txn) AS shared_txn_count
    RETURN 
        ch.customer_id,
        ch.risk_tier,
        d.device_fingerprint,
        d.is_emulator,
        shared_txn_count,
        collect(DISTINCT other_txn.txn_id)[0..10] AS sample_txns
    ORDER BY shared_txn_count DESC
    LIMIT 100
    """
    results = graph.query(query, {"txn_ids": txn_ids})
    return {
        "linked_cardholders": results,
        "unique_devices": len(set(r["device_fingerprint"] for r in results)),
        "total_linked_accounts": len(results)
    }

@tool
def get_merchant_risk_context(merchant_ids: List[str]) -> dict:
    """Retrieves merchant risk profile and historical fraud patterns."""
    query = """
    MATCH (m:Merchant)
    WHERE m.merchant_id IN $merchant_ids
    OPTIONAL MATCH (m)-[:FLAGGED_FOR]->(ft:FraudType)
    OPTIONAL MATCH (m)<-[:PROCESSED_AT]-(hist_txn:Transaction)
    WHERE hist_txn.response_code IN ['FRAUD', 'CHARGEBACK']
      AND hist_txn.timestamp > datetime() - duration('P90D')
    RETURN 
        m.merchant_id,
        m.mcc_code,
        m.risk_score,
        collect(DISTINCT ft.name) AS fraud_types,
        count(hist_txn) AS recent_fraud_txns
    """
    results = graph.query(query, {"merchant_ids": merchant_ids})
    return {"merchants": results}

@tool
def retrieve_fraud_policy(scenario_keywords: List[str]) -> list:
    """
    Hybrid RAG: Vector search + graph-filtered policy retrieval.
    Ensures only current, applicable policies are returned.
    """
    # In production, this calls Qdrant with metadata filtering
    # Here we simulate the hybrid approach via Cypher fulltext + vector
    query = """
    CALL db.index.fulltext.queryNodes('policySearch', $query_text) YIELD node, score
    WHERE node.effective_date <= datetime()
      AND node.status = 'ACTIVE'
      AND score > 0.7
    RETURN 
        node.doc_id AS policy_id,
        node.title,
        substring(node.content, 0, 500) AS excerpt,
        score AS relevance
    ORDER BY score DESC
    LIMIT 5
    """
    query_text = " ".join(scenario_keywords)
    return graph.query(query, {"query_text": query_text})

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)

# Specialized agent bindings
triage_llm = llm.bind_tools([find_device_contagion, get_merchant_risk_context])
policy_llm = llm.bind_tools([retrieve_fraud_policy])

def triage_agent(state: CardFraudState) -> CardFraudState:
    """Initial assessment: graph traversal for structural risk signals."""
    prompt = f"""You are a Fraud Triage Analyst. Given trigger transactions {state['trigger_txn_ids']},
    determine if there is evidence of coordinated fraud by investigating device sharing 
    and merchant clusters. Call the appropriate tools."""
    response = triage_llm.invoke([{"role": "user", "content": prompt}] + state["messages"])
    return {"messages": [response], "investigation_phase": "triage"}

def policy_retrieval_agent(state: CardFraudState) -> CardFraudState:
    """RAG agent: fetches relevant fraud policies based on graph findings."""
    # Extract keywords from triage findings
    keywords = ["fraud ring", "device sharing", "coordinated"]
    if state.get("merchant_cluster", {}).get("mcc_code"):
        keywords.append(f"MCC {state['merchant_cluster']['mcc_code']}")
    
    prompt = f"""Based on investigation findings, retrieve applicable fraud policies.
    Keywords: {keywords}. Only return ACTIVE policies."""
    response = policy_llm.invoke([{"role": "user", "content": prompt}])
    return {"messages": [response], "investigation_phase": "deep_dive"}

def synthesis_agent(state: CardFraudState) -> CardFraudState:
    """Combines graph intelligence + policy into actionable recommendation."""
    synthesis_prompt = f"""
    You are a Senior Fraud Investigator. Synthesize this card fraud investigation:
    
    CASE: {state['case_id']}
    TRIGGER TXNS: {len(state['trigger_txn_ids'])} transactions
    LINKED ACCOUNTS: {len(state.get('linked_cardholders', []))}
    SHARED DEVICES: {len(state.get('shared_devices', []))}
    ESTIMATED EXPOSURE: ${state.get('estimated_exposure', 0):,.2f}
    FRAUD RING CONFIDENCE: {state.get('fraud_ring_confidence', 0):.2%}
    
    APPLICABLE POLICIES:
    {chr(10).join(f"- {p['title']}: {p['excerpt'][:200]}" for p in state.get('applicable_policies', []))}
    
    Produce:
    1. Fraud Ring Determination (Confirmed / Suspected / Not Indicated)
    2. Total Exposure Calculation Methodology
    3. Recommended Actions (per policy)
    4. Evidence Summary for SAR filing if applicable
    5. Confidence Level & Caveats
    """
    response = llm.invoke(synthesis_prompt)
    requires_approval = state.get("estimated_exposure", 0) > 50000
    return {
        "messages": [response],
        "investigation_phase": "complete",
        "human_approval_required": requires_approval
    }

# Build workflow with conditional routing
workflow = StateGraph(CardFraudState)

workflow.add_node("triage", triage_agent)
workflow.add_node("triage_tools", ToolNode([find_device_contagion, get_merchant_risk_context]))
workflow.add_node("policy_retrieval", policy_retrieval_agent)
workflow.add_node("policy_tools", ToolNode([retrieve_fraud_policy]))
workflow.add_node("synthesis", synthesis_agent)

workflow.set_entry_point("triage")
workflow.add_edge("triage", "triage_tools")
workflow.add_edge("triage_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("human_approval_required") else "human_review"
)

# Persistent checkpointing for audit & resume capability
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://${PG_CONN}")
app = workflow.compile(checkpointer=checkpointer)

Step 4: Real-Time Execution with Streaming & Memory Resume

import uuid

async def investigate_fraud_case(trigger_txn_ids: List[str], analyst_id: str):
    case_id = f"FRAUD-{uuid.uuid4().hex[:8].upper()}"
    config = {"configurable": {"thread_id": case_id}}
    
    initial_state: CardFraudState = {
        "messages": [],
        "case_id": case_id,
        "trigger_txn_ids": trigger_txn_ids,
        "analyst_id": analyst_id,
        "investigation_phase": "triage",
        "human_approval_required": False,
        "error_log": []
    }
    
    # Stream events for real-time analyst dashboard
    async for event in app.astream_events(initial_state, config=config, version="v2"):
        kind = event["event"]
        
        if kind == "on_tool_end":
            yield {"type": "graph_signal", "tool": event["name"], "data": event["data"]["output"]}
        elif kind == "on_chat_model_stream":
            yield {"type": "synthesis", "token": event["data"]["chunk"].content}
        elif kind == "on_chain_end" and event["name"] == "synthesis":
            yield {"type": "case_complete", "case_id": case_id}

# RESUME EXAMPLE: Human approves/rejects after pause
async def approve_case(case_id: str, approved: bool, analyst_notes: str):
    config = {"configurable": {"thread_id": case_id}}
    # Inject human decision back into state
    await app.aupdate_state(config, {
        "messages": [{"role": "user", "content": f"Human decision: {'APPROVED' if approved else 'REJECTED'}. Notes: {analyst_notes}"}],
        "human_approval_required": False
    })
    # Resume execution
    async for event in app.astream(None, config=config):
        yield event

4. Performance Engineering for Card-Scale Data

Card transaction volumes dwarf lending data. Specific optimizations are non-negotiable:

Challenge	Solution	Expected Impact
Billions of Txn Nodes	Time-partitioned labels + TTL archival	Query scope reduced 99%
Device Fingerprint Hotspots	Composite index `(device_fingerprint, timestamp)`	p99 < 30ms for device traversal
Real-Time Auth Latency	Pre-computed risk scores cached in Redis; graph only for investigation	Auth path untouched
Fulltext Policy Search	Dedicated fulltext index with synonym expansion	RAG recall +25%
State Bloat	Message trimming + structured field extraction	Checkpoint size < 10KB

Critical Indexes for Card Fraud

// Device-centric fraud ring detection
CREATE INDEX device_txn_time IF NOT EXISTS 
FOR (t:Transaction) ON (t.timestamp);

CREATE INDEX device_fingerprint_lookup IF NOT EXISTS 
FOR (d:Device) ON (d.device_fingerprint);

// Merchant risk context
CREATE INDEX merchant_id_lookup IF NOT EXISTS 
FOR (m:Merchant) ON (m.merchant_id);

// Policy RAG
CREATE FULLTEXT INDEX policySearch FOR (p:PolicyDocument) 
ON EACH [p.title, p.content, p.keywords];

5. Compliance & Governance for Card Fraud

PCI-DSS: Graph never stores PANs. All card references use tokenized values from your PCI-compliant vault. Audit logs confirm zero PAN access.
FCRA/Adverse Action: When graph-derived signals contribute to account closure or limit reduction, the synthesis_agent output includes explicit citation of which nodes/relationships drove the decision.
SAR Filing: The FraudCase → Transaction relationship provides ready-made evidence chains for FinCEN reporting.
Model Risk (SR 11-7): LangGraph traces serve as model documentation. Each agent’s reasoning is logged and reviewable.
Right to Explanation: Structured state fields (linked_cardholders, shared_devices) enable customer-facing explanations without exposing internal graph topology.

Conclusion

The Cards domain demands a fundamentally different graph architecture than lending: temporal, device-centric, and velocity-aware. By modeling transactions as first-class nodes with time partitioning, centering fraud detection on device fingerprints, and orchestrating specialized agents through LangGraph’s stateful workflow, fintech enterprises can:

Detect fraud rings that evade rule-based systems
Conduct investigations in minutes instead of hours
Maintain full audit trails for regulatory compliance
Scale to billions of transactions without sacrificing latency

This architecture separates the real-time authorization path (which remains ultra-low-latency and rule/ML-driven) from the investigative path (which leverages graph depth and LLM reasoning). This separation is critical: never put a graph database in the hot path of card authorization.