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 可撤稿）。
241 lines
8.8 KiB
Python
241 lines
8.8 KiB
Python
"""kb_rule 的 CRUD、draft→published 状态机与已发布规则缓存。
|
||
|
||
状态机（同 T6 语义）：draft → published → archived；archived 终态；draft 可
|
||
直接 archived（撤稿）。只有 draft 可修改内容且每次修改 version+1——直接改
|
||
published 规则等于线上建议瞬间变味，必须重新走发布动作。
|
||
|
||
规则不做物理删除：命中链（rule_hits）是报告与 Agent 可解释性的数据来源，
|
||
历史规则要可追溯，下线一律走 archived。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
from pydantic import ValidationError
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from packages.domain.ids import next_id
|
||
from packages.rules import dsl, engine, repo
|
||
from packages.shared.errors import AppError, ErrorCode
|
||
from packages.shared.logging import get_logger
|
||
from packages.shared.models import KbRule
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
STATUS_DRAFT = "draft"
|
||
STATUS_PUBLISHED = "published"
|
||
STATUS_ARCHIVED = "archived"
|
||
|
||
#: 合法状态跃迁全集（自跃迁也非法：状态机不接受无操作变更）。
|
||
LEGAL_TRANSITIONS: frozenset[tuple[str, str]] = frozenset(
|
||
{
|
||
(STATUS_DRAFT, STATUS_PUBLISHED),
|
||
(STATUS_PUBLISHED, STATUS_ARCHIVED),
|
||
(STATUS_DRAFT, STATUS_ARCHIVED),
|
||
}
|
||
)
|
||
|
||
|
||
def ensure_transition(current: str, target: str) -> None:
|
||
"""非法跃迁抛 STATE_CONFLICT——状态错误必须响亮失败，不能悄悄吞掉。"""
|
||
if (current, target) not in LEGAL_TRANSITIONS:
|
||
raise AppError(
|
||
ErrorCode.STATE_CONFLICT,
|
||
message=f"规则状态不能从 {current} 变更为 {target}",
|
||
data={"current": current, "target": target},
|
||
)
|
||
|
||
|
||
def validate_rule_payload(payload: dict[str, Any]) -> dsl.Rule:
|
||
"""DSL 写入前校验：非法结构 100% 拒绝（未识别键不静默忽略，见 dsl 模块）。"""
|
||
try:
|
||
return dsl.Rule.model_validate(payload)
|
||
except ValidationError as exc:
|
||
raise AppError(
|
||
ErrorCode.VALIDATION_FAILED,
|
||
message="规则 DSL 不合法",
|
||
data={
|
||
"errors": [
|
||
f"{'.'.join(str(loc) for loc in err['loc'])}: {err['msg']}"
|
||
for err in exc.errors()
|
||
]
|
||
},
|
||
) from exc
|
||
|
||
|
||
def row_to_rule(row: KbRule) -> dsl.Rule:
|
||
"""ORM 行 → 域规则。
|
||
|
||
已入库数据不合法属于知识库数据错误（5xxxx），必须立刻暴露——静默跳过
|
||
一条 published 规则等于线上建议悄悄缺了一块，没人会发现。
|
||
"""
|
||
try:
|
||
return dsl.Rule.model_validate(
|
||
{
|
||
"rule_key": row.rule_key,
|
||
"category": row.category,
|
||
"priority": row.priority,
|
||
"conditions": row.conditions,
|
||
"effects": row.effects,
|
||
"is_hard": row.is_hard,
|
||
"confidence": float(row.confidence),
|
||
}
|
||
)
|
||
except ValidationError as exc:
|
||
logger.error("kb_rule_dsl_invalid", rule_key=row.rule_key, errors=exc.errors())
|
||
raise AppError(
|
||
ErrorCode.CONFIG_ERROR,
|
||
message="知识库中存在不合法的规则数据",
|
||
data={"rule_key": row.rule_key},
|
||
) from exc
|
||
|
||
|
||
class PublishedRules:
|
||
"""进程内已发布规则缓存。
|
||
|
||
量级 ~400、读极多写极少：整表载入 + 管理动作后失效即可满足 p95 < 50ms
|
||
（求值本身是纯内存操作）。跨进程失效广播（Redis，架构 §10）属后续接入
|
||
项，先留 ``invalidate()`` 单一入口。并发首载最多重复读一次库，无害。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
self._rules: tuple[dsl.Rule, ...] | None = None
|
||
|
||
async def load(self, session: AsyncSession) -> tuple[dsl.Rule, ...]:
|
||
if self._rules is None:
|
||
rows = await repo.list_published(session)
|
||
self._rules = tuple(row_to_rule(row) for row in rows)
|
||
logger.info("rules_cache_loaded", count=len(self._rules))
|
||
return self._rules
|
||
|
||
def invalidate(self) -> None:
|
||
self._rules = None
|
||
|
||
|
||
_published_rules = PublishedRules()
|
||
|
||
|
||
def invalidate_published_cache() -> None:
|
||
"""发布/下线/修改规则后调用，进程内缓存立刻重建。"""
|
||
_published_rules.invalidate()
|
||
|
||
|
||
class RuleService:
|
||
"""kb_rule 的读写门面：CRUD + 状态机 + 缓存求值入口。"""
|
||
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self._session = session
|
||
|
||
# -- 读 -------------------------------------------------------------------
|
||
|
||
async def get(self, rule_key: str) -> KbRule:
|
||
return await self._require(rule_key)
|
||
|
||
async def list(self, status: str | None = None) -> list[KbRule]:
|
||
if status is None:
|
||
return await repo.list_all(self._session)
|
||
return await repo.list_by_status(self._session, status)
|
||
|
||
async def evaluate(self, ctx: dsl.RuleContext) -> engine.RuleVerdict:
|
||
"""用进程内已发布规则缓存求值（命中链 + 硬拦截 + 冲突仲裁）。"""
|
||
rules = await _published_rules.load(self._session)
|
||
return engine.evaluate(rules, ctx)
|
||
|
||
# -- 写 -------------------------------------------------------------------
|
||
|
||
async def create(
|
||
self,
|
||
payload: dict[str, Any],
|
||
*,
|
||
author: str | None = None,
|
||
evidence: str | None = None,
|
||
) -> KbRule:
|
||
"""新建 draft 规则。DSL 先校验再落库；rule_key 全局唯一。"""
|
||
rule = validate_rule_payload(payload)
|
||
if await repo.get_by_key(self._session, rule.rule_key) is not None:
|
||
raise AppError(
|
||
ErrorCode.BAD_REQUEST,
|
||
message="规则键已存在",
|
||
data={"rule_key": rule.rule_key},
|
||
)
|
||
row = repo.add_rule(
|
||
self._session,
|
||
KbRule(
|
||
rule_id=next_id(),
|
||
rule_key=rule.rule_key,
|
||
category=rule.category,
|
||
version=1,
|
||
status=STATUS_DRAFT,
|
||
priority=rule.priority,
|
||
conditions=rule.conditions.model_dump()["clauses"],
|
||
effects=rule.effects.model_dump(),
|
||
is_hard=rule.is_hard,
|
||
confidence=Decimal(str(rule.confidence)),
|
||
author=author,
|
||
evidence=evidence,
|
||
),
|
||
)
|
||
await self._session.flush()
|
||
logger.info("rule_created", rule_key=rule.rule_key, author=author)
|
||
return row
|
||
|
||
async def update(self, rule_key: str, payload: dict[str, Any]) -> KbRule:
|
||
"""修改规则内容（仅 draft）：version+1；rule_key 是命中链锚点，不可改。"""
|
||
rule = validate_rule_payload(payload)
|
||
if rule.rule_key != rule_key:
|
||
raise AppError(
|
||
ErrorCode.BAD_REQUEST,
|
||
message="rule_key 不可修改",
|
||
data={"rule_key": rule_key},
|
||
)
|
||
row = await self._require(rule_key)
|
||
if row.status != STATUS_DRAFT:
|
||
raise AppError(
|
||
ErrorCode.STATE_CONFLICT,
|
||
message="仅 draft 规则可修改，published 规则请先下线重发",
|
||
data={"rule_key": rule_key, "status": row.status},
|
||
)
|
||
row.category = rule.category
|
||
row.priority = rule.priority
|
||
row.conditions = rule.conditions.model_dump()["clauses"]
|
||
row.effects = rule.effects.model_dump()
|
||
row.is_hard = rule.is_hard
|
||
row.confidence = Decimal(str(rule.confidence))
|
||
row.version += 1
|
||
await self._session.flush()
|
||
logger.info("rule_updated", rule_key=rule_key, version=row.version)
|
||
return row
|
||
|
||
async def publish(self, rule_key: str) -> KbRule:
|
||
row = await self._require(rule_key)
|
||
ensure_transition(row.status, STATUS_PUBLISHED)
|
||
row.status = STATUS_PUBLISHED
|
||
row.published_at = datetime.now(UTC)
|
||
await self._session.flush()
|
||
invalidate_published_cache()
|
||
logger.info("rule_published", rule_key=rule_key, version=row.version)
|
||
return row
|
||
|
||
async def archive(self, rule_key: str) -> KbRule:
|
||
"""下线/撤稿（draft 与 published 均可）。规则不物理删除，保留可追溯性。"""
|
||
row = await self._require(rule_key)
|
||
ensure_transition(row.status, STATUS_ARCHIVED)
|
||
row.status = STATUS_ARCHIVED
|
||
await self._session.flush()
|
||
invalidate_published_cache()
|
||
logger.info("rule_archived", rule_key=rule_key)
|
||
return row
|
||
|
||
async def _require(self, rule_key: str) -> KbRule:
|
||
row = await repo.get_by_key(self._session, rule_key)
|
||
if row is None:
|
||
raise AppError(
|
||
ErrorCode.NOT_FOUND,
|
||
message="规则不存在",
|
||
data={"rule_key": rule_key},
|
||
)
|
||
return row
|