银行AI智能体架构实战：从单体到Skill协同的技术演进

痛点：银行IT架构的三重困境

走在任何一家银行的科技部走廊里，你都能听到同样的叹息：系统又慢了、需求又排不上、监管又来查了。这不是某一家银行的困境，而是整个银行业IT架构的共性问题。我们把它拆解为三重困境。

困境一：单体系统的"牵一发而动全身"

一家城商行的核心信贷系统，单一代码仓库已经膨胀到230万行。改一个授信规则的逻辑分支，回归测试要跑三天。业务提需求时说的"小改一下"，到了开发这边就变成"你确定这不会影响其他47个调用的地方？"

┌─────────────────────────────────────────────┐
│            传统单体架构（230万行代码）           │
│  ┌───────┐ ┌───────┐ ┌───────┐ ┌───────┐    │
│  │授信   │ │风控   │ │贷后   │ │报表   │    │
│  │模块A  │ │模块B  │ │模块C  │ │模块D  │    │
│  └───┬───┘ └───┬───┘ └───┬───┘ └───┬───┘    │
│      └─────┬───┴───────┬┘         │         │
│         ┌──┴───────────┴──┐       │         │
│         │   共享数据库层    │───────┘         │
│         └─────────────────┘                 │
│  痛点：改一处 → 全量回归 → 三天发布周期        │
└─────────────────────────────────────────────┘

困境二：规则引擎与大模型的"各自为战"

风控团队用规则引擎跑决策，客服团队用大模型做意图识别，合规团队用Excel手工比对——三套体系，三种语言，数据标准不统一，模型版本管不住。当监管要求"解释这个审批决策的依据"时，风控说"规则命中了第37条"，大模型说"根据语义分析判断"，合规说"需要人工复核"——三条逻辑链对不上。

困境三：场景碎片化下的"重复造轮子"

零售条线做了个KYC智能体，对公条线也做了个KYC智能体，同业条线还在做第三个。三个团队互不知情，能力重复建设，连实体抽取的Prompt都是各写各的。更糟的是，当反洗钱规则更新时，三个KYC智能体要分别改三轮。

核心观点：银行IT困境的本质不是技术落后，而是架构范式没有跟上AI能力的进化。单体架构服务于确定性逻辑，但AI时代需要的是可组合、可替换、可评估的智能体协同架构。

技术要点总结：三重困境对应三个架构缺陷——耦合度高（单体）、异构系统集成难（规则+模型割裂）、能力复用率低（场景碎片化）。解法的共同方向是：松耦合、标准化接口、技能化封装。

从"买系统"到"养智能体"：数字化转型的三阶段

银行的数字化转型不是一蹴而就的，它经历了三个阶段，每个阶段对应不同的技术架构和治理模式。

阶段一：系统采购期（2010-2018）

这个阶段的特征是"买"。核心系统买IBM的，信贷系统买恒生电子的，CRM买Siebel的。好处是上线快，坏处是——你买了一堆黑箱。当业务提定制需求时，厂商报的高昂定制费让你怀疑人生。

更深层的问题是：数据被锁死在各个系统中。想要跨系统做一个"客户360度画像"，需要从6个系统抽取数据，ETL链路跑4个小时，数据新鲜度T+1。

阶段二：平台建设期（2018-2023）

中台概念兴起，银行开始建数据中台、AI中台、业务中台。方向对了，执行走偏了。很多银行的中台变成了"第二个核心系统"——重建设轻运营，有平台无场景。

以某股份制银行为例，AI中台上线一年，接入场景7个，平均调用量每天不到200次。根本原因：中台提供的是"原始能力"（OCR、NLP、语音识别），而不是"业务技能"（信贷资料智能审查、合规文档比对）。业务团队拿到API后，还要自己封装业务逻辑，门槛没降下来。

阶段三：智能体养成期（2023-至今）

大模型改变了游戏规则。当LLM能够理解业务语言、执行多步骤任务时，银行第一次可以不用"买系统"，而是"养智能体"。

# 阶段三的典型架构：智能体 = LLM + Skills + Memory
class BankingAgent:
    def __init__(self, name: str, skills: list, memory_store: dict):
        self.name = name
        self.llm = LLMEngine(model="qwen2.5-72b")
        self.skills = {
   s.name: s for s in skills}
        self.memory = MemoryStore(memory_store)

    async def execute(self, task: str, context: dict):
        # 第一步：LLM理解任务，选择Skills
        skill_plan = await self.llm.plan(task, available_skills=list(self.skills.keys()))

        # 第二步：按计划依次执行Skills
        results = {
   }
        for step in skill_plan.steps:
            skill = self.skills[step.skill_name]
            results[step.id] = await skill.run(step.params, context)
            # 每步结果写入记忆，供后续步骤使用
            self.memory.set(step.id, results[step.id])

        # 第三步：汇总输出
        return await self.llm.synthesize(task, results)

"养"的含义是：智能体从简单任务起步，在实战中积累经验（Memory），逐步解锁新能力（Skills），最终从"只会一招"变成"独当一面"。这比"买系统"灵活，比"建中台"务实。

核心观点：三阶段的关键转变不是技术换代，而是能力供给模式的根本变化——从"系统交付"到"能力组装"，从"一次性建设"到"持续进化"。

技术要点总结：阶段一的问题在于黑箱不可控；阶段二的问题在于中台离业务太远；阶段三的答案是Skill化封装——把AI能力拆成可复用的业务技能，通过智能体编排实现场景交付。

大模型+规则引擎：金融场景的Hybrid架构

在金融场景中，纯大模型方案和纯规则方案都有致命缺陷。纯大模型不可解释、不可控；纯规则不灵活、维护成本高。两者结合才是正解。

为什么必须Hybrid？

先看一个真实案例。某银行用纯大模型做信贷审批，测试集准确率95%，但上线后被监管叫停。原因：无法解释为什么客户A被拒贷。大模型说"模型综合评估风险较高"，但监管要求的是"命中了哪条规则、权重是多少、依据是什么数据"。

反过来，纯规则引擎也有瓶颈。某银行风控规则从最初的120条增长到4700条，规则之间的冲突越来越多，维护团队从3人膨胀到17人，每次规则更新都要花两周做冲突检测。

Hybrid架构设计

┌──────────────────────────────────────────────────┐
│              Hybrid 决策架构                       │
│                                                    │
│  请求 ──→ ┌──────────┐    ┌──────────┐           │
│            │ 规则引擎  │    │ 大模型    │           │
│            │ (确定性)  │    │ (语义层)  │           │
│            └────┬─────┘    └────┬─────┘           │
│                 │               │                  │
│                 ▼               ▼                  │
│           ┌─────────────────────────┐             │
│           │    决策仲裁层            │             │
│           │  规则优先 → 模型补充     │             │
│           │  冲突时 → 规则胜出       │             │
│           └──────────┬──────────────┘             │
│                      ▼                            │
│           ┌─────────────────────────┐             │
│           │    可解释性输出          │             │
│           │  规则命中 + 模型因子     │             │
│           │  双重审计链路            │             │
│           └─────────────────────────┘             │
└──────────────────────────────────────────────────┘

核心设计原则：规则兜底，模型增强，仲裁决策，全程可解释。

from typing import TypedDict, Optional
from enum import Enum

class DecisionSource(Enum):
    RULE = "rule"
    LLM = "llm"
    HYBRID = "hybrid"

class HybridDecision(TypedDict):
    result: str           # 审批结果：approve/reject/review
    confidence: float     # 置信度 0-1
    source: DecisionSource # 决策来源
    rule_hits: list       # 命中的规则列表
    llm_factors: list     # 大模型贡献的因子
    explanation: str      # 可解释性输出

class HybridDecisionEngine:
    def __init__(self, rule_engine, llm_engine, arbitration_policy="rule_first"):
        self.rule_engine = rule_engine
        self.llm_engine = llm_engine
        self.policy = arbitration_policy

    async def decide(self, application: dict) -> HybridDecision:
        # 第一步：规则引擎先跑，确定性逻辑不进模型
        rule_result = self.rule_engine.evaluate(application)

        # 规则明确命中（硬规则直接决策）
        if rule_result["hard_hit"]:
            return {
   
                "result": rule_result["action"],
                "confidence": 1.0,
                "source": DecisionSource.RULE,
                "rule_hits": rule_result["hits"],
                "llm_factors": [],
                "explanation": self._rule_explanation(rule_result["hits"])
            }

        # 第二步：规则灰色地带，模型补充判断
        llm_result = await self.llm_engine.analyze(
            application=application,
            rule_context=rule_result,  # 把规则结果传给模型，避免重复判断
            prompt="基于申请信息和规则评估结果，补充风险分析"
        )

        # 第三步：仲裁
        final = self._arbitrate(rule_result, llm_result)
        return final

    def _arbitrate(self, rule_result: dict, llm_result: dict) -> HybridDecision:
        # 规则建议拒绝 + 模型建议通过 → 规则优先（监管安全）
        if rule_result["soft_action"] == "reject" and llm_result["tendency"] == "approve":
            return {
   
                "result": "review",  # 升级为人工审核
                "confidence": 0.6,
                "source": DecisionSource.HYBRID,
                "rule_hits": rule_result["hits"],
                "llm_factors": llm_result["factors"],
                "explanation": f"规则引擎标记{rule_result['hits']}，模型倾向通过，建议人工审核"
            }

        # 规则通过 + 模型有疑虑 → 模型补充因子
        if rule_result["soft_action"] == "approve" and llm_result["tendency"] == "reject":
            return {
   
                "result": "review",
                "confidence": 0.7,
                "source": DecisionSource.HYBRID,
                "rule_hits": rule_result["hits"],
                "llm_factors": llm_result["factors"],
                "explanation": f"规则通过但模型识别{llm_result['factors']}，建议人工审核"
            }

        # 双方一致 → 高置信度输出
        return {
   
            "result": rule_result["soft_action"],
            "confidence": 0.95,
            "source": DecisionSource.HYBRID,
            "rule_hits": rule_result["hits"],
            "llm_factors": llm_result["factors"],
            "explanation": f"规则与模型判断一致：{rule_result['soft_action']}"
        }

实战数据

某城商行在消费贷场景部署Hybrid架构后，关键指标变化：

纯大模型准确率虽高，但23%的可解释通过率直接导致它无法上线。Hybrid方案在保持100%可解释的同时，准确率比纯规则提升近8个百分点。

核心观点：在金融场景中，可解释性不是可选项，而是准入条件。Hybrid架构的精髓不是规则和模型的简单堆叠，而是"让规则做规则的强项（确定性、可解释），让模型做模型的强项（语义理解、模式发现），用仲裁层解决冲突"。

技术要点总结：Hybrid架构四要素——规则兜底（硬规则零容忍）、模型增强（灰色地带补充判断）、仲裁决策（规则优先策略）、全程可解释（双重审计链路）。仲裁策略的选择取决于业务场景的监管强度。

7个Skill如何协同：银行智能体架构设计

回到架构核心问题：7个Skill如何协同工作？我们以一个完整的对公信贷场景为例，展示从客户准入到贷后监控的全链路Skill编排。

7个Skill的职责划分

┌─────────────────────────────────────────────────────────┐
│                   银行智能体Skill架构                     │
│                                                           │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐  ┌─────────┐    │
│  │Skill 1  │  │Skill 2  │  │Skill 3  │  │Skill 4  │    │
│  │KYC识别  │  │授信评估  │  │合规审查  │  │文档比对  │    │
│  └────┬────┘  └────┬────┘  └────┬────┘  └────┬────┘    │
│       │             │             │             │         │
│       ▼             ▼             ▼             ▼         │
│  ┌─────────────────────────────────────────────────┐     │
│  │              编排层（Orchestrator）               │     │
│  │   任务分解 → 依赖解析 → 并行调度 → 结果聚合      │     │
│  └─────────────────────┬───────────────────────────┘     │
│                         │                                 │
│  ┌─────────┐  ┌────────┴───┐  ┌─────────┐               │
│  │Skill 5  │  │Skill 6     │  │Skill 7  │               │
│  │风险定价  │  │贷后监控    │  │报告生成  │               │
│  └─────────┘  └────────────┘  └─────────┘               │
└─────────────────────────────────────────────────────────┘

协同编排的核心：依赖图与并行调度

7个Skill不是串行执行的。有些可以并行，有些存在依赖关系。以一笔对公贷款申请为例：

from dataclasses import dataclass
from typing import Dict, Any, List
import asyncio

@dataclass
class SkillResult:
    skill_name: str
    status: str  # success / failed / skipped
    data: Dict[str, Any]
    duration_ms: int

class SkillOrchestrator:
    def __init__(self):
        # 定义Skill依赖关系图
        self.dependency_graph = {
   
            "kyc": [],                    # KYC无依赖，最先执行
            "credit_assess": ["kyc"],     # 授信依赖KYC结果
            "compliance": ["kyc"],        # 合规依赖KYC结果
            "doc_compare": [],            # 文档比对无依赖，可与KYC并行
            "risk_pricing": ["credit_assess", "compliance"],  # 定价依赖评级+合规
            "post_loan": ["credit_assess"],  # 贷后依赖评级
            "report_gen": ["risk_pricing", "doc_compare", "compliance"]  # 报告汇总所有
        }
        self.skill_instances = {
   }

    def register(self, name: str, skill):
        self.skill_instances[name] = skill

    async def execute_pipeline(self, application: dict) -> Dict[str, SkillResult]:
        results = {
   }
        completed = set()

        # 拓扑排序 + 并行调度
        while len(completed) < len(self.dependency_graph):
            # 找出当前可执行的Skill（依赖已全部完成）
            ready = [
                name for name, deps in self.dependency_graph.items()
                if name not in completed 
                and all(d in completed for d in deps)
            ]

            if not ready:
                raise RuntimeError("依赖图存在循环，无法继续执行")

            # 并行执行所有就绪的Skill
            tasks = []
            for skill_name in ready:
                # 构造该Skill的输入：原始申请 + 依赖Skill的输出
                skill_input = self._build_input(skill_name, application, results)
                tasks.append(self._run_skill(skill_name, skill_input))

            # 等待本轮所有Skill完成
            batch_results = await asyncio.gather(*tasks, return_exceptions=True)

            for skill_name, result in zip(ready, batch_results):
                if isinstance(result, Exception):
                    results[skill_name] = SkillResult(
                        skill_name=skill_name, status="failed",
                        data={
   "error": str(result)}, duration_ms=0
                    )
                else:
                    results[skill_name] = result
                completed.add(skill_name)

        return results

    def _build_input(self, skill_name: str, application: dict, 
                     completed_results: dict) -> dict:
        """将依赖Skill的输出注入当前Skill的输入"""
        deps = self.dependency_graph[skill_name]
        enriched = {
   "application": application, "upstream": {
   }}
        for dep in deps:
            if dep in completed_results and completed_results[dep].status == "success":
                enriched["upstream"][dep] = completed_results[dep].data
        return enriched

    async def _run_skill(self, name: str, input_data: dict) -> SkillResult:
        import time
        start = time.time()
        skill = self.skill_instances[name]
        result_data = await skill.run(input_data)
        duration = int((time.time() - start) * 1000)
        return SkillResult(
            skill_name=name, status="success",
            data=result_data, duration_ms=duration
        )

数据流：从申请到审批的完整链路

一笔对公贷款申请的典型数据流：

客户提交申请
    │
    ├──→ [并行] KYC识别 ──→ 输出：客户画像{行业, 规模, 关联企业, 风险标签}
    │         │
    │         ├──→ 授信评估 ──→ 输出：信用等级AA, 建议额度500万
    │         │                    │
    │         │                    ├──→ 风险定价 ──→ 输出：基准利率+0.5%, 利润率2.1%
    │         │                    │                    │
    │         │                    └──→ 贷后监控配置 ──→ 输出：监控频率月度, 预警阈值
    │         │
    │         └──→ 合规审查 ──→ 输出：3项合规提示, 0项硬性违规
    │
    └──→ [并行] 文档比对 ──→ 输出：2处条款差异, 1处风险提示

    最终汇总 ──→ 报告生成 ──→ 输出：审批报告(12页, 含数据溯源)

在这个流程中，KYC和文档比对并行启动，KYC完成后，授信评估和合规审查并行启动，授信评估完成后，风险定价和贷后监控配置并行启动。最终报告生成等待所有前置Skill完成。

串行执行需要约4.2秒，并行调度后缩短到1.8秒，性能提升57%。

核心观点：Skill协同的关键不是"谁调谁"，而是依赖图驱动的并行调度。每个Skill只声明自己的依赖，编排器自动解析并行度。这种设计让新增Skill只需注册依赖关系，无需修改编排逻辑。

技术要点总结：7-Skill架构的核心设计——依赖图声明式编排、并行调度提升吞吐、上游输出自动注入下游输入、失败隔离（单个Skill失败不阻塞整条链路）。新增Skill只需定义依赖，零代码侵入。

评估你的智能体：AI能力度量框架

智能体"够不够聪明"不能只凭感觉。在金融场景中，我们需要一套可量化、可比较、可追踪的评估框架。

四维评估模型

                准确性(Accuracy)
                     ▲
                     │
    可解释性 ────────┼──────── 稳定性
    (Explainability) │   (Stability)
                     │
                     ▼
                效率性(Efficiency)

评估体系实现

from dataclasses import dataclass
from typing import List, Dict
import statistics

@dataclass
class EvalCase:
    input_data: dict
    expected_output: dict
    category: str  # ky_c / credit / compliance / ...
    difficulty: str  # easy / medium / hard

class AgentEvaluator:
    def __init__(self, agent, eval_dataset: List[EvalCase]):
        self.agent = agent
        self.dataset = eval_dataset

    async def run_evaluation(self) -> Dict:
        results = {
   
            "accuracy": await self._eval_accuracy(),
            "explainability": await self._eval_explainability(),
            "stability": await self._eval_stability(),
            "efficiency": await self._eval_efficiency(),
        }
        # 综合评分：金融场景可解释性权重最高
        weights = {
   "accuracy": 0.3, "explainability": 0.3, 
                   "stability": 0.25, "efficiency": 0.15}
        results["composite_score"] = sum(
            results[k]["score"] * weights[k] for k in weights
        )
        return results

    async def _eval_accuracy(self) -> Dict:
        correct = 0
        false_positive = 0  # 误报：正常判为异常
        false_negative = 0  # 漏报：异常判为正常
        total = len(self.dataset)

        for case in self.dataset:
            actual = await self.agent.execute(
                task=case.input_data["task"],
                context=case.input_data["context"]
            )
            if actual["result"] == case.expected_output["result"]:
                correct += 1
            elif case.expected_output["result"] == "normal":
                false_positive += 1  # 正常被误判
            else:
                false_negative += 1  # 异常被漏判

        return {
   
            "score": correct / total,
            "accuracy": correct / total,
            "false_positive_rate": false_positive / total,
            "false_negative_rate": false_negative / total,
            "detail": f"{correct}/{total}正确, {false_positive}误报, {false_negative}漏报"
        }

    async def _eval_explainability(self) -> Dict:
        traceable = 0
        rule_covered = 0
        total = len(self.dataset)

        for case in self.dataset:
            result = await self.agent.execute(
                task=case.input_data["task"],
                context=case.input_data["context"]
            )
            # 检查决策是否可追溯到规则或模型因子
            if result.get("rule_hits") or result.get("llm_factors"):
                traceable += 1
            # 检查是否覆盖了预期的规则
            expected_rules = case.expected_output.get("expected_rules", [])
            if expected_rules:
                actual_rules = [r["rule_id"] for r in result.get("rule_hits", [])]
                if all(r in actual_rules for r in expected_rules):
                    rule_covered += 1

        return {
   
            "score": (traceable / total + rule_covered / total) / 2,
            "traceable_rate": traceable / total,
            "rule_coverage": rule_covered / total,
            "detail": f"可追溯率{traceable/total:.1%}, 规则覆盖率{rule_covered/total:.1%}"
        }

    async def _eval_stability(self) -> Dict:
        # 同一输入多次执行，检查输出一致性
        import random
        sample_cases = random.sample(self.dataset, min(20, len(self.dataset)))
        score_variances = []

        for case in sample_cases:
            scores = []
            for _ in range(5):  # 每个case跑5次
                result = await self.agent.execute(
                    task=case.input_data["task"],
                    context=case.input_data["context"]
                )
                scores.append(result.get("confidence", 0))

            # 计算变异系数
            if statistics.mean(scores) > 0:
                cv = statistics.stdev(scores) / statistics.mean(scores)
                score_variances.append(cv)

        avg_cv = statistics.mean(score_variances) if score_variances else 0
        return {
   
            "score": max(0, 1 - avg_cv * 10),  # CV越小越好
            "avg_cv": avg_cv,
            "detail": f"平均变异系数{avg_cv:.4f}, 目标<0.05"
        }

    async def _eval_efficiency(self) -> Dict:
        import time
        latencies = []

        for case in self.dataset[:50]:  # 取前50个case做性能测试
            start = time.time()
            await self.agent.execute(
                task=case.input_data["task"],
                context=case.input_data["context"]
            )
            latencies.append((time.time() - start) * 1000)

        latencies.sort()
        p95 = latencies[int(len(latencies) * 0.95)] if latencies else 0
        avg = statistics.mean(latencies)

        return {
   
            "score": min(1, 2000 / p95) if p95 > 0 else 1,  # 2秒以内满分
            "p95_ms": p95,
            "avg_ms": avg,
            "detail": f"P95={p95:.0f}ms, 平均={avg:.0f}ms, 目标P95<2000ms"
        }

评估结果的解读

评估结果不是一堆数字，而是行动指南：

• 准确性<90%：先检查数据质量，再优化模型，最后调整规则权重
• 可解释性<100%：不可接受，必须补全决策链路，这是监管红线
• 稳定性CV>0.1：检查Prompt是否有随机性（temperature过高？），加强Few-shot样本
• P95>2s：检查是否有不必要的串行调用，优化Skill编排图

核心观点：金融AI的评估不能用"感觉还行"做结论。四维模型让每个维度都有明确的阈值和行动建议，评估不是为了打分，而是为了定位瓶颈和指导优化方向。

技术要点总结：四维评估模型——准确性（误报+漏报双控）、可解释性（100%红线）、稳定性（变异系数<0.05）、效率性（P95<2s）。综合评分按金融场景加权，可解释性与准确性各占30%。

开源金融AI技能库：56个场景的生态地图

单个银行不可能自己开发所有场景的AI能力。开源金融AI技能库的目标是：让银行聚焦业务逻辑，而不是重复造轮子。

56个场景的六大分类

┌─────────────────────────────────────────────────┐
│           金融AI技能库 · 56场景生态地图           │
│                                                   │
│  ┌──────────────┐  ┌──────────────┐              │
│  │ 风控与合规    │  │ 客户与营销    │              │
│  │ 12个场景     │  │ 10个场景     │              │
│  │ ·反洗钱识别   │  │ ·KYC智能核验 │              │
│  │ ·授信评估     │  │ ·客户画像    │              │
│  │ ·关联交易     │  │ ·精准营销    │              │
│  │ ·监管报送     │  │ ·流失预警    │              │
│  └──────────────┘  └──────────────┘              │
│                                                   │
│  ┌──────────────┐  ┌──────────────┐              │
│  │ 运营与效率    │  │ 产品与定价    │              │
│  │ 10个场景     │  │ 8个场景      │              │
│  │ ·文档比对     │  │ ·利率定价    │              │
│  │ ·智能审批     │  │ ·产品推荐    │              │
│  │ ·流程自动化   │  │ ·组合设计    │              │
│  │ ·知识管理     │  │ ·风险定价    │              │
│  └──────────────┘  └──────────────┘              │
│                                                   │
│  ┌──────────────┐  ┌──────────────┐              │
│  │ 资产与投资    │  │ 安全与审计    │              │
│  │ 8个场景      │  │ 8个场景      │              │
│  │ ·智能投顾     │  │ ·行为审计    │              │
│  │ ·组合优化     │  │ ·数据脱敏    │              │
│  │ ·市场预测     │  │ ·权限管控    │              │
│  │ ·信评分析     │  │ ·模型监控    │              │
│  └──────────────┘  └──────────────┘              │
└─────────────────────────────────────────────────┘

技能库的技术标准

每个Skill入库必须通过三类验收：

# 技能库验收标准（简化版）
SKILL_ACCEPTANCE_CRITERIA = {
   
    # 1. 功能验收：必须有输入输出schema和至少3个测试用例
    "functional": {
   
        "input_schema": True,    # 必须声明输入JSON Schema
        "output_schema": True,   # 必须声明输出JSON Schema
        "test_cases_min": 3,     # 不少于3个测试用例
        "edge_cases_min": 1,     # 至少1个边界用例
    },
    # 2. 安全验收：金融场景的硬性要求
    "security": {
   
        "no_hardcoded_secrets": True,  # 禁止硬编码密钥
        "pii_redaction": True,         # 输出必须脱敏
        "audit_log": True,             # 必须有操作审计日志
        "max_data_retention_days": 30, # 数据保留不超过30天
    },
    # 3. 性能验收：基本SLA
    "performance": {
   
        "p95_latency_ms": 3000,  # P95延迟 < 3秒
        "error_rate_max": 0.01,  # 错误率 < 1%
        "timeout_ms": 10000,     # 超时阈值10秒
    }
}

贡献指南

银行参与开源技能库的路径：

1. 使用方：直接从技能库安装，按需配置参数即可上线
2. 定制方：Fork后修改业务参数（规则阈值、Prompt模板），贡献回上游
3. 贡献方：开发新Skill并提交PR，通过三类验收后合入主线

核心观点：56个场景不是56个独立系统，而是56个标准化的Skill积木。银行的核心竞争力不在积木本身，而在于用积木拼出差异化业务流程的能力。

技术要点总结：技能库生态的三层保障——功能验收（Schema+测试用例）、安全验收（脱敏+审计+密钥管理）、性能验收（SLA红线）。贡献路径分使用、定制、贡献三级，降低参与门槛。

工程化落地：从设计到部署的实战清单

文章最后，给出一份可以直接落地的工程化清单。

架构设计阶段

开发实现阶段

# 工程化开发的核心骨架
from abc import ABC, abstractmethod
from pydantic import BaseModel

class SkillInput(BaseModel):
    """每个Skill必须定义输入Schema"""
    application: dict
    upstream: dict = {
   }

class SkillOutput(BaseModel):
    """每个Skill必须定义输出Schema"""
    result: str
    confidence: float
    rule_hits: list = []
    llm_factors: list = []
    explanation: str = ""
    metadata: dict = {
   }

class BaseSkill(ABC):
    """技能基类：所有金融AI技能必须继承"""

    name: str
    version: str

    @abstractmethod
    async def run(self, input_data: SkillInput) -> SkillOutput:
        pass

    @abstractmethod
    def validate_input(self, input_data: dict) -> bool:
        """输入校验：防止注入和越权"""
        pass

    @abstractmethod
    def redact_output(self, output: SkillOutput) -> SkillOutput:
        """输出脱敏：移除PII和敏感字段"""
        pass

    async def safe_run(self, input_data: SkillInput) -> SkillOutput:
        """安全执行：校验→执行→脱敏→审计"""
        if not self.validate_input(input_data.dict()):
            raise ValueError(f"[{self.name}] 输入校验失败")

        output = await self.run(input_data)
        redacted = self.redact_output(output)

        # 审计日志（结构化，不可篡改）
        self._audit_log(input_data, redacted)

        return redacted

测试验证阶段

三层测试策略：

• 单元测试：每个Skill独立测试，覆盖正常+边界+异常三种case
• 集成测试：2-3个Skill编组测试，验证数据流和依赖解析
• 端到端测试：全链路7个Skill编排测试，含对抗样本和压力测试

部署上线阶段

核心观点：工程化落地的核心不是技术实现，而是每一环都有检查点、每一环都有回退方案。金融AI上线不是"发布"，而是"有监管保障的渐进式交付"。

技术要点总结：工程化四阶段保障——设计阶段（Skill粒度+依赖无环+Hybrid覆盖）、开发阶段（基类约束+输入校验+输出脱敏+审计日志）、测试阶段（单元→集成→端到端三层递进）、部署阶段（灰度+对账+可解释验证+监管报备+应急预案）。

五重困境，一个答案：从单体系统到Skill协同架构，不是技术选型问题，而是银行AI化的必经之路。买系统是阶段一的解法，建中台是阶段二的探索，养智能体才是阶段三的正确答案。而让智能体真正可落地、可监管、可进化的关键，是Hybrid架构保安全、Skill编排提效率、四维评估定方向、生态复用降成本。把这四件事做扎实，银行AI就不是PPT里的愿景，而是实实在在的业务生产力。