禁止任何转载或商用包含规则引擎、机器学习评估和人工复核流程pythonAI审批机制 - 智能审批系统包含规则引擎初筛 ML模型评估 人工复核闭环import numpy as npimport pandas as pdfrom datetime import datetime, timedeltafrom enum import Enumfrom typing import Dict, List, Optional, Tuplefrom dataclasses import dataclass, fieldfrom sklearn.ensemble import RandomForestClassifierfrom sklearn.preprocessing import StandardScalerimport joblibimport jsonimport loggingfrom collections import deque# 配置日志logging.basicConfig(levellogging.INFO)logger logging.getLogger(__name__)class ApprovalStatus(Enum):审批状态枚举PENDING pending # 待审批AUTO_APPROVED auto_approved # 自动通过AUTO_REJECTED auto_rejected # 自动拒绝MANUAL_REVIEW manual_review # 需人工审核MANUAL_APPROVED manual_approved # 人工通过MANUAL_REJECTED manual_rejected # 人工拒绝class RiskLevel(Enum):风险等级LOW lowMEDIUM mediumHIGH highCRITICAL criticaldataclassclass Application:申请数据模型id: strapplicant_id: strapplication_type: str # loan, credit, refund, etc.amount: floatcredit_score: intincome: floatdebt_ratio: floatemployment_years: floatage: intpurpose: strhistorical_defaults: int 0recent_applications: int 0created_at: datetime field(default_factorydatetime.now)# 扩展特征可动态添加extra_features: Dict field(default_factorydict)dataclassclass ApprovalResult:审批结果application_id: strstatus: ApprovalStatusrisk_level: RiskLevelscore: float # 0-100reason: strrules_triggered: List[str] field(default_factorylist)ml_confidence: float 0.0reviewed_by: Optional[str] Nonereviewed_at: Optional[datetime] Nonemetadata: Dict field(default_factorydict)class RuleEngine:规则引擎 - 硬性规则和业务规则def __init__(self):self.rules []self._register_default_rules()def _register_default_rules(self):注册默认规则# 信用分规则self.add_rule(namecredit_score_min,conditionlambda app: app.credit_score 600,actionauto_approve,priority1,description信用分不低于600)self.add_rule(namecredit_score_max,conditionlambda app: app.credit_score 350,actionauto_reject,priority1,description信用分低于350自动拒绝)# 负债率规则self.add_rule(namedebt_ratio_high,conditionlambda app: app.debt_ratio 0.6,actionmanual_review,priority2,description负债率超过60%需人工审核)# 收入与贷款比例self.add_rule(nameloan_to_income,conditionlambda app: app.amount / app.income 5,actionmanual_review,priority2,description贷款金额超过年收入5倍需人工审核)# 历史违约self.add_rule(namehistorical_defaults,conditionlambda app: app.historical_defaults 2,actionauto_reject,priority1,description历史违约超过2次自动拒绝)# 年龄限制self.add_rule(nameage_limit,conditionlambda app: 18 app.age 65,actionauto_reject if False else auto_approve, # 实际逻辑在condition中priority1,description年龄在18-65岁之间)# 近期申请过多self.add_rule(namerecent_applications,conditionlambda app: app.recent_applications 5,actionmanual_review,priority2,description近期申请超过5次需人工审核)def add_rule(self, name: str, condition, action: str, priority: int, description: str):添加规则self.rules.append({name: name,condition: condition,action: action,priority: priority,description: description})# 按优先级排序self.rules.sort(keylambda x: x[priority])def evaluate(self, application: Application) - Tuple[Optional[str], List[str]]:评估规则返回(action, triggered_rules)triggered_rules []action Nonefor rule in self.rules:try:if rule[condition](application):triggered_rules.append(rule[name])# 高优先级规则覆盖低优先级if action is None or rule[priority] self._get_priority(action):action rule[action]except Exception as e:logger.error(f规则 {rule[name]} 执行失败: {e})continuereturn action, triggered_rulesdef _get_priority(self, action: str) - int:获取动作的优先级数字越小优先级越高priority_map {auto_reject: 1,auto_approve: 1,manual_review: 3}return priority_map.get(action, 4)class MLApprovalModel:机器学习审批模型def __init__(self):self.model Noneself.scaler StandardScaler()self.is_trained Falseself.feature_names [credit_score, amount, income, debt_ratio,employment_years, age, historical_defaults,recent_applications]def _extract_features(self, application: Application) - np.ndarray:提取特征向量features [application.credit_score,application.amount,application.income,application.debt_ratio,application.employment_years,application.age,application.historical_defaults,application.recent_applications]return np.array(features).reshape(1, -1)def train(self, X_train: np.ndarray, y_train: np.ndarray):训练模型X_train: 特征矩阵y_train: 标签 (0拒绝, 1通过)# 标准化X_scaled self.scaler.fit_transform(X_train)# 训练随机森林self.model RandomForestClassifier(n_estimators100,max_depth10,random_state42,class_weightbalanced)self.model.fit(X_scaled, y_train)self.is_trained Truelogger.info(ML模型训练完成)def predict(self, application: Application) - Tuple[float, float, str]:预测审批结果返回(分数, 置信度, 建议)if not self.is_trained:raise ValueError(模型未训练请先调用 train() 方法)features self._extract_features(application)features_scaled self.scaler.transform(features)# 预测概率proba self.model.predict_proba(features_scaled)[0]score proba[1] * 100 # 通过概率转为0-100分confidence max(proba) # 置信度# 根据分数给出建议if score 70:suggestion approveelif score 40:suggestion manual_reviewelse:suggestion rejectreturn score, confidence, suggestiondef save_model(self, path: str):保存模型joblib.dump({model: self.model,scaler: self.scaler,feature_names: self.feature_names}, path)logger.info(f模型已保存至: {path})def load_model(self, path: str):加载模型data joblib.load(path)self.model data[model]self.scaler data[scaler]self.feature_names data[feature_names]self.is_trained Truelogger.info(f模型已加载: {path})class HumanReviewQueue:人工审核队列管理def __init__(self, max_size: int 100):self.queue deque(maxlenmax_size)self.review_history []def add_for_review(self, application: Application, ml_score: float, reason: str):添加到审核队列self.queue.append({application: application,ml_score: ml_score,reason: reason,added_at: datetime.now(),priority: self._calculate_priority(ml_score, application)})# 按优先级排序self._sort_queue()def _calculate_priority(self, ml_score: float, application: Application) - int:计算优先级数字越小越优先priority 3 # 默认if ml_score 30 or ml_score 85:priority 1 # 高风险或高分优先审核elif application.amount 100000:priority 2 # 大额申请次优先return prioritydef _sort_queue(self):按优先级排序self.queue deque(sorted(self.queue, keylambda x: x[priority]),maxlenself.queue.maxlen)def get_next(self) - Optional[Dict]:获取下一个待审核项if self.queue:return self.queue.popleft()return Nonedef review_complete(self, application_id: str, decision: str, reviewer: str):完成审核self.review_history.append({application_id: application_id,decision: decision,reviewer: reviewer,reviewed_at: datetime.now()})logger.info(f人工审核完成: {application_id} - {decision})class AIApprovalSystem:AI审批系统主控制器def __init__(self):self.rule_engine RuleEngine()self.ml_model MLApprovalModel()self.review_queue HumanReviewQueue()self.approval_history []self.stats {total: 0,auto_approved: 0,auto_rejected: 0,manual_review: 0,manual_approved: 0,manual_rejected: 0}# 配置参数self.config {ml_threshold_approve: 70, # ML分数高于此值自动通过ml_threshold_reject: 30, # ML分数低于此值自动拒绝ml_confidence_threshold: 0.75, # 置信度阈值enable_auto_learning: True}def process_application(self, application: Application) - ApprovalResult:处理申请 - 核心审批流程self.stats[total] 1logger.info(f处理申请: {application.id})# Step 1: 规则引擎初筛rule_action, triggered_rules self.rule_engine.evaluate(application)# Step 2: ML模型评估try:ml_score, ml_confidence, ml_suggestion self.ml_model.predict(application)except Exception as e:logger.error(fML预测失败: {e})ml_score 50 # 默认中等分数ml_confidence 0.5ml_suggestion manual_review# Step 3: 综合决策status, risk_level, reason self._make_decision(application, rule_action, triggered_rules,ml_score, ml_confidence, ml_suggestion)# Step 4: 需要人工审核的加入队列if status ApprovalStatus.MANUAL_REVIEW:self.review_queue.add_for_review(application, ml_score, reason)# Step 5: 记录结果result ApprovalResult(application_idapplication.id,statusstatus,risk_levelrisk_level,scoreml_score,reasonreason,rules_triggeredtriggered_rules,ml_confidenceml_confidence)# 更新统计self._update_stats(status)self.approval_history.append(result)logger.info(f审批结果: {application.id} - {status.value}, 分数: {ml_score:.2f})return resultdef _make_decision(self, app: Application, rule_action: Optional[str],triggered_rules: List[str], ml_score: float,ml_confidence: float, ml_suggestion: str) - Tuple[ApprovalStatus, RiskLevel, str]:综合决策逻辑# 1. 规则引擎强制决策优先级最高if rule_action auto_reject:return (ApprovalStatus.AUTO_REJECTED,RiskLevel.HIGH,f规则拒绝: {, .join(triggered_rules)})if rule_action auto_approve:# 即使规则通过也检查ML分数是否太低if ml_score self.config[ml_threshold_reject]:return (ApprovalStatus.MANUAL_REVIEW,RiskLevel.MEDIUM,规则通过但ML分数过低转人工审核)return (ApprovalStatus.AUTO_APPROVED,RiskLevel.LOW,f规则通过: {, .join(triggered_rules)})# 2. ML模型决策if ml_confidence self.config[ml_confidence_threshold]:if ml_score self.config[ml_threshold_approve]:return (ApprovalStatus.AUTO_APPROVED,RiskLevel.LOW,fML分数 {ml_score:.1f}置信度 {ml_confidence:.2%})if ml_score self.config[ml_threshold_reject]:return (ApprovalStatus.AUTO_REJECTED,RiskLevel.HIGH,fML分数 {ml_score:.1f}置信度 {ml_confidence:.2%})# 3. 需要人工审核risk_level self._calculate_risk_level(app, ml_score)return (ApprovalStatus.MANUAL_REVIEW,risk_level,f需人工审核: ML分数 {ml_score:.1f}建议: {ml_suggestion})def _calculate_risk_level(self, app: Application, ml_score: float) - RiskLevel:计算风险等级if ml_score 20 or app.historical_defaults 2:return RiskLevel.CRITICALelif ml_score 40 or app.debt_ratio 0.5:return RiskLevel.HIGHelif ml_score 60 or app.amount 50000:return RiskLevel.MEDIUMelse:return RiskLevel.LOWdef _update_stats(self, status: ApprovalStatus):更新统计信息stats_map {ApprovalStatus.AUTO_APPROVED: auto_approved,ApprovalStatus.AUTO_REJECTED: auto_rejected,ApprovalStatus.MANUAL_REVIEW: manual_review,ApprovalStatus.MANUAL_APPROVED: manual_approved,ApprovalStatus.MANUAL_REJECTED: manual_rejected}key stats_map.get(status)if key:self.stats[key] 1def manual_review_decision(self, application_id: str, approved: bool, reviewer: str) - bool:人工审核决策# 在队列中找到该申请for item in self.review_queue.queue:if item[application].id application_id:# 从队列中移除self.review_queue.queue.remove(item)# 记录决策status ApprovalStatus.MANUAL_APPROVED if approved else ApprovalStatus.MANUAL_REJECTEDself._update_stats(status)# 更新历史for record in self.approval_history:if record.application_id application_id:record.status statusrecord.reviewed_by reviewerrecord.reviewed_at datetime.now()breaklogger.info(f人工审核: {application_id} - {通过 if approved else 拒绝}, 审核人: {reviewer})return Truelogger.warning(f申请 {application_id} 不在审核队列中)return Falsedef get_statistics(self) - Dict:获取统计信息stats self.stats.copy()stats[approval_rate] (stats[auto_approved] stats[manual_approved]) / max(stats[total], 1)stats[auto_rate] (stats[auto_approved] stats[auto_rejected]) / max(stats[total], 1)stats[queue_size] len(self.review_queue.queue)return statsdef generate_report(self) - str:生成审批报告stats self.get_statistics()report f AI审批系统报告 总申请数: {stats[total]}自动通过: {stats[auto_approved]} ({stats[auto_approved]/max(stats[total],1)*100:.1f}%)自动拒绝: {stats[auto_rejected]} ({stats[auto_rejected]/max(stats[total],1)*100:.1f}%)人工审核: {stats[manual_review]} ({stats[manual_review]/max(stats[total],1)*100:.1f}%)人工通过: {stats[manual_approved]}人工拒绝: {stats[manual_rejected]}审批通过率: {stats[approval_rate]*100:.1f}%自动化率: {stats[auto_rate]*100:.1f}%当前排队: {stats[queue_size]}return report# 使用示例 def demo():演示AI审批系统# 1. 初始化系统system AIApprovalSystem()# 2. 模拟训练数据实际场景中从数据库获取print(训练ML模型...)X_train np.random.randn(1000, 8) # 模拟特征y_train np.random.randint(0, 2, 1000) # 模拟标签system.ml_model.train(X_train, y_train)# 3. 模拟申请applications [Application(idAPP001,applicant_idU1001,application_typeloan,amount50000,credit_score720,income120000,debt_ratio0.25,employment_years5,age30,purpose购房,historical_defaults0,recent_applications1),Application(idAPP002,applicant_idU1002,application_typecredit,amount300000,credit_score580,income80000,debt_ratio0.65,employment_years2,age22,purpose创业,historical_defaults1,recent_applications3),Application(idAPP003,applicant_idU1003,application_typeloan,amount15000,credit_score450,income40000,debt_ratio0.7,employment_years1,age19,purpose消费,historical_defaults3,recent_applications6),Application(idAPP004,applicant_idU1004,application_typerefund,amount2000,credit_score800,income200000,debt_ratio0.15,employment_years10,age40,purpose退款,historical_defaults0,recent_applications0)]# 4. 处理申请print(\n开始处理申请...)for app in applications:result system.process_application(app)print(f\n申请 {app.id}:)print(f 状态: {result.status.value})print(f 风险等级: {result.risk_level.value})print(f 分数: {result.score:.2f})print(f 原因: {result.reason})# 5. 模拟人工审核print(\n模拟人工审核...)system.manual_review_decision(APP002, approvedTrue, reviewer张三)# 6. 查看统计print(system.generate_report())# 7. 查看审核队列print(f\n待审核队列大小: {len(system.review_queue.queue)})for item in system.review_queue.queue:app item[application]print(f - {app.id}: ML分数 {item[ml_score]:.2f}, 优先级 {item[priority]})if __name__ __main__:demo()核心设计特点1. 三层决策机制· 规则引擎硬性规则快速决策信用分、负债率、历史违约等· ML模型基于历史数据学习复杂模式· 人工审核处理边界情况和复杂案例2. 动态风险控制· 多维度风险评估信用、收入、负债、历史行为· 风险等级分级低/中/高/严重· 自动阈值调节3. 人工审核队列· 优先级排序高风险优先· 审核历史追踪· 闭环反馈4. 监控与统计· 实时统计自动化率、通过率· 详细审批日志· 可生成审计报告5. 扩展性设计· 规则可动态注册· ML模型可在线更新· 支持A/B测试不同决策策略这个框架可以快速适配各种审批场景贷款、信贷、退款、权限申请等且具备自我优化能力。
禁止任何转载或商用包含规则引擎、机器学习评估和人工复核流程pythonAI审批机制 - 智能审批系统包含规则引擎初筛 ML模型评估 人工复核闭环import numpy as npimport pandas as pdfrom datetime import datetime, timedeltafrom enum import Enumfrom typing import Dict, List, Optional, Tuplefrom dataclasses import dataclass, fieldfrom sklearn.ensemble import RandomForestClassifierfrom sklearn.preprocessing import StandardScalerimport joblibimport jsonimport loggingfrom collections import deque# 配置日志logging.basicConfig(levellogging.INFO)logger logging.getLogger(__name__)class ApprovalStatus(Enum):审批状态枚举PENDING pending # 待审批AUTO_APPROVED auto_approved # 自动通过AUTO_REJECTED auto_rejected # 自动拒绝MANUAL_REVIEW manual_review # 需人工审核MANUAL_APPROVED manual_approved # 人工通过MANUAL_REJECTED manual_rejected # 人工拒绝class RiskLevel(Enum):风险等级LOW lowMEDIUM mediumHIGH highCRITICAL criticaldataclassclass Application:申请数据模型id: strapplicant_id: strapplication_type: str # loan, credit, refund, etc.amount: floatcredit_score: intincome: floatdebt_ratio: floatemployment_years: floatage: intpurpose: strhistorical_defaults: int 0recent_applications: int 0created_at: datetime field(default_factorydatetime.now)# 扩展特征可动态添加extra_features: Dict field(default_factorydict)dataclassclass ApprovalResult:审批结果application_id: strstatus: ApprovalStatusrisk_level: RiskLevelscore: float # 0-100reason: strrules_triggered: List[str] field(default_factorylist)ml_confidence: float 0.0reviewed_by: Optional[str] Nonereviewed_at: Optional[datetime] Nonemetadata: Dict field(default_factorydict)class RuleEngine:规则引擎 - 硬性规则和业务规则def __init__(self):self.rules []self._register_default_rules()def _register_default_rules(self):注册默认规则# 信用分规则self.add_rule(namecredit_score_min,conditionlambda app: app.credit_score 600,actionauto_approve,priority1,description信用分不低于600)self.add_rule(namecredit_score_max,conditionlambda app: app.credit_score 350,actionauto_reject,priority1,description信用分低于350自动拒绝)# 负债率规则self.add_rule(namedebt_ratio_high,conditionlambda app: app.debt_ratio 0.6,actionmanual_review,priority2,description负债率超过60%需人工审核)# 收入与贷款比例self.add_rule(nameloan_to_income,conditionlambda app: app.amount / app.income 5,actionmanual_review,priority2,description贷款金额超过年收入5倍需人工审核)# 历史违约self.add_rule(namehistorical_defaults,conditionlambda app: app.historical_defaults 2,actionauto_reject,priority1,description历史违约超过2次自动拒绝)# 年龄限制self.add_rule(nameage_limit,conditionlambda app: 18 app.age 65,actionauto_reject if False else auto_approve, # 实际逻辑在condition中priority1,description年龄在18-65岁之间)# 近期申请过多self.add_rule(namerecent_applications,conditionlambda app: app.recent_applications 5,actionmanual_review,priority2,description近期申请超过5次需人工审核)def add_rule(self, name: str, condition, action: str, priority: int, description: str):添加规则self.rules.append({name: name,condition: condition,action: action,priority: priority,description: description})# 按优先级排序self.rules.sort(keylambda x: x[priority])def evaluate(self, application: Application) - Tuple[Optional[str], List[str]]:评估规则返回(action, triggered_rules)triggered_rules []action Nonefor rule in self.rules:try:if rule[condition](application):triggered_rules.append(rule[name])# 高优先级规则覆盖低优先级if action is None or rule[priority] self._get_priority(action):action rule[action]except Exception as e:logger.error(f规则 {rule[name]} 执行失败: {e})continuereturn action, triggered_rulesdef _get_priority(self, action: str) - int:获取动作的优先级数字越小优先级越高priority_map {auto_reject: 1,auto_approve: 1,manual_review: 3}return priority_map.get(action, 4)class MLApprovalModel:机器学习审批模型def __init__(self):self.model Noneself.scaler StandardScaler()self.is_trained Falseself.feature_names [credit_score, amount, income, debt_ratio,employment_years, age, historical_defaults,recent_applications]def _extract_features(self, application: Application) - np.ndarray:提取特征向量features [application.credit_score,application.amount,application.income,application.debt_ratio,application.employment_years,application.age,application.historical_defaults,application.recent_applications]return np.array(features).reshape(1, -1)def train(self, X_train: np.ndarray, y_train: np.ndarray):训练模型X_train: 特征矩阵y_train: 标签 (0拒绝, 1通过)# 标准化X_scaled self.scaler.fit_transform(X_train)# 训练随机森林self.model RandomForestClassifier(n_estimators100,max_depth10,random_state42,class_weightbalanced)self.model.fit(X_scaled, y_train)self.is_trained Truelogger.info(ML模型训练完成)def predict(self, application: Application) - Tuple[float, float, str]:预测审批结果返回(分数, 置信度, 建议)if not self.is_trained:raise ValueError(模型未训练请先调用 train() 方法)features self._extract_features(application)features_scaled self.scaler.transform(features)# 预测概率proba self.model.predict_proba(features_scaled)[0]score proba[1] * 100 # 通过概率转为0-100分confidence max(proba) # 置信度# 根据分数给出建议if score 70:suggestion approveelif score 40:suggestion manual_reviewelse:suggestion rejectreturn score, confidence, suggestiondef save_model(self, path: str):保存模型joblib.dump({model: self.model,scaler: self.scaler,feature_names: self.feature_names}, path)logger.info(f模型已保存至: {path})def load_model(self, path: str):加载模型data joblib.load(path)self.model data[model]self.scaler data[scaler]self.feature_names data[feature_names]self.is_trained Truelogger.info(f模型已加载: {path})class HumanReviewQueue:人工审核队列管理def __init__(self, max_size: int 100):self.queue deque(maxlenmax_size)self.review_history []def add_for_review(self, application: Application, ml_score: float, reason: str):添加到审核队列self.queue.append({application: application,ml_score: ml_score,reason: reason,added_at: datetime.now(),priority: self._calculate_priority(ml_score, application)})# 按优先级排序self._sort_queue()def _calculate_priority(self, ml_score: float, application: Application) - int:计算优先级数字越小越优先priority 3 # 默认if ml_score 30 or ml_score 85:priority 1 # 高风险或高分优先审核elif application.amount 100000:priority 2 # 大额申请次优先return prioritydef _sort_queue(self):按优先级排序self.queue deque(sorted(self.queue, keylambda x: x[priority]),maxlenself.queue.maxlen)def get_next(self) - Optional[Dict]:获取下一个待审核项if self.queue:return self.queue.popleft()return Nonedef review_complete(self, application_id: str, decision: str, reviewer: str):完成审核self.review_history.append({application_id: application_id,decision: decision,reviewer: reviewer,reviewed_at: datetime.now()})logger.info(f人工审核完成: {application_id} - {decision})class AIApprovalSystem:AI审批系统主控制器def __init__(self):self.rule_engine RuleEngine()self.ml_model MLApprovalModel()self.review_queue HumanReviewQueue()self.approval_history []self.stats {total: 0,auto_approved: 0,auto_rejected: 0,manual_review: 0,manual_approved: 0,manual_rejected: 0}# 配置参数self.config {ml_threshold_approve: 70, # ML分数高于此值自动通过ml_threshold_reject: 30, # ML分数低于此值自动拒绝ml_confidence_threshold: 0.75, # 置信度阈值enable_auto_learning: True}def process_application(self, application: Application) - ApprovalResult:处理申请 - 核心审批流程self.stats[total] 1logger.info(f处理申请: {application.id})# Step 1: 规则引擎初筛rule_action, triggered_rules self.rule_engine.evaluate(application)# Step 2: ML模型评估try:ml_score, ml_confidence, ml_suggestion self.ml_model.predict(application)except Exception as e:logger.error(fML预测失败: {e})ml_score 50 # 默认中等分数ml_confidence 0.5ml_suggestion manual_review# Step 3: 综合决策status, risk_level, reason self._make_decision(application, rule_action, triggered_rules,ml_score, ml_confidence, ml_suggestion)# Step 4: 需要人工审核的加入队列if status ApprovalStatus.MANUAL_REVIEW:self.review_queue.add_for_review(application, ml_score, reason)# Step 5: 记录结果result ApprovalResult(application_idapplication.id,statusstatus,risk_levelrisk_level,scoreml_score,reasonreason,rules_triggeredtriggered_rules,ml_confidenceml_confidence)# 更新统计self._update_stats(status)self.approval_history.append(result)logger.info(f审批结果: {application.id} - {status.value}, 分数: {ml_score:.2f})return resultdef _make_decision(self, app: Application, rule_action: Optional[str],triggered_rules: List[str], ml_score: float,ml_confidence: float, ml_suggestion: str) - Tuple[ApprovalStatus, RiskLevel, str]:综合决策逻辑# 1. 规则引擎强制决策优先级最高if rule_action auto_reject:return (ApprovalStatus.AUTO_REJECTED,RiskLevel.HIGH,f规则拒绝: {, .join(triggered_rules)})if rule_action auto_approve:# 即使规则通过也检查ML分数是否太低if ml_score self.config[ml_threshold_reject]:return (ApprovalStatus.MANUAL_REVIEW,RiskLevel.MEDIUM,规则通过但ML分数过低转人工审核)return (ApprovalStatus.AUTO_APPROVED,RiskLevel.LOW,f规则通过: {, .join(triggered_rules)})# 2. ML模型决策if ml_confidence self.config[ml_confidence_threshold]:if ml_score self.config[ml_threshold_approve]:return (ApprovalStatus.AUTO_APPROVED,RiskLevel.LOW,fML分数 {ml_score:.1f}置信度 {ml_confidence:.2%})if ml_score self.config[ml_threshold_reject]:return (ApprovalStatus.AUTO_REJECTED,RiskLevel.HIGH,fML分数 {ml_score:.1f}置信度 {ml_confidence:.2%})# 3. 需要人工审核risk_level self._calculate_risk_level(app, ml_score)return (ApprovalStatus.MANUAL_REVIEW,risk_level,f需人工审核: ML分数 {ml_score:.1f}建议: {ml_suggestion})def _calculate_risk_level(self, app: Application, ml_score: float) - RiskLevel:计算风险等级if ml_score 20 or app.historical_defaults 2:return RiskLevel.CRITICALelif ml_score 40 or app.debt_ratio 0.5:return RiskLevel.HIGHelif ml_score 60 or app.amount 50000:return RiskLevel.MEDIUMelse:return RiskLevel.LOWdef _update_stats(self, status: ApprovalStatus):更新统计信息stats_map {ApprovalStatus.AUTO_APPROVED: auto_approved,ApprovalStatus.AUTO_REJECTED: auto_rejected,ApprovalStatus.MANUAL_REVIEW: manual_review,ApprovalStatus.MANUAL_APPROVED: manual_approved,ApprovalStatus.MANUAL_REJECTED: manual_rejected}key stats_map.get(status)if key:self.stats[key] 1def manual_review_decision(self, application_id: str, approved: bool, reviewer: str) - bool:人工审核决策# 在队列中找到该申请for item in self.review_queue.queue:if item[application].id application_id:# 从队列中移除self.review_queue.queue.remove(item)# 记录决策status ApprovalStatus.MANUAL_APPROVED if approved else ApprovalStatus.MANUAL_REJECTEDself._update_stats(status)# 更新历史for record in self.approval_history:if record.application_id application_id:record.status statusrecord.reviewed_by reviewerrecord.reviewed_at datetime.now()breaklogger.info(f人工审核: {application_id} - {通过 if approved else 拒绝}, 审核人: {reviewer})return Truelogger.warning(f申请 {application_id} 不在审核队列中)return Falsedef get_statistics(self) - Dict:获取统计信息stats self.stats.copy()stats[approval_rate] (stats[auto_approved] stats[manual_approved]) / max(stats[total], 1)stats[auto_rate] (stats[auto_approved] stats[auto_rejected]) / max(stats[total], 1)stats[queue_size] len(self.review_queue.queue)return statsdef generate_report(self) - str:生成审批报告stats self.get_statistics()report f AI审批系统报告 总申请数: {stats[total]}自动通过: {stats[auto_approved]} ({stats[auto_approved]/max(stats[total],1)*100:.1f}%)自动拒绝: {stats[auto_rejected]} ({stats[auto_rejected]/max(stats[total],1)*100:.1f}%)人工审核: {stats[manual_review]} ({stats[manual_review]/max(stats[total],1)*100:.1f}%)人工通过: {stats[manual_approved]}人工拒绝: {stats[manual_rejected]}审批通过率: {stats[approval_rate]*100:.1f}%自动化率: {stats[auto_rate]*100:.1f}%当前排队: {stats[queue_size]}return report# 使用示例 def demo():演示AI审批系统# 1. 初始化系统system AIApprovalSystem()# 2. 模拟训练数据实际场景中从数据库获取print(训练ML模型...)X_train np.random.randn(1000, 8) # 模拟特征y_train np.random.randint(0, 2, 1000) # 模拟标签system.ml_model.train(X_train, y_train)# 3. 模拟申请applications [Application(idAPP001,applicant_idU1001,application_typeloan,amount50000,credit_score720,income120000,debt_ratio0.25,employment_years5,age30,purpose购房,historical_defaults0,recent_applications1),Application(idAPP002,applicant_idU1002,application_typecredit,amount300000,credit_score580,income80000,debt_ratio0.65,employment_years2,age22,purpose创业,historical_defaults1,recent_applications3),Application(idAPP003,applicant_idU1003,application_typeloan,amount15000,credit_score450,income40000,debt_ratio0.7,employment_years1,age19,purpose消费,historical_defaults3,recent_applications6),Application(idAPP004,applicant_idU1004,application_typerefund,amount2000,credit_score800,income200000,debt_ratio0.15,employment_years10,age40,purpose退款,historical_defaults0,recent_applications0)]# 4. 处理申请print(\n开始处理申请...)for app in applications:result system.process_application(app)print(f\n申请 {app.id}:)print(f 状态: {result.status.value})print(f 风险等级: {result.risk_level.value})print(f 分数: {result.score:.2f})print(f 原因: {result.reason})# 5. 模拟人工审核print(\n模拟人工审核...)system.manual_review_decision(APP002, approvedTrue, reviewer张三)# 6. 查看统计print(system.generate_report())# 7. 查看审核队列print(f\n待审核队列大小: {len(system.review_queue.queue)})for item in system.review_queue.queue:app item[application]print(f - {app.id}: ML分数 {item[ml_score]:.2f}, 优先级 {item[priority]})if __name__ __main__:demo()核心设计特点1. 三层决策机制· 规则引擎硬性规则快速决策信用分、负债率、历史违约等· ML模型基于历史数据学习复杂模式· 人工审核处理边界情况和复杂案例2. 动态风险控制· 多维度风险评估信用、收入、负债、历史行为· 风险等级分级低/中/高/严重· 自动阈值调节3. 人工审核队列· 优先级排序高风险优先· 审核历史追踪· 闭环反馈4. 监控与统计· 实时统计自动化率、通过率· 详细审批日志· 可生成审计报告5. 扩展性设计· 规则可动态注册· ML模型可在线更新· 支持A/B测试不同决策策略这个框架可以快速适配各种审批场景贷款、信贷、退款、权限申请等且具备自我优化能力。
