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

A transaction monitoring agent for lending watches borrower activity, flags suspicious or policy-breaking patterns, and turns raw transaction streams into actionable alerts for credit and fraud teams. In lending, that matters because missed anomalies can mean early default risk, synthetic identity abuse, or compliance failures tied to AML, KYC, and fair-lending controls.

Architecture

• •

  Transaction ingestion layer

  • •Pulls card, ACH, bank transfer, and loan repayment events from your ledger or event bus.
  • •Normalizes records into a consistent schema before they hit the model.
• •

  Risk rules engine

  • •Applies deterministic lending rules first: velocity checks, repayment reversals, cash-in/cash-out patterns, and high-risk merchant categories.
  • •Keeps the agent from hallucinating on obvious policy breaches.
• •

  LangChain reasoning layer

  • •Uses ChatOpenAI with structured output to classify events and explain why they are suspicious.
  • •Converts free-form reasoning into machine-readable alerts.
• •

  Case management output

  • •Writes alerts to your internal queue, ticketing system, or SIEM.
  • •Includes evidence fields for audit trails and analyst review.
• •

  Audit and observability layer

  • •Logs prompts, model outputs, rule hits, and final decisions.
  • •Supports model governance, especially important for lending compliance reviews.

Implementation

1) Define the transaction schema and the risk taxonomy

Start by making the input explicit. Lending teams need stable fields for auditability and downstream controls.

from typing import Literal, Optional
from pydantic import BaseModel, Field

class Transaction(BaseModel):
    transaction_id: str
    borrower_id: str
    amount: float
    currency: str = "USD"
    channel: Literal["ach", "card", "wire", "cash", "internal_transfer"]
    merchant_category: Optional[str] = None
    timestamp: str
    country: str
    description: Optional[str] = None

class RiskAssessment(BaseModel):
    risk_level: Literal["low", "medium", "high"]
    reasons: list[str] = Field(default_factory=list)
    recommended_action: Literal["ignore", "review", "escalate"]

This gives you a contract for both the model input and output. In production lending systems, structured outputs beat plain-text summaries every time.

2) Build deterministic pre-checks before the LLM

Do not send every transaction straight to the model. Filter obvious cases first so you reduce cost and improve consistency.

def rule_based_flags(txn: Transaction) -> list[str]:
    flags = []

    if txn.amount >= 5000:
        flags.append("high_amount")
    if txn.channel == "cash":
        flags.append("cash_activity")
    if txn.country not in {"US", "CA", "GB"}:
        flags.append("cross_border_risk")
    if txn.merchant_category in {"gambling", "crypto_exchange"}:
        flags.append("restricted_merchant")

    return flags

These rules are easy to explain to auditors. They also give the LLM context so it can focus on nuanced patterns like structuring or repayment manipulation.

3) Use LangChain structured output for the final assessment

This is the core pattern. ChatOpenAI plus .with_structured_output() gives you validated JSON instead of brittle text parsing.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "You are a transaction monitoring analyst for a lending platform. "
     "Assess whether this transaction is suspicious for fraud, AML, or credit abuse. "
     "Be conservative. Return only valid structured data."),
    ("human",
     "Transaction:\n{transaction}\n\nRule flags:\n{flags}\n\n"
     "Consider lending-specific risks like early default indicators, account takeover,"
     "synthetic identity behavior, repayment manipulation, cash-out patterns,"
     "and compliance concerns.")
])

structured_llm = llm.with_structured_output(RiskAssessment)

def assess_transaction(txn: Transaction) -> RiskAssessment:
    flags = rule_based_flags(txn)
    chain = prompt | structured_llm
    return chain.invoke({
        "transaction": txn.model_dump_json(indent=2),
        "flags": ", ".join(flags) if flags else "none"
    })

txn = Transaction(
    transaction_id="tx_10001",
    borrower_id="b_7781",
    amount=7200,
    channel="wire",
    merchant_category="crypto_exchange",
    timestamp="2026-04-21T10:15:00Z",
    country="NG",
    description="Payment to external wallet provider"
)

result = assess_transaction(txn)
print(result.model_dump())

That pattern is production-friendly because:

• •The prompt is versionable.
• •The output schema is enforced.
• •The model can only return low, medium, or high.

4) Wrap it in an alerting workflow

Once you have a structured assessment, route it into your case system.

def create_alert(txn: Transaction, assessment: RiskAssessment) -> dict:
    return {
        "transaction_id": txn.transaction_id,
        "borrower_id": txn.borrower_id,
        "risk_level": assessment.risk_level,
        "recommended_action": assessment.recommended_action,
        "reasons": assessment.reasons,
        "compliance_tags": ["lending_monitoring", "audit_trail"],
    }

alert = create_alert(txn, result)
print(alert)

In a real deployment this would publish to Kafka, SQS, or your case management API. Keep the alert payload small but complete enough for analysts to reconstruct why the agent fired.

Production Considerations

• •

  Add a hard rule layer before the LLM

  • •Lending teams need deterministic controls for thresholds like repayment reversals, cash deposits near due dates, and repeated transfers to linked accounts.
  • •Use the model only for ambiguous cases.
• •

  Log everything needed for audit

  • •Store prompt version, input features, rule hits, output schema version, and final action.
  • •This matters when compliance asks why a borrower was escalated.
• •

  Respect data residency

  • •Borrower financial data may need to stay in-region depending on jurisdiction.
  • •If you operate across markets, isolate model endpoints and storage by geography.
• •

  Monitor drift by portfolio segment

  • •Track false positives separately for secured loans, unsecured personal loans, SMB lending, and thin-file borrowers.
  • •A single threshold will not behave well across all segments.

Common Pitfalls

1. •

   Sending raw transactions directly to the model

   • •This creates noisy outputs and inconsistent decisions.
   • •Fix it by precomputing rule flags and normalizing fields first.
2. •

   Using free-form text responses

   • •Analysts cannot reliably consume prose at scale.
   • •Fix it with with_structured_output() and a Pydantic schema.
3. •

   Ignoring compliance constraints

   • •A useful alert is not enough if you cannot explain it during an audit.
   • •Fix it by storing prompt versions, evidence fields, and decision traces with every case.
4. •

   Treating all borrowers the same

   • •Lending risk signals vary by product type and customer segment.
   • •Fix it by tuning rules and prompts per portfolio instead of one global threshold.

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.