How to Build a transaction monitoring Agent Using LangGraph in Python for banking

A transaction monitoring agent watches payment activity, scores suspicious patterns, and decides when to escalate a case for review. In banking, that matters because you need fast detection of fraud and AML signals without turning every customer transfer into a manual investigation.

Architecture

• •

  Transaction ingestion layer

  • •Pulls events from core banking, card rails, or streaming topics.
  • •Normalizes fields like account IDs, merchant category, amount, country, and timestamp.
• •

  Risk scoring node

  • •Applies deterministic checks first: velocity, structuring, geo-velocity, threshold breaches.
  • •Can call an LLM only for explanation or case summarization, not for raw risk math.
• •

  Policy and compliance rules

  • •Encodes bank policy: SAR/STR thresholds, sanctions flags, high-risk jurisdictions.
  • •Keeps decisions auditable and consistent across cases.
• •

  LangGraph orchestration

  • •Routes each transaction through classify → score → decide → escalate.
  • •Uses state to carry evidence, scores, and final disposition.
• •

  Case management output

  • •Writes alerts to SIEM, case management systems, or investigator queues.
  • •Stores the full decision trail for audit and model governance.
• •

  Audit logging and observability

  • •Captures inputs, outputs, rule hits, and human overrides.
  • •Supports compliance reviews and post-incident reconstruction.

Implementation

1) Define the state and decision schema

Use a typed state object so every node knows exactly what it can read and write. For banking workflows, keep the evidence trail inside the state so you can persist it later for audit.

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class TxState(TypedDict):
    transaction_id: str
    amount: float
    country: str
    customer_risk: int
    alerts: list[str]
    score: int
    decision: Literal["allow", "review", "block"]
    rationale: str

def init_state(tx: dict) -> TxState:
    return {
        "transaction_id": tx["transaction_id"],
        "amount": tx["amount"],
        "country": tx["country"],
        "customer_risk": tx["customer_risk"],
        "alerts": [],
        "score": 0,
        "decision": "allow",
        "rationale": "",
    }

2) Build deterministic risk nodes

Start with rules that are easy to explain to auditors. In regulated environments, a simple threshold hit is often more defensible than a black-box score alone.

def rule_engine(state: TxState) -> dict:
    alerts = []
    score = 0

    if state["amount"] >= 10000:
        alerts.append("large_transaction")
        score += 40

    if state["country"] in {"IR", "KP", "SY"}:
        alerts.append("sanctions_jurisdiction")
        score += 100

    if state["customer_risk"] >= 80:
        alerts.append("high_customer_risk")
        score += 30

    return {"alerts": alerts, "score": score}

def decide(state: TxState) -> dict:
    if "sanctions_jurisdiction" in state["alerts"]:
        return {
            "decision": "block",
            "rationale": "Sanctions jurisdiction detected; block per policy."
        }

    if state["score"] >= 50:
        return {
            "decision": "review",
            "rationale": f"Risk score {state['score']} exceeded review threshold."
        }

    return {
        "decision": "allow",
        "rationale": f"Risk score {state['score']} below threshold."
    }

3) Wire the graph with LangGraph’s actual API

This is the core pattern. StateGraph defines the workflow; add_node, add_edge, and compile turn it into an executable agent.

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(TxState)

workflow.add_node("rule_engine", rule_engine)
workflow.add_node("decide", decide)

workflow.add_edge(START, "rule_engine")
workflow.add_edge("rule_engine", "decide")
workflow.add_edge("decide", END)

app = workflow.compile()

tx = {
    "transaction_id": "tx_1001",
    "amount": 12500.0,
    "country": "GB",
    "customer_risk": 72,
}

result = app.invoke(init_state(tx))
print(result["decision"])
print(result["rationale"])
print(result["alerts"])

4) Add branching for escalation and human review

For production banking systems, you usually want a conditional path. High-risk cases should go to investigators; low-risk cases should exit cleanly.

def route(state: TxState) -> str:
    if state["decision"] == "block":
        return END
    if state["decision"] == "review":
        return END
    return END

workflow = StateGraph(TxState)
workflow.add_node("rule_engine", rule_engine)
workflow.add_node("decide", decide)

workflow.add_edge(START, "rule_engine")
workflow.add_edge("rule_engine", "decide")
workflow.add_conditional_edges("decide", route)

app = workflow.compile()

If you want an LLM involved, use it only after deterministic scoring. A good pattern is to have the model generate an investigator summary from already-approved facts:

• •Transaction facts
• •Triggered rules
• •Historical context from internal systems
• •Final rationale template

That keeps the model out of the decision path while still reducing analyst workload.

Production Considerations

• •

  Deploy in-region

  • •Keep transaction data in approved regions to satisfy data residency requirements.
  • •If you operate across jurisdictions, split graphs by region or legal entity.
• •

  Log everything needed for audit

  • •Persist input features, triggered rules, final decision, model version, prompt version if used.
  • •Make logs immutable or append-only for compliance review.
• •

  Put guardrails around LLM usage

  • •Never let an LLM directly approve or block payments.
  • •Restrict it to summarization or case notes with strict redaction of PII where required.
• •

  Monitor drift and false positives

  • •Track alert rates by segment: customer tier, geography, payment rail.
  • •Watch for sudden spikes after policy changes or model updates.

Common Pitfalls

1. •

   Using the LLM as the decision engine

   • •Don’t do this in banking workflows.
   • •Use deterministic rules for decisions and reserve the model for explanations or triage summaries.
2. •

   Not persisting the evidence trail

   • •If you only store the final decision, compliance will hate you during audits.
   • •Save all triggered rules and intermediate scores with transaction IDs and timestamps.
3. •

   Ignoring jurisdictional constraints

   • •Data residency and cross-border transfer rules are not optional.
   • •Design your graph deployment so regulated data stays inside approved boundaries.
4. •

   Mixing policy logic into prompt text

   • •Policy belongs in code or configuration files with change control.
   • •Prompts should not be the source of truth for AML thresholds or sanctions handling.

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.