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

A transaction monitoring agent watches payment activity, scores it against policy and risk rules, and decides whether to clear, enrich, escalate, or freeze a case for human review. In retail banking, that matters because you need to catch fraud, AML patterns, sanctions hits, and unusual customer behavior without burying operations teams in false positives.

Architecture

• •

  Transaction intake layer

  • •Receives card payments, ACH transfers, wire events, and internal ledger movements.
  • •Normalizes fields like customer_id, merchant_category, amount, country, channel, and timestamp.
• •

  Policy and rules engine

  • •Applies deterministic checks first: velocity limits, country blocks, amount thresholds, watchlist matches.
  • •Keeps obvious violations out of the model path.
• •

  LangGraph workflow

  • •Orchestrates the agent steps: enrich → score → decide → escalate.
  • •Uses StateGraph to keep state explicit and auditable.
• •

  Risk scoring service

  • •Produces a risk score from features like transaction history, device fingerprint, geo-distance, and merchant profile.
  • •Can be an internal model endpoint or a feature-based heuristic during phase one.
• •

  Case management output

  • •Writes decisions to a case queue with reason codes.
  • •Supports analyst review, SAR/STR preparation, and audit trails.
• •

  Audit and compliance store

  • •Persists every node input/output, model version, policy version, and final decision.
  • •Needed for AML governance, model validation, and regulator review.

Implementation

1. Define the graph state and domain objects

Keep the state small and explicit. For banking workflows, that makes audit logging easier and reduces accidental data leakage across nodes.

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

class TxnState(TypedDict):
    transaction_id: str
    customer_id: str
    amount: float
    country: str
    channel: str
    merchant_category: str
    risk_score: Optional[float]
    decision: Optional[Literal["approve", "review", "block"]]
    reason_code: Optional[str]

def enrich_transaction(state: TxnState) -> TxnState:
    # Replace with real enrichment calls: customer profile, velocity counters,
    # sanctions screening results, device reputation.
    if state["country"] not in {"US", "GB", "CA"}:
        state["reason_code"] = "HIGH_RISK_COUNTRY"
    return state

2. Add scoring and decision nodes

Use deterministic routing for compliance-sensitive decisions. If your bank later adds an ML model, keep the thresholding logic here so risk policy stays versioned outside the model.

def score_transaction(state: TxnState) -> TxnState:
    score = 0.1
    if state["amount"] > 5000:
        score += 0.4
    if state["merchant_category"] in {"money_services", "crypto"}:
        score += 0.3
    if state.get("reason_code") == "HIGH_RISK_COUNTRY":
        score += 0.3
    state["risk_score"] = min(score, 1.0)
    return state

def decide_action(state: TxnState) -> TxnState:
    if state.get("reason_code") == "HIGH_RISK_COUNTRY" or state["risk_score"] >= 0.8:
        state["decision"] = "block"
        if not state.get("reason_code"):
            state["reason_code"] = "HIGH_RISK_SCORE"
    elif state["risk_score"] >= 0.5:
        state["decision"] = "review"
        state["reason_code"] = "MANUAL_REVIEW_REQUIRED"
    else:
        state["decision"] = "approve"
        state["reason_code"] = "LOW_RISK"
    return state

3. Wire the LangGraph workflow

This is the actual pattern you want in production: explicit nodes, explicit edges, explicit terminal states.

def build_graph():
    graph = StateGraph(TxnState)

    graph.add_node("enrich", enrich_transaction)
    graph.add_node("score", score_transaction)
    graph.add_node("decide", decide_action)

    graph.set_entry_point("enrich")
    graph.add_edge("enrich", "score")
    graph.add_edge("score", "decide")
    graph.add_edge("decide", END)

    return graph.compile()

app = build_graph()

result = app.invoke({
    "transaction_id": "txn_10001",
    "customer_id": "cust_7781",
    "amount": 7200.00,
    "country": "NG",
    "channel": "card_present",
    "merchant_category": "retail",
    "risk_score": None,
    "decision": None,
    "reason_code": None,
})

print(result)

4. Add a review branch when manual investigation is required

Retail banking needs analyst escalation paths. Use conditional routing so high-risk transactions can go to a separate review workflow instead of being hard-coded into a single linear chain.

from langgraph.graph import StateGraph

def route_after_scoring(state: TxnState):
    if state.get("reason_code") == "HIGH_RISK_COUNTRY" or state["risk_score"] >= 0.8:
        return "block"
    if state["risk_score"] >= 0.5:
        return "review"
    return "approve"

def send_to_case_queue(state: TxnState) -> TxnState:
    # Persist to case management system with audit metadata.
    print(f"CASE CREATED: {state['transaction_id']} {state['reason_code']}")
    return state

def approve_txn(state: TxnState) -> TxnState:
    print(f"APPROVED: {state['transaction_id']}")
    return state

graph = StateGraph(TxnState)
graph.add_node("enrich", enrich_transaction)
graph.add_node("score", score_transaction)
graph.add_node("decide", decide_action)
graph.add_node("case_queue", send_to_case_queue)
graph.add_node("approve", approve_txn)

graph.set_entry_point("enrich")
graph.add_edge("enrich", "score")
graph.add_edge("score", "decide")
graph.add_conditional_edges(
    "decide",
    
### Production Considerations

- **Persist every decision with full lineage**
  - Store transaction payload hash, node outputs, policy version, model version, analyst override status.
  - Regulators will ask why a transaction was blocked; your logs need to answer that without reconstructing memory from application code.

- **Separate data residency by region**
  - Keep EU customer data in EU infrastructure and domestic customer data in-country where required.
  - If you call external LLMs or hosted tools for enrichment summaries, ensure they do not move regulated PII across borders.

- **Put hard guardrails before any model call**
  - Sanctions hits, prohibited geographies, velocity breaches, and account takeover indicators should be deterministic.
  - The model can rank ambiguity; it should not override mandatory controls.

- **Monitor false positives by segment**
  - Track approval/review/block rates by product line, geography, customer tier, and channel.
  - A rule that looks fine globally can still crush legitimate cross-border remittances for one segment.

## Common Pitfalls

- **Using one giant prompt-driven agent for everything**
  - This creates non-deterministic behavior where compliance rules get mixed with narrative reasoning.
  - Fix it by separating rules from enrichment and using LangGraph nodes for each stage.

- **Not versioning policies and thresholds**
  - If a threshold changes from `0.7` to `0.6`, you need to know which transactions were evaluated under which rule set.
  - Store policy versions alongside every decision record.

- **Skipping analyst feedback loops**
  - Without review outcomes feeding back into tuning metrics, your false positive rate will drift upward.
  - Capture analyst dispositions like `true_positive`, `false_positive`, and `needs_more_info`, then use them to recalibrate rules and scores.

---

## Keep learning

- [The complete AI Agents Roadmap](/blog/ai-agents-roadmap-2026) — my full 8-step breakdown
- [Free: The AI Agent Starter Kit](/starter-kit) — PDF checklist + starter code
- [Work with me](/contact) — I build AI for banks and insurance companies

*By Cyprian Aarons, AI Consultant at [Topiax](https://topiax.xyz).*