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

A transaction monitoring agent flags suspicious activity in trade and payment flows, then routes the right cases to compliance with enough context to act fast. In investment banking, that matters because false negatives become regulatory risk, and false positives burn analyst time across large volumes of high-value transactions.

Architecture

• •

  Transaction ingestion layer

  • •Pulls trades, payments, counterparty metadata, and reference data from internal systems.
  • •Normalizes records into a single schema before the agent sees them.
• •

  Risk rules engine

  • •Applies deterministic checks first: sanctions hits, threshold breaches, rapid movement patterns, unusual counterparties.
  • •Keeps obvious cases out of the LLM path.
• •

  LangChain reasoning layer

  • •Uses an LLM to summarize why a transaction looks normal or suspicious.
  • •Produces structured outputs for case management, not free-form prose.
• •

  Evidence retrieval layer

  • •Fetches policy docs, AML typologies, prior alerts, and customer KYC notes.
  • •Grounds the agent in bank-specific context using RetrievalQA-style retrieval or a retriever tool.
• •

  Case management output

  • •Writes alert payloads to downstream systems with severity, rationale, and supporting evidence.
  • •Preserves auditability for model review and compliance sign-off.
• •

  Human review gate

  • •Keeps final disposition with analysts for escalations.
  • •The agent recommends; it does not clear regulated activity on its own.

Implementation

1. •Define the transaction schema and deterministic pre-checks

Start with structured input. In banking, you want typed fields because downstream controls depend on them.

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

class Transaction(BaseModel):
    transaction_id: str
    account_id: str
    counterparty: str
    amount_usd: float
    currency: str
    country: str
    timestamp: str
    channel: str
    description: Optional[str] = None
    flags: List[str] = Field(default_factory=list)

def rule_based_checks(tx: Transaction) -> List[str]:
    flags = []
    if tx.amount_usd >= 1000000:
        flags.append("HIGH_VALUE")
    if tx.country in {"IR", "KP", "SY"}:
        flags.append("SANCTIONS_JURISDICTION")
    if "cash" in (tx.description or "").lower():
        flags.append("CASH_KEYWORD")
    return flags

2. •Load policy documents into a retriever

Use LangChain’s document loaders and vector store so the agent can cite internal policy instead of guessing.

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

loader = TextLoader("bank_aml_policy.txt", encoding="utf-8")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=120)
chunks = splitter.split_documents(docs)

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

3. •Build the LangChain prompt and structured output chain

For production alerts, use ChatPromptTemplate and PydanticOutputParser. That gives you stable JSON-like output that can be stored in an audit trail.

from pydantic import BaseModel, Field
from typing import Literal, List
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

class AlertDecision(BaseModel):
    decision: Literal["clear", "review", "escalate"]
    risk_score: int = Field(ge=0, le=100)
    rationale: str
    evidence: List[str]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(AlertDecision)

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "You are a transaction monitoring analyst for an investment bank. "
     "Use only provided policy context and transaction data. "
     "Never invent facts. Return conservative decisions when evidence is incomplete."),
    ("human",
     "Transaction:\n{transaction}\n\nPolicy context:\n{policy_context}\n\n"
     "Assess whether this should be cleared, reviewed, or escalated.")
])

def build_context(tx):
    docs = retriever.invoke(f"{tx.country} {tx.channel} {tx.amount_usd} suspicious activity")
    policy_context = "\n\n".join(d.page_content for d in docs)
    return policy_context

def analyze_transaction(tx):
    rules = rule_based_checks(tx)
    policy_context = build_context(tx)
    result = (prompt | structured_llm).invoke({
        "transaction": tx.model_dump_json(indent=2),
        "policy_context": policy_context + f"\n\nDeterministic flags: {rules}"
    })
    return result

4. •Run the agent and persist an auditable case record

The output should include both machine-readable fields and enough detail for compliance review.

sample_tx = Transaction(
    transaction_id="TXN-88421",
    account_id="ACCT-9912",
    counterparty="Omega Trading Ltd",
    amount_usd=1250000,
    currency="USD",
    country="AE",
    timestamp="2026-04-21T10:15:00Z",
    channel="wire",
    description="consulting settlement"
)

alert = analyze_transaction(sample_tx)

case_record = {
    "transaction_id": sample_tx.transaction_id,
    "flags": rule_based_checks(sample_tx),
    "decision": alert.decision,
    "risk_score": alert.risk_score,
    "rationale": alert.rationale,
    "evidence": alert.evidence,
}

print(case_record)

Production Considerations

• •

  Keep deterministic controls ahead of the LLM

  • •Sanctions screening, thresholding, velocity checks, and watchlist matches should run first.
  • •Use the model for explanation and triage, not primary detection.
• •

  Log every prompt and response with immutable retention

  • •Store input payloads, retrieved documents IDs, model version, timestamps, and final decisions.
  • •Compliance teams will ask why a case was escalated months later.
• •

  Respect data residency and confidentiality

  • •Keep customer data inside approved regions.
  • •If you use hosted models, confirm where prompts are processed and whether logs are retained outside your control.
• •

  Add guardrails for analyst trust

  • •Force structured outputs.
  • •Reject responses that contain unsupported claims or missing evidence references.
  • •Route low-confidence cases to human review automatically.

Common Pitfalls

• •

  Letting the LLM decide without rules

  • •This creates inconsistent outcomes and weakens control coverage.
  • •Fix it by using a rules-first pipeline with the LLM only as a second-stage reviewer.
• •

  Using ungrounded prompts

  • •If the model does not see policy text or prior typologies, it will hallucinate rationales.
  • •Fix it with retrieval over approved internal documents and strict instructions to cite only provided context.
• •

  Skipping audit-friendly output design

  • •Free-form text is hard to store in case systems and hard to defend in reviews.
  • •Fix it with with_structured_output, stable schemas like Pydantic models, and explicit evidence fields.

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.