用 Amazon Q 和 Snowflake Cortex 把 AML 警告排查从 90 分钟压到 5 分钟以内

-- 在 Snowflake SQL Worksheet 中执行

CREATE OR REPLACE PROCEDURE aml_risk_summary(customer_id STRING)
RETURNS STRING
LANGUAGE SQL
AS
$$
    DECLARE
        customer_info STRING;
        summary_result STRING;
    BEGIN
        -- 先拼接客户上下文：KYC + 近 90 天交易汇总
        SELECT
            OBJECT_CONSTRUCT(
                'name', c.full_name,
                'risk_score', c.risk_score,
                'total_txn_90d', COUNT(t.transaction_id),
                'total_amount_90d', SUM(t.amount)
            )::STRING
        INTO customer_info
        FROM customers c
        LEFT JOIN transactions t
            ON c.customer_id = t.customer_id
            AND t.transaction_date >= DATEADD(day, -90, CURRENT_DATE())
        WHERE c.customer_id = :customer_id
        GROUP BY c.customer_id, c.full_name, c.risk_score;

        -- 调用 Cortex COMPLETE 生成摘要
        SELECT SNOWFLAKE.CORTEX.COMPLETE(
            'snowflake-arctic',
            CONCAT(
                'You are an AML analyst. Based on the following customer data, ',
                'write a concise risk assessment summary in 3-5 sentences. ',
                'Highlight any red flags.\n\nCustomer data: ', customer_info
            )
        )::STRING
        INTO summary_result;

        RETURN summary_result;
    END;
$$;

# snowflake_mcp_server_config.yaml
server:
  name: aml-triage-tools
  version: "1.0"

tools:
  - name: get_customer_risk_summary
    description: >
      Given a customer_id, returns an AI-generated AML risk assessment
      summary based on KYC data and recent transaction history.
    inputSchema:
      type: object
      properties:
        customer_id:
          type: string
          description: "The customer identifier from the AML alert"
      required:
        - customer_id
    handler:
      type: snowflake-procedure
      procedure: aml_risk_summary
      database: AML_DB
      schema: ANALYTICS

  - name: search_related_alerts
    description: >
      Search for other AML alerts involving the same customer or
      related accounts in the past 12 months.
    inputSchema:
      type: object
      properties:
        customer_id:
          type: string
        lookback_months:
          type: integer
          default: 12
      required:
        - customer_id
    handler:
      type: snowflake-procedure
      procedure: search_related_alerts
      database: AML_DB
      schema: ANALYTICS

{
  "flow_name": "aml-alert-triage",
  "description": "Automated AML alert triage: gather context, analyze risk, produce summary",
  "trigger": {
    "type": "api",
    "input": {
      "alert_id": "string",
      "customer_id": "string",
      "alert_reason": "string"
    }
  },
  "steps": [
    {
      "id": "fetch_risk_summary",
      "tool": "get_customer_risk_summary",
      "input_mapping": {
        "customer_id": "{{trigger.customer_id}}"
      }
    },
    {
      "id": "fetch_related_alerts",
      "tool": "search_related_alerts",
      "input_mapping": {
        "customer_id": "{{trigger.customer_id}}"
      }
    },
    {
      "id": "generate_triage_report",
      "type": "llm",
      "model": "amazon-q-business",
      "prompt_template": [
        "You are an AML compliance analyst reviewing an alert.",
        "Alert reason: {{trigger.alert_reason}}",
        "Customer risk summary: {{fetch_risk_summary.output}}",
        "Related past alerts: {{fetch_related_alerts.output}}",
        "",
        "Based on the above, produce a triage recommendation:",
        "- Is this alert likely a false positive or requires escalation?",
        "- List the key red flags (if any).",
        "- Suggest next steps for the human analyst.",
        "Keep the output under 300 words."
      ]
    }
  ],
  "output": {
    "triage_report": "{{generate_triage_report.output}}"
  }
}

import boto3
import json

q_business = boto3.client("qbusiness", region_name="us-east-1")

# 假设你的 Q Business application ID 和 Flow ID 已配置好
APP_ID = "your-q-business-app-id"
FLOW_ID = "aml-alert-triage"

def triage_alert(alert_id: str, customer_id: str, alert_reason: str) -> str:
    response = q_business.execute_flow(
        applicationId=APP_ID,
        flowId=FLOW_ID,
        input={
            "alert_id": alert_id,
            "customer_id": customer_id,
            "alert_reason": alert_reason,
        }
    )
    return response["output"]["triage_report"]

# 实际调用
report = triage_alert(
    alert_id="ALM-20250618-0042",
    customer_id="CUST-88321",
    alert_reason="Large wire transfer to high-risk jurisdiction followed by rapid fund dispersal"
)

print(report)