Introduction
In this eighth article of the AI Agents series, we focus on the Fraud Detection & Prevention Agent, a vital agent for any business that handles sensitive data, payments, or user transactions. This agent’s main job is to automatically flag fraudulent activities, detect suspicious patterns, and prevent financial losses. It leverages advanced machine learning techniques, as well as established policies, to monitor transactions in real-time, alerting teams or blocking potentially fraudulent transactions. Just like the previous agents, this system never claims success unless all actions are fully executed and receipts are generated.
The Use Case
In financial sectors, e-commerce, and other data-sensitive industries, fraud detection is a critical aspect of the business. The agent examines transactions based on certain patterns, such as unusual spending behavior, location inconsistencies, or account activity that deviates from the norm. It integrates with payment gateways, banks, and other third-party services to detect irregularities in real-time and alert the relevant teams or automatically block fraudulent transactions. By using machine learning and predefined rules, the agent can prevent significant losses before they occur.
Prompt Contract (agent interface)
# file: contracts/fraud_detection_v1.yaml
role: "FraudDetectionAgent"
scope: >
Detect fraudulent activities by analyzing transaction data, customer behavior, and account anomalies.
Ask once for missing fields (transaction_id, customer_id, amount, location, transaction_type).
Propose tool calls; never assert success without a receipt.
output:
type: object
required: [summary, decision, fraud_alert, citations, next_steps, tool_proposals]
properties:
summary: {type: string, maxWords: 100}
decision: {type: string, enum: ["approve","block","need_approval","need_more_info"]}
fraud_alert:
type: object
required: [transaction_id, customer_id, amount, risk_level, flagged_reason]
properties:
transaction_id: {type: string}
customer_id: {type: string}
amount: {type: integer}
risk_level: {type: string, enum: ["low", "medium", "high"]}
flagged_reason: {type: string}
citations: {type: array, items: {type: string}}
next_steps: {type: array, items: {type: string}, maxItems: 6}
tool_proposals:
type: array
items:
type: object
required: [name, args, preconditions, idempotency_key]
properties:
name: {type: string, enum: [AnalyzeTransaction, CheckAnomalyPatterns, BlockTransaction, RequestApproval]}
args: {type: object}
preconditions: {type: string}
idempotency_key: {type: string}
policy_id: "fraud_detection_policy.v4"
citation_rule: "1–2 minimal-span claim_ids per factual sentence"
decoding:
narrative: {top_p: 0.92, temperature: 0.72, stop: ["\n\n## "]}
bullets: {top_p: 0.82, temperature: 0.45}
Example claims (context provided to the model)
[
{"claim_id":"policy:fraud:location_mismatch","text":"Transactions from new locations or IP addresses need further scrutiny.",
"effective_date":"2025-01-01","source_id":"doc:fraud_detection_policy_v4","span":"new locations or IP addresses"},
{"claim_id":"policy:fraud:high_amount","text":"Transactions exceeding $500 should be flagged for review if the user hasn't made similar payments in the past.",
"effective_date":"2025-01-01","source_id":"doc:fraud_detection_policy_v4","span":"exceeding $500 should be flagged"},
{"claim_id":"policy:fraud:unusual_behavior","text":"Transactions with frequent logins from different devices or IP addresses must be flagged.",
"effective_date":"2025-01-01","source_id":"doc:fraud_detection_policy_v4","span":"frequent logins from different devices"}
]
Tool Interfaces (typed, with receipts)
# tools.py
from pydantic import BaseModel
from typing import Optional, List, Dict
from datetime import datetime
class AnalyzeTransactionArgs(BaseModel):
transaction_id: str
customer_id: str
amount: int
location: str
transaction_type: str
class CheckAnomalyPatternsArgs(BaseModel):
customer_id: str
transaction_history: List[Dict[str, str]] # e.g., [{amount: int, location: str, transaction_type: str}]
class BlockTransactionArgs(BaseModel):
transaction_id: str
customer_id: str
amount: int
flagged_reason: str
class RequestApprovalArgs(BaseModel):
transaction_id: str
approver_id: str
class ToolReceipt(BaseModel):
tool: str
ok: bool
ref: str
message: str = ""
data: Optional[Dict] = None
# adapters.py (demo logic)
from tools import *
from datetime import datetime
TRANSACTIONS = {
"trans-001": {"customer_id": "C001", "amount": 1500, "location": "New York", "transaction_type": "purchase", "approved": False},
"trans-002": {"customer_id": "C002", "amount": 100, "location": "California", "transaction_type": "transfer", "approved": False}
}
def analyze_transaction(a: AnalyzeTransactionArgs) -> ToolReceipt:
if a.amount > 1000:
risk_level = "high"
flagged_reason = "High transaction amount"
else:
risk_level = "low"
flagged_reason = "Normal transaction"
return ToolReceipt(tool="AnalyzeTransaction", ok=True, ref=f"trans-{a.transaction_id}",
message="Transaction analyzed", data={"risk_level": risk_level, "flagged_reason": flagged_reason})
def check_anomaly_patterns(a: CheckAnomalyPatternsArgs) -> ToolReceipt:
if len(a.transaction_history) > 5: # Simple anomaly detection logic
return ToolReceipt(tool="CheckAnomalyPatterns", ok=False, ref="pattern-mismatch",
message="Customer has frequent logins from different locations", data={"flagged": True})
return ToolReceipt(tool="CheckAnomalyPatterns", ok=True, ref="pattern-ok", message="No anomalies detected")
def block_transaction(a: BlockTransactionArgs) -> ToolReceipt:
TRANSACTIONS[a.transaction_id]["approved"] = False
return ToolReceipt(tool="BlockTransaction", ok=True, ref=f"block-{a.transaction_id}",
message=f"Transaction {a.transaction_id} blocked", data={"reason": a.flagged_reason})
def request_approval(a: RequestApprovalArgs) -> ToolReceipt:
return ToolReceipt(tool="RequestApproval", ok=True, ref=f"approval-{a.transaction_id}",
message=f"Approval requested for transaction {a.transaction_id}", data={"approver_id": a.approver_id})
Agent Loop (proposal → verification → execution → receipts)
# agent_fraud_detection.py
import uuid, json
from typing import Any, Dict, List
from tools import *
from adapters import *
ALLOWED_TOOLS = {"AnalyzeTransaction", "CheckAnomalyPatterns", "BlockTransaction", "RequestApproval"}
def new_idem() -> str:
return f"idem-{uuid.uuid4()}"
def verify_proposal(p: Dict[str, Any]) -> str:
required = {"name","args","preconditions","idempotency_key"}
if not required.issubset(p): return "Missing proposal fields"
if p["name"] not in ALLOWED_TOOLS: return "Tool not allowed"
return ""
def execute(p: Dict[str, Any]) -> ToolReceipt:
n, a = p["name"], p["args"]
if n == "AnalyzeTransaction": return analyze_transaction(AnalyzeTransactionArgs(**a))
if n == "CheckAnomalyPatterns": return check_anomaly_patterns(CheckAnomalyPatternsArgs(**a))
if n == "BlockTransaction": return block_transaction(BlockTransactionArgs(**a))
if n == "RequestApproval": return request_approval(RequestApprovalArgs(**a))
return ToolReceipt(tool=n, ok=False, ref="none", message="Unknown tool")
# --- Model shim returning a plan per contract (replace with your LLM call) ---
def call_model(contract_yaml: str, claims: List[Dict[str,Any]], transaction_data: Dict[str,Any]) -> Dict[str,Any]:
decision = "approve" if transaction_data["amount"] <= 1000 else "block"
return {
"summary": f"Transaction for {transaction_data['customer_id']} reviewed for fraud.",
"decision": decision,
"fraud_alert": {
"transaction_id": transaction_data["transaction_id"],
"customer_id": transaction_data["customer_id"],
"amount": transaction_data["amount"],
"risk_level": "high" if decision == "block" else "low",
"flagged_reason": "High transaction amount" if decision == "block" else "Normal transaction"
},
"citations": ["policy:fraud:location_mismatch", "policy:fraud:high_amount", "policy:fraud:unusual_behavior"],
"next_steps": ["Check transaction patterns", "Request approval for blocked transactions", "Block or approve transaction"],
"tool_proposals": [
{"name":"AnalyzeTransaction","args":{"transaction_id":transaction_data["transaction_id"],"customer_id":transaction_data["customer_id"],
"amount":transaction_data["amount"],"location":transaction_data["location"],
"transaction_type":transaction_data["transaction_type"]},
"preconditions":"Analyze transaction for fraud risk.","idempotency_key": new_idem()},
{"name":"CheckAnomalyPatterns","args":{"customer_id":transaction_data["customer_id"],
"transaction_history":[{"amount":transaction_data["amount"], "location":transaction_data["location"], "transaction_type":transaction_data["transaction_type"}]},
"preconditions":"Check for unusual transaction patterns.","idempotency_key": new_idem()},
{"name":"BlockTransaction","args":{"transaction_id":transaction_data["transaction_id"],"customer_id":transaction_data["customer_id"],
"amount":transaction_data["amount"],"flagged_reason":"High transaction amount"},
"preconditions":"Block transaction if fraud risk is high.","idempotency_key": new_idem()},
{"name":"RequestApproval","args":{"transaction_id":transaction_data["transaction_id"],"approver_id":"M002"},
"preconditions":"Request manager approval if the transaction is blocked.","idempotency_key": new_idem()}
]
}
def render_response(model_json: Dict[str,Any], receipts: List[ToolReceipt]) -> str:
idx = {r.tool:r for r in receipts}
lines = [model_json["summary"], ""]
lines.append(f"Decision: {model_json['decision']}")
lines.append(f"Fraud Alert: {model_json['fraud_alert']}")
lines.append("")
lines.append("Next steps:")
for s in model_json["next_steps"]:
lines.append(f"• {s}")
if idx.get("BlockTransaction") and idx["BlockTransaction"].ok:
lines.append(f"\nTransaction blocked: {idx['BlockTransaction'].message}")
if idx.get("RequestApproval") and idx["RequestApproval"].ok:
lines.append(f"Approval requested: {idx['RequestApproval'].message}")
lines.append("\nCitations: " + ", ".join(model_json["citations"]))
return "\n".join(lines)
def handle(transaction_data: Dict[str,Any]) -> str:
contract = open("contracts/fraud_detection_v1.yaml").read()
claims: List[Dict[str,Any]] = [] # load real claims from policy
plan = call_model(contract, claims, transaction_data)
receipts: List[ToolReceipt] = []
for prop in plan["tool_proposals"]:
reason = verify_proposal(prop)
if reason:
receipts.append(ToolReceipt(tool=prop["name"], ok=False, ref="blocked", message=reason)); continue
rec = execute(prop)
receipts.append(rec)
if not rec.ok and prop["name"] in {"BlockTransaction"}:
break
return render_response(plan, receipts)
if __name__ == "__main__":
example_transaction_data = {
"transaction_id": "trans-001",
"customer_id": "C001",
"amount": 1500,
"location": "New York",
"transaction_type": "purchase"
}
print(handle(example_transaction_data))
The Prompt You’d Send to the Model (concise and testable)
System:
You are FraudDetectionAgent. Follow the contract:
- Ask once if transaction_id, customer_id, amount, location, or transaction_type is missing.
- Cite 1–2 claim_ids per factual sentence using provided claims.
- Propose tools; never assert success without a receipt.
- Output JSON with keys: summary, decision, fraud_alert, citations[], next_steps[], tool_proposals[].
Claims (eligible only):
[ ... JSON array of fraud policy claims like above ... ]
User:
Review this transaction for fraud: {"transaction_id":"trans-001","customer_id":"C001","amount":1500,"location":"New York","transaction_type":"purchase"}
How to adapt quickly
Integrate with your fraud detection systems, bank transaction gateways, and AML (Anti-Money Laundering) systems to detect anomalies in transactions. Implement idempotency and transactional integrity to ensure no accidental double-blocks. Load claims from fraud prevention policies related to payment types, geolocation checks, and transaction patterns. Add a validation layer to ensure compliance with regulatory guidelines such as PCI-DSS or GDPR. Deploy the contract, policy bundle, and decoder settings behind a feature flag for testing, canary deployments, and rollback support.