Building a reliable AI agent banking compliance monitoring system means pairing hard-coded regulatory filters with language-model reasoning for messy free-text fields. This tutorial walks through a runnable Python agent that ingests transactions, applies deterministic rules, and calls an LLM only when needed to flag policy violations.
Prerequisites
- Python 3.11 or newer
pip install openai pydantic- An API key for an OpenAI-compatible inference gateway. We point the client at n4n.ai’s endpoint to get automatic fallback across providers and per-token usage metering.
- No external dataset required; we synthesize sample transactions inline.
Data model and policy
Start with explicit structures. A compliance policy is not a prompt—it is a versioned config.
from pydantic import BaseModel
from enum import Enum
from datetime import datetime
class TxType(str, Enum):
WIRE = "wire"
ACH = "ach"
CARD = "card"
class Transaction(BaseModel):
id: str
amount: float
currency: str
counterparty_country: str
timestamp: datetime
type: TxType
class CompliancePolicy(BaseModel):
max_wire_without_review: float = 10000.0
sanctioned_countries: set[str] = {"IR", "KP", "SY", "RU"} # illustrative
sensitive_keywords: list[str] = ["cash", "crypto", "offshore"]The sanctioned list above is an example only. In production, load it from your regulator feed.
Deterministic pre-filters
Run cheap, exact checks first. They cut LLM spend and latency to near zero for clear violations.
def deterministic_checks(tx: Transaction, policy: CompliancePolicy) -> list[str]:
violations: list[str] = []
if tx.type == TxType.WIRE and tx.amount > policy.max_wire_without_review:
violations.append(f"Wire exceeds {policy.max_wire_without_review} threshold")
if tx.counterparty_country in policy.sanctioned_countries:
violations.append(f"Counterparty in sanctioned country {tx.counterparty_country}")
desc = tx.description.lower()
for kw in policy.sensitive_keywords:
if kw in desc:
violations.append(f"Description contains sensitive keyword '{kw}'")
return violationsIf this returns anything, we never call the model.
LLM semantic screening
Some descriptions evade keyword matches but still signal structuring or layering. The AI agent banking compliance monitoring logic uses an LLM as a second opinion, not a first line.
from openai import OpenAI
import json
client = OpenAI(
base_url="https://api.n4n.ai/v1", # OpenAI-compatible, 240+ models
api_key="YOUR_KEY",
)
MODEL = "gpt-4o-mini"
def llm_screen(tx: Transaction, policy: CompliancePolicy) -> dict:
prompt = f"""You are a compliance officer. Decide if the transaction likely
violates AML or fraud policy. Sensitive keywords: {policy.sensitive_keywords}.
Return JSON only: {{"flag": bool, "reason": str}}
Transaction: {tx.model_dump_json()}"""
resp = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
max_tokens=150,
)
return json.loads(resp.choices[0].message.content)Expected LLM payload for a suspicious but keyword-clean description:
{"flag": true, "reason": "Description indicates splitting payments to evade reporting thresholds"}Agent orchestration
Combine both layers. The agent returns a structured verdict.
def monitor_transaction(tx: Transaction, policy: CompliancePolicy) -> dict:
det = deterministic_checks(tx, policy)
if det:
return {"tx_id": tx.id, "status": "blocked", "reasons": det}
llm = llm_screen(tx, policy)
if llm.get("flag"):
return {"tx_id": tx.id, "status": "review", "reasons": [llm.get("reason")]}
return {"tx_id": tx.id, "status": "approved", "reasons": []}Why deterministic first
The AI agent banking compliance monitoring pipeline must be cost-predictable. Deterministically blocking 80% of bad transactions before any model call keeps per-token metering bounded and reduces tail latency.
Run on sample data
from datetime import datetime
policy = CompliancePolicy()
samples = [
Transaction(id="t1", amount=5000, currency="USD", counterparty_country="US",
description="Payroll deposit", timestamp=datetime.now(), type=TxType.ACH),
Transaction(id="t2", amount=15000, currency="USD", counterparty_country="GB",
description="Invoice payment", timestamp=datetime.now(), type=TxType.WIRE),
Transaction(id="t3", amount=300, currency="USD", counterparty_country="IR",
description="Gift", timestamp=datetime.now(), type=TxType.CARD),
Transaction(id="t4", amount=800, currency="USD", counterparty_country="US",
description="Transfer to offshore crypto exchange", timestamp=datetime.now(), type=TxType.WIRE),
Transaction(id="t5", amount=900, currency="USD", counterparty_country="US",
description="Split payment into multiple small amounts to avoid limits",
timestamp=datetime.now(), type=TxType.WIRE),
]
for tx in samples:
print(json.dumps(monitor_transaction(tx, policy)))Checkpoint output:
{"tx_id": "t1", "status": "approved", "reasons": []}
{"tx_id": "t2", "status": "blocked", "reasons": ["Wire exceeds 10000.0 threshold"]}
{"tx_id": "t3", "status": "blocked", "reasons": ["Counterparty in sanctioned country IR"]}
{"tx_id": "t4", "status": "blocked", "reasons": ["Description contains sensitive keyword 'offshore'", "Description contains sensitive keyword 'crypto'"]}
{"tx_id": "t5", "status": "review", "reasons": ["Description indicates splitting payments to evade reporting thresholds"]}t5 is the only case that reaches LLM review. That is the point.
Async batch processing
Production volumes demand concurrency. Wrap the screen in asyncio and use the async client.
import asyncio
from openai import AsyncOpenAI
aclient = AsyncOpenAI(base_url="https://api.n4n.ai/v1", api_key="YOUR_KEY")
async def llm_screen_async(tx, policy):
# same prompt construction as above
resp = await aclient.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
max_tokens=150,
)
return json.loads(resp.choices[0].message.content)
async def monitor_batch(txs, policy):
tasks = [asyncio.create_task(_wrap(tx, policy)) for tx in txs]
return await asyncio.gather(*tasks)
async def _wrap(tx, policy):
det = deterministic_checks(tx, policy)
if det:
return {"tx_id": tx.id, "status": "blocked", "reasons": det}
llm = await llm_screen_async(tx, policy)
if llm.get("flag"):
return {"tx_id": tx.id, "status": "review", "reasons": [llm.get("reason")]}
return {"tx_id": tx.id, "status": "approved", "reasons": []}Audit logging
Every verdict must be replayable. Append to JSONL:
def log_verdict(verdict: dict, path: str = "audit.log.jsonl"):
with open(path, "a") as f:
f.write(json.dumps(verdict) + "\n")For regulated environments, also store the raw LLM response and the model ID. Because the gateway honors client routing directives and forwards provider cache-control hints, you can pin a specific model version and cache policy prompts across calls to keep records consistent.
Testing the agent
Unit-test the deterministic layer at minimum.
def test_deterministic_sanctioned():
p = CompliancePolicy()
tx = Transaction(id="x", amount=10, currency="USD", counterparty_country="KP",
description="test", timestamp=datetime.now(), type=TxType.CARD)
assert deterministic_checks(tx, p) == ["Counterparty in sanctioned country KP"]The LLM layer should be tested with recorded fixtures or a mock client to avoid nondeterminism in CI.
Deployment notes
- Run the agent as a sidecar to your payment service, not inline on the critical path.
- Cache policy prompts; they change weekly, not per request.
- Alert on
reviewstatus to a human queue; never auto-block on LLM output alone. - Meter cost per token via the gateway dashboard to spot prompt drift.
A solid AI agent banking compliance monitoring design keeps the model constrained, logged, and secondary to law. Ship the deterministic core first, then tune the semantic layer against real false-positive rates.