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

A transaction monitoring agent for lending watches borrower activity, scores it against policy rules and model signals, and decides whether to pass, flag, or escalate a case. For lenders, this matters because the cost of missing suspicious payment behavior is not just fraud loss; it also affects compliance, portfolio risk, collections quality, and auditability.

Architecture

• •Transaction intake
  • •Pulls payments, disbursements, chargebacks, ACH returns, and account events from your ledger or event bus.
• •Policy engine
  • •Encodes lending-specific rules like missed-payment patterns, rapid repayment after drawdown, velocity spikes, and account ownership mismatches.
• •Risk scoring node
  • •Combines deterministic rules with model outputs such as anomaly score, delinquency risk, or synthetic identity indicators.
• •Case decision node
  • •Produces one of a small set of actions: approve, review, escalate, freeze.
• •Audit logger
  • •Persists every input, rule hit, score, and final decision for model risk management and regulator review.
• •Human review handoff
  • •Routes high-risk cases to an analyst queue with enough context to make a decision quickly.

Implementation

1) Define the state and nodes

Use a typed LangGraph state so every step has explicit inputs and outputs. In lending workflows, keep the raw transaction data separate from derived flags so you can explain decisions later.

from typing import TypedDict, List, Dict, Any
from langgraph.graph import StateGraph, START, END

class TxnState(TypedDict):
    transaction: Dict[str, Any]
    policy_flags: List[str]
    risk_score: float
    decision: str
    audit: List[Dict[str, Any]]

def ingest_txn(state: TxnState) -> TxnState:
    txn = state["transaction"]
    audit = state.get("audit", [])
    audit.append({"step": "ingest", "transaction_id": txn["id"]})
    return {**state, "audit": audit}

def apply_policies(state: TxnState) -> TxnState:
    txn = state["transaction"]
    flags = []

    if txn["amount"] > 5000 and txn["channel"] == "cash":
        flags.append("high_cash_amount")
    if txn.get("days_since_last_payment", 999) < 2:
        flags.append("rapid_repeat_payment")
    if txn.get("country") not in {"US", "CA"}:
        flags.append("cross_border_activity")

    audit = state.get("audit", [])
    audit.append({"step": "policy_check", "flags": flags})
    return {**state, "policy_flags": flags, "audit": audit}

def score_risk(state: TxnState) -> TxnState:
    base = 0.1
    base += 0.35 * len(state["policy_flags"])
    if state["transaction"].get("delinquency_days", 0) > 30:
        base += 0.25

    risk_score = min(base, 1.0)
    audit = state.get("audit", [])
    audit.append({"step": "score", "risk_score": risk_score})
    return {**state, "risk_score": risk_score, "audit": audit}

2) Add the decision logic with LangGraph routing

Use add_conditional_edges to route cases based on score thresholds. This keeps the graph explicit and easy to test.

def decide_route(state: TxnState) -> str:
    if state["risk_score"] >= 0.8:
        return "escalate"
    if state["risk_score"] >= 0.4:
        return "review"
    return "approve"

def approve_case(state: TxnState) -> TxnState:
    audit = state.get("audit", [])
    audit.append({"step": "decision", "value": "approve"})
    return {**state, "decision": "approve", "audit": audit}

def review_case(state: TxnState) -> TxnState:
    audit = state.get("audit", [])
    audit.append({"step": "decision", "value": "review"})
    return {**state, "decision": "review", "audit": audit}

def escalate_case(state: TxnState) -> TxnState:
    audit = state.get("audit", [])
    audit.append({"step": "decision", "value": "escalate"})
    return {**state, "decision": "escalate", "audit": audit}

3) Build and compile the graph

This is the actual LangGraph pattern you want in production: a small DAG with deterministic branching and auditable outputs.

graph = StateGraph(TxnState)

graph.add_node("ingest_txn", ingest_txn)
graph.add_node("apply_policies", apply_policies)
graph.add_node("score_risk", score_risk)
graph.add_node("approve_case", approve_case)
graph.add_node("review_case", review_case)
graph.add_node("escalate_case", escalate_case)

graph.add_edge(START, "ingest_txn")
graph.add_edge("ingest_txn", "apply_policies")
graph.add_edge("apply_policies", "score_risk")

graph.add_conditional_edges(
    "score_risk",
    decide_route,
    {
        "approve": "approve_case",
        "review": "review_case",
        "escalate": "escalate_case",
    },
)

graph.add_edge("approve_case", END)
graph.add_edge("review_case", END)
graph.add_edge("escalate_case", END)

app = graph.compile()

4) Run it on a lending transaction

Keep the transaction payload realistic. You want fields that map to real lending controls: payment history, delinquency status, channel type, geography, and amount.

sample_state = {
    "transaction": {
        "id": "txn_10021",
        "amount": 7800,
        "channel": "cash",
        "country": "NG",
        "days_since_last_payment": 1,
        "delinquency_days": 45,
        # add only fields your jurisdiction allows you to process
        # keep residency constraints in mind for PII storage
    },
    "policy_flags": [],
    # include an initial list so downstream steps can append safely
    # this also makes replay/audit easier
   ,"risk_score": 0.0,
   ,"decision": "",
   ,"audit": []
}

result = app.invoke(sample_state)
print(result["decision"])
print(result["risk_score"])
print(result["policy_flags"])
print(result["audit"])

Production Considerations

• •Data residency
  • •Keep borrower PII and transaction logs in-region if your lending program operates under local residency rules. If the graph calls external models or tools, pass only redacted features.
• •Auditability
  • •Persist every node input/output with timestamps and versioned policy logic. Regulators care about why a case was escalated as much as the final label.
• •Guardrails
  • •Hard-limit actions available to the agent. A lending monitor should not invent new outcomes; it should only classify into approved states like approve, review, or escalate.
• •Deployment
  • •Run the graph as a stateless service behind a queue consumer for bursty payment streams. Use idempotency keys on transaction IDs so replays do not duplicate cases.

Common Pitfalls

• •Using an LLM for deterministic policy checks
  • •Don’t ask a model to decide whether a borrower missed two payments or whether an amount exceeds a threshold. Put those checks in code so they are stable and auditable.
• •Mixing raw PII into prompts or traces
  • •Redact account numbers, names, addresses, and national IDs before any model call or logging sink. Store sensitive data in your core system of record.
• •Building one giant agent node
  • •Keep ingestion, policy evaluation, scoring, and routing separate. Smaller nodes are easier to test under model risk governance and much easier to certify during audits.

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.