# GCP Aml AI > - Skill: `frank-luongt/gcp-aml-ai` (Agent Skill) - Install (CLI): `npx skillmds@latest add frank-luongt/gcp-aml-ai` - Raw SKILL.md: https://api.skillmd.com/api/skills/frank-luongt/gcp-aml-ai/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - Author: frank-luongt (https://skillmd.com/u/frank-luongt) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/frank-luongt/gcp-aml-ai --- --- name: gcp-aml-ai description: Google Cloud Anti Money Laundering AI for financial crime detection. Use when implementing AML transaction monitoring, risk scoring, SAR prioritization, or integrating Google's pre-trained AML models with existing compliance systems. tags: [gcp, aml, fraud, compliance, financial-crime] --- # Google Cloud AML AI Leverage Google Cloud's purpose-built Anti Money Laundering AI for transaction monitoring, risk scoring, and suspicious activity detection with explainable results. ## When to Use - Implementing or modernizing AML transaction monitoring systems - Replacing rules-based AML alerting with ML-based risk scoring - Reducing false positives in existing AML alert pipelines (up to 60% reduction) - Integrating AML risk scores into AI agent workflows for compliance investigation - Building SAR prioritization and investigation tools ## Google AML AI Capabilities | Capability | Description | |---|---| | **Risk Scoring** | ML-based customer and transaction risk scores (0-1 scale) | | **Alert Prioritization** | Rank alerts by actual risk vs. false positives | | **Network Analysis** | Detect suspicious fund flow patterns across entities | | **Explainability** | Human-readable reasons for each risk score | | **Back-testing** | Validate model performance against historical SARs | | **Continuous Learning** | Model improves with feedback from investigation outcomes | ## Patterns ### 1. AML AI Instance Setup ```python from google.cloud import financialservices_v1 def create_aml_instance(project_id: str, location: str, instance_id: str) -> dict: """Create an AML AI instance for a financial institution.""" client = financialservices_v1.AMLClient() instance = financialservices_v1.Instance( display_name=f"AML Instance - {instance_id}", kms_key=f"projects/{project_id}/locations/{location}/keyRings/aml-keyring/cryptoKeys/aml-key", ) operation = client.create_instance( parent=f"projects/{project_id}/locations/{location}", instance=instance, instance_id=instance_id, ) return operation.result() ``` ### 2. Ingest Transaction Data ```python from google.cloud import bigquery def prepare_aml_dataset(project_id: str, dataset_id: str): """Prepare BigQuery dataset with AML AI required schema. AML AI expects transaction data in a specific schema including: - Party table (customers/entities) - Transaction table (financial transactions) - Account table (bank accounts) - Party-Account link table """ bq_client = bigquery.Client(project=project_id) # Party (customer) table party_schema = [ bigquery.SchemaField("party_id", "STRING", mode="REQUIRED"), bigquery.SchemaField("type", "STRING"), # INDIVIDUAL, ORGANIZATION bigquery.SchemaField("registration_date", "TIMESTAMP"), bigquery.SchemaField("country", "STRING"), bigquery.SchemaField("risk_category", "STRING"), bigquery.SchemaField("pep_status", "BOOLEAN"), ] # Transaction table transaction_schema = [ bigquery.SchemaField("transaction_id", "STRING", mode="REQUIRED"), bigquery.SchemaField("type", "STRING"), # WIRE, ACH, CASH, CHECK bigquery.SchemaField("direction", "STRING"), # DEBIT, CREDIT bigquery.SchemaField("amount", "FLOAT64"), bigquery.SchemaField("currency", "STRING"), bigquery.SchemaField("timestamp", "TIMESTAMP"), bigquery.SchemaField("originator_account", "STRING"), bigquery.SchemaField("beneficiary_account", "STRING"), bigquery.SchemaField("originator_country", "STRING"), bigquery.SchemaField("beneficiary_country", "STRING"), ] table_configs = [ (f"{dataset_id}.party", party_schema), (f"{dataset_id}.transaction", transaction_schema), ] for table_id, schema in table_configs: table = bigquery.Table(f"{project_id}.{table_id}", schema=schema) bq_client.create_table(table, exists_ok=True) ``` ### 3. Run Risk Scoring ```python from google.cloud import financialservices_v1 def run_risk_scoring( project_id: str, location: str, instance_id: str, dataset: str, engine_config_id: str, ) -> dict: """Execute AML AI risk scoring on transaction data. Returns risk scores for all parties with explainability signals. """ client = financialservices_v1.AMLClient() parent = f"projects/{project_id}/locations/{location}/instances/{instance_id}" # Create engine config (defines model and scoring parameters) engine_config = financialservices_v1.EngineConfig( display_name="Production AML Scoring", engine_version="latest", dataset=f"projects/{project_id}/locations/{location}/instances/{instance_id}/datasets/{dataset}", ) # Run prediction prediction_result = client.create_prediction_result( parent=parent, prediction_result=financialservices_v1.PredictionResult( engine_config=f"{parent}/engineConfigs/{engine_config_id}", ), ) return prediction_result.result() def get_risk_scores(project_id: str, prediction_table: str, threshold: float = 0.7) -> list[dict]: """Query high-risk scores from AML AI prediction results. Args: project_id: GCP project ID prediction_table: BigQuery table with prediction results threshold: Risk score threshold (0-1, default 0.7 for high risk) """ from google.cloud import bigquery bq = bigquery.Client(project=project_id) query = f""" SELECT party_id, risk_score, risk_category, explanation, top_signals FROM `{prediction_table}` WHERE risk_score >= {threshold} ORDER BY risk_score DESC LIMIT 100 """ results = bq.query(query).result() return [dict(row) for row in results] ``` ### 4. Integrate Risk Scores into Agent Workflow ```python def check_customer_aml_risk(customer_id: str) -> str: """Check AML risk score for a customer from Google AML AI. Args: customer_id: The customer/party identifier """ from google.cloud import bigquery bq = bigquery.Client() query = """ SELECT risk_score, risk_category, explanation, scored_at FROM `project.aml_results.latest_scores` WHERE party_id = @customer_id ORDER BY scored_at DESC LIMIT 1 """ job_config = bigquery.QueryJobConfig( query_parameters=[bigquery.ScalarQueryParameter("customer_id", "STRING", customer_id)] ) rows = list(bq.query(query, job_config=job_config).result()) if not rows: return f"No AML risk score found for customer {customer_id}." row = rows[0] risk_level = "HIGH" if row["risk_score"] >= 0.7 else "MEDIUM" if row["risk_score"] >= 0.4 else "LOW" result = ( f"Customer {customer_id} AML Risk Assessment:\n" f" Risk Score: {row['risk_score']:.3f} ({risk_level})\n" f" Category: {row['risk_category']}\n" f" Last Scored: {row['scored_at']}\n" f" Explanation: {row['explanation']}" ) if risk_level == "HIGH": result += "\n ACTION REQUIRED: Escalate to compliance officer for enhanced due diligence." return result ``` ### 5. Back-Testing Model Performance ```python def backtest_aml_model(project_id: str, location: str, instance_id: str, engine_config_id: str) -> dict: """Run back-test to validate AML AI model against historical SARs. Compares model predictions against known suspicious activity to measure: - True positive rate (caught SARs) - False positive reduction vs. rules-based system - Coverage of different typologies """ client = financialservices_v1.AMLClient() parent = f"projects/{project_id}/locations/{location}/instances/{instance_id}" backtest = client.create_backtest_result( parent=parent, backtest_result=financialservices_v1.BacktestResult( engine_config=f"{parent}/engineConfigs/{engine_config_id}", ), ) result = backtest.result() return { "true_positive_rate": result.metrics.get("true_positive_rate"), "false_positive_reduction": result.metrics.get("false_positive_reduction"), "coverage": result.metrics.get("typology_coverage"), } ``` ## Architecture ``` Core Banking / Payment System --> BigQuery (transaction + party data, AML AI schema) --> Google AML AI (ML risk scoring + explainability) --> BigQuery (risk scores + signals) --> AI Agent (investigation assistant, SAR drafting) --> Compliance Dashboard (human review + filing) ``` ## Anti-Patterns - Using AML AI scores as final decisions without human review -- regulatory requirement for human oversight - Skipping back-testing before production deployment -- model validation is a regulatory expectation - Not feeding investigation outcomes back to the model -- reduces model improvement over time - Storing raw PII in AML AI without encryption -- use CMEK and data masking - Running AML scoring infrequently (monthly) -- high-risk customers need near-real-time monitoring ## References - [Google Cloud AML AI](https://cloud.google.com/financial-services/anti-money-laundering/docs/overview) - [AML AI API Reference](https://cloud.google.com/financial-services/anti-money-laundering/docs/reference) - [AML AI Architecture](https://cloud.google.com/architecture/anti-money-laundering-ai-architecture) - [FATF Guidance on AI/ML in AML](https://www.fatf-gafi.org/)