规则定义与抽象：规则接口设计（IRule）
好，咱们进入第三章。这一章，我打算聊聊规则的定义与抽象。
你想想看，一个风控系统里，规则可能有成百上千条。有简单的黑白名单，有复杂的设备指纹评分，还有基于机器学习的模型输出。如果每条规则都自己玩自己的，那系统就乱套了。所以，我们需要一个统一的抽象。
这个抽象，就是规则接口。我习惯叫它 IRule。
规则接口设计（IRule）
接口是什么？说白了，就是一份合同。它规定了一条规则「必须」做什么，以及「可以」提供什么信息。
我个人习惯，一个最简的规则接口，至少包含三个核心方法：
- 评估（evaluate）：这是规则的核心。给它一个上下文（比如用户请求、设备信息），它返回一个决策结果（通过、拒绝、人工审核）。
- 获取元数据（getMetadata）：返回规则的描述信息，比如名字、版本、优先级。这方便我们做日志、监控和动态管理。
- 初始化（initialize）：有些规则需要加载配置、建立数据库连接。这个方法就是干这个的。
来看一个简单的 Python 接口定义：
from abc import ABC, abstractmethod
from typing import Any, Dict
class IRule(ABC):
"""规则接口，所有规则都必须实现这个接口。"""
@abstractmethod
def initialize(self, config: Dict[str, Any]) -> None:
"""初始化规则，加载配置。"""
pass
@abstractmethod
def evaluate(self, context: Dict[str, Any]) -> str:
"""
评估规则。
参数 context: 包含请求上下文，如用户ID、IP、设备指纹等。
返回值: 'pass', 'reject', 'review'
"""
pass
@abstractmethod
def get_metadata(self) -> Dict[str, Any]:
"""返回规则的元数据。"""
pass
嗯，这里要注意。接口设计得越简单，扩展性就越强。我在项目中遇到过，有人把规则接口设计得特别「胖」，塞了几十个方法进去。结果新加一条规则，就得实现一堆用不上的方法，痛苦得很。
规则元数据：规则的名字、版本、优先级
元数据，就是「关于数据的数据」。对于规则来说，元数据就是它的身份证。
我一般会包含这几个字段：
| 字段 | 类型 | 说明 |
|---|---|---|
| name | str | 规则名称，用于日志和监控。比如「IP_黑名单_检查」 |
| version | str | 规则版本号，用于灰度发布和回滚。比如 "1.2.0" |
| priority | int | 优先级，数值越小越优先。用于规则排序 |
| description | str | 规则描述，方便运维同学理解 |
| enabled | bool | 是否启用。动态开关规则就靠它 |
为什么版本和优先级这么重要？
我曾经踩过一个坑。有一次上线新规则，没注意版本号，结果旧规则和新规则混在一起执行，导致线上误杀了一大批正常用户。从那以后，我强制要求每条规则必须带版本号，并且在加载时做版本校验。
优先级呢？说白了就是规则执行的顺序。比如「白名单」规则优先级最高，一旦命中直接放行，后面的规则就不用跑了。这能大幅提升性能。
来看一个元数据类的实现：
from dataclasses import dataclass, field
from typing import Any, Dict
@dataclass
class RuleMetadata:
"""规则元数据"""
name: str
version: str = "1.0.0"
priority: int = 100
description: str = ""
enabled: bool = True
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"version": self.version,
"priority": self.priority,
"description": self.description,
"enabled": self.enabled
}
你看，用 dataclass 定义，简洁又清晰。默认优先级设为 100，这样新加的规则默认排在后面，不会影响已有的高优规则。
规则条件与动作的抽象
一条规则，本质上就是「如果条件满足，就执行动作」。所以，把条件和动作抽象出来，是复用规则逻辑的关键。
我习惯把条件抽象成一个接口 ICondition，动作抽象成 IAction。
条件接口：
class ICondition(ABC):
@abstractmethod
def evaluate(self, context: Dict[str, Any]) -> bool:
"""判断条件是否满足。返回 True 表示条件成立。"""
pass
动作接口：
class IAction(ABC):
@abstractmethod
def execute(self, context: Dict[str, Any]) -> None:
"""执行动作，比如记录日志、发送告警、拒绝请求。"""
pass
有了这两个抽象，我们就可以组合出各种规则。比如：
- 条件：IP 在黑名单中、用户注册时间小于 7 天、设备指纹异常。
- 动作：拒绝请求、标记为高风险、发送短信验证码。
你想想看，这样做的好处是什么？
条件可以复用。比如「用户注册时间小于 7 天」这个条件，既可以用在「新用户注册风控」规则里，也可以用在「新用户支付风控」规则里。动作也可以复用。比如「拒绝请求」这个动作，几乎所有规则都能用。
来看一个完整的规则实现示例：
class IpBlacklistCondition(ICondition):
def __init__(self, blacklist: set):
self.blacklist = blacklist
def evaluate(self, context: Dict[str, Any]) -> bool:
ip = context.get("ip", "")
return ip in self.blacklist
class RejectAction(IAction):
def execute(self, context: Dict[str, Any]) -> None:
print(f"拒绝请求: {context.get('ip')}")
# 实际项目中，这里会调用风控决策引擎的拒绝接口
class IpBlacklistRule(IRule):
def __init__(self):
self.metadata = RuleMetadata(
name="IP黑名单检查",
version="1.0.0",
priority=10,
description="检查请求IP是否在黑名单中"
)
self.condition = None
self.action = None
def initialize(self, config: Dict[str, Any]) -> None:
blacklist = set(config.get("blacklist_ips", []))
self.condition = IpBlacklistCondition(blacklist)
self.action = RejectAction()
def evaluate(self, context: Dict[str, Any]) -> str:
if self.condition.evaluate(context):
self.action.execute(context)
return "reject"
return "pass"
def get_metadata(self) -> Dict[str, Any]:
return self.metadata.to_dict()
你看，这条规则的结构非常清晰。条件负责判断，动作负责执行，规则本身只负责编排。这就是抽象的力量。
本章小结
这一章，我们聊了规则的定义与抽象。核心就三件事：
- 规则接口（IRule）：定义了规则的生命周期方法，包括初始化、评估、获取元数据。
- 规则元数据：给规则一个身份，包括名称、版本、优先级等。这是动态加载和灰度发布的基础。
- 条件与动作的抽象：把规则拆成「条件」和「动作」两个独立的部分，实现逻辑复用。
说白了，好的抽象，能让你的规则系统像搭积木一样灵活。每块积木都很简单，但组合起来能应对各种复杂的业务场景。
下一章，我们会聊聊如何把这些规则动态加载到系统中。嗯，那才是真正有意思的部分。
公众号:蓝海资料掘金营，微信deep3321