AI Agent for Compliance: Automate Risk Assessment, Policy Monitoring & Regulatory Reporting

Compliance teams at regulated institutions are drowning. The average financial firm tracks over 200 regulatory bodies, processes thousands of rule changes annually, and spends 60-70% of compliance budgets on manual monitoring and reporting tasks. Meanwhile, regulators keep raising the bar—fines for non-compliance exceeded $6.6 billion globally in 2025, and enforcement actions are accelerating.

AI agents are transforming compliance from a reactive, labor-intensive function into a proactive, automated system. Unlike traditional GRC platforms that require constant human input, an AI compliance agent can continuously monitor regulatory changes, score risks in real time, manage policy lifecycles, detect suspicious transactions, and generate audit-ready reports—all with minimal human oversight.

This guide covers six core compliance capabilities you can build with AI agents, complete with Python code, architecture decisions, and a detailed ROI breakdown for a 500-employee financial institution.

1. Regulatory Change Management

Keeping up with regulatory changes is one of the most time-consuming tasks in compliance. The Federal Register publishes 70,000+ pages per year. The EU Official Journal, SEC filings, FINRA notices, OCC bulletins—the volume is staggering. A single missed rule change can result in millions in fines and months of remediation work.

An AI agent can automate this entire pipeline: ingest regulatory feeds, assess impact against your current policies and controls, identify gaps, and maintain a regulatory calendar with upcoming deadlines.

Regulation Tracking

The first step is building a multi-source regulatory feed ingester. You want to pull from the Federal Register API, EU Official Journal RSS, SEC EDGAR filings, and industry-specific sources. The agent parses each update and classifies it by jurisdiction, topic, and affected business lines.

import feedparser
import requests
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class RegulatoryUpdate:
    source: str
    title: str
    summary: str
    published: datetime
    jurisdiction: str
    url: str
    category: Optional[str] = None
    impact_score: float = 0.0
    affected_policies: list = field(default_factory=list)


class RegulatoryChangeTracker:
    """Tracks regulatory changes across multiple sources."""

    SOURCES = {
        "federal_register": {
            "url": "https://www.federalregister.gov/api/v1/documents.json",
            "type": "api",
            "jurisdiction": "US-Federal",
        },
        "eu_official_journal": {
            "url": "https://eur-lex.europa.eu/rss/search-result.xml",
            "type": "rss",
            "jurisdiction": "EU",
        },
        "sec_filings": {
            "url": "https://efts.sec.gov/LATEST/search-index?q=rule&dateRange=custom",
            "type": "api",
            "jurisdiction": "US-SEC",
        },
        "occ_bulletins": {
            "url": "https://www.occ.gov/topics/laws-and-regulations/bulletins/rss.xml",
            "type": "rss",
            "jurisdiction": "US-OCC",
        },
    }

    def __init__(self, llm_client, policy_store):
        self.llm = llm_client
        self.policy_store = policy_store
        self.tracked_updates = []

    def fetch_all_sources(self, lookback_days=7):
        updates = []
        cutoff = datetime.now() - timedelta(days=lookback_days)

        for name, source in self.SOURCES.items():
            if source["type"] == "rss":
                updates.extend(self._fetch_rss(name, source, cutoff))
            elif source["type"] == "api":
                updates.extend(self._fetch_api(name, source, cutoff))

        return sorted(updates, key=lambda u: u.published, reverse=True)

    def _fetch_rss(self, name, source, cutoff):
        feed = feedparser.parse(source["url"])
        updates = []
        for entry in feed.entries:
            pub_date = datetime(*entry.published_parsed[:6])
            if pub_date >= cutoff:
                updates.append(RegulatoryUpdate(
                    source=name,
                    title=entry.title,
                    summary=entry.get("summary", ""),
                    published=pub_date,
                    jurisdiction=source["jurisdiction"],
                    url=entry.link,
                ))
        return updates

    def _fetch_api(self, name, source, cutoff):
        resp = requests.get(source["url"], params={
            "conditions[publication_date][gte]": cutoff.strftime("%Y-%m-%d"),
            "per_page": 50,
            "order": "newest",
        })
        results = resp.json().get("results", [])
        return [
            RegulatoryUpdate(
                source=name,
                title=r["title"],
                summary=r.get("abstract", ""),
                published=datetime.strptime(r["publication_date"], "%Y-%m-%d"),
                jurisdiction=source["jurisdiction"],
                url=r["html_url"],
            )
            for r in results
        ]

    def assess_impact(self, update: RegulatoryUpdate) -> dict:
        """Use LLM to assess regulatory impact on current policies."""
        current_policies = self.policy_store.get_relevant_policies(
            update.summary, top_k=10
        )
        prompt = f"""Analyze this regulatory update and assess its impact:

REGULATION: {update.title}
SUMMARY: {update.summary}
JURISDICTION: {update.jurisdiction}

CURRENT POLICIES:
{chr(10).join(p.title for p in current_policies)}

Return JSON with:
- impact_level: "critical" | "high" | "medium" | "low" | "informational"
- affected_policies: list of policy IDs that need updating
- gap_analysis: what is NOT covered by current policies
- action_items: specific steps the compliance team must take
- deadline: estimated compliance deadline (if mentioned)
"""
        return self.llm.generate_json(prompt)

Impact Assessment & Gap Analysis

Raw tracking is only half the battle. The real value comes from mapping each regulatory change to your internal policy framework. The agent compares new requirements against existing controls using vector similarity search, then identifies gaps—areas where your current policies fall short of the new requirements.

For gap analysis, the agent produces a structured report: which policies need updating, what new controls must be implemented, and what the compliance deadline is. This transforms a task that typically takes a compliance analyst 4-8 hours per regulation into a 15-minute review.

Regulatory Calendar Management

The agent also maintains a forward-looking regulatory calendar. It tracks comment periods, effective dates, phase-in schedules, and filing deadlines across all jurisdictions. When a deadline approaches, it triggers alerts and escalation workflows automatically. No more missed comment periods or surprise effective dates.

2. Risk Assessment & Scoring

Traditional risk assessments happen quarterly or annually—a snapshot that is stale by the time the report is printed. AI agents enable continuous risk assessment that updates in real time as conditions change. Entity-level risk profiles, dynamic heat maps, emerging risk detection, and KRI monitoring all feed into a living risk picture.

Entity-Level Risk Profiling

Every entity in your organization—business units, products, geographies, counterparties—carries inherent risk. The agent calculates inherent risk based on the entity's characteristics, evaluates control effectiveness from audit findings and testing results, and derives residual risk as the net exposure after controls.

import numpy as np
from enum import Enum
from dataclasses import dataclass


class RiskLevel(Enum):
    CRITICAL = 5
    HIGH = 4
    MEDIUM = 3
    LOW = 2
    MINIMAL = 1


@dataclass
class RiskFactor:
    name: str
    weight: float
    score: float  # 1-5
    evidence: str


class ComplianceRiskAssessor:
    """Continuous risk assessment engine for compliance entities."""

    INHERENT_FACTORS = {
        "regulatory_complexity": 0.20,
        "transaction_volume": 0.15,
        "geographic_exposure": 0.15,
        "product_complexity": 0.15,
        "customer_risk_profile": 0.15,
        "historical_violations": 0.10,
        "third_party_dependencies": 0.10,
    }

    def __init__(self, llm_client, data_store):
        self.llm = llm_client
        self.data_store = data_store

    def compute_entity_risk(self, entity_id: str) -> dict:
        entity = self.data_store.get_entity(entity_id)

        # Step 1: Inherent risk scoring
        inherent_factors = self._score_inherent_risk(entity)
        inherent_score = sum(f.weight * f.score for f in inherent_factors)

        # Step 2: Control effectiveness
        controls = self.data_store.get_controls(entity_id)
        control_scores = self._evaluate_controls(controls)
        control_effectiveness = np.mean(control_scores) if control_scores else 0.5

        # Step 3: Residual risk = inherent * (1 - control_effectiveness)
        residual_score = inherent_score * (1 - control_effectiveness * 0.8)

        return {
            "entity_id": entity_id,
            "inherent_risk": round(inherent_score, 2),
            "control_effectiveness": round(control_effectiveness, 2),
            "residual_risk": round(residual_score, 2),
            "risk_level": self._classify_risk(residual_score),
            "factors": inherent_factors,
            "recommendations": self._generate_recommendations(
                inherent_factors, control_effectiveness, residual_score
            ),
        }

    def generate_risk_heatmap(self, entity_ids: list) -> dict:
        """Build likelihood x impact matrix across all entities."""
        matrix = {"critical": [], "high": [], "medium": [], "low": []}

        for eid in entity_ids:
            risk = self.compute_entity_risk(eid)
            likelihood = risk["inherent_risk"] / 5.0
            impact = self._estimate_impact(eid)

            entry = {
                "entity_id": eid,
                "likelihood": round(likelihood, 2),
                "impact": round(impact, 2),
                "combined": round(likelihood * impact, 2),
                "residual_risk": risk["residual_risk"],
            }

            if likelihood * impact > 0.75:
                matrix["critical"].append(entry)
            elif likelihood * impact > 0.50:
                matrix["high"].append(entry)
            elif likelihood * impact > 0.25:
                matrix["medium"].append(entry)
            else:
                matrix["low"].append(entry)

        return matrix

    def detect_emerging_risks(self, industry: str) -> list:
        """Scan news, social sentiment, and industry trends."""
        signals = self.data_store.get_risk_signals(industry, days=30)

        prompt = f"""Analyze these signals for emerging compliance risks:

INDUSTRY: {industry}
SIGNALS:
{chr(10).join(f"- [{s['source']}] {s['headline']}" for s in signals[:50])}

Identify emerging risks not yet in our risk register.
Return JSON list with: risk_name, description, likelihood (1-5),
potential_impact (1-5), affected_regulations, recommended_actions.
"""
        return self.llm.generate_json(prompt)

    def monitor_kris(self, entity_id: str) -> list:
        """Monitor Key Risk Indicators against thresholds."""
        kris = self.data_store.get_kri_definitions(entity_id)
        alerts = []

        for kri in kris:
            current = self.data_store.get_kri_value(kri["id"])
            threshold = kri["threshold"]
            trend = self.data_store.get_kri_trend(kri["id"], periods=6)

            if current > threshold:
                alerts.append({
                    "kri": kri["name"],
                    "current": current,
                    "threshold": threshold,
                    "breach_pct": round((current - threshold) / threshold * 100, 1),
                    "trend": "worsening" if trend[-1] > trend[0] else "improving",
                    "action": kri["escalation_action"],
                })

        return alerts

    def _score_inherent_risk(self, entity) -> list:
        factors = []
        for name, weight in self.INHERENT_FACTORS.items():
            score = self._compute_factor_score(entity, name)
            factors.append(RiskFactor(name=name, weight=weight,
                                      score=score, evidence=""))
        return factors

    def _classify_risk(self, score: float) -> str:
        if score >= 4.0: return "CRITICAL"
        if score >= 3.0: return "HIGH"
        if score >= 2.0: return "MEDIUM"
        if score >= 1.0: return "LOW"
        return "MINIMAL"

    def _evaluate_controls(self, controls) -> list:
        return [c.effectiveness_score for c in controls if c.last_tested]

    def _compute_factor_score(self, entity, factor_name) -> float:
        scoring_data = getattr(entity, factor_name, None)
        return min(5.0, max(1.0, float(scoring_data or 2.5)))

    def _estimate_impact(self, entity_id) -> float:
        entity = self.data_store.get_entity(entity_id)
        revenue_pct = entity.revenue_share / 100
        regulatory_severity = entity.max_penalty_exposure / 1_000_000
        return min(1.0, (revenue_pct * 0.5 + min(regulatory_severity, 1.0) * 0.5))

    def _generate_recommendations(self, factors, ctrl_eff, residual):
        weak_factors = [f for f in factors if f.score >= 4.0]
        recs = []
        for f in weak_factors:
            recs.append(f"Mitigate {f.name} (score: {f.score}/5)")
        if ctrl_eff < 0.6:
            recs.append("Control effectiveness below threshold - review control design")
        return recs

Risk Heat Maps & Emerging Risk Detection

The generate_risk_heatmap method builds a likelihood-times-impact matrix across all entities. This gives compliance leadership a single view of where risk concentrates. The agent updates this continuously—not just during quarterly reviews.

For emerging risks, the agent scans news feeds, social media sentiment, and industry trend reports. It identifies risks that are not yet in your risk register—the kind of blind spots that lead to surprise enforcement actions. Think of it as an early warning system that gives you months of lead time instead of days.

Key Risk Indicator Monitoring

KRIs are the vital signs of your compliance program. The agent monitors them against predefined thresholds, detects trends, and triggers escalation workflows when breaches occur. A rising complaint rate, increasing exception approvals, or declining training completion—the agent catches it before it becomes an audit finding.

3. Policy Lifecycle Management

Most organizations manage policies through a painful combination of Word documents, SharePoint sites, and email chains. Policies drift out of date, version control is unreliable, and nobody knows which employees have actually read and attested to which versions. AI agents can automate the entire policy lifecycle from creation through retirement.

Template-Based Policy Drafting

When a new regulation requires a new policy (or a significant update to an existing one), the agent drafts the first version. It pulls from policy templates, cross-references the specific regulation, maps to existing controls, and produces a draft that is typically 80-90% ready for human review. This cuts weeks of drafting time down to hours.

from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
import hashlib


class PolicyStatus(Enum):
    DRAFT = "draft"
    IN_REVIEW = "in_review"
    APPROVED = "approved"
    PUBLISHED = "published"
    UNDER_REVISION = "under_revision"
    RETIRED = "retired"


class PolicyLifecycleManager:
    """Manages the full lifecycle of compliance policies."""

    def __init__(self, llm_client, policy_store, regulation_store):
        self.llm = llm_client
        self.policies = policy_store
        self.regulations = regulation_store

    def draft_policy(self, regulation_id: str, template_id: str = None) -> dict:
        """AI-assisted policy drafting from regulatory requirements."""
        regulation = self.regulations.get(regulation_id)
        template = self.policies.get_template(template_id) if template_id else None
        existing = self.policies.find_related(regulation.keywords, top_k=5)

        prompt = f"""Draft a compliance policy for the following regulation:

REGULATION: {regulation.title}
REQUIREMENTS: {regulation.key_requirements}
JURISDICTION: {regulation.jurisdiction}
EFFECTIVE DATE: {regulation.effective_date}

EXISTING RELATED POLICIES:
{chr(10).join(f"- {p.title}: {p.summary}" for p in existing)}

{"TEMPLATE: " + template.content if template else ""}

Draft a policy that:
1. States purpose, scope, and applicability
2. Maps each regulatory requirement to a specific control
3. Defines roles and responsibilities (three lines of defense)
4. Includes monitoring and reporting requirements
5. Sets review frequency and triggers for ad-hoc review

Return structured JSON with: title, purpose, scope, definitions,
policy_statements (list), controls_mapping (list), roles,
monitoring_requirements, review_schedule.
"""
        draft = self.llm.generate_json(prompt)
        draft["status"] = PolicyStatus.DRAFT.value
        draft["version"] = "0.1"
        draft["regulation_ids"] = [regulation_id]
        draft["created_at"] = datetime.now().isoformat()
        draft["content_hash"] = hashlib.sha256(str(draft).encode()).hexdigest()

        return self.policies.save_draft(draft)

    def submit_for_approval(self, policy_id: str, approvers: list) -> dict:
        """Route policy through approval workflow."""
        policy = self.policies.get(policy_id)
        policy["status"] = PolicyStatus.IN_REVIEW.value

        workflow = {
            "policy_id": policy_id,
            "version": policy["version"],
            "submitted_at": datetime.now().isoformat(),
            "approvers": [
                {"user_id": uid, "role": role, "status": "pending",
                 "due_date": (datetime.now() + timedelta(days=5)).isoformat()}
                for uid, role in approvers
            ],
            "changes_summary": self._generate_changes_summary(policy),
        }

        self.policies.update(policy_id, policy)
        return self.policies.create_workflow(workflow)

    def track_attestations(self, policy_id: str) -> dict:
        """Track which employees have attested to a published policy."""
        policy = self.policies.get(policy_id)
        required = self.policies.get_applicable_employees(policy["scope"])
        attested = self.policies.get_attestations(policy_id, policy["version"])

        attested_ids = {a["employee_id"] for a in attested}
        missing = [e for e in required if e["id"] not in attested_ids]
        overdue = [
            e for e in missing
            if (datetime.now() - datetime.fromisoformat(policy["published_at"])).days
            > policy.get("attestation_deadline_days", 30)
        ]

        return {
            "policy_id": policy_id,
            "version": policy["version"],
            "total_required": len(required),
            "attested": len(attested),
            "completion_rate": round(len(attested) / max(len(required), 1) * 100, 1),
            "missing": [{"id": e["id"], "name": e["name"], "dept": e["department"]}
                        for e in missing],
            "overdue": [{"id": e["id"], "name": e["name"], "dept": e["department"]}
                        for e in overdue],
            "send_reminders": len(overdue) > 0,
        }

    def detect_policy_gaps(self) -> list:
        """Find regulations not adequately covered by current policies."""
        all_regs = self.regulations.get_active()
        all_policies = self.policies.get_published()

        # Build regulation-to-policy coverage map
        coverage = {}
        for reg in all_regs:
            mapped_policies = [
                p for p in all_policies
                if reg["id"] in p.get("regulation_ids", [])
            ]
            coverage[reg["id"]] = {
                "regulation": reg["title"],
                "jurisdiction": reg["jurisdiction"],
                "mapped_policies": len(mapped_policies),
                "policy_titles": [p["title"] for p in mapped_policies],
            }

        # Identify gaps
        gaps = []
        for reg_id, cov in coverage.items():
            if cov["mapped_policies"] == 0:
                gaps.append({
                    "regulation_id": reg_id,
                    "regulation": cov["regulation"],
                    "jurisdiction": cov["jurisdiction"],
                    "gap_type": "no_coverage",
                    "severity": "critical",
                    "recommendation": f"Create new policy for {cov['regulation']}",
                })

        # Use LLM to find partial coverage gaps
        prompt = f"""Review this regulation-to-policy mapping and identify
partial coverage gaps (regulations that ARE mapped to policies but
where the policy likely does not fully address the regulatory requirements):

{chr(10).join(
    f"- {c['regulation']} -> {', '.join(c['policy_titles']) or 'NONE'}"
    for c in coverage.values()
)}

Return JSON list of gaps with: regulation, gap_description, severity, recommendation.
"""
        partial_gaps = self.llm.generate_json(prompt)
        gaps.extend(partial_gaps)

        return sorted(gaps, key=lambda g: {"critical": 0, "high": 1,
                                            "medium": 2, "low": 3}
                       .get(g["severity"], 4))

Version Control & Approval Workflows

Every policy change is versioned with a content hash, ensuring a tamper-proof audit trail. The approval workflow routes drafts to the right stakeholders—policy owners, legal, business line heads—with due dates and automatic escalation if approvals stall. The agent tracks who approved what and when, which is exactly what examiners want to see.

Employee Attestation Tracking

Publishing a policy is meaningless if employees have not read it. The agent monitors attestation completion rates by department, sends targeted reminders to overdue employees, and escalates to management when completion drops below thresholds. During audits, you can instantly produce a report showing attestation status for any policy version.

Policy Gap Detection

This is where the agent provides its highest value. It continuously maps every active regulation to published policies and identifies two types of gaps: zero coverage (regulations with no corresponding policy) and partial coverage (policies that exist but do not fully address the regulatory requirements). The LLM performs nuanced analysis that rule-based systems miss—detecting when a policy covers the letter but not the spirit of a regulation.

4. Transaction Monitoring & AML

Anti-money laundering compliance is one of the most expensive areas for financial institutions. Transaction monitoring systems generate thousands of alerts daily, and 95%+ are false positives. Compliance analysts spend their days closing false alerts instead of investigating real suspicious activity. AI agents can dramatically improve this ratio while also automating SAR narrative generation and customer due diligence.

Suspicious Activity Detection

Modern transaction monitoring combines rules-based detection (structuring, rapid movement, round-tripping) with ML anomaly detection that catches patterns rules miss. The agent runs both in parallel and consolidates results into a single prioritized alert queue.

import numpy as np
from datetime import datetime, timedelta
from sklearn.ensemble import IsolationForest
from collections import defaultdict


class TransactionMonitoringAgent:
    """AML transaction monitoring with rules + ML anomaly detection."""

    STRUCTURING_THRESHOLD = 10_000  # CTR filing threshold
    RAPID_MOVEMENT_HOURS = 24
    VELOCITY_LOOKBACK_DAYS = 30

    def __init__(self, llm_client, tx_store, customer_store):
        self.llm = llm_client
        self.tx_store = tx_store
        self.customers = customer_store
        self.anomaly_model = None

    def monitor_transactions(self, transactions: list) -> list:
        """Run rules-based + ML detection on transaction batch."""
        alerts = []

        # Rules-based detection
        for tx in transactions:
            rule_alerts = self._apply_rules(tx)
            alerts.extend(rule_alerts)

        # ML anomaly detection
        ml_alerts = self._detect_anomalies(transactions)
        alerts.extend(ml_alerts)

        # Deduplicate and prioritize
        alerts = self._deduplicate(alerts)
        alerts = self._prioritize(alerts)

        return alerts

    def _apply_rules(self, tx: dict) -> list:
        alerts = []
        customer_id = tx["customer_id"]

        # Rule 1: Structuring detection (just-below-CTR transactions)
        if 8_000 <= tx["amount"] < self.STRUCTURING_THRESHOLD:
            recent = self.tx_store.get_customer_transactions(
                customer_id, days=self.VELOCITY_LOOKBACK_DAYS
            )
            just_below = [t for t in recent
                          if 8_000 <= t["amount"] < self.STRUCTURING_THRESHOLD]
            if len(just_below) >= 3:
                alerts.append({
                    "type": "structuring",
                    "customer_id": customer_id,
                    "transaction_id": tx["id"],
                    "details": f"{len(just_below)} transactions between "
                               f"$8K-$10K in {self.VELOCITY_LOOKBACK_DAYS} days",
                    "severity": "high",
                    "rule": "STRUCT-001",
                })

        # Rule 2: Rapid movement (funds in and out within 24h)
        if tx["type"] == "credit":
            debits = self.tx_store.get_debits_after(
                customer_id, tx["timestamp"],
                hours=self.RAPID_MOVEMENT_HOURS
            )
            moved = sum(d["amount"] for d in debits)
            if moved >= tx["amount"] * 0.8:
                alerts.append({
                    "type": "rapid_movement",
                    "customer_id": customer_id,
                    "transaction_id": tx["id"],
                    "details": f"${tx['amount']:,.0f} in, ${moved:,.0f} out "
                               f"within {self.RAPID_MOVEMENT_HOURS}h",
                    "severity": "high",
                    "rule": "RAPID-001",
                })

        # Rule 3: Geographic risk (high-risk jurisdictions)
        if tx.get("counterparty_country") in self._high_risk_countries():
            alerts.append({
                "type": "geographic_risk",
                "customer_id": customer_id,
                "transaction_id": tx["id"],
                "details": f"Transaction with {tx['counterparty_country']}",
                "severity": "medium",
                "rule": "GEO-001",
            })

        return alerts

    def _detect_anomalies(self, transactions: list) -> list:
        """Isolation Forest anomaly detection on transaction features."""
        if not transactions:
            return []

        features = np.array([
            [
                tx["amount"],
                tx.get("hour_of_day", 12),
                tx.get("day_of_week", 3),
                self._customer_avg_tx(tx["customer_id"]),
                self._customer_tx_frequency(tx["customer_id"]),
                1 if tx.get("is_international", False) else 0,
            ]
            for tx in transactions
        ])

        model = IsolationForest(contamination=0.02, random_state=42)
        predictions = model.fit_predict(features)
        scores = model.decision_function(features)

        alerts = []
        for i, (pred, score) in enumerate(zip(predictions, scores)):
            if pred == -1:  # Anomaly
                alerts.append({
                    "type": "ml_anomaly",
                    "customer_id": transactions[i]["customer_id"],
                    "transaction_id": transactions[i]["id"],
                    "details": f"Anomaly score: {abs(score):.3f}",
                    "severity": "medium" if abs(score) > 0.3 else "low",
                    "rule": "ML-ISO-001",
                })

        return alerts

    def generate_sar_narrative(self, alert_id: str) -> str:
        """Auto-generate SAR narrative from investigation findings."""
        alert = self.tx_store.get_alert(alert_id)
        customer = self.customers.get(alert["customer_id"])
        transactions = self.tx_store.get_alert_transactions(alert_id)
        prior_sars = self.tx_store.get_prior_sars(alert["customer_id"])

        prompt = f"""Generate a FinCEN SAR narrative for this case:

SUBJECT: {customer['name']} (ID: {customer['id']})
ACCOUNT TYPE: {customer['account_type']}
OCCUPATION: {customer.get('occupation', 'Unknown')}
ACCOUNT OPENED: {customer['opened_date']}

SUSPICIOUS ACTIVITY:
Alert Type: {alert['type']}
Detection Rule: {alert['rule']}
Details: {alert['details']}

TRANSACTIONS ({len(transactions)} total):
{chr(10).join(
    f"- {t['date']} | {t['type']} | ${t['amount']:,.2f} | {t.get('description','')}"
    for t in transactions[:25]
)}

PRIOR SARs: {len(prior_sars)} filed

Write a professional SAR narrative following FinCEN guidance:
1. Who is conducting the suspicious activity
2. What instruments or mechanisms are being used
3. When did the activity occur (date range)
4. Where did the activity take place
5. Why is the activity suspicious
"""
        return self.llm.generate(prompt)

    def screen_customer(self, customer_id: str) -> dict:
        """KYC screening: PEP, sanctions, adverse media."""
        customer = self.customers.get(customer_id)
        results = {
            "customer_id": customer_id,
            "screening_date": datetime.now().isoformat(),
            "pep_match": self._check_pep(customer),
            "sanctions_match": self._check_sanctions(customer),
            "adverse_media": self._check_adverse_media(customer),
            "risk_rating": None,
        }

        # Compute overall CDD risk rating
        if results["sanctions_match"]["hit"]:
            results["risk_rating"] = "prohibited"
        elif results["pep_match"]["hit"]:
            results["risk_rating"] = "high"
        elif results["adverse_media"]["hit_count"] > 2:
            results["risk_rating"] = "high"
        else:
            results["risk_rating"] = "standard"

        return results

    def _high_risk_countries(self):
        return {"IR", "KP", "SY", "MM", "AF", "YE", "SO", "LY", "SD", "VE"}

    def _customer_avg_tx(self, cid):
        txs = self.tx_store.get_customer_transactions(cid, days=90)
        return np.mean([t["amount"] for t in txs]) if txs else 0

    def _customer_tx_frequency(self, cid):
        return len(self.tx_store.get_customer_transactions(cid, days=30))

    def _check_pep(self, customer):
        return self.customers.screen_pep(customer["name"], customer.get("dob"))

    def _check_sanctions(self, customer):
        return self.customers.screen_sanctions(customer["name"], customer.get("dob"))

    def _check_adverse_media(self, customer):
        return self.customers.screen_adverse_media(customer["name"])

    def _deduplicate(self, alerts):
        seen = set()
        unique = []
        for a in alerts:
            key = (a["customer_id"], a["type"], a.get("transaction_id"))
            if key not in seen:
                seen.add(key)
                unique.append(a)
        return unique

    def _prioritize(self, alerts):
        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        return sorted(alerts, key=lambda a: severity_order.get(a["severity"], 4))

SAR Narrative Generation

Writing SAR narratives is one of the most tedious tasks in AML compliance. Each narrative requires summarizing who, what, when, where, and why—pulling together customer information, transaction details, and investigation findings. The agent generates a first draft that follows FinCEN guidance, which analysts then review and finalize. This cuts SAR filing time from 3-4 hours to 30-45 minutes.

Customer Due Diligence

The screen_customer method performs automated KYC screening: PEP (Politically Exposed Persons) matching, sanctions list checking (OFAC, EU, UN), and adverse media scanning. The agent assigns a risk rating based on screening results and triggers enhanced due diligence workflows for high-risk customers.

Transaction Pattern Analysis

Beyond individual transaction monitoring, the agent analyzes behavioral patterns over time. It detects gradual changes in transaction behavior that might indicate layering, identifies networks of related accounts moving funds in coordinated patterns, and flags sudden changes in transaction geography or counterparty profiles. The Isolation Forest model captures patterns that fixed rules cannot express.

5. Audit & Reporting Automation

Regulatory reporting is a recurring burden that consumes enormous amounts of time. SOX compliance, GDPR Article 30 records of processing activities, Basel III capital adequacy reports, CCAR stress testing documentation—each requires collecting evidence from multiple systems, formatting data to specific templates, and ensuring accuracy under tight deadlines. AI agents can automate the collection, formatting, and initial review.

Regulatory Report Generation

The agent maintains templates for each report type and knows which data sources feed into each. When a reporting deadline approaches, it automatically collects the required data, populates the template, runs validation checks, and produces a draft for human review.

from datetime import datetime
from pathlib import Path
import json


class AuditReportingAgent:
    """Automated regulatory reporting and audit evidence collection."""

    REPORT_TEMPLATES = {
        "sox_302": {
            "name": "SOX Section 302 Certification",
            "frequency": "quarterly",
            "data_sources": ["financial_controls", "material_weaknesses",
                             "disclosure_controls"],
            "template": "sox_302_template.json",
        },
        "gdpr_art30": {
            "name": "GDPR Article 30 Records of Processing",
            "frequency": "on_change",
            "data_sources": ["processing_activities", "data_flows",
                             "legal_bases", "retention_schedules"],
            "template": "gdpr_art30_template.json",
        },
        "basel_iii": {
            "name": "Basel III Capital Adequacy Report",
            "frequency": "quarterly",
            "data_sources": ["capital_ratios", "risk_weighted_assets",
                             "liquidity_coverage", "leverage_ratio"],
            "template": "basel_iii_template.json",
        },
        "bsa_ctr": {
            "name": "BSA Currency Transaction Report",
            "frequency": "per_event",
            "data_sources": ["large_cash_transactions", "customer_info"],
            "template": "bsa_ctr_template.json",
        },
    }

    def __init__(self, llm_client, data_store, evidence_store):
        self.llm = llm_client
        self.data = data_store
        self.evidence = evidence_store

    def generate_report(self, report_type: str, period: str) -> dict:
        """Generate a regulatory report with automated data collection."""
        config = self.REPORT_TEMPLATES[report_type]

        # Step 1: Collect data from all required sources
        collected_data = {}
        collection_log = []
        for source in config["data_sources"]:
            try:
                data = self.data.query(source, period=period)
                collected_data[source] = data
                collection_log.append({
                    "source": source,
                    "records": len(data) if isinstance(data, list) else 1,
                    "status": "success",
                    "timestamp": datetime.now().isoformat(),
                })
            except Exception as e:
                collection_log.append({
                    "source": source,
                    "status": "failed",
                    "error": str(e),
                    "timestamp": datetime.now().isoformat(),
                })

        # Step 2: Validate data completeness
        validation = self._validate_data(report_type, collected_data)

        # Step 3: Populate report template
        report = self._populate_template(config["template"], collected_data)

        # Step 4: LLM review for inconsistencies
        review = self._ai_review(report, report_type)

        return {
            "report_type": report_type,
            "period": period,
            "generated_at": datetime.now().isoformat(),
            "status": "draft" if validation["issues"] else "ready_for_review",
            "report_content": report,
            "validation": validation,
            "ai_review": review,
            "collection_log": collection_log,
            "evidence_refs": self._attach_evidence(report_type, period),
        }

    def collect_control_evidence(self, control_id: str) -> dict:
        """Automated control testing and evidence collection."""
        control = self.data.get_control(control_id)

        evidence_items = []

        # Automated test execution based on control type
        if control["type"] == "access_control":
            evidence_items.append(self._test_access_control(control))
        elif control["type"] == "segregation_of_duties":
            evidence_items.append(self._test_sod(control))
        elif control["type"] == "reconciliation":
            evidence_items.append(self._test_reconciliation(control))
        elif control["type"] == "approval_workflow":
            evidence_items.append(self._test_approvals(control))

        # Screenshot capture for UI-based controls
        if control.get("requires_screenshot"):
            screenshot = self._capture_evidence_screenshot(control)
            evidence_items.append(screenshot)

        # Compile evidence package
        package = {
            "control_id": control_id,
            "control_name": control["name"],
            "test_date": datetime.now().isoformat(),
            "tester": "AI Compliance Agent",
            "evidence_items": evidence_items,
            "conclusion": self._assess_control_effectiveness(evidence_items),
            "exceptions": [e for e in evidence_items if e.get("result") == "fail"],
        }

        self.evidence.store(package)
        return package

    def manage_audit_trail(self, entity_type: str, entity_id: str,
                           action: str, details: dict) -> str:
        """Immutable audit trail for all compliance actions."""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "entity_type": entity_type,
            "entity_id": entity_id,
            "action": action,
            "details": details,
            "user": details.get("performed_by", "system"),
            "ip_address": details.get("ip_address"),
            "previous_hash": self.evidence.get_latest_hash(entity_type, entity_id),
        }
        # Hash chain for tamper evidence
        entry["hash"] = self._compute_hash(entry)
        return self.evidence.append_trail(entry)

    def generate_board_report(self, period: str) -> dict:
        """Executive-level compliance report for board/committee."""
        metrics = {
            "risk_posture": self.data.get_risk_summary(period),
            "regulatory_changes": self.data.get_reg_change_summary(period),
            "policy_health": self.data.get_policy_metrics(period),
            "aml_stats": self.data.get_aml_summary(period),
            "audit_findings": self.data.get_audit_findings(period),
            "training_completion": self.data.get_training_metrics(period),
            "incidents": self.data.get_incident_summary(period),
        }

        prompt = f"""Generate an executive compliance report for {period}.

DATA:
{json.dumps(metrics, indent=2, default=str)}

Write a board-level summary that covers:
1. Overall compliance posture (improving/stable/deteriorating)
2. Key regulatory developments and their impact
3. Risk assessment highlights (top 5 risks)
4. AML program effectiveness (alert volumes, SAR filings, false positive rate)
5. Audit findings requiring attention
6. Recommended actions for the board

Tone: concise, factual, action-oriented. No jargon.
"""
        narrative = self.llm.generate(prompt)

        return {
            "period": period,
            "generated_at": datetime.now().isoformat(),
            "metrics": metrics,
            "executive_summary": narrative,
            "status": "draft",
        }

    def _validate_data(self, report_type, data):
        issues = []
        for source, content in data.items():
            if content is None or (isinstance(content, list) and len(content) == 0):
                issues.append({"source": source, "issue": "No data returned"})
        return {"complete": len(issues) == 0, "issues": issues}

    def _populate_template(self, template_name, data):
        template = self.data.load_template(template_name)
        for field in template.get("fields", []):
            source = field.get("data_source")
            if source and source in data:
                field["value"] = data[source]
        return template

    def _ai_review(self, report, report_type):
        prompt = f"""Review this {report_type} report for inconsistencies,
missing data, or potential compliance issues.
Report: {json.dumps(report, indent=2, default=str)[:3000]}
Return: list of issues found, each with severity and recommendation."""
        return self.llm.generate_json(prompt)

    def _attach_evidence(self, report_type, period):
        return self.evidence.get_refs(report_type, period)

    def _test_access_control(self, control):
        users = self.data.get_users_with_access(control["system"], control["role"])
        authorized = self.data.get_authorized_users(control["role"])
        unauthorized = [u for u in users if u["id"] not in authorized]
        return {
            "test": "access_review",
            "result": "pass" if not unauthorized else "fail",
            "total_users": len(users),
            "unauthorized": unauthorized,
        }

    def _test_sod(self, control):
        conflicts = self.data.detect_sod_conflicts(control["conflicting_roles"])
        return {
            "test": "segregation_of_duties",
            "result": "pass" if not conflicts else "fail",
            "conflicts_found": len(conflicts),
            "details": conflicts,
        }

    def _test_reconciliation(self, control):
        recon = self.data.run_reconciliation(control["source_a"], control["source_b"])
        return {
            "test": "reconciliation",
            "result": "pass" if recon["variance_pct"] < 0.01 else "fail",
            "variance": recon["variance_pct"],
            "unmatched_items": recon["unmatched"],
        }

    def _test_approvals(self, control):
        samples = self.data.sample_transactions(control["tx_type"], n=25)
        missing = [s for s in samples if not s.get("approved_by")]
        return {
            "test": "approval_workflow",
            "result": "pass" if not missing else "fail",
            "sampled": len(samples),
            "missing_approval": len(missing),
        }

    def _capture_evidence_screenshot(self, control):
        return {"test": "screenshot", "result": "captured", "path": f"/evidence/{control['id']}.png"}

    def _assess_control_effectiveness(self, items):
        failures = sum(1 for i in items if i.get("result") == "fail")
        if failures == 0: return "effective"
        if failures <= 1: return "needs_improvement"
        return "ineffective"

    def _compute_hash(self, entry):
        import hashlib
        content = json.dumps(entry, sort_keys=True, default=str)
        return hashlib.sha256(content.encode()).hexdigest()

Evidence Collection

The agent automates control testing across four common control types: access controls (who has access vs. who should), segregation of duties (detecting role conflicts), reconciliations (matching data across systems), and approval workflows (verifying transactions were properly authorized). Each test produces structured evidence with pass/fail results, which feeds directly into audit workpapers.

Audit Trail Management

Every action in the compliance system is logged in a hash-chained audit trail. Each entry references the previous entry's hash, creating a tamper-evident chain. Regulators and auditors can verify the integrity of the entire audit history by validating the hash chain—any modification to historical records would break the chain.

Board & Committee Reporting

The generate_board_report method pulls metrics from across the compliance program—risk posture, regulatory changes, policy health, AML statistics, audit findings, training completion, and incidents—and uses the LLM to produce a concise executive summary. Board members get a clear picture of compliance posture without wading through hundreds of pages of detail. The agent highlights trends, flags deterioration, and recommends specific actions.

6. ROI Analysis

Let us quantify the business case for a regulated financial institution with 500 employees, a compliance team of 40, and annual compliance spending of approximately $12 million.

class ComplianceROICalculator:
    """ROI model for AI compliance agents at a 500-employee institution."""

    def __init__(self):
        self.baseline = {
            "total_employees": 500,
            "compliance_team_size": 40,
            "avg_compliance_salary": 115_000,
            "annual_compliance_budget": 12_000_000,
            "regulatory_bodies_tracked": 45,
            "policies_managed": 180,
            "monthly_alerts": 8_500,
            "annual_sar_filings": 320,
            "annual_audit_hours": 12_000,
            "avg_regulatory_fine": 4_500_000,
            "fine_probability_per_year": 0.15,
            "external_audit_cost": 850_000,
        }

    def calculate_full_roi(self) -> dict:
        savings = {}

        # 1. Regulatory Change Management
        # Before: 6 analysts full-time tracking regulations
        # After: 1 analyst reviewing AI output
        reg_change_before = 6 * self.baseline["avg_compliance_salary"]
        reg_change_after = 1.5 * self.baseline["avg_compliance_salary"]
        savings["regulatory_change_mgmt"] = {
            "before": reg_change_before,
            "after": reg_change_after,
            "annual_savings": reg_change_before - reg_change_after,
            "fte_freed": 4.5,
            "details": "Automated tracking of 45 regulatory bodies, "
                       "AI impact assessment, gap analysis",
        }

        # 2. Risk Assessment
        # Before: quarterly manual assessments, 3 analysts, 4 weeks each
        # After: continuous automated, 1 analyst oversight
        risk_before = 3 * self.baseline["avg_compliance_salary"] * 0.5
        risk_after = 1 * self.baseline["avg_compliance_salary"] * 0.25
        savings["risk_assessment"] = {
            "before": risk_before,
            "after": risk_after,
            "annual_savings": risk_before - risk_after,
            "fte_freed": 1.25,
            "details": "Continuous risk scoring vs quarterly manual, "
                       "real-time KRI monitoring, emerging risk detection",
        }

        # 3. Policy Lifecycle
        # Before: 2 analysts managing 180 policies, attestation chasing
        # After: AI drafting, automated attestation tracking
        policy_before = 2 * self.baseline["avg_compliance_salary"]
        policy_after = 0.5 * self.baseline["avg_compliance_salary"]
        savings["policy_management"] = {
            "before": policy_before,
            "after": policy_after,
            "annual_savings": policy_before - policy_after,
            "fte_freed": 1.5,
            "details": "AI-assisted policy drafting (80% time reduction), "
                       "automated attestation tracking, gap detection",
        }

        # 4. Transaction Monitoring & AML
        # Before: 15 analysts reviewing 8,500 alerts/month (95% false positive)
        # After: AI pre-screening reduces human review by 70%
        aml_before = 15 * self.baseline["avg_compliance_salary"]
        aml_after = 6 * self.baseline["avg_compliance_salary"]
        sar_time_savings = self.baseline["annual_sar_filings"] * 2.5 * 75  # hours saved * rate
        savings["aml_monitoring"] = {
            "before": aml_before,
            "after": aml_after,
            "annual_savings": (aml_before - aml_after) + sar_time_savings,
            "fte_freed": 9,
            "details": "70% alert reduction via ML pre-screening, "
                       "SAR narrative auto-generation (3h -> 30min per SAR), "
                       "automated KYC screening",
        }

        # 5. Audit & Reporting
        # Before: 12,000 hours of audit prep + $850K external audit
        # After: automated evidence collection cuts prep by 60%
        audit_before = self.baseline["annual_audit_hours"] * 75  # internal rate
        audit_external_savings = self.baseline["external_audit_cost"] * 0.20
        audit_after = audit_before * 0.40
        savings["audit_reporting"] = {
            "before": audit_before + self.baseline["external_audit_cost"],
            "after": audit_after + self.baseline["external_audit_cost"] * 0.80,
            "annual_savings": (audit_before - audit_after) + audit_external_savings,
            "fte_freed": 3,
            "details": "60% reduction in audit prep time, 20% reduction "
                       "in external audit fees, automated report generation",
        }

        # 6. Regulatory Fine Avoidance
        # Better compliance = lower probability of fines
        fine_risk_before = (self.baseline["avg_regulatory_fine"]
                           * self.baseline["fine_probability_per_year"])
        fine_risk_after = fine_risk_before * 0.35  # 65% risk reduction
        savings["fine_avoidance"] = {
            "before": fine_risk_before,
            "after": fine_risk_after,
            "annual_savings": fine_risk_before - fine_risk_after,
            "details": "Expected value of fine reduction from 15% to ~5% "
                       "annual probability through better monitoring",
        }

        # Implementation costs
        implementation = {
            "software_licenses": 180_000,
            "llm_api_costs": 96_000,
            "integration_development": 350_000,
            "training_and_change_mgmt": 75_000,
            "ongoing_maintenance": 120_000,
        }

        total_savings = sum(s["annual_savings"] for s in savings.values())
        total_implementation = sum(implementation.values())
        total_ongoing = implementation["llm_api_costs"] + implementation["ongoing_maintenance"]

        return {
            "savings_breakdown": savings,
            "total_annual_savings": total_savings,
            "total_fte_freed": sum(s.get("fte_freed", 0) for s in savings.values()),
            "implementation_costs": implementation,
            "total_year_1_cost": total_implementation,
            "total_ongoing_annual": total_ongoing,
            "net_annual_benefit": total_savings - total_ongoing,
            "payback_months": round(total_implementation
                                    / (total_savings / 12), 1),
            "three_year_roi": round(
                ((total_savings * 3 - total_implementation - total_ongoing * 2)
                 / total_implementation) * 100, 1
            ),
        }


# Run the calculation
calc = ComplianceROICalculator()
roi = calc.calculate_full_roi()

print(f"Annual Savings: ${roi['total_annual_savings']:,.0f}")
print(f"FTEs Freed: {roi['total_fte_freed']}")
print(f"Year 1 Implementation: ${roi['total_year_1_cost']:,.0f}")
print(f"Ongoing Annual Cost: ${roi['total_ongoing_annual']:,.0f}")
print(f"Net Annual Benefit: ${roi['net_annual_benefit']:,.0f}")
print(f"Payback Period: {roi['payback_months']} months")
print(f"3-Year ROI: {roi['three_year_roi']}%")

Savings Breakdown

Implementation Costs

What the Numbers Do Not Capture

These calculations are conservative. They do not include several significant benefits that are harder to quantify:

• Faster time-to-compliance — New regulations are analyzed in hours instead of weeks, reducing exposure windows
• Better risk decisions — Continuous monitoring catches issues that quarterly reviews miss, potentially avoiding major incidents
• Examiner confidence — Regulators view automated, well-documented compliance programs more favorably, leading to less intrusive exams
• Talent retention — Compliance professionals prefer strategic work over manual data gathering; automation reduces turnover
• Scalability — Growing transaction volumes, new products, and geographic expansion do not require proportional headcount increases

Implementation Roadmap

Do not attempt to build all six capabilities at once. Follow this phased approach:

1. Month 1-2: Regulatory Change Tracking — Start with the feed ingester and impact assessment. This provides immediate visibility and builds confidence in the AI approach. Low risk, high visibility.
2. Month 2-4: Policy Gap Detection — Layer in the policy lifecycle manager, starting with gap detection against your existing policy library. This often surfaces surprises that justify the entire program.
3. Month 4-7: AML Alert Triage — Deploy ML-based alert scoring in shadow mode alongside your existing transaction monitoring system. Validate the false positive reduction rate before routing alerts through the AI agent.
4. Month 7-9: Risk Assessment — Build the continuous risk scoring engine. Integrate KRI monitoring and emerging risk detection. Connect outputs to the board reporting module.
5. Month 9-12: Audit Automation — Automate evidence collection for your most-tested controls first. Build the report generation pipeline for your highest-volume regulatory reports.

Critical Guardrails

Compliance AI requires stricter guardrails than most AI applications. Here are the non-negotiables:

• Human-in-the-loop for all decisions — The AI agent recommends and drafts; humans approve and file. Never auto-file a SAR or auto-certify a regulatory report.
• Explainability requirements — Every risk score, alert, and recommendation must come with supporting evidence and reasoning. Black-box outputs are not acceptable in regulated environments.
• Model validation — AML models require independent validation per OCC/Fed guidance. Document model methodology, assumptions, limitations, and ongoing monitoring procedures.
• Data privacy — Compliance data often includes PII and sensitive financial information. Ensure your LLM deployment (self-hosted or private API) meets data residency and security requirements.
• Audit trail for AI actions — Log every AI decision, including the input data, model version, prompt, and output. Regulators will ask how the AI reached its conclusions.