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

A transaction monitoring agent watches payment and trade activity, scores each event against risk rules and model outputs, and escalates suspicious cases for human review. In investment banking, that matters because you need to catch AML, sanctions, market abuse, and unusual flow patterns fast enough to reduce regulatory exposure without flooding analysts with false positives.

Architecture

A production transaction monitoring agent in investment banking usually needs these components:

• •

  Ingestion layer

  • •Pulls transactions from Kafka, SFTP drops, REST APIs, or internal ledgers.
  • •Normalizes payloads into a common schema before any analysis.
• •

  Risk enrichment layer

  • •Adds customer KYC data, counterparty risk, jurisdiction, historical behavior, and sanctions hits.
  • •This is where you make the agent useful instead of just pattern-matching raw transfers.
• •

  Policy/rules engine

  • •Applies deterministic controls like thresholds, velocity checks, watchlist matches, and country restrictions.
  • •Banking teams still need explicit rule traces for audit.
• •

  LLM reasoning node

  • •Summarizes why a transaction is suspicious or benign using structured evidence only.
  • •Use it for triage and narrative generation, not as the final decision-maker.
• •

  Case management node

  • •Creates an alert object with reason codes, evidence links, and analyst-ready notes.
  • •Pushes to an internal queue or case platform.
• •

  Audit and observability layer

  • •Logs every state transition, prompt input/output, and decision path.
  • •Required for model governance and regulator review.

Implementation

1) Define the state and the graph nodes

LangGraph works best when you keep the state explicit. For transaction monitoring, store the transaction payload, enrichment data, rule results, model output, and final disposition in a typed state object.

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
import operator

class TxState(TypedDict):
    transaction: dict
    enrichment: dict
    rule_flags: list[str]
    risk_score: int
    llm_summary: str
    disposition: str
    audit_log: Annotated[list[str], operator.add]

def enrich_transaction(state: TxState) -> TxState:
    tx = state["transaction"]
    enrichment = {
        "customer_segment": "institutional",
        "country_risk": "high" if tx["country"] in ["IR", "RU"] else "normal",
        "sanctions_hit": tx["counterparty"] in {"ACME_FRONTIER_LTD"},
    }
    return {
        **state,
        "enrichment": enrichment,
        "audit_log": [f"Enriched tx {tx['id']}"],
    }

def apply_rules(state: TxState) -> TxState:
    tx = state["transaction"]
    enr = state["enrichment"]
    flags = []
    score = 0

    if tx["amount"] > 1000000:
        flags.append("HIGH_VALUE")
        score += 30
    if enr["country_risk"] == "high":
        flags.append("HIGH_COUNTRY_RISK")
        score += 40
    if enr["sanctions_hit"]:
        flags.append("SANCTIONS_MATCH")
        score += 100

    return {
        **state,
        "rule_flags": flags,
        "risk_score": score,
        "audit_log": [f"Rules applied: {flags}"],
    }

2) Add an LLM triage node with structured evidence

The LLM should not invent facts. Feed it only the transaction fields plus rule outputs and ask for a short analyst-facing summary. In regulated environments, this keeps explanations grounded in evidence.

from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI

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

def summarize_risk(state: TxState) -> TxState:
    tx = state["transaction"]
    prompt = [
        SystemMessage(content=(
            "You are a transaction monitoring analyst. "
            "Summarize suspicious indicators using only provided facts. "
            "Do not speculate."
        )),
        HumanMessage(content=f"""
Transaction:
{tx}

Enrichment:
{state['enrichment']}

Rule flags:
{state['rule_flags']}

Risk score:
{state['risk_score']}
""")
    ]
    resp = llm.invoke(prompt)
    return {
        **state,
        "llm_summary": resp.content,
        "audit_log": ["LLM summary generated"],
    }

def decide_case(state: TxState) -> TxState:
    if state["risk_score"] >= 100:
        disposition = "ESCALATE_IMMEDIATELY"
    elif state["risk_score"] >= 50:
        disposition = "REVIEW"
    else:
        disposition = "CLEAR"

    return {
        **state,
        "disposition": disposition,
        "audit_log": [f"Disposition set to {disposition}"],
    }

3) Wire the workflow with StateGraph

This is the actual LangGraph pattern: define nodes, connect them with edges, compile once, then invoke per transaction.

workflow = StateGraph(TxState)

workflow.add_node("enrich", enrich_transaction)
workflow.add_node("rules", apply_rules)
workflow.add_node("summarize", summarize_risk)
workflow.add_node("decide", decide_case)

workflow.add_edge(START, "enrich")
workflow.add_edge("enrich", "rules")
workflow.add_edge("rules", "summarize")
workflow.add_edge("summarize", "decide")
workflow.add_edge("decide", END)

graph = workflow.compile()

sample_tx = {
    "id": "TX-9001",
    "amount": 2500000,
    "country": "RU",
    "counterparty": "ACME_FRONTIER_LTD",
}

result = graph.invoke({
    "transaction": sample_tx,
    "enrichment": {},
    "rule_flags": [],
    "risk_score": 0,
    "llm_summary": "",
    "disposition": "",
    "audit_log": [],
})

print(result["disposition"])
print(result["llm_summary"])
print(result["audit_log"])

4) Add branching for analyst escalation

If you want analyst review only on high-risk cases, use add_conditional_edges. That keeps your graph explicit and easier to audit than burying branching logic inside one giant function.

def route_case(state: TxState) -> str:
    return "escalate" if state["disposition"] != "CLEAR" else END

def escalate_to_case_management(state: TxState) -> TxState:
    return {
        **state,
        "audit_log": [f"Case created for {state['transaction']['id']}"],
    }

workflow2 = StateGraph(TxState)
workflow2.add_node("enrich", enrich_transaction)
workflow2.add_node("rules", apply_rules)
workflow2.add_node("summarize", summarize_risk)
workflow2.add_node("decide", decide_case)
workflow2.add_node("escalate", escalate_to_case_management)

workflow2.add_edge(START, "enrich")
workflow2.add_edge("enrich", "rules")
workflow2.add_edge("rules", "summarize")
workflow2.add_edge("summarize", "decide")
workflow2.add_conditional_edges("decide", route_case)
workflow2.add_edge("escalate", END)

Production Considerations

• •

  Keep PII and trading data inside your residency boundary

  • •If your bank requires EU-only processing or on-prem deployment, run the model endpoint in-region.
  • •Do not send raw account numbers or client identifiers to external APIs unless policy explicitly allows it.
• •

  Log every decision path

  • •Persist input features, rule hits, LLM prompt version, model response hash, and final disposition.
  • •Auditors will ask why a case was escalated months later.
• •

  Use guardrails around the LLM

  • •Treat it as a summarizer/triager.
  • •Hard-code deterministic thresholds for escalation; never let the model override sanctions or AML rules by itself.
• •

  Monitor false positives by desk and corridor

  • •A good alert rate globally can still be bad for a specific trading desk or jurisdiction.
  • •Track precision by product line: FX spot, derivatives margin flows, securities settlement, correspondent banking.

Common Pitfalls

1. •

   Letting the LLM make final compliance decisions

   • •Avoid this by making decide_case() deterministic.
   • •The model should explain; rules should decide.
2. •

   Using unstructured prompts without evidence boundaries

   • •Don’t pass free-form context dumps.
   • •Pass normalized fields and explicit rule outputs so explanations stay auditable.
3. •

   Skipping replayability

   • •If you cannot replay a case with the same graph version and inputs, your audit trail is weak.
   • •Version prompts, rulesets, enrichment sources, and graph code together.

A transaction monitoring agent built this way gives you clear control points: deterministic policy enforcement where regulation demands it, LLM assistance where analysts need speed. That’s the right split for investment banking systems that have to survive both production load and compliance review.

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.