银行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架构后，关键指标变化：
| 指标 | 纯规则 | 纯大模型 | Hybrid |
|---|
| 审批准确率 | 87.3% | 94.8% | 95.2% |
|---|
| 可解释通过率 | 100% | 23% | 100% |
|---|
| 人工介入率 | 31% | 12% | 8% |
|---|
| 规则维护成本 | 高 | 低 | 中 |
|---|
| 监管检查通过 | 是 | 否 | 是 |
|---|
纯大模型准确率虽高，但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 │ │ │ │风险定价 │ │贷后监控 │ │报告生成 │ │ │ └─────────┘ └────────────┘ └─────────┘ │ └─────────────────────────────────────────────────────────┘| Skill | 职责 | 输入 | 输出 | 关键能力 |
|---|
| KYC识别 | 客户身份核验与画像 | 企业工商信息+法人信息 | 客户画像+风险标签 | 实体抽取、关系图谱 |
|---|
| 授信评估 | 信用评级与额度测算 | 财务报表+征信数据 | 信用等级+建议额度 | 规则引擎+模型评分 |
|---|
| 合规审查 | 监管合规性检查 | 业务资料+法规库 | 合规报告+违规清单 | 法规匹配、条款解读 |
|---|
| 文档比对 | 合同/协议智能比对 | 标准模板+实际文档 | 差异清单+风险提示 | OCR+语义相似度 |
|---|
| 风险定价 | 利率与费率测算 | 评级结果+市场数据 | 定价方案+利润模拟 | 蒙特卡洛+敏感性分析 |
|---|
| 贷后监控 | 还款跟踪与预警 | 账户流水+行为数据 | 预警信号+处置建议 | 时序异常检测 |
|---|
| 报告生成 | 审批报告自动生成 | 前序Skill输出 | 结构化审批报告 | 模板填充+逻辑校验 |
|---|
协同编排的核心：依赖图与并行调度
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)| 维度 | 核心指标 | 金融场景要求 | 评估方法 |
|---|
| 准确性 | 端到端准确率、误报率、漏报率 | 误报<5%，漏报<1% | 标注数据集+AB测试 |
|---|
| 可解释性 | 决策可追溯率、规则覆盖率 | 100%可追溯 | 自动化审计脚本 |
|---|
| 稳定性 | 输出方差、对抗样本鲁棒性 | CV<0.05 | 压力测试+对抗样本 |
|---|
| 效率性 | P95延迟、吞吐量、成本/千次 | P95<2s | 性能基准测试 |
|---|
评估体系实现
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秒 } }贡献指南
银行参与开源技能库的路径：
- 使用方：直接从技能库安装，按需配置参数即可上线
- 定制方：Fork后修改业务参数（规则阈值、Prompt模板），贡献回上游
- 贡献方：开发新Skill并提交PR，通过三类验收后合入主线
核心观点：56个场景不是56个独立系统，而是56个标准化的Skill积木。银行的核心竞争力不在积木本身，而在于用积木拼出差异化业务流程的能力。
技术要点总结：技能库生态的三层保障——功能验收（Schema+测试用例）、安全验收（脱敏+审计+密钥管理）、性能验收（SLA红线）。贡献路径分使用、定制、贡献三级，降低参与门槛。
工程化落地：从设计到部署的实战清单
文章最后，给出一份可以直接落地的工程化清单。
架构设计阶段
| 序号 | 检查项 | 通过标准 | 常见坑 |
|---|
| 1 | Skill拆分粒度 | 单Skill可独立测试、独立部署 | 拆得太细→编排开销大；拆得太粗→无法复用 |
|---|
| 2 | 依赖图无环 | 拓扑排序可执行 | A依赖B、B依赖A的死循环 |
|---|
| 3 | Hybrid架构覆盖 | 每个决策节点都有规则兜底 | 只靠大模型做决策→不可解释 |
|---|
| 4 | 数据流可追溯 | 每条输出可溯源到输入+规则+模型 | 中间结果丢失→审计不过 |
|---|
开发实现阶段
# 工程化开发的核心骨架 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编排测试，含对抗样本和压力测试
部署上线阶段
| 步骤 | 动作 | 关键检查点 |
|---|
| 1 | 灰度发布 | 5%流量→20%→50%→100%，每阶段至少运行3天 |
|---|
| 2 | 精度对账 | 智能体决策 vs 人工决策，偏差>5%暂停放量 |
|---|
| 3 | 可解释性验证 | 随机抽取100条决策，人工复核可追溯性 |
|---|
| 4 | 监管报备 | 提交模型说明文档（含算法逻辑、数据来源、风险控制） |
|---|
| 5 | 应急预案 | 一键回退到规则引擎模式，确保业务不中断 |
|---|
核心观点：工程化落地的核心不是技术实现，而是每一环都有检查点、每一环都有回退方案。金融AI上线不是"发布"，而是"有监管保障的渐进式交付"。
技术要点总结：工程化四阶段保障——设计阶段（Skill粒度+依赖无环+Hybrid覆盖）、开发阶段（基类约束+输入校验+输出脱敏+审计日志）、测试阶段（单元→集成→端到端三层递进）、部署阶段（灰度+对账+可解释验证+监管报备+应急预案）。
五重困境，一个答案：从单体系统到Skill协同架构，不是技术选型问题，而是银行AI化的必经之路。买系统是阶段一的解法，建中台是阶段二的探索，养智能体才是阶段三的正确答案。而让智能体真正可落地、可监管、可进化的关键，是Hybrid架构保安全、Skill编排提效率、四维评估定方向、生态复用降成本。把这四件事做扎实，银行AI就不是PPT里的愿景，而是实实在在的业务生产力。
