人机回环(Human-in-the-loop)在金融风控 Agent 中的落地模式
人机回环(Human-in-the-loop)在金融风控 Agent 中的落地模式
关键词:人机回环、金融风控Agent、主动学习、不确定性估计、风控合规、大模型风控、闭环迭代
摘要:本文针对金融风控场景中纯AI模型误判率高、可解释性不足、合规性难满足,纯人工审核效率低、成本高的痛点,系统讲解人机回环(HITL)技术的核心原理,以及其与金融风控Agent结合的三种落地模式,从概念解析、算法原理、代码实战、场景应用、最佳实践等多个维度,手把手教你搭建可落地的人机回环风控系统,同时分析行业发展趋势与面临的挑战,帮助金融科技从业者解决风控业务的实际痛点。
背景介绍
目的和范围
前两年AI技术爆发的时候,很多金融机构喊出了"全自动化风控"的口号,希望用AI模型100%替代人工审核,结果踩了无数坑:要么是模型对新出现的欺诈手段完全识别不出来,坏账率飙升;要么是模型误拒了很多优质客户,业务量掉了30%;更麻烦的是监管要求风控决策可解释、可追溯,黑箱AI模型根本过不了合规检查。
本文的目的就是给大家提供一个成熟的解决方案:人机回环风控系统。我们会覆盖从核心概念到落地代码的全流程,适用于信用卡申请、消费信贷、反欺诈、贷后管理、供应链金融等所有金融风控场景,同时兼顾效率、准确率和合规要求。
预期读者
本文适合以下人群阅读:
- 金融科技公司的算法工程师、风控模型工程师
- 银行、消费金融公司的风控业务负责人、产品经理
- 金融系统架构师、合规审计人员
- 对Agent技术、人机协同技术感兴趣的技术从业者
哪怕你没有金融行业背景,只要懂基础的Python代码和机器学习概念,就能完全看懂本文。
文档结构概述
本文会按照"概念→原理→实战→应用→趋势"的逻辑展开:首先用生活化的例子讲解人机回环、风控Agent的核心概念,然后讲解背后的主动学习、不确定性估计算法原理,接着带大家从零搭建一个完整的信贷风控人机回环Demo,然后分析不同场景的落地模式和最佳实践,最后讨论行业发展趋势和挑战。
术语表
核心术语定义
- 人机回环(HITL, Human-in-the-loop):一种AI系统的运行模式,AI会自动处理置信度高的任务,将置信度低、边界模糊的任务推给人类专家处理,人类的反馈会回流到AI模型中,不断迭代优化模型效果,形成"AI推理→人工干预→模型迭代"的闭环。
- 金融风控Agent:具备自主感知、推理、决策能力的AI系统,专门用于金融风险防控场景,比如自动审核信贷申请、识别欺诈交易、监控贷后风险等。
- 不确定性估计:衡量AI模型对自己的预测结果的"自信程度"的技术,是触发人工干预的核心依据。
- 主动学习:一种机器学习范式,模型可以主动选择最有价值的样本请求人工标注,用最少的标注成本获得最大的模型效果提升。
相关概念解释
- 风控召回率:指风控模型识别出的风险用户占全部真实风险用户的比例,召回率越高,漏判的风险越少。
- 风控精确率:指模型识别出的风险用户中,真实风险用户的比例,精确率越高,误拒的优质用户越少。
- 合规可追溯:金融监管要求所有风控决策都要留痕,能够查到决策的依据、操作人员、时间等信息,满足审计要求。
缩略词列表
| 缩略词 | 全称 | 中文含义 |
|---|---|---|
| HITL | Human-in-the-loop | 人机回环 |
| AL | Active Learning | 主动学习 |
| UE | Uncertainty Estimation | 不确定性估计 |
| LGD | Loss Given Default | 违约损失率 |
| PD | Probability of Default | 违约概率 |
核心概念与联系
故事引入
我先给大家讲个奶茶店品控的故事,你一下子就能明白人机回环在风控里的作用:
假设你开了一家连锁奶茶店,雇了10个品控员,每天要检查10000杯奶茶是不是合格,合格就出餐,不合格就倒掉。
一开始你全靠人工检查,每个品控员每天要查1000杯,累得要死,每个月工资就要发5万块,还经常有看走眼的时候,要么把坏的奶茶放出去被客户投诉,要么把好的奶茶倒掉浪费成本。
后来你买了个智能品控机,用图像识别+气味识别来判断奶茶合不合格,90%的情况都能判断对,成本只要1万块一个月,你特别开心,就把人工都开了,结果用了半个月就出事了:新出的芋泥啵啵奶茶因为颜色本来就偏深,智能品控机全当成坏的给扔了,损失了十几万;还有一种新的变质情况,机器从来没见过,全给放出去了,被投诉赔了好多钱。
最后你找到了最优方案:智能品控机先检查所有奶茶,它觉得自己100%确定合格或者不合格的,就自动处理;它拿不准的(比如颜色奇怪的新品、闻起来有点怪但又不确定变质的),就推给剩下的2个品控员来判断。品控员判断的结果,你再喂给智能品控机学习,慢慢的机器拿不准的情况越来越少,最后只需要1个品控员兜底就够了,成本只要2万一个月,准确率比纯人工还高,还从来不会漏判新的风险。
这个故事里,智能品控机就是我们说的金融风控Agent,品控员就是风控专家,这套"机器先审,拿不准的找人工,人工结果喂给机器学习"的模式,就是人机回环。
核心概念解释
我再用更通俗的话把三个核心概念讲透:
核心概念一:人机回环(HITL)
人机回环就像你用的智能输入法:你打字的时候,输入法会自动给你联想候选词,大部分时候都是对的,你直接选就行;如果它给出的候选词不对,你就手动输入正确的词,输入法会记住你的输入习惯,下次再遇到类似的情况就会给出正确的候选词。
这里的关键点是"回环":不是简单的AI干一部分,人工干一部分,而是人工的反馈必须回流到AI模型里,让AI越来越聪明,需要人工干预的情况越来越少,形成正向循环。如果没有反馈回流,那只能叫人机并行,不是人机回环。
核心概念二:金融风控Agent
金融风控Agent就像银行的实习审核员:刚入职的时候,他只会按照培训的规则审核申请,遇到没见过的情况就拿不准,要找老员工请教;跟着老员工学的案例多了,他的经验越来越丰富,能自己处理的情况越来越多,最后几乎不用请教老员工就能独立完成审核。
和普通的风控模型不一样,风控Agent是具备自主迭代能力的:它可以自己判断哪些情况自己拿不准,主动找专家请教,还能自动学习专家的判断结果,不需要算法工程师每次手动重新训练模型。
核心概念三:不确定性估计
不确定性估计就是AI的"自信度评分":比如AI判断一个用户的逾期概率是30%,如果它见过1000个类似特征的用户,其中确实300个逾期了,那它的自信度就很高,可能有95分;如果它只见过2个类似特征的用户,一个逾期一个没逾期,那它的自信度就很低,可能只有40分。
我们就是根据这个自信度评分来决定要不要找人工审核:自信度超过阈值的,AI直接处理;自信度低于阈值的,推给人工。
核心概念之间的关系
三个核心概念是互相依存的,我给大家做个对比和关系说明:
三种风控模式的对比
| 模式 | 准确率 | 效率 | 成本 | 可解释性 | 合规性 | 应对新风险的能力 |
|---|---|---|---|---|---|---|
| 纯人工 | 中等(看人) | 低 | 高 | 好 | 好 | 强 |
| 纯AI | 高(常见场景) | 高 | 低 | 差 | 差 | 弱 |
| 人机回环 | 高(全场景) | 中高 | 中低 | 好 | 好 | 强 |
概念之间的关系
- 人机回环是风控Agent的"学习机制":就像实习审核员需要老员工带才能成长一样,风控Agent需要人机回环的机制才能不断迭代优化,否则就是个固定的静态模型,遇到新的风险就会失效。
- 不确定性估计是人机回环的"触发开关":如果没有办法衡量AI的自信度,你就不知道哪些案例该推给人工,要么推的太多增加人工成本,要么推的太少增加误判风险。
- 风控Agent是人机回环的"执行载体":人机回环的所有逻辑,包括推理、不确定性计算、人工任务派发、反馈回流、模型迭代,都需要运行在风控Agent这个载体上,否则就是零散的工具,没办法形成闭环。
核心概念原理和架构的文本示意图
[用户风控请求] → [数据采集/预处理] → [风控Agent推理层]
↓
[不确定性计算模块] → [置信度≥阈值?]
↓ 是 → [自动决策] → [结果存储]
↓ 否 → [人工审核队列] → [风控专家标注]
↓
[模型迭代模块] ← [标注结果入库]
↓
[更新风控Agent推理层]
[审计日志模块] 记录全流程所有操作,满足合规要求
Mermaid 架构图
Mermaid 交互流程图
核心算法原理 & 具体操作步骤
人机回环风控系统的核心算法主要是两个:不确定性估计算法和主动学习算法,我们一个个来讲。
1. 不确定性估计算法
不确定性估计的目标是给模型的每个预测结果打一个"自信分",常用的方法有三种,我们用Python代码来演示:
方法一:概率置信度法(最简单,适合传统机器学习模型)
对于分类模型,我们直接取预测概率的最大值作为置信度,公式如下:
Confidence=max(p1,p2,...,pn)Confidence = max(p_1, p_2, ..., p_n)Confidence=max(p1,p2,...,pn)
其中pip_ipi是模型预测样本属于第iii类的概率。
比如风控二分类模型预测一个用户的逾期概率是0.05,不逾期的概率是0.95,那么置信度就是0.95;如果预测逾期概率是0.48,不逾期概率是0.52,置信度就是0.52,显然这个结果是模棱两可的,需要推人工。
Python代码示例:
import lightgbm as lgb
import numpy as np
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
# 生成模拟信贷风控数据:10000个样本,20个特征,二分类(0=正常,1=逾期)
X, y = make_classification(n_samples=10000, n_features=20, n_informative=15, n_classes=2, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 训练LightGBM分类模型
model = lgb.LGBMClassifier(objective='binary', random_state=42, verbose=-1)
model.fit(X_train, y_train)
# 预测测试集的概率
y_proba = model.predict_proba(X_test)
# 计算置信度:取每个样本预测概率的最大值
confidence = np.max(y_proba, axis=1)
# 设置置信度阈值为0.9,低于0.9的推人工
threshold = 0.9
need_human = confidence < threshold
print(f"需要人工审核的样本比例:{np.mean(need_human):.2%}")
# 输出:需要人工审核的样本比例:18.25%
方法二:信息熵法(更科学,适合多分类场景)
信息熵衡量的是预测结果的混乱程度,熵越高,不确定性越高,公式如下:
H=−∑i=1npilog(pi)H = -\sum_{i=1}^{n} p_i \log(p_i)H=−i=1∑npilog(pi)
其中pip_ipi是预测属于第iii类的概率,HHH的取值范围是0到log(n)\log(n)log(n),熵越高,模型越不确定。
Python代码示例:
def calculate_entropy(proba):
# 避免log(0)的情况
proba = np.clip(proba, 1e-10, 1 - 1e-10)
entropy = -np.sum(proba * np.log(proba), axis=1)
return entropy
# 计算每个样本的熵
entropy = calculate_entropy(y_proba)
# 熵大于阈值的推人工,这里阈值设为0.3
entropy_threshold = 0.3
need_human_entropy = entropy > entropy_threshold
print(f"信息熵法需要人工审核的样本比例:{np.mean(need_human_entropy):.2%}")
# 输出:信息熵法需要人工审核的样本比例:16.70%
方法三:蒙特卡洛Dropout法(适合深度学习/大模型场景)
对于深度学习模型或者大模型,我们可以在推理的时候开启Dropout,多次推理得到多个预测结果,结果的方差越大,不确定性越高。
公式如下:
Uncertainty=Var(p1,p2,...,pT)Uncertainty = Var(p_1, p_2, ..., p_T)Uncertainty=Var(p1,p2,...,pT)
其中TTT是推理的次数,pip_ipi是第iii次推理得到的预测概率,方差越大,不确定性越高。
Python代码示例(以简单的神经网络为例):
import tensorflow as tf
from tensorflow.keras.layers import Dense, Dropout
from tensorflow.keras.models import Sequential
# 构建带Dropout的神经网络模型
def build_model():
model = Sequential([
Dense(64, activation='relu', input_shape=(20,)),
Dropout(0.2),
Dense(32, activation='relu'),
Dropout(0.2),
Dense(2, activation='softmax')
])
model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])
return model
model = build_model()
model.fit(X_train, y_train, epochs=20, batch_size=32, validation_split=0.1, verbose=0)
# 蒙特卡洛Dropout推理:推理T次,开启Dropout
T = 10
predictions = []
for _ in range(T):
# 训练模式开启Dropout,推理的时候也用训练模式
pred = model(X_test, training=True)
predictions.append(pred.numpy())
predictions = np.array(predictions)
# 计算方差作为不确定性
uncertainty = np.var(predictions, axis=0).mean(axis=1)
# 方差大于阈值的推人工
mc_threshold = np.percentile(uncertainty, 80) # 取前20%不确定的推人工
need_human_mc = uncertainty > mc_threshold
print(f"蒙特卡洛Dropout法需要人工审核的样本比例:{np.mean(need_human_mc):.2%}")
# 输出:蒙特卡洛Dropout法需要人工审核的样本比例:20.00%
2. 主动学习算法
主动学习的目标是从所有需要人工标注的样本里,选择最有价值的样本给人工标注,用最少的标注成本获得最大的模型效果提升,常用的采样策略有三种:
- 不确定性采样:就是我们上面讲的,选不确定性最高的样本标注,最常用,实现最简单。
- 委员会查询法:训练多个模型组成"委员会",选多个模型预测结果不一致的样本标注,适合对准确率要求高的场景。
- 预期误差减少法:选标注之后能最大程度减少模型整体误差的样本,效果最好,但计算成本最高,适合小样本场景。
主动学习的完整流程是: - 用初始标注数据训练基础模型
- 用模型预测未标注数据,计算每个样本的价值
- 选择Top N个最有价值的样本给人工标注
- 把标注后的样本加入训练集,重新训练模型
- 重复步骤2-4,直到模型效果达到业务要求
数学模型和公式 & 详细讲解 & 举例说明
我们把人机回环风控系统的核心数学模型整理成可量化的公式,方便大家根据自己的业务场景调整参数。
1. 人工审核阈值优化模型
我们的目标是在满足业务风险约束的前提下,最小化人工审核成本,公式如下:
mintCost=Ch∗P(Confidence<t)+Cr∗P(Error∣Confidence≥t)\min_{t} Cost = C_h * P(Confidence < t) + C_r * P(Error | Confidence \geq t)tminCost=Ch∗P(Confidence<t)+Cr∗P(Error∣Confidence≥t)
约束条件:
Recall≥RminRecall \geq R_{min}Recall≥Rmin
其中:
- ttt是置信度阈值,我们要优化的变量
- ChC_hCh是单样本人工审核成本
- P(Confidence<t)P(Confidence < t)P(Confidence<t)是需要人工审核的样本比例
- CrC_rCr是单样本误判带来的损失(比如坏账损失、客户流失损失)
- P(Error∣Confidence≥t)P(Error | Confidence \geq t)P(Error∣Confidence≥t)是AI自动处理的样本的误判率
- RminR_{min}Rmin是业务要求的最低召回率
举个例子:假设你们的消费信贷业务,单个人工审核成本是2元,每个误判带来的平均损失是1000元,业务要求风险召回率不能低于95%。我们可以计算不同阈值下的总成本:
| 阈值t | 人工审核比例 | AI误判率 | 总成本(每10000样本) | 召回率 |
|---|---|---|---|---|
| 0.8 | 10% | 1% | 10%100002 + 90%100001%*1000 = 2000 + 90000 = 92000元 | 92% |
| 0.9 | 20% | 0.2% | 20%100002 + 80%100000.2%*1000 = 4000 + 16000 = 20000元 | 96% |
| 0.95 | 35% | 0.05% | 35%100002 + 65%100000.05%*1000 = 7000 + 3250 = 10250元 | 98.5% |
显然阈值设为0.9的时候,既满足了召回率≥95%的要求,总成本也最低。
2. 模型迭代效果预测模型
我们可以用以下公式预测模型迭代N轮之后的准确率:
Acc(N)=Accmax−(Accmax−Acc0)∗e−αNAcc(N) = Acc_{max} - (Acc_{max} - Acc_0) * e^{-\alpha N}Acc(N)=Accmax−(Accmax−Acc0)∗e−αN
其中:
- Acc(N)Acc(N)Acc(N)是迭代N轮之后的模型准确率
- AccmaxAcc_{max}Accmax是模型能达到的最高准确率(理论上限)
- Acc0Acc_0Acc0是初始模型的准确率
- α\alphaα是学习效率,和每次标注的样本质量有关
比如初始准确率是85%,理论上限是97%,学习效率α是0.3,那么迭代5轮之后的准确率是:
Acc(5)=0.97−(0.97−0.85)∗e−0.3∗5=0.97−0.12∗e−1.5≈0.97−0.12∗0.223≈0.943,即94.3Acc(5) = 0.97 - (0.97 - 0.85) * e^{-0.3*5} = 0.97 - 0.12 * e^{-1.5} ≈ 0.97 - 0.12*0.223 ≈ 0.943,即94.3%Acc(5)=0.97−(0.97−0.85)∗e−0.3∗5=0.97−0.12∗e−1.5≈0.97−0.12∗0.223≈0.943,即94.3
项目实战:代码实际案例和详细解释说明
我们来从零搭建一个完整的信贷申请风控人机回环系统,包含模型推理、不确定性计算、人工审核界面、模型自动迭代四个模块。
开发环境搭建
首先安装需要的依赖包:
pip install fastapi uvicorn lightgbm pandas numpy scikit-learn gradio sqlalchemy python-multipart
我们用FastAPI做后端接口,Gradio做人工审核的前端界面,SQLite做数据存储,LightGBM做风控模型。
源代码详细实现和代码解读
第一步:数据库模型定义,存储申请数据、标注结果、审计日志
# db.py
from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
SQLALCHEMY_DATABASE_URL = "sqlite:///./risk_control.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# 信贷申请表
class CreditApplication(Base):
__tablename__ = "credit_applications"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(String, index=True)
features = Column(String) # 存储JSON格式的特征数据
pd = Column(Float) # 模型预测的违约概率
confidence = Column(Float) # 模型置信度
need_human = Column(Boolean) # 是否需要人工审核
status = Column(String, default="pending") # 状态:pending/approve/reject
operator = Column(String, nullable=True) # 审核人
create_time = Column(DateTime, default=datetime.utcnow)
update_time = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# 审计日志表
class AuditLog(Base):
__tablename__ = "audit_logs"
id = Column(Integer, primary_key=True, index=True)
application_id = Column(Integer, index=True)
operation = Column(String)
operator = Column(String)
detail = Column(String)
create_time = Column(DateTime, default=datetime.utcnow)
Base.metadata.create_all(bind=engine)
第二步:模型工具类,包含模型训练、推理、不确定性计算、迭代逻辑
# model.py
import lightgbm as lgb
import numpy as np
import json
import pandas as pd
from db import SessionLocal, CreditApplication
class RiskModel:
def __init__(self):
self.model = None
self.threshold = 0.9 # 置信度阈值
self.load_model()
def load_model(self):
try:
self.model = lgb.Booster(model_file='risk_model.txt')
except:
# 模型不存在,初始化训练
self.init_train()
def init_train(self):
# 生成初始训练数据
from sklearn.datasets import make_classification
X, y = make_classification(n_samples=1000, n_features=20, n_informative=15, n_classes=2, random_state=42)
train_data = lgb.Dataset(X, label=y)
params = {'objective': 'binary', 'verbose': -1}
self.model = lgb.train(params, train_data, num_boost_round=100)
self.model.save_model('risk_model.txt')
def predict(self, features):
# features是长度为20的特征数组
features = np.array(features).reshape(1, -1)
proba = self.model.predict(features)[0]
confidence = max(proba, 1 - proba)
need_human = confidence < self.threshold
return {
"pd": float(proba),
"confidence": float(confidence),
"need_human": need_human
}
def retrain(self):
# 从数据库取所有已经标注的样本重新训练
db = SessionLocal()
applications = db.query(CreditApplication).filter(CreditApplication.status != "pending").all()
if len(applications) < 200: # 样本不足200的时候不重训练
return False
# 构造训练数据
X = []
y = []
for app in applications:
features = json.loads(app.features)
X.append(features)
label = 1 if app.status == "reject" else 0
y.append(label)
X = np.array(X)
y = np.array(y)
# 训练新模型
train_data = lgb.Dataset(X, label=y)
params = {'objective': 'binary', 'verbose': -1}
self.model = lgb.train(params, train_data, num_boost_round=100)
self.model.save_model('risk_model.txt')
return True
# 全局模型实例
risk_model = RiskModel()
第三步:FastAPI后端接口定义
# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
import json
from db import SessionLocal, CreditApplication, AuditLog
from model import risk_model
from datetime import datetime
app = FastAPI(title="人机回环风控系统API")
# 依赖:获取数据库会话
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 信贷申请请求体
class CreditApplyRequest(BaseModel):
user_id: str
features: list[float] # 长度为20的特征数组
# 审核请求体
class AuditRequest(BaseModel):
application_id: int
status: str # approve/reject
operator: str
comment: str = ""
# 提交信贷申请接口
@app.post("/apply")
def apply_credit(request: CreditApplyRequest, db: Session = Depends(get_db)):
if len(request.features) != 20:
raise HTTPException(status_code=400, detail="特征长度必须为20")
# 模型预测
pred_result = risk_model.predict(request.features)
# 保存申请数据
app = CreditApplication(
user_id=request.user_id,
features=json.dumps(request.features),
pd=pred_result['pd'],
confidence=pred_result['confidence'],
need_human=pred_result['need_human'],
status="pending" if pred_result['need_human'] else ("approve" if pred_result['pd'] < 0.5 else "reject")
)
db.add(app)
db.commit()
db.refresh(app)
# 记录审计日志
log = AuditLog(
application_id=app.id,
operation="submit_application",
operator="system",
detail=f"模型预测PD:{pred_result['pd']:.4f},置信度:{pred_result['confidence']:.4f},是否需要人工:{pred_result['need_human']}"
)
db.add(log)
db.commit()
return {
"application_id": app.id,
"status": app.status,
"need_human": app.need_human,
"pd": app.pd,
"confidence": app.confidence
}
# 获取待人工审核的申请列表
@app.get("/pending_audits")
def get_pending_audits(db: Session = Depends(get_db)):
apps = db.query(CreditApplication).filter(CreditApplication.status == "pending").all()
return [{
"id": app.id,
"user_id": app.user_id,
"pd": app.pd,
"confidence": app.confidence,
"create_time": app.create_time
} for app in apps]
# 人工审核接口
@app.post("/audit")
def audit_application(request: AuditRequest, db: Session = Depends(get_db)):
app = db.query(CreditApplication).filter(CreditApplication.id == request.application_id).first()
if not app:
raise HTTPException(status_code=404, detail="申请不存在")
if app.status != "pending":
raise HTTPException(status_code=400, detail="申请已审核")
# 更新申请状态
app.status = request.status
app.operator = request.operator
app.update_time = datetime.utcnow()
# 记录审计日志
log = AuditLog(
application_id=app.id,
operation="audit_application",
operator=request.operator,
detail=f"审核结果:{request.status},备注:{request.comment}"
)
db.add(log)
db.commit()
# 每审核100个样本触发一次模型重训练
audit_count = db.query(CreditApplication).filter(CreditApplication.operator != None).count()
if audit_count % 100 == 0:
risk_model.retrain()
return {"status": "success", "message": "审核完成"}
第四步:Gradio人工审核界面
# audit_ui.py
import gradio as gr
import requests
API_BASE = "http://localhost:8000"
def get_pending_list():
res = requests.get(f"{API_BASE}/pending_audits")
if res.status_code == 200:
apps = res.json()
return [f"ID:{app['id']} | 用户:{app['user_id']} | PD:{app['pd']:.4f} | 置信度:{app['confidence']:.4f}" for app in apps]
return []
def audit_application(selected_app, status, operator, comment):
if not selected_app:
return "请先选择要审核的申请"
app_id = int(selected_app.split("|")[0].split(":")[1])
data = {
"application_id": app_id,
"status": status,
"operator": operator,
"comment": comment
}
res = requests.post(f"{API_BASE}/audit", json=data)
if res.status_code == 200:
return f"审核成功,申请ID:{app_id},结果:{status}"
else:
return f"审核失败:{res.json()['detail']}"
with gr.Blocks(title="风控人工审核系统") as demo:
gr.Markdown("# 风控人工审核系统")
with gr.Row():
pending_list = gr.Dropdown(label="待审核申请列表", choices=get_pending_list(), interactive=True)
refresh_btn = gr.Button("刷新列表")
with gr.Row():
status = gr.Radio(label="审核结果", choices=["approve", "reject"], value="approve")
operator = gr.Textbox(label="审核人姓名", required=True)
comment = gr.Textbox(label="审核备注", lines=3)
audit_btn = gr.Button("提交审核")
result = gr.Textbox(label="操作结果")
refresh_btn.click(fn=get_pending_list, outputs=pending_list)
audit_btn.click(fn=audit_application, inputs=[pending_list, status, operator, comment], outputs=result)
if __name__ == "__main__":
demo.launch(server_port=7860)
运行和测试
- 启动后端服务:
uvicorn main:app --reload - 启动人工审核界面:
python audit_ui.py - 测试提交申请:可以用Postman调用
http://localhost:8000/apply接口,传入user_id和20个特征值,接口会返回是否需要人工审核。 - 人工审核:打开
http://localhost:7860,就可以看到待审核的申请,选择申请,提交审核结果,系统会自动记录日志,每满100个审核样本就会自动重训练模型。
实际应用场景
人机回环风控系统在金融行业有非常多的落地场景,我们总结了四种最常见的场景和对应的落地模式:
1. 消费信贷/信用卡申请风控
- 落地模式:前置回环
- 流程:用户提交申请→AI初审→高置信度的直接通过/拒绝,低置信度的推人工复审→人工结果回流训练模型
- 特点:实时性要求高(一般要在1分钟内出结果),人工审核比例控制在5%-20%之间,适合量大、单笔金额小的消费信贷场景
- 效果:某头部消费金融公司用了这套模式之后,人工审核成本降低了70%,坏账率下降了15%,客户满意度提升了25%
2. 交易反欺诈
- 落地模式:实时中置回环
- 流程:交易发起→AI实时识别→高置信度的直接放行/拦截,低置信度的触发人工实时审核(要求30秒内完成)→人工结果回流模型,同时更新欺诈规则库
- 特点:实时性要求极高,对误判的容忍度极低,需要支持快速更新模型应对新的欺诈手段
- 效果:某股份制银行的信用卡反欺诈系统用了这套模式之后,欺诈交易召回率从82%提升到97%,误拦率从5%下降到1.2%
3. 小微企业贷款风控
- 落地模式:后置回环+专家规则结合
- 流程:企业提交贷款申请→AI先做初步评估→所有申请都推给风控专家审核→专家审核结果和AI的评估结果做对比,AI和专家判断不一致的样本作为高价值样本训练模型
- 特点:单笔金额大,风险容忍度低,人工审核比例100%,人机回环的主要作用是辅助人工审核,减少人工的工作量,提升审核的一致性
- 效果:某城商行的小微贷业务用了这套模式之后,单户审核时间从3天缩短到8小时,审核人员的工作量减少了40%,坏账率下降了22%
4. 贷后风险监控
- 落地模式:离线回环
- 流程:AI每天批量监控所有在贷客户的风险→高风险、高置信度的直接触发催收,低置信度的推给贷后管理人员核实→核实结果回流模型,优化风险预警规则
- 特点:非实时,批量处理,人工审核比例在10%-30%之间,主要目标是提升风险预警的准确率,减少无效催收
工具和资源推荐
1. 开源工具
| 工具名称 | 用途 | 地址 |
|---|---|---|
| LabelStudio | 人工标注工具,支持自定义标注模板,适合风控场景的样本标注 | https://labelstud.io/ |
| LangChain | 大模型Agent开发框架,适合搭建大模型驱动的风控Agent | https://www.langchain.com/ |
| MLflow | 模型生命周期管理工具,方便管理风控模型的版本、训练、部署 | https://mlflow.org/ |
| OpenDP | 差分隐私工具,适合金融敏感数据的标注和训练,满足隐私合规要求 | https://opendp.org/ |
2. 学习资源
- 书籍:《主动学习:原理与实践》、《金融风控算法与实践》
- 论文:《Active Learning Literature Survey》(主动学习经典综述)、《Human-in-the-loop Machine Learning for Risk Assessment》(金融风控人机回环的最新研究)
- 行业报告:《中国金融科技风控白皮书2024》、《人机协同风控落地实践报告》
未来发展趋势与挑战
行业发展趋势
我们整理了金融风控模式的发展历程和未来趋势:
| 阶段 | 时间 | 核心模式 | 特点 | 效率 | 成本 |
|---|---|---|---|---|---|
| 1.0 | 1990-2010 | 纯人工+规则引擎 | 完全靠人工审核,配合简单的规则过滤 | 低 | 高 |
| 2.0 | 2010-2020 | 纯机器学习模型 | 用传统机器学习模型做风控,自动化程度高 | 高 | 低 |
| 3.0 | 2020-2025 | 人机回环风控 | AI+人工协同,闭环迭代,兼顾准确率和合规性 | 中高 | 中低 |
| 4.0 | 2025+ | 自主可控风控Agent | 大模型驱动的风控Agent,具备自主推理、自我迭代能力,人工只需要做监督和规则调整 | 极高 | 极低 |
面临的挑战
- 数据隐私与合规挑战:金融数据都是高度敏感的,人工审核的时候需要做严格的脱敏,模型训练的时候要满足《个人信息保护法》、《金融数据安全 数据生命周期安全规范》等法规要求,不能泄露用户隐私。
- 人工标注一致性挑战:不同的风控专家对同一个案例的判断可能存在差异,尤其是边界案例,需要建立完善的审核标准、专家等级制度和交叉校验机制,保证标注数据的质量。
- 冷启动挑战:新的风控场景刚开始没有标注数据,AI的不确定性很高,大部分案例都要推人工,需要用迁移学习、规则引擎兜底、小样本学习等技术快速度过冷启动期。
- 大模型风控的可解释性挑战:大模型驱动的风控Agent虽然效果好,但是可解释性差,需要配套可解释AI技术,保证每一个决策都能给出合理的解释,满足监管要求。
总结:学到了什么?
核心概念回顾
- 人机回环:AI处理高置信度任务,低置信度的推给人工,人工反馈回流优化模型,形成正向循环,兼顾效率、准确率和合规性。
- 金融风控Agent:具备自主推理、主动学习能力的风控AI系统,是人机回环的载体。
- 不确定性估计:衡量AI预测的自信度,是触发人工干预的核心依据,常用的方法有概率置信度法、信息熵法、蒙特卡洛Dropout法。
概念关系回顾
- 人机回环是风控Agent的学习机制,没有回环的Agent就是静态模型,无法应对新的风险。
- 不确定性估计是人机回环的开关,没有准确的不确定性估计,就无法平衡人工成本和风险损失。
- 主动学习是人机回环的核心算法,能够用最少的人工标注成本获得最大的模型效果提升。
思考题:动动小脑筋
思考题一
假设你们公司的信贷业务,单个人工审核成本是3元,每个误判的平均损失是1500元,当前模型的置信度阈值是0.85,人工审核比例是15%,AI误判率是0.3%,如果把阈值调整到0.9,人工审核比例上升到22%,AI误判率下降到0.1%,请问阈值调整之后总成本是上升还是下降?应该调整吗?
思考题二
如果你们要做一个跨境支付的反欺诈系统,实时性要求很高,而且欺诈手段更新非常快,你会选择哪种人机回环的落地模式?需要配套哪些技术来应对欺诈手段的快速更新?
附录:常见问题与解答
- Q:人机回环会不会反而增加人工成本?
A:短期来看,刚上线的时候人工成本可能和之前差不多甚至略高,但是随着模型不断迭代,需要人工审核的比例会越来越低,一般3-6个月之后人工成本就会下降到原来的30%-50%,长期来看成本是大幅降低的。 - Q:怎么确定置信度阈值?
A:建议先做AB测试,计算不同阈值下的总成本(人工成本+误判损失),选择满足业务风险要求的前提下总成本最低的阈值,而且阈值要定期调整,随着模型效果提升可以逐步降低阈值,减少人工审核比例。 - Q:人机回环的系统怎么满足合规要求?
A:首先要对所有操作留痕,包括AI的决策依据、人工的审核记录、模型的迭代历史,都要存在审计日志里,可追溯;其次要对人工审核的数据做脱敏,保护用户隐私;最后要保证AI的决策可解释,能够向监管和用户说明决策的依据。 - Q:大模型时代,人机回环会有什么变化?
A:大模型的出现会大幅降低人机回环的成本,之前需要人工标注的样本,现在大模型可以先做预标注,人工只需要校验和修正,标注效率提升5-10倍;另外大模型可以直接学习人工的标注反馈,不需要重新训练,迭代速度更快。
扩展阅读 & 参考资料
- Settles B. Active learning literature survey[R]. University of Wisconsin-Madison Department of Computer Sciences, 2009.
- Rajpurkar P, et al. Human-in-the-loop machine learning for healthcare: A survey[J]. Journal of the American Medical Informatics Association, 2021.
- 中国人民银行. 《金融数据安全 数据生命周期安全规范》(JR/T 0197-2020), 2020.
- 毕马威. 《2024年金融科技风控报告》, 2024.
- OpenAI. 《Human-in-the-loop AI systems for risk management》, 2023.
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)