从桌面推演到韧性智能:金融机构如何用智能体 AI 建设可审计的运营韧性

2026-08-18 43 预计阅读时间: 1 分钟
来源: cloud.google.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:13 分钟

金融机构的运营韧性不再只是定期填写一份演练报告。随着欧盟《数字运营韧性法案》(DORA)落地,银行需要更明确地证明:关键业务服务及其数字基础设施能够承受中断,团队可以协同响应,系统能够受控恢复,而且整个过程留下了可复核、可审计的证据。

德意志银行正在探索一种更适合复杂金融系统的路径:用智能体 AI 把架构、数据流、日志、历史事件、告警和运行遥测统一成企业上下文,再自动生成贴近真实生产环境的桌面推演场景、决策节点、模拟证据和审计记录。变化的重点不只是“让 AI 写演练脚本”,而是把韧性测试从人工准备的周期性活动,推进为持续更新的运营智能能力。

为什么静态演练脚本不够用了

大型银行的关键服务通常跨越多个应用、数据流、基础设施组件和第三方服务。一个支付服务的中断,可能同时牵涉身份认证、消息队列、风控系统、数据库、云平台和外部供应商。若桌面推演只依赖几个月前手工维护的系统清单,场景很容易建立在过时假设上。

面向 DORA 或类似监管框架的韧性演练,至少需要回答以下问题:

  • 受影响的关键业务服务是什么,业务影响如何传播?
  • 哪些应用、数据流和第三方依赖位于故障路径上?
  • 团队何时发现问题,谁拥有升级和决策权限?
  • 当前的恢复目标、控制措施和应急手册是否仍然适用?
  • 演练输入、参与者决策、系统响应和改进项能否形成完整证据链?

智能体平台的价值在于把这些问题连接起来。它可以从受控数据源组装上下文,根据实际依赖关系生成带时间线的场景,并把每次推演中的输入、输出、决策和审查结果持久化。

双重编排:既要可追溯,也要能调查

德意志银行采用了双重编排思路,将两类执行模式分开:

  1. 受监管的确定性流程:使用 LangGraph 组织场景生成和审查步骤,让每个输出都能追溯到输入上下文、工具调用和版本信息。这类流程适合监管对可解释性、稳定性和审计留痕的要求。
  2. 自适应的调查流程:使用 Google Agent Development Kit(ADK)支持智能体根据当前条件动态选择工具、分析信号和组织响应。这类流程适合根因分析、异常调查和开放式运营问题。

两者共享同一套企业上下文和工具层,但不强迫所有任务采用同一种执行方式。场景生成需要稳定、可复现;根因调查则往往需要根据新证据临时扩大分析范围。把二者混在一起,会让监管流程难以复核,也会限制调查型任务的灵活性。

可以把平台抽象为下面这条链路:

架构与数据流
      + 日志、告警、历史事件、运行遥测
                    |
                    v
          企业上下文组装与版本化
                    |
          +---------+---------+
          |                   |
          v                   v
  LangGraph 受控场景流      ADK 自适应调查流
          |                   |
          +---------+---------+
                    v
      场景、证据、会话记录、审查结果
                    |
                    v
         Cloud SQL 持久化与持续改进

这个架构的关键不是某个单独的模型,而是边界清晰的治理方式:哪些步骤必须固定,哪些步骤可以由智能体探索,哪些工具允许访问生产数据,以及哪些输出必须经过人工审批。

一个可改造的最小场景生成器

下面的 Python 示例是一个可运行的最小实现,用于演示“受控上下文 + 确定性步骤 + 证据记录”的基本形态。它不代表德意志银行的内部实现,也不连接真实生产系统。实际部署时,可以把 load_context 替换为经过授权的架构目录、日志平台或 CMDB 接口,把规则节点替换为 LangGraph 工作流,把生成结果写入 Cloud SQL 或其他受控存储。

运行前只需要 Python 3.10 或更高版本:

python3 --version
python3 resilience_demo.py

将下面内容保存为 resilience_demo.py

from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import json


@dataclass
class Scenario:
    scenario_id: str
    service: str
    trigger: str
    dependencies: list[str]
    timeline: list[dict[str, str]]
    evidence: list[dict[str, str]]
    context_version: str


def load_context() -> dict:
    # 示例数据;生产环境应从受控且可审计的数据源读取。
    return {
        "context_version": "payments-context-2025-01-15",
        "service": "card-payment-authorisation",
        "dependencies": [
            "identity-api",
            "risk-engine",
            "payment-ledger",
            "message-bus",
            "third-party-fraud-provider",
        ],
        "signals": [
            "authorisation latency above 2 seconds",
            "message-bus consumer lag increasing",
            "fraud-provider timeout rate above 20%",
        ],
    }


def generate_scenario(context: dict) -> Scenario:
    trigger = (
        "The third-party fraud provider has elevated timeout rates, "
        "causing message-bus lag and delayed payment authorisation."
    )

    timeline = [
        {"t_plus": "00:00", "event": "Alert threshold is breached"},
        {"t_plus": "00:10", "event": "Incident commander is assigned"},
        {"t_plus": "00:20", "event": "Fallback and customer-impact decisions are reviewed"},
        {"t_plus": "00:45", "event": "Recovery validation and evidence capture begin"},
    ]

    evidence = [
        {"type": "context", "value": context["context_version"]},
        {"type": "signal", "value": signal}
        for signal in context["signals"]
    ]

    return Scenario(
        scenario_id="tbl-2025-001",
        service=context["service"],
        trigger=trigger,
        dependencies=context["dependencies"],
        timeline=timeline,
        evidence=evidence,
        context_version=context["context_version"],
    )


if __name__ == "__main__":
    context = load_context()
    scenario = generate_scenario(context)
    record = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "scenario": asdict(scenario),
    }
    print(json.dumps(record, indent=2))

这个示例体现了几个适合生产化的原则:上下文有版本号,场景包含明确的服务和依赖,时间线与决策点结构化保存,输入信号进入证据记录。接入模型后,模型可以帮助生成自然语言描述、候选响应和调查问题,但关键元数据、权限边界和审计记录不应只存在于模型输出中。

从桌面推演形成反馈闭环

智能体韧性平台的作用不应止于“生成场景”。每次真实事件都可以为后续演练提供反馈:哪些依赖实际受到影响,告警是否及时,升级路径是否有效,恢复步骤是否与生产系统一致。反过来,桌面推演中的决策结果也可以用于改进应急手册、升级规则和恢复准备度评估。

一个可持续运行的闭环可以这样设计:

# 以下命令展示一种部署和审查流程;镜像名、项目名和区域需要替换为实际值。
gcloud builds submit --tag REGION-docker.pkg.dev/PROJECT/resilience/scenario-generator:latest

gcloud run deploy scenario-generator \
  --image REGION-docker.pkg.dev/PROJECT/resilience/scenario-generator:latest \
  --region REGION \
  --service-account resilience-runtime@PROJECT.iam.gserviceaccount.com \
  --no-allow-unauthenticated

# 生成的场景进入人工审查队列后,再发布给演练参与团队。
gcloud logging read \
  'resource.type="cloud_run_revision" AND severity>=ERROR' \
  --limit=20 \
  --format='table(timestamp,severity,textPayload)'

在一个类似的云端实现中,Cloud Run 可以承载弹性的场景和证据生成服务;Gemini Enterprise Agent Platform 可以将运营上下文转换为结构化韧性场景;ADK 可以编排自适应智能体;Cloud SQL 则可以保存场景、会话工件和审查记录。具体组件选择应服从企业已有的身份、网络、数据保留和审计要求。

落地时需要守住的边界

这类平台处理的是高敏感度运营信息,部署重点不能只放在模型能力上。建议在推广前明确以下控制:

  • 数据分级:限制日志、代码、架构和第三方信息的访问范围,避免把不必要的生产数据送入生成流程。
  • 工具授权:为每个智能体配置最小权限,调查型智能体默认只读,涉及变更或恢复操作时必须经过人工批准。
  • 流程分层:监管演练采用可复现、可追溯的工作流;开放调查允许动态推理,但必须记录工具调用和证据来源。
  • 输出审查:AI 生成的场景、影响判断和恢复建议不能直接视为事实,需要由服务所有者、风险团队或运营团队确认。
  • 证据保留:保存上下文版本、提示词或规则版本、模型版本、工具调用、人工修改、决策和最终批准状态。
  • 质量评估:持续检查场景是否覆盖关键依赖、是否产生误报、是否遗漏第三方风险,以及演练结果能否真正改善恢复能力。

采用清单

金融机构可以从一个关键业务服务开始,而不必一开始就覆盖整个企业:

  • 建立该服务的依赖图和数据源清单。
  • 定义场景生成、调查分析和人工审批的边界。
  • 先实现结构化场景与证据记录,再增加更复杂的智能体推理。
  • 让每次真实事件和每次演练都能更新上下文与改进项。
  • 用可追溯性、响应质量、恢复准备度和审计完整性衡量成效。

运营韧性的成熟方向,是从“按季度做一次演练”转向“持续理解系统当前状态,并用受控方式验证关键能力”。智能体 AI 能够降低上下文整理和场景准备的成本,但真正决定平台能否用于金融监管场景的,仍然是数据治理、编排边界、人工责任和证据质量。


相关推荐