为金融机构构建 AI 驱动的合规监控系统

我们将深入剖析为一家中型银行构建的合规监控系统的架构。不是那种策略演示稿版本，是工程版本，包含数据流水线设计、智能体架构、审计追踪实现，以及那些实际耗时是我们预估三倍的部分。

如果你是为金融服务行业构建系统的工程师，合规监控很可能就是你下个季度会被 CTO 问到的项目。监管压力在加剧，人工监控方式已触及容量上限，监管机构预期与批量审查流程之间的差距每月都在扩大。

金融领域的 AI 智能体已经过了概念验证阶段。各银行正在生产环境中运行基于智能体的合规系统。但关于如何构建这些系统的工程内容非常匮乏，发表的大多是供应商营销文案或监管理论。这是来自构建者的视角。

工程术语描述的问题

金融机构中的合规监控意味着根据适用的法规、内部政策和风险阈值评估每笔交易和运营事件。关键词是"每"。

一家中型银行每天处理 200,000 笔交易，涉及约 150 条不同的监管规则，涵盖反洗钱（AML）、制裁、合规贷款、消费者保护和内部政策，这产生了每天 3000 万次评估的监控矩阵。人工方式抽取 5% 的交易进行人工审查，覆盖了 150 万次评估。剩余的 2850 万次在引起关注之前都处于未监控状态。

工程挑战不在于 AI 的 sophistication，而在于构建一个每天执行 3000 万次评估、对新交易具有亚分钟级延迟、为每次评估维护完整审计追踪、在规则变更时无需系统停机、能以不会压垮合规团队的比率区分真实违规与误报、并满足监管机构对监控的全面性、文档化和可解释性要求的系统。

这是一个带有 AI 组件的系统工程问题，而非带有系统需求的 AI 问题。

四智能体架构

系统通过中央编排层协调四个专业智能体运行。每个智能体都有明确的范围、定义的输入输出，以及各自的审计追踪流。

                    ┌─────────────────────┐
                    │   Orchestration      │
                    │   Engine             │
                    └──────────┬──────────┘
                               │
            ┌──────────────────┼──────────────────┐
            │                  │                   │
            ▼                  ▼                   ▼
   ┌────────────────┐ ┌────────────────┐ ┌────────────────┐
   │  Transaction    │ │  Regulatory    │ │  Policy        │
   │  Monitor       │ │  Change Agent  │ │  Compliance    │
   │  Agent         │ │                │ │  Agent         │
   └────────────────┘ └────────────────┘ └────────────────┘
            │                  │                   │
            └──────────────────┼──────────────────┘
                               │
                    ┌──────────▼──────────┐
                    │   Audit Trail       │
                    │   Engine            │
                    └─────────────────────┘

第四个智能体——审计就绪智能体，采用批量调度而非实时运行，从审计追踪中组装检查证据包。我将单独介绍它，因为它的架构与实时智能体有根本性不同。

智能体 1：交易监控

交易监控器实时根据完整的适用规则集评估每笔交易。这是吞吐量最高的组件，架构决策对其性能的影响最直接。

简单做法——将每笔交易都通过 LLM 进行合规评估——在这个规模下会在三个维度上失败：延迟（LLM 推理每次评估增加 500ms 到 2s）、成本（按推理定价每天 200,000 笔交易会烧穿预算）和确定性（对于明确的监管规则，LLM 输出并非完全一致，监管机构不会接受）。

可行的架构将评估分为两层。

确定性层处理有明确、二元答案的规则。交易金额超过 10,000 美元 CTR 报告阈值，是或否。交易对手出现在 OFAC 制裁名单上，是或否。交易地理来源在受限司法管辖区，是或否。这些评估是基于规则的，而非基于 AI 的。它们作为流式规则引擎针对每个交易事件运行。

# Deterministic compliance checks - no AI needed
class DeterministicComplianceEngine:
    def __init__(self, rule_set: RuleSet):
        self.rules = rule_set

def evaluate(self, transaction: Transaction) -> list[ComplianceResult]:
        results = []
        for rule in self.rules.get_applicable(transaction):
            result = ComplianceResult(
                rule_id=rule.id,
                transaction_id=transaction.id,
                passed=rule.evaluate(transaction),
                evaluation_type="deterministic",
                timestamp=datetime.utcnow(),
                evidence=rule.get_evidence(transaction)
            )
            results.append(result)
        return results

AI 层处理需要解读的评估。这组交易序列是与正常商业活动一致还是暗示结构性拆分？这位客户近期行为代表财务模式的合法变化还是潜在账户被盗用？这笔交易的叙述是否包含与声明交易目的相矛盾的信息？

这些评估使用 LLM，包含交易数据、相关监管上下文和客户行为基线的结构化提示。输出是结构化的：分类、置信度分数和推理链。

# AI-assisted compliance evaluation for pattern-based rules
class AIComplianceEvaluator:
    def __init__(self, model_client, prompt_templates):
        self.client = model_client
        self.templates = prompt_templates

def evaluate_pattern(
        self, 
        transaction: Transaction,
        customer_history: CustomerHistory,
        rule: PatternRule
    ) -> ComplianceResult:
        prompt = self.templates.render(
            rule_type=rule.type,
            transaction=transaction.to_context(),
            history=customer_history.recent_summary(),
            regulatory_context=rule.regulatory_text,
            output_schema="classification, confidence, reasoning"
        )

response = self.client.complete(
            prompt=prompt,
            temperature=0.1,  # Near-deterministic for compliance
            max_tokens=500
        )

parsed = self.parse_structured_response(response)

return ComplianceResult(
            rule_id=rule.id,
            transaction_id=transaction.id,
            passed=parsed.classification == "compliant",
            confidence=parsed.confidence,
            reasoning=parsed.reasoning,
            evaluation_type="ai_assisted",
            timestamp=datetime.utcnow(),
            model_version=self.client.model_version,
            prompt_hash=hash(prompt)
        )

temperature=0.1 是经过慎重考虑且在合规评估中不可妥协的设置。我们需要近乎确定性的输出。我们还在每个评估结果中存储 prompt_hash 和 model_version，因为监管机构需要重现任何特定评估时的条件。

确定性层与 AI 层的划分是使系统在规模化下可行的设计决策。在我们的部署中，大约 85% 的评估是确定性的——对结构化数据应用的明确规则。剩余 15% 进入 AI 层。这意味着 LLM 推理每天在 30,000 笔交易上运行，而非 200,000 笔，这完全改变了成本和延迟状况。

智能体 2：监管动态监控

该智能体扫描监管出版物并识别与机构运营相关的变更。这是系统中最简单的 AI 应用，也是能最快交付价值的部分。

数据流水线从机构适用的监管机构、联邦登记条目、监管函、指导文件、执法行动和同意令获取出版物。对于美国银行业务，最低限度包括 OCC、FDIC、联邦储备委员会、CFPB 和 FinCEN，加上机构运营司法管辖区的州级监管机构。

Regulatory change monitoring pipeline

class RegulatoryChangeAgent:
    def __init__(self, sources, relevance_model, institution_profile):
        self.sources = sources  # RSS feeds, API endpoints, scrapers
        self.model = relevance_model
        self.profile = institution_profile

def scan(self) -> list[RegulatoryChange]:
        new_publications = []
        for source in self.sources:
            new_publications.extend(source.fetch_since_last_scan())

relevant_changes = []
        for pub in new_publications:
            assessment = self.model.assess_relevance(
                publication=pub,
                institution_products=self.profile.products,
                institution_jurisdictions=self.profile.jurisdictions,
                institution_charter=self.profile.charter_type
            )

if assessment.relevance_score > 0.6:
                change = RegulatoryChange(
                    source=pub.source,
                    publication_date=pub.date,
                    summary=assessment.summary,
                    affected_products=assessment.affected_products,
                    affected_rules=assessment.affected_rules,
                    recommended_actions=assessment.actions,
                    relevance_score=assessment.relevance_score,
                    urgency=assessment.urgency_classification
                )
                relevant_changes.append(change)

return relevant_changes

相关性模型并不是在尝试解释法律条文的含义。它的作用是判断某份公告是否与该机构特定的产品、服务和管辖区相关，并识别哪些内部政策和监控规则可能需要更新。解释性决策由合规团队做出，AI 智能体只是将需要他们关注的内容呈现出来。

该智能体按计划运行，联邦级来源每六小时一次，州级来源每日一次。延迟容忍度以小时计而非秒级，这意味着它可以使用更大的模型和更长的推理时间，以获得更好的分类质量。

Agent 3: Policy compliance monitoring

策略合规智能体负责监控内部运营是否符合机构自身的政策和程序、员工交易限制、信息隔离墙、审批权限限制以及客户沟通标准。

其架构与交易监控器的双层方法类似。确定性检查用于明确的政策规则（审批权限阈值、强制冷静期、禁止活动清单）。AI 辅助评估用于需要解释的政策（沟通语气合规性、利益冲突评估、跨通信渠道的信息隔离墙监控）。

数据源比交易数据更广泛，包括：邮件元数据、内部消息、访问日志、交易活动、文档访问模式。策略合规的 AI 合规监控架构需要将通信平台、HR 系统和访问管理基础设施与金融系统整合在一起。

# Policy compliance - information barrier monitoring
class InformationBarrierMonitor:
    def __init__(self, barrier_config, communication_feed, model):
        self.barriers = barrier_config
        self.feed = communication_feed
        self.model = model

def evaluate_communication(
        self, event: CommunicationEvent
    ) -> Optional[PolicyAlert]:
        # Deterministic check: are participants in restricted groups?
        sender_group = self.barriers.get_group(event.sender)
        recipient_groups = [
            self.barriers.get_group(r) for r in event.recipients
        ]

barrier_crossed = any(
            self.barriers.is_restricted(sender_group, rg)
            for rg in recipient_groups
        )

if not barrier_crossed:
            return None  # No barrier concern

# AI assessment: does the communication content
        # contain material non-public information?
        content_assessment = self.model.assess(
            content_summary=event.content_summary,  # Never full content
            sender_role=event.sender_role,
            context="information_barrier_evaluation",
            output_schema="risk_level, reasoning, recommended_action"
        )

if content_assessment.risk_level in ["high", "critical"]:
            return PolicyAlert(
                alert_type="information_barrier",
                severity=content_assessment.risk_level,
                participants=event.participants,
                reasoning=content_assessment.reasoning,
                recommended_action=content_assessment.recommended_action,
                timestamp=datetime.utcnow()
            )

上述代码中有一个关键的设计注意事项：AI 模型收到的是 content_summary（内容摘要），而非完整的通信内容。在受监管环境中，合规监控系统本身必须遵守数据处理限制。模型评估的是风险指标，而非阅读每个人的邮件。这种隐私优先的设计是监管要求，而非工程上的便利。

The audit trail engine

审计追踪不是一项日志功能。它是主要交付物。系统产生的所有内容——每条确定性评估、每项 AI 辅助评估、每个生成的告警、每个升级案例的人工决策——都会流入一个不可变的审计存储。

# Audit trail, immutable, complete, reproducible
@dataclass
class AuditRecord:
    record_id: str
    timestamp: datetime
    event_type: str  # evaluation, alert, escalation, resolution
    agent_id: str

# What was evaluated
    subject_id: str  # transaction_id, communication_id, etc.
    subject_data_hash: str  # Hash of the input data

# How it was evaluated
    evaluation_type: str  # deterministic, ai_assisted
    rule_id: str
    model_version: Optional[str]  # For AI evaluations
    prompt_hash: Optional[str]  # For reproducibility

# What was concluded
    result: str  # compliant, non_compliant, escalated
    confidence: Optional[float]
    reasoning: Optional[str]

# What happened next
    action_taken: str
    human_reviewer: Optional[str]
    human_decision: Optional[str]
    human_decision_timestamp: Optional[datetime]

subject_data_hash 和 prompt_hash 字段是实现可重现性的机制。如果监管机构问："为什么这笔交易在 3 月 15 日被评估为合规？"审计记录会包含被评估数据的哈希值以及所使用的精确提示词。原始数据可以被检索回来，相同的模型版本可以重新执行该评估。

审计存储采用只追加（append-only）存储方式，记录从不修改或删除。保留期限与监管要求相匹配，银行业通常为七到十年。我们使用热温分层组合：热层（最近 90 天存放在 PostgreSQL 中以支持快速查询）和冷层（更早的记录存放在 S3 中，使用 Parquet 格式以实现经济高效的长期存储）。

The audit readiness agent

第四个智能体按批处理计划运行（每周一次和按需触发），而非实时运行。它从审计追踪中汇编审查证据包，按照特定的监管审查模块组织。

当审查通知到达时，合规团队指定适用的审查模块。该智能体查询相应时间段的审计追踪，为每个模块汇编证据——交易监控覆盖率统计数据、告警数量和处置结果、规则变更历史、策略合规指标——并生成结构化的证据包，供审查团队审阅。

对于我们合作过的一家机构来说，正是这个智能体将六周的审查准备工作缩短到了四天。证据已经存在于审计追踪中。智能体的工作是汇编和格式化，而非调查。

The engineering lessons that cost us the most time

有三件事比预期花费了更长时间，值得在你开始之前了解。

确定性规则引擎比看起来更难，因为监管规则在法规文本中并不像看起来那样具有确定性。一条写着"超过 10,000 美元的交易"的规则看似二元的。但随后你发现：10,000 美元的阈值适用于同一客户 24 小时内的聚合交易；聚合逻辑必须考虑同一受益人持有的多个账户之间的交易；"同一客户"有一个特定的法律定义，无法干净地映射到你的客户 ID 字段。原本看起来简单的阈值检查变成了包含实体解析的多步聚合查询。

模型一致性监控消耗的工程工作量超过了模型开发本身。在合规场景下，你需要证明 AI 层在长期内产生一致的结果，即今天评估的相同输入与一个月前评估的结果一致。我们构建了一个回归测试管道，每次模型更新和每次提示词变更时，都会对包含 500 条标注交易的黄金测试集进行重新评估。如果评估结果偏离超过定义的阈值，该更新将被阻止。该管道花费了三周时间构建，自动运行，已经阻止了两次可能导致评估行为发生变化（而合规团队尚未审查过这些变化）的更新。

与遗留通信系统集成以进行政策监控是最长的单项工程任务。该银行的内部通信基础设施包括一个现代电子邮件系统、一个遗留消息平台，以及一个具有专有格式的语音录音系统。构建从这三个来源标准化通信元数据并转换为政策合规 Agent 可评估格式的摄入管道花费了六周时间，比构建 Agent 本身还要长。

运营架构

┌──────────────────────────────────────────────────┐
│              Event Sources                        │
│  Core banking │ Payments │ CRM │ Comms │ Trading │
└───────────────────────┬──────────────────────────┘
                        │ (Event-driven)
                        ▼
┌──────────────────────────────────────────────────┐
│           Stream Processing Layer                 │
│  (Kafka / event bus - normalisation, routing)    │
└───────────────────────┬──────────────────────────┘
                        │
         ┌──────────────┼──────────────┐
         ▼              ▼              ▼
┌──────────────┐ ┌────────────┐ ┌────────────────┐
│ Deterministic│ │ AI-Assisted│ │ Policy         │
│ Rules Engine │ │ Evaluator  │ │ Monitor        │
│ (85% volume) │ │ (15% vol)  │ │ (async)        │
└──────┬───────┘ └─────┬──────┘ └───────┬────────┘
       │               │               │
       └───────────────┼───────────────┘
                       ▼
┌──────────────────────────────────────────────────┐
│           Audit Trail Engine                      │
│  (Append-only │ Immutable │ 7-10yr retention)    │
└───────────────────────┬──────────────────────────┘
                        │
         ┌──────────────┼──────────────┐
         ▼              ▼              ▼
┌──────────────┐ ┌────────────┐ ┌────────────────┐
│ Alert Queue  │ │ Dashboard  │ │ Audit Readiness │
│ (compliance  │ │ (real-time │ │ Agent (batch    │
│  team)       │ │  metrics)  │ │  assembly)      │
└──────────────┘ └────────────┘ └────────────────┘

流处理层是基础架构的核心投入。我们使用 Kafka 进行事件摄入，因为吞吐量需求（每天 200,000+ 事件，子分钟级处理延迟）超过了基于 REST 的架构所能从容处理的范围。每个源系统向 Kafka topics 产生事件。标准化层将特定源的事件格式转换为通用评估 schema。

确定性和 AI 辅助评估路径是同一 Kafka topics 的独立消费者。路由决策——哪些交易需要 AI 评估，哪些仅需确定性处理——由流处理层中的一个轻量级分类器根据交易特征（金额、类型、对手方风险等级、客户风险画像）做出。

构建和运行成本

开发时间线：核心系统（交易监控、审计跟踪、基本监管变更监控）约五个月。政策合规监控和审计就绪 Agent 额外花费两个月。总计：从启动到全面生产部署七个月。

基础设施成本：中型部署每月约 $4,000 到 $7,000。包括 Kafka 集群、AI 层推理计算、PostgreSQL 热存储层和 S3 冷存储。LLM 推理成本是最大的可变组成部分，随路由到 AI 层的交易量而伸缩。

与人工监控的成本对比：一个十二人的合规团队进行抽样审查，年度完全补偿成本约为 $120 万至 $180 万。Agent 系统的构建成本约为 $25 万，年度运营成本约为 $6 万至 $8.5 万。该系统监控 100% 的交易。人工团队仅监控 5%。数学上没有悬念，且随着交易量增长对系统越发有利——因为系统成本亚线性增长，而人工团队成本线性增长。

对于准备构建合规监控的金融机构——对每笔交易依据每条适用规则进行评估，并附带完整的审计跟踪文档——Dextra Labs 的 AI Agent 开发合作伙伴实践覆盖全栈，从流处理架构到确定性规则引擎、AI 辅助评估管道、审计跟踪实现，以及随着监管演进保持规则集时效性的监管变更监控。

合规监控问题本质上是工程问题。AI 组件重要但有界——它们处理需要解读的 15% 的评估。其余 85% 是在流式规模下应用确定性规则，并具备完整可审计性。为合规而构建意味着同时构建两者，并了解哪种评估类型适用于哪条规则。

Published by Dextra Labs, AI Consulting and Enterprise Agent Development