feat(rules): data-driven decision table engine with hard constraints

kb_rule DSL 解析 + 纯域求值（优先级分组/硬约束/用户禁忌压过专业推荐） + 400 条基准 + 黄金集雏形（T-P2-02，TD-10）。实现：dsl.py 封闭词汇表 pydantic 校验（未识别键拒绝）+ match 纯逻辑（等值/include/exclude/min-max，缺字段不命中）；engine.py 纯域求值（priority 降序、confidence 次序、is_hard avoid → hard_blocks、高优先级 avoid 剔除冲突 recommend）；service.py+repo.py published 整表载入进程内缓存 + CRUD + draft→published→archived 状态机（同 T6 语义）。测试：DSL 解析与 match 矩阵、排序/硬约束/禁忌优先、状态机矩阵、黄金集 11 组（含用户禁忌压过专业推荐与多规则叠加）、400 条 × 1000 次基准（实测均值 0.26ms/次、p95 0.41ms，预算 50ms）；集成 3 用例（published 载入命中拦截、draft 不载入、DSL 写入拒绝与状态机守卫）。单测 310 通过（+72），集成 19 通过（+3）。偏差说明： 1. 冲突仲裁按通用优先级语义实现（高优先级 avoid 剔除"严格更低"优先级的冲突 recommend），任务文档的"用户禁忌（≥95）永远压过专业推荐（≤90）" 是该规则的特例，黄金集已含该场景用例。 2. include 子句按字段类型取语义：标量字段=命中列表任一值；集合字段= 列表全部在场（子集，精确命中）；文档未区分，此实现两类字段均可用。 3. 单测引擎文件命名为 tests/unit/test_rules_evaluation.py（而非 test_rules_engine.py）：与契约指定的集成文件 tests/integration/ test_rules_engine.py 同名会导致 pytest 收集冲突（tests 目录非包）。 4. 集成测试每用例独立 engine 并 dispose：pytest-asyncio 每用例一个事件循环，直接用全局 session 工厂会让连接跨 loop 污染其它集成用例。 5. 规则不提供物理删除（命中链需可追溯，下线走 archived）；kb_rule 契约无 reviewing 状态，状态机为 draft→published→archived（draft 可撤稿）。
2026-08-14 20:10:21 +08:00
parent d5f366fa4c
commit e50a9424a1
11 changed files with 1984 additions and 9 deletions
@@ -0,0 +1,165 @@
+"""纯域求值引擎（架构 §8.2 rules-engine，TD-10）：无 I/O。
+
+防幻觉红线：**结论由规则引擎产出，LLM 只负责表达**。因此本模块是"钱与
+正确性"代码——输出必须完全确定、可解释、可回放。
+
+求值流程：
+
+1. ``dsl.match`` 过滤命中规则（子句 AND，缺字段不命中）；
+2. 按 priority 降序、confidence 降序排序（再按 rule_key 兜底，输出完全确定）；
+3. 汇总 recommend/avoid（保序去重）与解释；
+4. 硬约束：``is_hard`` 规则的 avoid 命中 → ``hard_blocks`` 非空。调用方必须
+   用其中的 avoid 约束对候选做硬过滤——禁忌违反 = 0 是发布红线；
+5. 冲突仲裁：recommend 与 avoid 同 slot 且共享维度上取值相交视为冲突，
+   **高优先级 avoid 剔除严格低优先级 recommend**。由此：用户禁忌（95）压过
+   体型（90）及以下的专业推荐，合规（100）压过一切——"数字越大越优先"在
+   规则之间同样成立，而不是只写死 95 这一个档位。
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+
+from packages.rules import dsl
+from packages.rules.dsl import SlotAttrs
+
+#: recommend/avoid 条目的去重键：(slot, 排序后的 (维度, 排序后取值) 序列)。
+ItemKey = tuple[str, tuple[tuple[str, tuple[str, ...]], ...]]
+
+
+@dataclass(frozen=True, slots=True)
+class RuleHit:
+    """命中链的一环：报告与 Agent 可解释性的数据来源（架构 §8.1）。"""
+
+    rule_key: str
+    priority: int
+    category: str
+    is_hard: bool
+    confidence: float
+    explanation: str | None
+
+
+@dataclass(frozen=True, slots=True)
+class HardBlock:
+    """一条必须执行的硬约束：命中规则元数据 + 待过滤的 avoid 约束。
+
+    调用方拿 ``avoid`` 里的 slot/attrs 对候选单品做过滤（同 slot 且任一
+    维度取值落在列表内即剔除），而不是只拿 rule_key 记日志。
+    """
+
+    rule_key: str
+    priority: int
+    category: str
+    confidence: float
+    explanation: str | None
+    avoid: tuple[SlotAttrs, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class RuleVerdict:
+    """一次求值的完整结论。"""
+
+    recommend: tuple[SlotAttrs, ...]
+    avoid: tuple[SlotAttrs, ...]
+    explanations: tuple[str, ...]
+    rule_hits: tuple[RuleHit, ...]
+    hard_blocks: tuple[HardBlock, ...]
+
+    @property
+    def explanation(self) -> str:
+        """命中规则的解释按优先级序拼接，供 LLM 直接表达。"""
+        return "；".join(self.explanations)
+
+    @property
+    def blocked(self) -> bool:
+        """是否触发硬拦截。"""
+        return bool(self.hard_blocks)
+
+
+def _item_key(item: SlotAttrs) -> ItemKey:
+    return (
+        item.slot,
+        tuple(sorted((dim, tuple(sorted(vals))) for dim, vals in item.attrs.items())),
+    )
+
+
+def _conflicts(rec: SlotAttrs, avoid: SlotAttrs) -> bool:
+    """同 slot 且至少一个共享维度上取值相交，才视为推荐与回避冲突。
+
+    不同维度的 avoid（如"回避修身"之于"推荐垫肩"）是互补信号而非冲突：
+    只有当推荐恰好落进回避圈定的事实集合时才需要仲裁。
+    """
+    if rec.slot != avoid.slot:
+        return False
+    return any(
+        dim in avoid.attrs and set(vals) & set(avoid.attrs[dim]) for dim, vals in rec.attrs.items()
+    )
+
+
+def evaluate(rules: Sequence[dsl.Rule], ctx: dsl.RuleContext) -> RuleVerdict:
+    """对一份画像求值全部规则，产出推荐/回避/命中链/硬拦截。"""
+    fired = [r for r in rules if dsl.match(r.conditions, ctx)]
+    fired.sort(key=lambda r: (-r.priority, -r.confidence, r.rule_key))
+
+    rule_hits = tuple(
+        RuleHit(
+            rule_key=r.rule_key,
+            priority=r.priority,
+            category=r.category,
+            is_hard=r.is_hard,
+            confidence=r.confidence,
+            explanation=r.effects.explanation,
+        )
+        for r in fired
+    )
+    hard_blocks = tuple(
+        HardBlock(
+            rule_key=r.rule_key,
+            priority=r.priority,
+            category=r.category,
+            confidence=r.confidence,
+            explanation=r.effects.explanation,
+            avoid=r.effects.avoid,
+        )
+        for r in fired
+        # 硬约束的拦截语义只由 avoid 表达：is_hard 规则没写 avoid 就无物可拦。
+        if r.is_hard and r.effects.avoid
+    )
+    explanations = tuple(r.effects.explanation for r in fired if r.effects.explanation)
+
+    # recommend 保序去重，并记住来源优先级供仲裁。
+    recommend: list[tuple[int, SlotAttrs]] = []
+    seen_rec: set[ItemKey] = set()
+    # avoid 保序去重；同一约束来自多条规则时取最高来源优先级参与仲裁。
+    avoid: list[SlotAttrs] = []
+    avoid_priority: dict[ItemKey, int] = {}
+    for r in fired:
+        for item in r.effects.recommend:
+            key = _item_key(item)
+            if key not in seen_rec:
+                seen_rec.add(key)
+                recommend.append((r.priority, item))
+        for item in r.effects.avoid:
+            key = _item_key(item)
+            if key in avoid_priority:
+                avoid_priority[key] = max(avoid_priority[key], r.priority)
+            else:
+                avoid_priority[key] = r.priority
+                avoid.append(item)
+
+    kept_recommend = tuple(
+        item
+        for prio, item in recommend
+        if not any(
+            avoid_priority[_item_key(blocker)] > prio and _conflicts(item, blocker)
+            for blocker in avoid
+        )
+    )
+    return RuleVerdict(
+        recommend=kept_recommend,
+        avoid=tuple(avoid),
+        explanations=explanations,
+        rule_hits=rule_hits,
+        hard_blocks=hard_blocks,
+    )