银行智能体架构：7个Skill如何协同工作
实战代码：基于
financial-ai-skills项目 | 架构设计 | Skill协同 | 数据流
单体架构 vs 微服务架构
银行系统常见的两种架构：
单体架构: 微服务架构:
┌─────────────┐ ┌─────┐ ┌─────┐ ┌─────┐
│ 核心系统 │ │信贷 │ │风控 │ │营销 │
│ ├─信贷 │ │服务 │ │服务 │ │服务 │
│ ├─风控 │ └─────┘ └─────┘ └─────┘
│ ├─营销 │ ↑ ↑ ↑
│ └─运营 │ ┌─────────────────────┐
└─────────────┘ │ API网关 │
└─────────────────────┘
问题：单体太臃肿，微服务太复杂。
方案：Skill化架构
我设计的架构：
┌─────────────────────────────────────────┐
│ 应用层 (业务系统) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 信贷系统 │ │ 风控系统 │ │ 营销系统 │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
└───────┼───────────┼───────────┼────────┘
│ │ │
└───────────┼───────────┘
↓
┌─────────────────────────────────────────┐
│ Skill层 (7个核心模块) │
│ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │财务 │ │风控 │ │尽调 │ │营销 │ │
│ │智能 │ │合规 │ │ │ │ │ │
│ └─────┘ └─────┘ └─────┘ └─────┘ │
│ ┌─────┐ ┌─────┐ ┌─────┐ │
│ │信贷 │ │运营 │ │财富 │ │
│ │审批 │ │自动化│ │管理 │ │
│ └─────┘ └─────┘ └─────┘ │
└─────────────────────────────────────────┘
│
┌───────────┼───────────┐
↓ ↓ ↓
┌───────────┐ ┌───────────┐ ┌───────────┐
│ 数据层 │ │ 规则引擎 │ │ LLM层 │
│ CSV/SQL │ │ 评分卡 │ │ (可选) │
└───────────┘ └───────────┘ └───────────┘
Skill协同示例
from financial_intelligence import CashflowForecaster
from risk_compliance import FraudDetector
from credit_approval import CreditScorer
# 场景：企业申请贷款
# 需要多个Skill协同
class LoanApprovalWorkflow:
def __init__(self):
self.cashflow = CashflowForecaster()
self.fraud = FraudDetector()
self.credit = CreditScorer()
def process(self, application: dict) -> dict:
"""处理贷款申请"""
results = {
}
# Step 1: 财务分析
print("🔍 Step 1: 财务分析")
cashflow_result = self.cashflow.forecast(
application["financial_data"]
)
results["cashflow"] = cashflow_result
# Step 2: 反欺诈检测
print("🔍 Step 2: 反欺诈检测")
fraud_result = self.fraud.check(
application["transaction_history"]
)
results["fraud"] = fraud_result
# Step 3: 信用评分
print("🔍 Step 3: 信用评分")
credit_result = self.credit.score(
application["credit_data"]
)
results["credit"] = credit_result
# Step 4: 综合决策
print("🔍 Step 4: 综合决策")
decision = self.make_decision(results)
return {
"application_id": application["id"],
"results": results,
"decision": decision,
"timestamp": datetime.now().isoformat()
}
def make_decision(self, results: dict) -> dict:
"""综合决策"""
score = 0
reasons = []
# 现金流评分 (权重30%)
cf_score = results["cashflow"]["health_score"]
score += cf_score * 0.3
# 反欺诈评分 (权重30%)
fraud_score = 100 - results["fraud"]["risk_score"]
score += fraud_score * 0.3
# 信用评分 (权重40%)
credit_score = results["credit"]["score"]
score += credit_score * 0.4
# 决策
if score >= 80:
decision = "APPROVE"
suggestion = "建议审批"
elif score >= 60:
decision = "REVIEW"
suggestion = "建议人工复核"
else:
decision = "REJECT"
suggestion = "建议拒绝"
return {
"score": score,
"decision": decision,
"suggestion": suggestion,
"reasons": reasons
}
# 使用
workflow = LoanApprovalWorkflow()
result = workflow.process({
"id": "APP001",
"financial_data": {
...},
"transaction_history": {
...},
"credit_data": {
...}
})
print(f"决策: {result['decision']['decision']}")
print(f"评分: {result['decision']['score']}")
print(f"建议: {result['decision']['suggestion']}")
数据流设计
# Skill间的数据流
class DataFlow:
"""数据流管理器"""
def __init__(self):
self.pipeline = []
def add_step(self, skill, input_mapping, output_mapping):
"""添加处理步骤"""
self.pipeline.append({
"skill": skill,
"input_mapping": input_mapping,
"output_mapping": output_mapping
})
def execute(self, initial_data: dict) -> dict:
"""执行数据流"""
data = initial_data.copy()
for step in self.pipeline:
skill = step["skill"]
# 映射输入
inputs = {
}
for key, path in step["input_mapping"].items():
inputs[key] = self._get_value(data, path)
# 执行Skill
result = skill.process(**inputs)
# 映射输出
for key, path in step["output_mapping"].items():
self._set_value(data, path, result[key])
return data
def _get_value(self, data: dict, path: str):
"""获取嵌套值"""
keys = path.split(".")
value = data
for key in keys:
value = value.get(key, {
})
return value
def _set_value(self, data: dict, path: str, value):
"""设置嵌套值"""
keys = path.split(".")
target = data
for key in keys[:-1]:
if key not in target:
target[key] = {
}
target = target[key]
target[keys[-1]] = value
# 定义贷款审批数据流
flow = DataFlow()
# Step 1: 财务分析
flow.add_step(
skill=CashflowForecaster(),
input_mapping={
"data": "application.financial_data"},
output_mapping={
"health_score": "results.cashflow.health_score"}
)
# Step 2: 反欺诈
flow.add_step(
skill=FraudDetector(),
input_mapping={
"transactions": "application.transaction_history"},
output_mapping={
"risk_score": "results.fraud.risk_score"}
)
# Step 3: 信用评分
flow.add_step(
skill=CreditScorer(),
input_mapping={
"credit_data": "application.credit_data"},
output_mapping={
"score": "results.credit.score"}
)
# 执行
result = flow.execute({
"application": {
...}
})
配置化Skill加载
# skill_config.yaml
skills:
financial-intelligence:
enabled: true
version: "1.0"
config:
forecast_horizon: 30
risk-compliance:
enabled: true
version: "1.1"
config:
min_transaction_amount: 100000
credit-approval:
enabled: true
version: "1.0"
config:
min_score: 75
max_auto_amount: 500000
# 加载配置
import yaml
class SkillManager:
def __init__(self, config_path: str):
with open(config_path) as f:
self.config = yaml.safe_load(f)
self.skills = {
}
def load_skills(self):
"""加载所有Skill"""
for name, cfg in self.config["skills"].items():
if cfg["enabled"]:
self.skills[name] = self._load_skill(name, cfg)
def _load_skill(self, name: str, cfg: dict):
"""加载单个Skill"""
module = __import__(f"skills.{name.replace('-', '_')}")
skill_class = getattr(module, name.replace("-", " ").title().replace(" ", ""))
return skill_class(**cfg.get("config", {
}))
def get_skill(self, name: str):
"""获取Skill"""
return self.skills.get(name)
# 使用
manager = SkillManager("skill_config.yaml")
manager.load_skills()
credit_skill = manager.get_skill("credit-approval")
result = credit_skill.score(application_data)
监控与日志
class SkillMonitor:
"""Skill监控器"""
def __init__(self):
self.metrics = {
}
def record(self, skill_name: str, operation: str, duration: float, success: bool):
"""记录指标"""
key = f"{skill_name}.{operation}"
if key not in self.metrics:
self.metrics[key] = {
"count": 0,
"total_duration": 0,
"success_count": 0,
"fail_count": 0
}
self.metrics[key]["count"] += 1
self.metrics[key]["total_duration"] += duration
if success:
self.metrics[key]["success_count"] += 1
else:
self.metrics[key]["fail_count"] += 1
def get_report(self) -> dict:
"""生成监控报告"""
report = {
}
for key, metrics in self.metrics.items():
report[key] = {
"total_calls": metrics["count"],
"avg_duration": metrics["total_duration"] / metrics["count"],
"success_rate": metrics["success_count"] / metrics["count"] * 100,
"error_rate": metrics["fail_count"] / metrics["count"] * 100
}
return report
# 装饰器自动记录
import time
import functools
def monitored(skill_name: str, operation: str):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
try:
result = func(*args, **kwargs)
monitor.record(skill_name, operation, time.time() - start, True)
return result
except Exception as e:
monitor.record(skill_name, operation, time.time() - start, False)
raise
return wrapper
return decorator
# 使用
class CreditScorer:
@monitored("credit-approval", "score")
def score(self, data: dict) -> dict:
# 评分逻辑
pass
完整代码：https://github.com/yuzhaopeng-up/financial-ai-skills
#架构设计 #Skill协同 #数据流 #银行系统 #微服务
