第1817篇：实时欺诈检测中的AI——毫秒级决策的架构设计
2026/4/30大约 11 分钟
第1817篇：实时欺诈检测中的AI——毫秒级决策的架构设计
有一次和一个做金融风控的朋友聊天，他说了一句话让我印象很深："风控工程师的最大挑战不是模型准不准，是在准的同时还得快。你的模型再厉害，决策慢了半秒，那笔钱可能已经被转走了。"
这就是金融欺诈检测最核心的矛盾：精度和速度的极致博弈。
今天聊的这个方向，是我研究了挺久、也在项目里实战过的思路：用规则引擎+ML模型+LLM三层架构，做到毫秒级决策。不是每层都快，但每层用在刀刃上。
为什么不能直接用LLM做欺诈检测
很多人第一反应是：LLM那么强，直接用LLM分析交易不就行了？
这个想法好，但行不通。原因很直接：
延迟。LLM调用最快也要100-300ms，高峰期可能更长。但金融交易对风控决策的要求通常是50ms以内，甚至更严。
成本。一家中型支付公司日均交易量可能是百万量级，如果每笔都调LLM，光API费用就能压垮预算。
一致性。LLM的输出有随机性，同样的交易可能今天判断通过、明天判断拦截，无法满足合规要求。
所以LLM在欺诈检测里的正确定位是：不做实时决策，做离线分析和模型迭代。
三层决策架构
这个架构的核心思想：
- 第一层规则引擎处理"显然的"情况（黑名单、限额、明显异常），成本最低
- 第二层ML模型处理大部分"普通"情况，快速风险评分
- 第三层策略引擎处理"两可"情况，综合多个维度做最终决策
- LLM只做离线分析，产出规则和模型优化方案
数据模型
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class TransactionRequest {
private String transactionId;
private String userId;
private String merchantId;
private String merchantCategory; // 商户类别（MCC码）
private double amount;
private String currency;
private String paymentMethod; // CARD/WALLET/TRANSFER
private String deviceId;
private String ipAddress;
private String location; // 经纬度或城市
private long timestamp;
private Map<String, Object> deviceFingerprint;
private Map<String, String> metadata;
}
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class RiskDecision {
private String transactionId;
private RiskLevel riskLevel;
private double riskScore; // 0.0-1.0
private DecisionResult decision;
private List<String> triggeredRules; // 触发的规则列表
private String decisionReason;
private long processingTimeMs;
public enum RiskLevel {
VERY_LOW, LOW, MEDIUM, HIGH, VERY_HIGH
}
public enum DecisionResult {
APPROVE, // 通过
REJECT, // 拒绝
REVIEW, // 人工审核
STEP_UP, // 二次验证（短信/人脸）
LIMIT // 限额通过（金额超限降额放行）
}
}第一层：规则引擎
规则引擎要做到微秒级，数据结构选型很重要：
@Component
@Slf4j
public class RuleEngine {
// 黑名单用BloomFilter做快速预筛（内存友好，允许极低误判率）
private final BloomFilter<String> userBlacklist;
private final BloomFilter<String> deviceBlacklist;
// 精确黑名单用HashSet（BloomFilter命中后再精确确认）
private final Set<String> exactUserBlacklist;
private final Set<String> exactDeviceBlacklist;
// 规则列表（有序，按优先级执行）
private final List<FraudRule> rules;
public RuleEngine() {
// BloomFilter：预计100万元素，误判率0.01%
this.userBlacklist = BloomFilter.create(
Funnels.stringFunnel(StandardCharsets.UTF_8), 1_000_000, 0.0001);
this.deviceBlacklist = BloomFilter.create(
Funnels.stringFunnel(StandardCharsets.UTF_8), 500_000, 0.0001);
this.exactUserBlacklist = ConcurrentHashMap.newKeySet();
this.exactDeviceBlacklist = ConcurrentHashMap.newKeySet();
this.rules = buildDefaultRules();
}
public RuleCheckResult check(TransactionRequest tx, UserRiskContext context) {
List<String> triggeredRules = new ArrayList<>();
for (FraudRule rule : rules) {
RuleResult result = rule.evaluate(tx, context);
if (result.isTriggered()) {
triggeredRules.add(rule.getName());
if (result.isTerminal()) {
// 终止规则：直接返回结论
return RuleCheckResult.builder()
.triggered(true)
.terminal(true)
.decision(result.getDecision())
.triggeredRules(triggeredRules)
.build();
}
}
}
return RuleCheckResult.builder()
.triggered(!triggeredRules.isEmpty())
.terminal(false)
.triggeredRules(triggeredRules)
.build();
}
private List<FraudRule> buildDefaultRules() {
List<FraudRule> rules = new ArrayList<>();
// 规则1：黑名单检查（终止规则）
rules.add(FraudRule.builder()
.name("USER_BLACKLIST")
.priority(1)
.terminal(true)
.evaluator((tx, ctx) -> {
// 两步检查：BloomFilter快速过滤 + 精确确认
if (userBlacklist.mightContain(tx.getUserId())) {
return exactUserBlacklist.contains(tx.getUserId())
? RuleResult.terminal(RiskDecision.DecisionResult.REJECT)
: RuleResult.pass();
}
return RuleResult.pass();
})
.build());
// 规则2：设备黑名单（终止规则）
rules.add(FraudRule.builder()
.name("DEVICE_BLACKLIST")
.priority(2)
.terminal(true)
.evaluator((tx, ctx) -> {
if (tx.getDeviceId() != null && deviceBlacklist.mightContain(tx.getDeviceId())) {
return exactDeviceBlacklist.contains(tx.getDeviceId())
? RuleResult.terminal(RiskDecision.DecisionResult.REJECT)
: RuleResult.pass();
}
return RuleResult.pass();
})
.build());
// 规则3：单笔金额超限（终止规则）
rules.add(FraudRule.builder()
.name("SINGLE_AMOUNT_LIMIT")
.priority(3)
.terminal(true)
.evaluator((tx, ctx) -> {
if (tx.getAmount() > 100_000) {
return RuleResult.terminal(RiskDecision.DecisionResult.REVIEW);
}
return RuleResult.pass();
})
.build());
// 规则4：高频交易（非终止，增加风险分）
rules.add(FraudRule.builder()
.name("HIGH_FREQUENCY")
.priority(4)
.terminal(false)
.evaluator((tx, ctx) -> {
// 1分钟内超过10笔
if (ctx.getTransactionCountLastMinute() > 10) {
return RuleResult.triggered("1分钟内交易频率过高");
}
return RuleResult.pass();
})
.build());
// 规则5：异地交易（非终止）
rules.add(FraudRule.builder()
.name("LOCATION_ANOMALY")
.priority(5)
.terminal(false)
.evaluator((tx, ctx) -> {
if (ctx.getLastTransactionLocation() != null
&& calculateDistance(tx.getLocation(), ctx.getLastTransactionLocation()) > 500
&& (tx.getTimestamp() - ctx.getLastTransactionTime()) < 3600_000) {
// 1小时内，地理距离超500km（不可能的速度）
return RuleResult.triggered("地理位置异常（不可能的移动速度）");
}
return RuleResult.pass();
})
.build());
return rules.stream()
.sorted(Comparator.comparingInt(FraudRule::getPriority))
.collect(Collectors.toList());
}
private double calculateDistance(String loc1, String loc2) {
// 简化的地理距离计算，实际应用中使用Haversine公式
// 省略实现
return 0.0;
}
// 动态更新规则（LLM分析后可能更新规则）
public void updateBlacklist(String userId) {
userBlacklist.put(userId);
exactUserBlacklist.add(userId);
}
}第二层：ML模型实时评分
@Service
@Slf4j
public class MLScoringService {
// 使用ONNX Runtime加载预训练模型
private final OrtEnvironment ortEnv;
private final OrtSession ortSession;
// 特征计算服务
private final FeatureEngineeringService featureService;
public MLScoringService(@Value("${fraud.model.path}") String modelPath,
FeatureEngineeringService featureService) throws OrtException {
this.ortEnv = OrtEnvironment.getEnvironment();
OrtSession.SessionOptions opts = new OrtSession.SessionOptions();
opts.setInterOpNumThreads(2); // 推理使用2个线程
opts.setIntraOpNumThreads(2);
opts.addConfigEntry("session.intra_op.allow_spinning", "0"); // 避免CPU空转
this.ortSession = ortEnv.createSession(modelPath, opts);
this.featureService = featureService;
log.info("ML fraud model loaded from: {}", modelPath);
}
public double score(TransactionRequest tx, UserRiskContext context) {
try {
// 特征工程
float[] features = featureService.extractFeatures(tx, context);
// ONNX推理
OnnxTensor inputTensor = OnnxTensor.createTensor(
ortEnv,
new float[][]{features} // batch_size=1
);
Map<String, OnnxTensorLike> inputs = Collections.singletonMap("input", inputTensor);
OrtSession.Result result = ortSession.run(inputs);
// 提取风险分数（假设模型输出[batch_size, 2]，索引1是欺诈概率）
float[][] output = (float[][]) result.get(0).getValue();
double fraudProbability = output[0][1];
result.close();
inputTensor.close();
return fraudProbability;
} catch (Exception e) {
log.error("ML scoring failed for transaction: {}", tx.getTransactionId(), e);
return 0.5; // 出错时返回中间值，由策略引擎保守处理
}
}
}
@Service
public class FeatureEngineeringService {
private final UserBehaviorFeatureStore featureStore;
/**
* 特征提取：将交易请求和用户上下文转换为模型输入向量
* 这个方法要极致优化，在整个链路里它会被高频调用
*/
public float[] extractFeatures(TransactionRequest tx, UserRiskContext context) {
// 特征维度固定为32维（和模型训练时对齐）
float[] features = new float[32];
// 交易金额相关特征
features[0] = normalizeAmount(tx.getAmount());
features[1] = (float)(tx.getAmount() / Math.max(context.getAvgTransactionAmount(), 1));
features[2] = tx.getAmount() > context.getMaxHistoricalAmount() ? 1.0f : 0.0f;
// 时间相关特征
int hour = LocalDateTime.ofEpochSecond(tx.getTimestamp()/1000, 0, ZoneOffset.UTC).getHour();
features[3] = (float) Math.sin(2 * Math.PI * hour / 24); // 时间周期编码
features[4] = (float) Math.cos(2 * Math.PI * hour / 24);
features[5] = isWeekend(tx.getTimestamp()) ? 1.0f : 0.0f;
// 频率特征
features[6] = normalize(context.getTransactionCountLastMinute(), 0, 20);
features[7] = normalize(context.getTransactionCountLastHour(), 0, 100);
features[8] = normalize(context.getTransactionCountLastDay(), 0, 500);
// 商户特征
features[9] = encodeMCC(tx.getMerchantCategory());
features[10] = context.getTransactionCountAtMerchant() > 0 ? 1.0f : 0.0f; // 是否新商户
// 设备和位置特征
features[11] = context.isNewDevice() ? 1.0f : 0.0f;
features[12] = context.isNewLocation() ? 1.0f : 0.0f;
features[13] = normalizeDistance(context.getDistanceFromLastTransaction());
// 历史行为特征
features[14] = normalize(context.getAccountAgedays(), 0, 1825);
features[15] = normalize(context.getTotalTransactionCount(), 0, 10000);
features[16] = (float) context.getFraudHistoryScore();
// 填充其余维度（可以是更多特征，这里简化）
// features[17-31] = ...
return features;
}
private float normalizeAmount(double amount) {
// 对数归一化，处理金额的长尾分布
return (float)(Math.log1p(amount) / Math.log1p(1_000_000));
}
private float normalize(double value, double min, double max) {
return (float) Math.max(0, Math.min(1, (value - min) / (max - min)));
}
private float normalizeDistance(double distanceKm) {
return (float)(Math.log1p(distanceKm) / Math.log1p(20000));
}
private float encodeMCC(String mcc) {
// 高风险商户类别编码
if (mcc == null) return 0.5f;
Map<String, Float> highRiskMCC = Map.of(
"7995", 1.0f, // 赌场/博彩
"5912", 0.8f, // 药店
"4829", 0.9f, // 资金转移
"6010", 0.7f // 金融机构
);
return highRiskMCC.getOrDefault(mcc, 0.3f);
}
private boolean isWeekend(long timestamp) {
DayOfWeek day = LocalDateTime.ofEpochSecond(timestamp/1000, 0, ZoneOffset.UTC).getDayOfWeek();
return day == DayOfWeek.SATURDAY || day == DayOfWeek.SUNDAY;
}
}第三层：策略引擎
综合规则引擎和ML评分，做最终决策：
@Service
@Slf4j
public class FraudDecisionService {
private final RuleEngine ruleEngine;
private final MLScoringService mlScoringService;
private final UserRiskContextService contextService;
private final MeterRegistry meterRegistry;
// 风险分数阈值配置（可动态调整）
@Value("${fraud.threshold.reject:0.85}")
private double rejectThreshold;
@Value("${fraud.threshold.review:0.60}")
private double reviewThreshold;
@Value("${fraud.threshold.stepup:0.40}")
private double stepupThreshold;
public RiskDecision decide(TransactionRequest tx) {
long startTime = System.nanoTime();
// Step 1: 获取用户风险上下文（从Redis读取，必须极快）
UserRiskContext context = contextService.getContext(tx.getUserId(), tx.getDeviceId());
// Step 2: 规则引擎检查
RuleCheckResult ruleResult = ruleEngine.check(tx, context);
// 如果规则引擎有终止决策，直接返回
if (ruleResult.isTerminal()) {
return buildDecision(tx, ruleResult, 0.0, startTime);
}
// Step 3: ML风险评分
double riskScore = mlScoringService.score(tx, context);
// Step 4: 综合决策
RiskDecision.DecisionResult decision = makeDecision(riskScore, ruleResult, tx, context);
// Step 5: 异步更新用户上下文（不阻塞主流程）
CompletableFuture.runAsync(() -> contextService.updateContext(tx, decision));
RiskDecision result = buildDecision(tx, ruleResult, riskScore, startTime);
result.setDecision(decision);
// 记录指标
long elapsedMs = (System.nanoTime() - startTime) / 1_000_000;
meterRegistry.timer("fraud.decision.latency").record(elapsedMs, TimeUnit.MILLISECONDS);
meterRegistry.counter("fraud.decision." + decision.name().toLowerCase()).increment();
return result;
}
private RiskDecision.DecisionResult makeDecision(double riskScore,
RuleCheckResult ruleResult,
TransactionRequest tx,
UserRiskContext context) {
// 规则命中会加成分数
double adjustedScore = riskScore;
if (!ruleResult.getTriggeredRules().isEmpty()) {
adjustedScore = Math.min(1.0, riskScore + 0.2 * ruleResult.getTriggeredRules().size());
}
if (adjustedScore >= rejectThreshold) {
return RiskDecision.DecisionResult.REJECT;
}
if (adjustedScore >= reviewThreshold) {
// 大额交易送审，小额直接拒绝
return tx.getAmount() > 10000
? RiskDecision.DecisionResult.REVIEW
: RiskDecision.DecisionResult.REJECT;
}
if (adjustedScore >= stepupThreshold) {
return RiskDecision.DecisionResult.STEP_UP;
}
return RiskDecision.DecisionResult.APPROVE;
}
private RiskDecision buildDecision(TransactionRequest tx, RuleCheckResult ruleResult,
double riskScore, long startNano) {
long elapsedMs = (System.nanoTime() - startNano) / 1_000_000;
RiskLevel riskLevel;
if (riskScore >= 0.8) riskLevel = RiskDecision.RiskLevel.VERY_HIGH;
else if (riskScore >= 0.6) riskLevel = RiskDecision.RiskLevel.HIGH;
else if (riskScore >= 0.4) riskLevel = RiskDecision.RiskLevel.MEDIUM;
else if (riskScore >= 0.2) riskLevel = RiskDecision.RiskLevel.LOW;
else riskLevel = RiskDecision.RiskLevel.VERY_LOW;
return RiskDecision.builder()
.transactionId(tx.getTransactionId())
.riskScore(riskScore)
.riskLevel(riskLevel)
.triggeredRules(ruleResult.getTriggeredRules())
.decisionReason(buildReason(ruleResult, riskScore))
.processingTimeMs(elapsedMs)
.build();
}
private String buildReason(RuleCheckResult ruleResult, double riskScore) {
if (!ruleResult.getTriggeredRules().isEmpty()) {
return "命中规则: " + String.join(", ", ruleResult.getTriggeredRules());
}
return String.format("ML风险评分: %.2f", riskScore);
}
}LLM的离线分析角色
LLM不参与实时决策，但在离线分析中价值巨大：
@Service
@Slf4j
public class FraudPatternAnalysisService {
private final ChatLanguageModel chatModel;
private final FraudCaseRepository caseRepository;
/**
* 每日凌晨分析前一天的欺诈案例，发现新的攻击模式
*/
@Scheduled(cron = "0 2 * * *") // 每天凌晨2点
public void analyzeNewFraudPatterns() {
log.info("Starting daily fraud pattern analysis");
// 获取昨天的确认欺诈案例
List<FraudCase> confirmedCases = caseRepository.getConfirmedFraudYesterday();
if (confirmedCases.size() < 10) {
log.info("Too few cases for pattern analysis: {}", confirmedCases.size());
return;
}
// 构造分析Prompt
String analysisPrompt = buildPatternAnalysisPrompt(confirmedCases);
try {
String analysis = chatModel.generate(analysisPrompt);
FraudPatternReport report = parsePatternReport(analysis);
// 将新发现的模式存档，由风控团队人工审核后决定是否加入规则
savePatternReport(report);
// 如果发现高置信度的新模式，自动发送告警给风控团队
if (!report.getHighConfidencePatterns().isEmpty()) {
notifyFraudTeam(report);
}
log.info("Pattern analysis completed, found {} new patterns",
report.getAllPatterns().size());
} catch (Exception e) {
log.error("Fraud pattern analysis failed", e);
}
}
private String buildPatternAnalysisPrompt(List<FraudCase> cases) {
StringBuilder sb = new StringBuilder();
sb.append("你是金融欺诈分析专家。分析以下欺诈案例，识别共同规律和攻击模式。\n\n");
sb.append("## 欺诈案例（").append(cases.size()).append("条，来自过去24小时）\n\n");
// 只发送关键字段，控制token消耗
cases.stream().limit(100).forEach(c -> {
sb.append(String.format(
"案例%s: 金额=%.0f, 商户类别=%s, 时间=%s, 设备=%s, 触发规则=%s\n",
c.getCaseId(), c.getAmount(), c.getMerchantCategory(),
c.getHour() + "时", c.isNewDevice() ? "新设备" : "已知设备",
String.join("+", c.getTriggeredRules())
));
});
sb.append("\n请识别：\n");
sb.append("1. 主要攻击模式（3-5种）\n");
sb.append("2. 每种模式的特征描述\n");
sb.append("3. 建议新增的检测规则\n");
sb.append("4. 需要关注的异常趋势\n\n");
sb.append("以JSON格式返回：");
sb.append("{\"patterns\": [{\"name\": \"\", \"description\": \"\", ");
sb.append("\"suggestedRule\": \"\", \"confidence\": 0.0}], \"trends\": []}");
return sb.toString();
}
/**
* 对人工审核的案例生成详细分析报告
* 帮助审核人员快速理解案例背景
*/
public FraudCaseReport analyzeCase(String caseId) {
FraudCase fraudCase = caseRepository.findById(caseId)
.orElseThrow(() -> new RuntimeException("Case not found: " + caseId));
// 获取该用户历史交易上下文
List<TransactionRecord> history = caseRepository
.getUserTransactionHistory(fraudCase.getUserId(), 30); // 最近30天
String prompt = String.format("""
分析以下疑似欺诈案例，提供详细审核报告。
## 可疑交易
交易ID: %s
金额: %.2f元
时间: %s
商户: %s (%s)
设备: %s
IP: %s
触发规则: %s
ML评分: %.2f
## 用户近30天交易历史（%d笔）
%s
请分析：
1. 这笔交易是否为欺诈（概率估计）
2. 关键可疑点
3. 建议的审核操作
""",
fraudCase.getTransactionId(),
fraudCase.getAmount(),
fraudCase.getTransactionTime(),
fraudCase.getMerchantName(), fraudCase.getMerchantCategory(),
fraudCase.getDeviceInfo(),
fraudCase.getIpAddress(),
String.join(", ", fraudCase.getTriggeredRules()),
fraudCase.getRiskScore(),
history.size(),
summarizeHistory(history)
);
String analysis = chatModel.generate(prompt);
return FraudCaseReport.builder()
.caseId(caseId)
.aiAnalysis(analysis)
.generatedAt(LocalDateTime.now())
.build();
}
private String summarizeHistory(List<TransactionRecord> history) {
if (history.isEmpty()) return "无历史记录";
double totalAmount = history.stream().mapToDouble(TransactionRecord::getAmount).sum();
double avgAmount = totalAmount / history.size();
long fraudCount = history.stream().filter(TransactionRecord::isFraud).count();
return String.format("总交易%d笔，平均金额%.0f元，历史欺诈%d笔",
history.size(), avgAmount, fraudCount);
}
private FraudPatternReport parsePatternReport(String json) {
// JSON解析逻辑
try {
ObjectMapper mapper = new ObjectMapper();
JsonNode root = mapper.readTree(extractJson(json));
// ... 解析逻辑
return FraudPatternReport.builder().build();
} catch (Exception e) {
return FraudPatternReport.empty();
}
}
private String extractJson(String text) {
int start = text.indexOf('{');
int end = text.lastIndexOf('}');
return (start >= 0 && end > start) ? text.substring(start, end + 1) : "{}";
}
private void savePatternReport(FraudPatternReport report) {
// 持久化报告
}
private void notifyFraudTeam(FraudPatternReport report) {
// 发送告警
}
}关键性能数据
这套架构在我们的测试环境（8核CPU、16GB内存）的关键指标：
- 规则引擎P99延迟：0.8ms
- ML评分P99延迟：3.2ms（含特征提取）
- 端到端决策P99延迟：6.1ms
- 吞吐量：在单节点上能支持约2000 TPS
实际生产上，这已经远低于50ms的SLA要求，留有充足的余量。
值得特别提一下的是ML模型加载。用ONNX Runtime比直接调Python服务快很多，省去了网络往返，而且ONNX Runtime的Java绑定非常成熟。模型从训练（Python/PyTorch）→导出ONNX→Java推理这条链路，是我目前见过在Java体系里做ML推理最工程化的方案。