4、规则引擎实战：硬规则与软规则定义、规则优先级与冲突处理、规则热更新机制

规则引擎，说白了就是风控系统的「交通警察」。

它不搞什么花里胡哨的模型推理，就干一件事：按规矩办事。但你别小看它，我见过太多项目，模型跑得飞起，最后却被几条写死的规则坑得血本无归。

嗯，今天我们就来聊聊规则引擎里最核心的三个问题：硬规则和软规则怎么分？规则打架了怎么办？以及，怎么让规则「活」起来？

4.1 硬规则与软规则：谁说了算？

先说说定义。我个人习惯把规则分成两类：

• 硬规则：触碰即死，没有商量余地。比如「单笔转账超过500万，直接拒绝」。
• 软规则：触碰后触发告警或降级，但不直接拒绝。比如「同一设备登录超过3个账号，标记为可疑」。

为什么要有这种区分？

我在项目中遇到过这样一个案例：某支付平台上线了一套反欺诈规则，其中有一条「新注册用户24小时内交易超过5笔，直接封号」。结果上线当天，大量真实用户被误杀——因为那是个电商大促活动，新用户疯狂下单。这就是典型的硬规则滥用。

来看一段简单的代码实现：

class Rule:
    def __init__(self, name, rule_type, condition, action):
        self.name = name
        self.rule_type = rule_type  # 'hard' or 'soft'
        self.condition = condition  # 条件函数
        self.action = action        # 动作函数

    def evaluate(self, transaction):
        if self.condition(transaction):
            return self.action(transaction)
        return None

# 硬规则示例
hard_rule = Rule(
    name="单笔超限",
    rule_type="hard",
    condition=lambda tx: tx['amount'] > 5000000,
    action=lambda tx: {"decision": "reject", "reason": "金额超限"}
)

# 软规则示例
soft_rule = Rule(
    name="设备异常",
    rule_type="soft",
    condition=lambda tx: tx['device_count'] > 3,
    action=lambda tx: {"decision": "review", "reason": "设备登录过多"}
)

你看，硬规则直接返回 reject，软规则返回 review。这就是最朴素的区分。

4.2 规则优先级与冲突处理

规则多了，难免打架。比如：

• 规则A说「金额大于10万，拒绝」
• 规则B说「VIP用户金额大于50万，放行」

一个VIP用户转账30万，该听谁的？

这就涉及到优先级和冲突处理。

我一般用两种方式解决：

4.2.1 静态优先级

给每条规则一个数字优先级，数字越小越优先。硬规则默认优先级高于软规则。

class PriorityRule(Rule):
    def __init__(self, name, rule_type, condition, action, priority=100):
        super().__init__(name, rule_type, condition, action)
        self.priority = priority

    def __lt__(self, other):
        return self.priority < other.priority

# 优先级定义
rules = [
    PriorityRule("VIP放行", "hard", lambda tx: tx['vip'] and tx['amount'] < 500000, 
                 lambda tx: {"decision": "approve"}, priority=1),
    PriorityRule("金额超限", "hard", lambda tx: tx['amount'] > 100000,
                 lambda tx: {"decision": "reject"}, priority=2)
]

# 按优先级排序执行
rules.sort()
for rule in rules:
    result = rule.evaluate(tx)
    if result:
        break  # 高优先级规则命中后，不再执行低优先级

4.2.2 冲突矩阵

当规则数量超过50条时，静态优先级容易出bug。我建议用冲突矩阵来显式定义规则之间的互斥关系。

规则A \ 规则B	VIP放行	金额超限	设备异常
VIP放行	-	VIP优先	同时触发
金额超限	VIP优先	-	金额优先
设备异常	同时触发	金额优先	-

4.3 规则热更新机制

规则不是一成不变的。黑产在变，业务在变，规则也得变。

但问题来了：每次改规则都要重启服务吗？

当然不。重启意味着断服，断服意味着损失。所以我们需要热更新——在不重启服务的情况下，动态加载新规则。

4.3.1 基于数据库的热更新

最常用的方式。规则存储在数据库或Redis中，引擎定时拉取。

import time
import threading

class HotReloadEngine:
    def __init__(self, db_connector):
        self.rules = []
        self.db = db_connector
        self._load_rules()
        # 启动后台线程，每60秒检查一次
        self._start_watcher()

    def _load_rules(self):
        # 从数据库加载规则
        raw_rules = self.db.query("SELECT * FROM rules WHERE status=1")
        self.rules = [self._parse_rule(r) for r in raw_rules]
        print(f"已加载 {len(self.rules)} 条规则")

    def _start_watcher(self):
        def watcher():
            while True:
                time.sleep(60)
                self._load_rules()
        threading.Thread(target=watcher, daemon=True).start()

    def evaluate(self, transaction):
        for rule in self.rules:
            result = rule.evaluate(transaction)
            if result:
                return result
        return {"decision": "approve"}

4.3.2 基于文件的热更新

如果你不想引入数据库，也可以用文件。我习惯用YAML格式，可读性好。

# rules.yaml
rules:
  - name: "金额超限"
    type: "hard"
    condition: "amount > 1000000"
    action: "reject"
    priority: 10
  - name: "设备异常"
    type: "soft"
    condition: "device_count > 5"
    action: "review"
    priority: 20

import yaml
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class FileRuleEngine:
    def __init__(self, file_path):
        self.file_path = file_path
        self.rules = []
        self._load_rules()
        self._watch_file()

    def _load_rules(self):
        with open(self.file_path, 'r') as f:
            data = yaml.safe_load(f)
        self.rules = [self._parse_rule(r) for r in data['rules']]

    def _watch_file(self):
        event_handler = FileSystemEventHandler()
        event_handler.on_modified = lambda e: self._load_rules()
        observer = Observer()
        observer.schedule(event_handler, os.path.dirname(self.file_path))
        observer.start()

4.4 知识体系总览

说了这么多，我们来画个图，把整个规则引擎的脉络理清楚。

这张图把整个流程串起来了：规则从输入进来，先分硬软，再处理冲突，最后通过热更新机制保持活力。说白了，规则引擎就是个「输入-判断-输出」的闭环，但细节里全是坑。

我个人建议，刚开始做规则引擎时，别追求大而全。先搞定硬规则和软规则的区分，再慢慢加优先级和热更新。步子迈大了，容易扯着蛋。

---

公众号:蓝海资料掘金营，微信deep3321