4. 规则引擎核心概念：规则、事实、规则集、上下文与冲突解决

规则引擎的五个核心概念构成了推理系统的骨架。理解它们之间的协作关系，是编写可维护、可扩展规则系统的前提。下面逐一拆解每个概念的定义、职责与交互方式。

4.1 核心概念总览

概念	英文	职责	类比（现实世界）
规则	Rule	定义条件-动作对，描述“当某条件成立时执行某操作”	法律条文
事实	Fact	表示已知的数据对象或状态，是规则推理的输入	案件证据
规则集	RuleSet	对规则进行分组、排序、命名空间管理	法律汇编
执行上下文	Context	承载规则执行过程中的临时数据、全局配置与生命周期	法庭庭审记录
冲突解决策略	Conflict Resolution	当多条规则同时满足条件时，决定执行顺序	法官裁决优先级

4.2 规则（Rule）

规则是规则引擎的最小逻辑单元。一个标准规则包含三个部分：名称、条件（LHS）、动作（RHS）。

• 名称：唯一标识，便于日志追踪和调试。
• 条件（LHS — Left Hand Side）：对事实的布尔表达式，支持比较、包含、正则、自定义函数。
• 动作（RHS — Right Hand Side）：条件满足时执行的操作，如修改事实、触发外部服务、记录日志。

# 规则定义示例（伪代码风格）
rule "折扣规则-满100减20"
    when
        order.total >= 100
        order.user_level == "VIP"
    then
        order.discount = 20
        context.add_log("VIP用户触发满减")

4.3 事实（Fact）

事实是引擎推理的原材料。在 Python 规则引擎中，事实通常是一个普通的 Python 对象（dict、dataclass、Pydantic 模型）。引擎通过模式匹配将事实与规则条件进行比对。

• 不可变性建议：事实在进入引擎后尽量不修改，如需变更应通过规则动作显式更新。
• 类型支持：数字、字符串、列表、嵌套对象、枚举。
• 事实插入：通过 engine.insert(fact) 或 context.add_fact(fact) 注入。

# 事实定义示例
from dataclasses import dataclass

@dataclass
class OrderFact:
    order_id: str
    total: float
    user_level: str
    items: list

# 插入事实
fact = OrderFact(order_id="ORD001", total=120.0, user_level="VIP", items=["A", "B"])
engine.insert(fact)

4.4 规则集（RuleSet）

规则集用于对规则进行逻辑分组和优先级管理。一个规则集可以包含多个规则，并支持继承、覆盖、条件激活等高级特性。

特性	说明	典型用途
分组	将同类规则放在一起	折扣规则、风控规则、校验规则
排序	定义规则集内规则的执行顺序	优先级高的规则先执行
命名空间	避免不同模块的规则名称冲突	订单模块 vs 用户模块
条件激活	根据上下文决定是否加载该规则集	仅当促销活动开启时加载促销规则

# 规则集定义示例
ruleset = RuleSet("order_discount_rules", priority=10)
ruleset.add_rule(rule_vip_discount)
ruleset.add_rule(rule_new_user_discount)
ruleset.add_rule(rule_seasonal_discount)

engine.register_ruleset(ruleset)

4.5 执行上下文（Context）

上下文是规则执行期间的“工作台”。它负责存储临时变量、全局配置、日志记录、事实变更追踪以及规则执行状态。

• 全局配置：如阈值、开关、环境标识。
• 临时数据：规则间共享的中间计算结果。
• 日志与监控：记录每条规则的触发次数、耗时。
• 生命周期管理：上下文在规则执行开始时创建，执行结束后销毁。

# 上下文使用示例
class RuleContext:
    def __init__(self, config: dict):
        self.config = config
        self.logs = []
        self.intermediate_results = {}
        self.execution_count = 0

    def add_log(self, message: str):
        self.logs.append(f"[{self.execution_count}] {message}")

    def get_config(self, key: str, default=None):
        return self.config.get(key, default)

# 使用
ctx = RuleContext({"max_discount": 50})
ctx.add_log("开始执行规则集")

4.6 冲突解决策略

当多条规则的条件同时满足时，引擎必须决定先执行哪一条。不同的策略适用于不同的业务场景。

策略名称	规则	适用场景
优先级优先	按规则显式指定的优先级降序执行	业务有明确的等级制度（如法律条款优先级）
最近更新优先	最近被修改过的事实所触发的规则先执行	实时监控、动态数据流
特定性优先	条件更具体（约束更多）的规则先执行	异常处理、精确匹配优于模糊匹配
先进先出	按规则加入引擎的顺序依次执行	简单线性校验流程
随机	随机选择一条满足条件的规则	A/B 测试、负载均衡

4.7 核心概念协作关系图

下图展示了五个核心概念在规则引擎执行过程中的协作关系：

4.8 实战案例：订单风控规则引擎

下面通过一个完整的 Python 代码片段，展示五个核心概念如何协同工作：

from dataclasses import dataclass, field
from typing import List, Dict

# 1. 事实定义
@dataclass
class OrderFact:
    order_id: str
    amount: float
    user_id: str
    ip_country: str
    is_new_user: bool

# 2. 规则定义
class Rule:
    def __init__(self, name: str, condition, action, priority: int = 0):
        self.name = name
        self.condition = condition
        self.action = action
        self.priority = priority

    def evaluate(self, fact: OrderFact, ctx: 'RuleContext') -> bool:
        return self.condition(fact, ctx)

    def execute(self, fact: OrderFact, ctx: 'RuleContext'):
        self.action(fact, ctx)

# 3. 上下文定义
class RuleContext:
    def __init__(self):
        self.logs: List[str] = []
        self.risk_score: float = 0.0
        self.blocked: bool = False

    def add_log(self, msg: str):
        self.logs.append(msg)

# 4. 规则集定义
class RuleSet:
    def __init__(self, name: str, strategy: str = "priority"):
        self.name = name
        self.rules: List[Rule] = []
        self.strategy = strategy

    def add_rule(self, rule: Rule):
        self.rules.append(rule)
        self.rules.sort(key=lambda r: r.priority, reverse=True)

    def fire(self, fact: OrderFact, ctx: RuleContext):
        for rule in self.rules:
            if rule.evaluate(fact, ctx):
                rule.execute(fact, ctx)
                ctx.add_log(f"触发规则: {rule.name}")

# 5. 实例化规则
def high_amount_condition(fact: OrderFact, ctx: RuleContext) -> bool:
    return fact.amount > 10000

def high_amount_action(fact: OrderFact, ctx: RuleContext):
    ctx.risk_score += 30
    ctx.add_log(f"高风险订单: {fact.order_id}")

def oversea_condition(fact: OrderFact, ctx: RuleContext) -> bool:
    return fact.ip_country != "CN"

def oversea_action(fact: OrderFact, ctx: RuleContext):
    ctx.risk_score += 20
    ctx.add_log(f"海外IP订单: {fact.order_id}")

rule1 = Rule("高金额风控", high_amount_condition, high_amount_action, priority=10)
rule2 = Rule("海外IP风控", oversea_condition, oversea_action, priority=5)

# 6. 组装并执行
ruleset = RuleSet("风控规则集", strategy="priority")
ruleset.add_rule(rule1)
ruleset.add_rule(rule2)

ctx = RuleContext()
fact = OrderFact("ORD999", 15000, "U001", "US", False)
ruleset.fire(fact, ctx)

print(ctx.logs)
# 输出: ['高风险订单: ORD999', '海外IP订单: ORD999']
print(f"最终风险分: {ctx.risk_score}")
# 输出: 最终风险分: 50.0