meida_backend/packages/rules/engine.py

"""纯域求值引擎（架构 §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,
    )