4. 规则优先级与冲突解决

任务说明

在授信规则引擎中，多条规则可能同时命中同一笔申请。本案例为规则增加 priority 字段，当多条规则命中时按优先级（数值越大越优先）选出最终决策，并记录冲突日志，便于审计与调优。

实现思路

• 规则结构：name、priority、condition（接收申请字典返回布尔）、decision。
• 评估流程：遍历所有规则，收集命中项；若命中数 > 1，写入冲突日志；按 priority 降序取第一条作为最终决策。
• 冲突日志记录：命中规则名、各自优先级、最终选中规则，便于回溯。

完整代码

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

@dataclass
class Rule:
    name: str
    priority: int
    condition: Callable[[Dict[str, Any]], bool]
    decision: str

@dataclass
class DecisionResult:
    decision: str
    matched: List[str] = field(default_factory=list)
    conflict_log: List[str] = field(default_factory=list)

class RuleEngine:
    def __init__(self, rules: List[Rule]):
        self.rules = rules

    def evaluate(self, application: Dict[str, Any]) -> DecisionResult:
        hits = [r for r in self.rules if r.condition(application)]
        if not hits:
            return DecisionResult(decision="REJECT", matched=[], conflict_log=[])

        # 按优先级降序排序，数值越大越优先
        hits.sort(key=lambda r: r.priority, reverse=True)
        matched = [r.name for r in hits]
        conflict_log: List[str] = []

        if len(hits) > 1:
            detail = ", ".join(f"{r.name}(p={r.priority})" for r in hits)
            conflict_log.append(
                f"冲突: 命中 {len(hits)} 条规则 [{detail}]，"
                f"最终选择 {hits[0].name}(p={hits[0].priority})"
            )

        return DecisionResult(
            decision=hits[0].decision,
            matched=matched,
            conflict_log=conflict_log,
        )

if __name__ == "__main__":
    rules = [
        Rule("高负债拒绝", priority=90,
             condition=lambda a: a["debt_ratio"] > 0.7,
             decision="REJECT"),
        Rule("黑名单拒绝", priority=100,
             condition=lambda a: a["blacklisted"],
             decision="REJECT"),
        Rule("低分人工审核", priority=50,
             condition=lambda a: a["score"] < 600,
             decision="MANUAL_REVIEW"),
        Rule("优质客户通过", priority=30,
             condition=lambda a: a["score"] >= 750 and a["debt_ratio"] < 0.3,
             decision="APPROVE"),
    ]

    engine = RuleEngine(rules)

    app = {"score": 580, "debt_ratio": 0.8, "blacklisted": True}
    result = engine.evaluate(app)
    print("决策:", result.decision)
    print("命中规则:", result.matched)
    for line in result.conflict_log:
        print("日志:", line)

运行与验证

直接运行脚本：python rule_engine.py。示例申请同时命中三条规则（黑名单 p=100、高负债 p=90、低分 p=50），预期输出：

决策: REJECT
命中规则: ['黑名单拒绝', '高负债拒绝', '低分人工审核']
日志: 冲突: 命中 3 条规则 [黑名单拒绝(p=100), 高负债拒绝(p=90), 低分人工审核(p=50)]，最终选择 黑名单拒绝(p=100)

判定通过条件：最终决策为优先级最高的命中规则对应决策；matched 按优先级降序排列；命中数 > 1 时 conflict_log 非空且包含最终选中规则。可再构造只命中一条规则的申请，验证 conflict_log 为空。