xsl
e50a9424a1
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 可撤稿）。
331 lines
13 KiB
Python
331 lines
13 KiB
Python
"""规则引擎黄金集雏形（架构 §1081）：固定画像 → 期望输出。
|
||
|
||
改引擎行为必须先过黄金集——这是"专业可信"的回归防线。必含两类场景：
|
||
硬禁忌（用户禁忌压过专业推荐）与多规则叠加。规则库是知识库文档 §3.2⑩ 的
|
||
迷你缩影（体型/场景/温度/风格/趋势/用户禁忌/合规硬约束各至少一条）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Iterable
|
||
from typing import Any
|
||
|
||
from packages.rules import dsl, engine
|
||
|
||
# ── 固定规则库（8 条）───────────────────────────────────────────────────────
|
||
|
||
_GOLDEN_RULES = [
|
||
# 体型（90）：文档 §3.2⑩ 原始示例规则
|
||
{
|
||
"rule_key": "R-BODY-A-002",
|
||
"category": "body_shape",
|
||
"priority": 90,
|
||
"confidence": 0.95,
|
||
"conditions": {
|
||
"body_shape": "A型",
|
||
"features": {"include": ["溜肩"]},
|
||
"goals": {"include": ["修饰溜肩"]},
|
||
},
|
||
"effects": {
|
||
"recommend": [
|
||
{"slot": "top", "attrs": {"shoulder": ["垫肩", "自然肩"], "sleeve": ["泡泡袖"]}}
|
||
],
|
||
"avoid": [{"slot": "top", "attrs": {"shoulder": ["落肩"], "silhouette": ["修身"]}}],
|
||
"explanation": "溜肩需要用垫肩与泡泡袖在肩部建立结构感。",
|
||
},
|
||
},
|
||
# 体型（90，同优先级低置信度：验证二级排序）
|
||
{
|
||
"rule_key": "R-BODY-A-010",
|
||
"category": "body_shape",
|
||
"priority": 90,
|
||
"confidence": 0.90,
|
||
"conditions": {"body_shape": "A型"},
|
||
"effects": {
|
||
"recommend": [{"slot": "bottom", "attrs": {"silhouette": ["直筒", "阔腿"]}}],
|
||
"avoid": [{"slot": "bottom", "attrs": {"silhouette": ["紧身"]}}],
|
||
"explanation": "A型下半身宜直线轮廓，避免紧身放大胯部。",
|
||
},
|
||
},
|
||
# 场景（80）
|
||
{
|
||
"rule_key": "R-OCCASION-INTERVIEW-001",
|
||
"category": "occasion",
|
||
"priority": 80,
|
||
"confidence": 0.90,
|
||
"conditions": {"occasion": "面试"},
|
||
"effects": {
|
||
"recommend": [
|
||
{"slot": "top", "attrs": {"collar": ["衬衫领"]}},
|
||
{"slot": "bottom", "attrs": {"silhouette": ["直筒"]}},
|
||
],
|
||
"avoid": [{"slot": "top", "attrs": {"neckline": ["深V"]}}],
|
||
"explanation": "面试场合需要干练可信的轮廓。",
|
||
},
|
||
},
|
||
# 温度（75）：闭区间
|
||
{
|
||
"rule_key": "R-TEMP-COMFORT-18-26",
|
||
"category": "temperature",
|
||
"priority": 75,
|
||
"confidence": 0.85,
|
||
"conditions": {"temperature": {"min": 18, "max": 26}},
|
||
"effects": {
|
||
"recommend": [{"slot": "outer", "attrs": {"thickness": ["薄款"]}}],
|
||
"avoid": [{"slot": "outer", "attrs": {"thickness": ["厚款"]}}],
|
||
"explanation": "18-26℃ 单层薄外套最舒适。",
|
||
},
|
||
},
|
||
# 风格（70）：与场景规则的推荐完全相同 → 验证去重
|
||
{
|
||
"rule_key": "R-STYLE-OL-001",
|
||
"category": "style",
|
||
"priority": 70,
|
||
"confidence": 0.80,
|
||
"conditions": {"style": "通勤"},
|
||
"effects": {
|
||
"recommend": [{"slot": "top", "attrs": {"collar": ["衬衫领"]}}],
|
||
"explanation": "通勤风格以衬衫领立住精神气。",
|
||
},
|
||
},
|
||
# 趋势（50）：推荐恰与用户禁忌冲突 → 必须被压过
|
||
{
|
||
"rule_key": "R-TREND-SWEET-001",
|
||
"category": "trend",
|
||
"priority": 50,
|
||
"confidence": 0.70,
|
||
"conditions": {"style": "甜酷"},
|
||
"effects": {
|
||
"recommend": [{"slot": "top", "attrs": {"length": ["短款"]}}],
|
||
"explanation": "短款上衣是当季甜酷风的核心单品。",
|
||
},
|
||
},
|
||
# 用户禁忌（95）：avoid 永远压过专业推荐
|
||
{
|
||
"rule_key": "R-TABOO-NAVEL-001",
|
||
"category": "user_taboo",
|
||
"priority": 95,
|
||
"confidence": 1.0,
|
||
"conditions": {"user_taboos": {"include": ["露脐"]}},
|
||
"effects": {
|
||
"avoid": [{"slot": "top", "attrs": {"length": ["短款", "露脐"]}}],
|
||
"explanation": "用户不接受露脐，已剔除相关单品。",
|
||
},
|
||
},
|
||
# 合规（100）：is_hard，avoid 命中即硬拦截
|
||
{
|
||
"rule_key": "R-HARD-EXOTIC-001",
|
||
"category": "compliance",
|
||
"priority": 100,
|
||
"confidence": 1.0,
|
||
"is_hard": True,
|
||
"conditions": {"occasion": {"include": ["红毯", "晚宴"]}},
|
||
"effects": {
|
||
"avoid": [
|
||
{"slot": "accessory", "attrs": {"material": ["真兽皮"]}},
|
||
{"slot": "shoes", "attrs": {"material": ["真皮"]}},
|
||
],
|
||
"explanation": "合规红线：不推荐真兽皮材质。",
|
||
},
|
||
},
|
||
]
|
||
|
||
_RULES = [dsl.Rule.model_validate(r) for r in _GOLDEN_RULES]
|
||
|
||
|
||
# ── 断言辅助：把 SlotAttrs 规整成可比较的集合 ────────────────────────────────
|
||
|
||
|
||
#: 单个 recommend/avoid 条目规整后的形状。
|
||
NormItem = tuple[str, tuple[tuple[str, tuple[str, ...]], ...]]
|
||
|
||
|
||
def _item(slot: str, **dims: tuple[str, ...]) -> NormItem:
|
||
# 与 _norm 相同的规整规则（维度与取值都排序），期望值不必手写排序
|
||
return (slot, tuple(sorted((d, tuple(sorted(v))) for d, v in dims.items())))
|
||
|
||
|
||
def _norm(items: Iterable[dsl.SlotAttrs]) -> frozenset[NormItem]:
|
||
return frozenset(
|
||
(i.slot, tuple(sorted((d, tuple(sorted(v))) for d, v in i.attrs.items()))) for i in items
|
||
)
|
||
|
||
|
||
def _evaluate(**ctx: Any) -> engine.RuleVerdict:
|
||
return engine.evaluate(_RULES, dsl.RuleContext(**ctx))
|
||
|
||
|
||
def _hit_keys(verdict: engine.RuleVerdict) -> list[str]:
|
||
return [h.rule_key for h in verdict.rule_hits]
|
||
|
||
|
||
# ── 1. 基础命中 ─────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_a_shape_with_sloped_shoulders() -> None:
|
||
verdict = _evaluate(body_shape="A型", features=("溜肩",), goals=("修饰溜肩",))
|
||
assert _hit_keys(verdict) == ["R-BODY-A-002", "R-BODY-A-010"]
|
||
assert _norm(verdict.recommend) == {
|
||
_item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)),
|
||
_item("bottom", silhouette=("阔腿", "直筒")),
|
||
}
|
||
assert _norm(verdict.avoid) == {
|
||
_item("top", shoulder=("落肩",), silhouette=("修身",)),
|
||
_item("bottom", silhouette=("紧身",)),
|
||
}
|
||
assert verdict.hard_blocks == ()
|
||
assert verdict.blocked is False
|
||
|
||
|
||
def test_missing_body_shape_does_not_fire_shape_rules() -> None:
|
||
# 缺字段 = 不匹配该子句：只有 features/goals 时体型规则不命中
|
||
verdict = _evaluate(features=("溜肩",), goals=("修饰溜肩",))
|
||
assert _hit_keys(verdict) == []
|
||
assert verdict.recommend == ()
|
||
assert verdict.avoid == ()
|
||
assert verdict.explanation == ""
|
||
|
||
|
||
def test_include_is_subset_semantics() -> None:
|
||
# features 是 include 列表的超集 → 命中
|
||
hit = _evaluate(body_shape="A型", features=("溜肩", "宽胯"), goals=("修饰溜肩",))
|
||
assert "R-BODY-A-002" in _hit_keys(hit)
|
||
# goals 缺一个 → 不命中
|
||
miss = _evaluate(body_shape="A型", features=("溜肩",), goals=("显瘦",))
|
||
assert "R-BODY-A-002" not in _hit_keys(miss)
|
||
assert _hit_keys(miss) == ["R-BODY-A-010"]
|
||
|
||
|
||
# ── 2. 多规则叠加 ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_body_plus_temperature_plus_occasion_stack() -> None:
|
||
verdict = _evaluate(
|
||
body_shape="A型",
|
||
features=("溜肩",),
|
||
goals=("修饰溜肩",),
|
||
temperature=20.0,
|
||
occasion="面试",
|
||
)
|
||
assert _hit_keys(verdict) == [
|
||
"R-BODY-A-002",
|
||
"R-BODY-A-010",
|
||
"R-OCCASION-INTERVIEW-001",
|
||
"R-TEMP-COMFORT-18-26",
|
||
]
|
||
# 四条规则的推荐全部叠加，互不冲突（不同维度/不同 slot；两条 bottom 推荐
|
||
# 的属性集不同，各自保留）
|
||
assert _norm(verdict.recommend) == {
|
||
_item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)),
|
||
_item("bottom", silhouette=("阔腿", "直筒")),
|
||
_item("top", collar=("衬衫领",)),
|
||
_item("bottom", silhouette=("直筒",)),
|
||
_item("outer", thickness=("薄款",)),
|
||
}
|
||
assert _norm(verdict.avoid) == {
|
||
_item("top", shoulder=("落肩",), silhouette=("修身",)),
|
||
_item("bottom", silhouette=("紧身",)),
|
||
_item("top", neckline=("深V",)),
|
||
_item("outer", thickness=("厚款",)),
|
||
}
|
||
assert verdict.blocked is False
|
||
|
||
|
||
def test_same_priority_sorted_by_confidence_and_dedupe_collapsed() -> None:
|
||
# A-002 不命中（缺 features/goals），A-010 与面试/通勤规则叠加；
|
||
# 场景(80)与风格(70)推荐同一件衬衫领 → 去重为一条
|
||
verdict = _evaluate(body_shape="A型", occasion="面试", style="通勤")
|
||
assert _hit_keys(verdict) == [
|
||
"R-BODY-A-010",
|
||
"R-OCCASION-INTERVIEW-001",
|
||
"R-STYLE-OL-001",
|
||
]
|
||
assert _norm(verdict.recommend) == {
|
||
_item("bottom", silhouette=("阔腿", "直筒")),
|
||
_item("bottom", silhouette=("直筒",)),
|
||
_item("top", collar=("衬衫领",)),
|
||
}
|
||
assert verdict.explanations == (
|
||
"A型下半身宜直线轮廓，避免紧身放大胯部。",
|
||
"面试场合需要干练可信的轮廓。",
|
||
"通勤风格以衬衫领立住精神气。",
|
||
)
|
||
|
||
|
||
def test_temperature_closed_interval_boundaries() -> None:
|
||
assert _hit_keys(_evaluate(temperature=18.0)) == ["R-TEMP-COMFORT-18-26"]
|
||
assert _hit_keys(_evaluate(temperature=26.0)) == ["R-TEMP-COMFORT-18-26"]
|
||
assert _hit_keys(_evaluate(temperature=17.9)) == []
|
||
assert _hit_keys(_evaluate(temperature=26.1)) == []
|
||
|
||
|
||
# ── 3. 用户禁忌压过专业推荐（硬性场景）──────────────────────────────────────
|
||
|
||
|
||
def test_user_taboo_overrides_trend_recommendation() -> None:
|
||
# 趋势(50)推荐短款上衣，用户禁忌(95)回避短款/露脐 → 推荐被剔除
|
||
verdict = _evaluate(style="甜酷", user_taboos=("露脐",))
|
||
assert _hit_keys(verdict) == ["R-TABOO-NAVEL-001", "R-TREND-SWEET-001"]
|
||
assert verdict.recommend == ()
|
||
assert _norm(verdict.avoid) == {_item("top", length=("露脐", "短款"))}
|
||
# 禁忌不是 is_hard：不构成硬拦截，但推荐侧已经干净
|
||
assert verdict.hard_blocks == ()
|
||
assert verdict.explanation.startswith("用户不接受露脐")
|
||
|
||
|
||
def test_user_taboo_keeps_unrelated_slot_recommendations() -> None:
|
||
# 禁忌只作用于 top.length：体型对 bottom 的推荐不受影响
|
||
verdict = _evaluate(body_shape="A型", user_taboos=("露脐",))
|
||
assert _hit_keys(verdict) == ["R-TABOO-NAVEL-001", "R-BODY-A-010"]
|
||
assert _norm(verdict.recommend) == {_item("bottom", silhouette=("阔腿", "直筒"))}
|
||
|
||
|
||
# ── 4. 硬约束拦截（is_hard）─────────────────────────────────────────────────
|
||
|
||
|
||
def test_hard_compliance_rule_blocks_gala_occasion() -> None:
|
||
verdict = _evaluate(occasion="红毯")
|
||
assert _hit_keys(verdict) == ["R-HARD-EXOTIC-001"]
|
||
assert verdict.blocked is True
|
||
assert [b.rule_key for b in verdict.hard_blocks] == ["R-HARD-EXOTIC-001"]
|
||
# 调用方用 hard_blocks.avoid 对候选做硬过滤：约束内容完整带出
|
||
assert _norm(verdict.hard_blocks[0].avoid) == {
|
||
_item("accessory", material=("真兽皮",)),
|
||
_item("shoes", material=("真皮",)),
|
||
}
|
||
assert verdict.hard_blocks[0].explanation == "合规红线：不推荐真兽皮材质。"
|
||
|
||
|
||
def test_hard_rule_not_fired_on_other_occasions() -> None:
|
||
assert _evaluate(occasion="通勤").hard_blocks == ()
|
||
assert _evaluate(occasion="通勤").blocked is False
|
||
|
||
|
||
# ── 5. 综合叠加：优先级链 + 禁忌压制 + 硬拦截同时发生 ────────────────────────
|
||
|
||
|
||
def test_composite_profile_full_priority_chain() -> None:
|
||
verdict = _evaluate(
|
||
body_shape="A型",
|
||
features=("溜肩",),
|
||
goals=("修饰溜肩",),
|
||
style="甜酷",
|
||
user_taboos=("露脐",),
|
||
occasion="红毯",
|
||
)
|
||
# 命中链严格按 priority 降序、同优先级按 confidence 降序
|
||
assert _hit_keys(verdict) == [
|
||
"R-HARD-EXOTIC-001", # 100
|
||
"R-TABOO-NAVEL-001", # 95
|
||
"R-BODY-A-002", # 90 / 0.95
|
||
"R-BODY-A-010", # 90 / 0.90
|
||
"R-TREND-SWEET-001", # 50
|
||
]
|
||
# 体型推荐保留（与禁忌不同维度）；趋势推荐短款被禁忌压掉
|
||
assert _norm(verdict.recommend) == {
|
||
_item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)),
|
||
_item("bottom", silhouette=("阔腿", "直筒")),
|
||
}
|
||
assert verdict.blocked is True
|
||
assert [b.rule_key for b in verdict.hard_blocks] == ["R-HARD-EXOTIC-001"]
|