3. Python规则引擎基础:实现方式与优先级设计
在深入实验案例之前,必须掌握Python中实现规则引擎的几种基础方式。本章聚焦于轻量级、无外部依赖的实现模式,这些模式是后续构建复杂规则引擎的基石。
3.1 规则引擎的核心要素
一个规则引擎无论复杂与否,都包含三个核心要素:
| 要素 | 说明 | 在Python中的常见表示 |
|---|---|---|
| 条件(Condition) | 触发规则执行的前提判断 | 布尔表达式、lambda函数、字典键值对 |
| 动作(Action) | 条件满足时执行的操作 | 函数调用、赋值、日志输出、状态变更 |
| 优先级(Priority) | 多条规则同时满足时的执行顺序 | 整数数值、列表顺序、排序键 |
实验视角: 本章所有实现方式均可在单个Python文件中完成,无需安装任何第三方库。建议读者在Jupyter Notebook或Python脚本中边读边写。
3.2 基于字典的简单规则
字典是Python中最灵活的数据结构之一,利用字典可以快速构建“条件→动作”的映射规则。这种方式适合规则数量少、条件简单的场景。
3.2.1 基础模式:条件-动作映射
将规则定义为字典,键为条件描述,值为对应的处理函数或结果。
# 基于字典的简单规则引擎示例
def rule_low_risk(data):
return f"低风险:{data['name']},信用良好"
def rule_medium_risk(data):
return f"中风险:{data['name']},需人工复核"
def rule_high_risk(data):
return f"高风险:{data['name']},拒绝申请"
# 规则字典:条件字符串 -> 动作函数
rules_dict = {
"score >= 700": rule_low_risk,
"500 <= score < 700": rule_medium_risk,
"score < 500": rule_high_risk
}
# 执行引擎
def execute_dict_engine(data, rules):
score = data.get("score", 0)
if score >= 700:
return rules["score >= 700"](data)
elif 500 <= score < 700:
return rules["500 <= score < 700"](data)
else:
return rules["score < 500"](data)
# 测试
applicant = {"name": "张三", "score": 680}
print(execute_dict_engine(applicant, rules_dict))
# 输出:中风险:张三,需人工复核
3.2.2 字典规则的优缺点
| 优点 | 缺点 |
|---|---|
| 代码直观,易于理解和维护 | 条件逻辑硬编码在引擎中,扩展性差 |
| 无需额外抽象,适合快速原型 | 规则数量增多时,字典管理混乱 |
| 字典本身可序列化(JSON) | 不支持复杂的组合条件(AND/OR) |
3.3 基于函数的规则
将每条规则封装为一个独立的函数,通过规则列表统一管理。这种方式比字典模式更灵活,支持复杂的条件判断和动作组合。
3.3.1 规则函数规范
定义统一的规则函数签名:rule(data) -> (bool, result),返回一个元组,第一个元素表示规则是否触发,第二个元素为执行结果。
# 基于函数的规则引擎
from typing import Callable, Any, List, Tuple
# 规则类型定义
RuleFunc = Callable[[dict], Tuple[bool, Any]]
def rule_vip_customer(data: dict) -> Tuple[bool, str]:
"""VIP客户自动通过"""
if data.get("level") == "VIP":
return True, f"VIP客户 {data['name']} 自动通过"
return False, None
def rule_high_value_order(data: dict) -> Tuple[bool, str]:
"""高价值订单需人工审核"""
if data.get("amount", 0) > 10000:
return True, f"高价值订单 {data['amount']}元,转人工"
return False, None
def rule_fraud_risk(data: dict) -> Tuple[bool, str]:
"""欺诈风险规则"""
if data.get("risk_flag") == "high":
return True, f"高风险欺诈标记,拒绝 {data['name']}"
return False, None
# 规则注册表
rules_registry: List[RuleFunc] = [
rule_vip_customer,
rule_high_value_order,
rule_fraud_risk
]
# 执行引擎:遍历所有规则,返回第一个触发的规则结果
def execute_function_engine(data: dict, rules: List[RuleFunc]) -> str:
for rule in rules:
triggered, result = rule(data)
if triggered:
return result
return "默认处理:无规则触发"
# 测试
order1 = {"name": "李四", "level": "VIP", "amount": 5000, "risk_flag": "low"}
order2 = {"name": "王五", "level": "normal", "amount": 15000, "risk_flag": "low"}
order3 = {"name": "赵六", "level": "normal", "amount": 200, "risk_flag": "high"}
print(execute_function_engine(order1, rules_registry)) # VIP客户
print(execute_function_engine(order2, rules_registry)) # 高价值订单
print(execute_function_engine(order3, rules_registry)) # 欺诈风险
关键设计模式: 函数式规则引擎的核心在于“规则即函数”。每个规则函数只负责一件事,通过列表组合实现规则链。这种模式天然支持单元测试,每个规则函数都可以独立验证。
3.4 规则优先级设计
当多条规则同时满足条件时,必须通过优先级决定执行顺序。优先级设计直接影响规则引擎的行为正确性。
3.4.1 优先级实现方式
| 方式 | 实现方法 | 适用场景 |
|---|---|---|
| 列表顺序 | 规则按优先级从高到低排列,引擎顺序遍历 | 规则数量少,优先级固定 |
| 优先级数值 | 每条规则绑定一个整数优先级,引擎排序后执行 | 规则可动态增删,优先级需灵活调整 |
| 权重累加 | 每条规则有权重值,引擎计算总分后决策 | 评分卡、信用评估等累积型场景 |
3.4.2 优先级数值模式示例
# 带优先级的规则引擎
from dataclasses import dataclass
from typing import List
@dataclass
class PriorityRule:
priority: int # 数值越小,优先级越高
name: str
condition: Callable[[dict], bool]
action: Callable[[dict], str]
def execute_priority_engine(data: dict, rules: List[PriorityRule]) -> str:
# 按优先级排序(升序)
sorted_rules = sorted(rules, key=lambda r: r.priority)
for rule in sorted_rules:
if rule.condition(data):
return rule.action(data)
return "默认处理"
# 定义规则
rules_with_priority = [
PriorityRule(priority=10, name="黑名单拦截",
condition=lambda d: d.get("blacklisted"),
action=lambda d: f"拦截:{d['name']} 在黑名单中"),
PriorityRule(priority=20, name="VIP优先",
condition=lambda d: d.get("level") == "VIP",
action=lambda d: f"VIP通道:{d['name']}"),
PriorityRule(priority=30, name="金额检查",
condition=lambda d: d.get("amount", 0) > 5000,
action=lambda d: f"大额订单:{d['name']} 金额 {d['amount']}"),
]
# 测试:同时满足VIP和金额检查,但VIP优先级更高
test_data = {"name": "钱七", "level": "VIP", "amount": 8000, "blacklisted": False}
print(execute_priority_engine(test_data, rules_with_priority))
# 输出:VIP通道:钱七
避坑提示: 优先级数值不要使用0或负数作为有效优先级,建议从10开始递增,预留调整空间。同时,务必处理优先级相同的情况——此时应保持规则注册时的原始顺序(稳定排序)。
3.5 三种实现方式对比
| 维度 | 字典规则 | 函数规则 | 优先级规则 |
|---|---|---|---|
| 抽象层级 | 低(键值对映射) | 中(函数封装) | 高(数据类+排序) |
| 条件复杂度 | 简单(单条件) | 中等(可组合) | 高(支持复杂逻辑) |
| 优先级支持 | 隐式(字典顺序) | 隐式(列表顺序) | 显式(数值排序) |
| 可测试性 | 一般 | 优秀 | 优秀 |
| 动态修改 | 容易(修改字典) | 中等(修改列表) | 容易(增删规则对象) |
3.6 规则引擎基础架构图
下图展示了从数据输入到规则匹配、优先级排序、动作执行的完整流程:
3.7 本章小结与实验准备
- 字典规则:适合快速原型,条件简单,规则数量少(<10条)。
- 函数规则:适合中等复杂度,每个规则独立可测,是实际项目中最常用的基础模式。
- 优先级设计:显式数值优先级是最灵活的方式,建议作为默认选择。