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

A fraud detection agent built with LangGraph takes a transaction, enriches it with customer and device context, scores risk, decides whether to approve, hold, or escalate, and writes an audit trail for later review. In fintech, that matters because fraud decisions need to be fast, explainable, and deterministic enough to satisfy compliance while still adapting to changing attack patterns.

Architecture

• •

  Input normalization node

  • •Validates the incoming transaction payload.
  • •Converts raw events into a consistent schema for downstream nodes.
• •

  Risk enrichment node

  • •Pulls in customer history, device fingerprint, IP reputation, velocity signals, and merchant metadata.
  • •Keeps enrichment separate so it can be tested and audited independently.
• •

  Fraud scoring node

  • •Produces a structured risk assessment.
  • •Can combine rules-based checks with an LLM only for explanation or triage, not as the final authority.
• •

  Decision node

  • •Applies policy thresholds to return approve, hold, or escalate.
  • •This is where compliance rules live.
• •

  Audit/logging node

  • •Persists every input, score, decision, and rationale.
  • •Required for dispute handling, model review, and regulatory evidence.
• •

  Human review handoff

  • •Routes borderline cases to an analyst queue.
  • •Prevents the agent from making irreversible decisions on weak signals.

Implementation

1) Define the state and graph nodes

Use a typed state so every step knows exactly what data it can read and write. In production systems, this is where you keep your schema tight; loose dictionaries turn into incident tickets later.

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

class FraudState(TypedDict):
    transaction_id: str
    amount: float
    currency: str
    customer_id: str
    ip_address: str
    device_id: str
    merchant_id: str
    risk_score: Optional[float]
    decision: Optional[Literal["approve", "hold", "escalate"]]
    rationale: Optional[str]

def normalize_transaction(state: FraudState) -> FraudState:
    state["currency"] = state["currency"].upper()
    return state

def enrich_context(state: FraudState) -> FraudState:
    # Replace with real calls to feature store / risk services
    velocity_flag = state["amount"] > 5000
    state["risk_score"] = 0.82 if velocity_flag else 0.21
    return state

def score_fraud(state: FraudState) -> FraudState:
    if state["risk_score"] is None:
        state["risk_score"] = 0.0
    return state

def decide(state: FraudState) -> FraudState:
    score = state["risk_score"] or 0.0
    if score >= 0.8:
        state["decision"] = "escalate"
        state["rationale"] = "High-risk transaction based on velocity and amount"
    elif score >= 0.5:
        state["decision"] = "hold"
        state["rationale"] = "Borderline risk; requires analyst review"
    else:
        state["decision"] = "approve"
        state["rationale"] = "Risk below threshold"
    return state

def audit(state: FraudState) -> FraudState:
    print(
        {
            "transaction_id": state["transaction_id"],
            "risk_score": state["risk_score"],
            "decision": state["decision"],
            "rationale": state["rationale"],
        }
    )
    return state

2) Wire the workflow with StateGraph

This is the core LangGraph pattern. StateGraph gives you explicit control over execution order, which is what you want when every branch must be defensible in a postmortem or regulator review.

builder = StateGraph(FraudState)

builder.add_node("normalize_transaction", normalize_transaction)
builder.add_node("enrich_context", enrich_context)
builder.add_node("score_fraud", score_fraud)
builder.add_node("decide", decide)
builder.add_node("audit", audit)

builder.add_edge(START, "normalize_transaction")
builder.add_edge("normalize_transaction", "enrich_context")
builder.add_edge("enrich_context", "score_fraud")
builder.add_edge("score_fraud", "decide")
builder.add_edge("decide", "audit")
builder.add_edge("audit", END)

graph = builder.compile()

3) Run the agent on a transaction

The compiled graph is callable via .invoke(). Keep the input minimal and let the graph enrich and decide deterministically.

result = graph.invoke(
    {
        "transaction_id": "txn_10001",
        "amount": 7400.00,
        "currency": "usd",
        "customer_id": "cust_42",
        "ip_address": "203.0.113.10",
        "device_id": "dev_91",
        "merchant_id": "m_778",
        "risk_score": None,
        "decision": None,
        "rationale": None,
    }
)

print(result["decision"])
print(result["rationale"])

4) Add branching for analyst escalation

For real fraud systems, hard thresholds are not enough. Use add_conditional_edges() when you need explicit routing based on risk bands or policy exceptions.

def route_after_scoring(state: FraudState) -> str:
    score = state.get("risk_score") or 0.0
    if score >= 0.8:
        return "escalate"
    if score >= 0.5:
        return "hold"
    return "approve"

def approve(state: FraudState) -> FraudState:
    state["decision"] = "approve"
    return state

def hold_for_review(state: FraudState) -> FraudState:
    state["decision"] = "hold"
    return state

def escalate_case(state: FraudState) -> FraudState:
    state["decision"] = "escalate"
    return state

builder2 = StateGraph(FraudState)
builder2.add_node("normalize_transaction", normalize_transaction)
builder2.add_node("enrich_context", enrich_context)
builder2.add_node("score_fraud", score_fraud)
builder2.add_node("approve", approve)
builder2.add_node("hold_for_review", hold_for_review)
builder2.add_node("escalate_case", escalate_case)

builder2.add_edge(START, "normalize_transaction")
builder2.add_edge("normalize_transaction", "enrich_context")
builder2.add_edge("enrich_context", "score_fraud")

builder2.add_conditional_edges(
    "score_fraud",
    route_after_scoring,
    {
        "approve": "approve",
        "hold": "hold_for_review",
        "escalate": "escalate_case",
    },
)

builder2.add_edge("approve", END)
builder2.add_edge("hold_for_review", END)
builder2.add_edge("escalate_case", END)

fraud_graph = builder2.compile()

Production Considerations

• •

  Auditability

  • •Persist the full input payload, derived features, score, decision path, and model/version metadata.
  • •For fintech audits, you need to explain why a transaction was blocked without replaying brittle external dependencies.
• •

  Data residency

  • •Keep customer PII and transaction data inside approved regions.
  • •If you call external APIs or hosted LLMs for explanations, ensure redaction happens before egress.
• •

  Guardrails

  • •Do not let an LLM make the final fraud decision.
  • •Use it for summarization or analyst notes only; final routing should be threshold-based or policy-based code.
• •

  Monitoring

  • •Track false positives, false negatives, analyst overrides, latency per node, and drift in feature distributions.
  • •Alert when decision rates shift suddenly across merchants, geographies, or card BIN ranges.

Common Pitfalls

• •

  Using the LLM as the decision engine

  • •This creates inconsistent outcomes and weak auditability.
  • •Keep deterministic logic in Python nodes and reserve LLMs for explanations or case summaries.
• •

  Skipping schema validation

  • •Bad inputs from upstream payment services will break your flow at runtime.
  • •Validate required fields before graph execution and reject incomplete events early.
• •

  Not versioning rules and thresholds

  • •A threshold change without versioning makes historical decisions impossible to reproduce.
  • •Store rule versions alongside each decision so compliance teams can reconstruct exact behavior later.

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.