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 规则引擎基础架构图

下图展示了从数据输入到规则匹配、优先级排序、动作执行的完整流程:

规则引擎基础架构流程图 规则引擎基础架构:数据→规则→动作 输入数据 (Data) 规则注册表 (Rule Registry) 规则1: 条件A 优先级: 10 规则2: 条件B 优先级: 20 规则3: 条件C 优先级: 30 优先级排序 (Priority Sort) 动作执行 (Action Execution)

3.7 本章小结与实验准备

  • 字典规则:适合快速原型,条件简单,规则数量少(<10条)。
  • 函数规则:适合中等复杂度,每个规则独立可测,是实际项目中最常用的基础模式。
  • 优先级设计:显式数值优先级是最灵活的方式,建议作为默认选择。