规则编排与冲突解决

当治理规则数量增长后，规则之间可能产生冲突或优先级混乱。本章构建规则编排系统来管理这种复杂性。

规则冲突场景

规则优先级系统

from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Any
class RuleAction(Enum):
ALLOW = "allow"
BLOCK = "block"
REVIEW = "review"
MODIFY = "modify"
LOG = "log"
class RulePriority(Enum):
"""优先级越小越优先"""
EMERGENCY = 0      # 紧急规则（不可覆盖）
COMPLIANCE = 10    # 合规规则
SECURITY = 20      # 安全规则
BUSINESS = 30      # 业务规则
DEFAULT = 100      # 默认规则
@dataclass
class GovernanceRule:
id: str
name: str
priority: RulePriority
action: RuleAction
condition: Callable[[dict], bool]
description: str = ""
enabled: bool = True
tags: list[str] = field(default_factory=list)
@dataclass
class RuleResult:
rule_id: str
action: RuleAction
priority: RulePriority
matched: bool
class RuleOrchestrator:
"""规则编排器——管理规则执行顺序和冲突解决"""
def __init__(self):
self.rules: list[GovernanceRule] = []
def add_rule(self, rule: GovernanceRule) -> None:
self.rules.append(rule)
# 按优先级排序（小值优先）
self.rules.sort(key=lambda r: r.priority.value)
def evaluate(self, context: dict) -> list[RuleResult]:
"""评估所有规则，返回匹配结果（按优先级排序）"""
results = []
for rule in self.rules:
if not rule.enabled:
continue
matched = rule.condition(context)
results.append(RuleResult(
rule_id=rule.id,
action=rule.action,
priority=rule.priority,
matched=matched,
))
return results
def resolve(self, context: dict) -> RuleAction:
"""冲突解决——最终决策"""
results = self.evaluate(context)
matched = [r for r in results if r.matched]
if not matched:
return RuleAction.ALLOW  # 无规则匹配，默认放行
# 策略：最高优先级的匹配规则获胜
winner = matched[0]
# 特殊处理：如果有 BLOCK 和 ALLOW 冲突
block_rules = [r for r in matched if r.action == RuleAction.BLOCK]
if block_rules:
# BLOCK 始终优先（安全第一）
return RuleAction.BLOCK
return winner.action
def detect_conflicts(self) -> list[tuple[str, str, str]]:
"""检测潜在冲突的规则对"""
conflicts = []
for i, r1 in enumerate(self.rules):
for r2 in self.rules[i + 1:]:
if r1.priority == r2.priority and r1.action != r2.action:
conflicts.append((
r1.id, r2.id,
f"同优先级 {r1.priority.name} 但动作不同: "
f"{r1.action.value} vs {r2.action.value}"
))
return conflicts
# 使用示例
orchestrator = RuleOrchestrator()
orchestrator.add_rule(GovernanceRule(
id="R001",
name="禁止敏感词",
priority=RulePriority.COMPLIANCE,
action=RuleAction.BLOCK,
condition=lambda ctx: any(
w in ctx.get("input", "") for w in ["暴力", "色情"]
),
))
orchestrator.add_rule(GovernanceRule(
id="R002",
name="VIP用户放行",
priority=RulePriority.BUSINESS,
action=RuleAction.ALLOW,
condition=lambda ctx: ctx.get("user_tier") == "vip",
))
# 即使VIP用户，合规规则也优先
action = orchestrator.resolve({"input": "暴力内容", "user_tier": "vip"})
print(f"决策: {action.value}")  # → block

规则链执行模式

规则组合模式

from dataclasses import dataclass
@dataclass
class RuleChain:
"""规则链——按类别分组执行"""
name: str
rules: list[GovernanceRule]
mode: str = "first_match"  # first_match | all_match | majority
def execute(self, context: dict) -> RuleAction | None:
matched = [r for r in self.rules if r.enabled and r.condition(context)]
if not matched:
return None  # 链内无匹配，传递给下一个链
if self.mode == "first_match":
return matched[0].action
if self.mode == "all_match":
# 所有匹配规则必须一致
actions = set(r.action for r in matched)
if len(actions) == 1:
return actions.pop()
return RuleAction.REVIEW  # 不一致则人工审核
if self.mode == "majority":
# 多数决
action_counts: dict[RuleAction, int] = {}
for r in matched:
action_counts[r.action] = action_counts.get(r.action, 0) + 1
return max(action_counts, key=action_counts.get)
return None

规则编排模式对比

本章小结

• 规则优先级分层——紧急 > 合规 > 安全 > 业务 > 默认
• BLOCK 始终优先——安全第一原则，任何一条安全规则触发即拦截
• 冲突检测——自动发现同优先级、不同动作的规则对
• 链式执行——规则按类别分组，短路退出提高效率
• 多种决策模式——首次匹配、全部一致、多数决适用不同场景

下一章：动态规则更新