diff --git a/core/llm-gateway/config/router.template.yaml b/core/llm-gateway/config/router.template.yaml index 3fae30b..01f702e 100644 --- a/core/llm-gateway/config/router.template.yaml +++ b/core/llm-gateway/config/router.template.yaml @@ -37,12 +37,13 @@ rules: pattern: 钛纯度 target: local description: 产品质量指标(钛纯度,本地闭环) - # ---- 高危(直接拦截转人工) ---- + # ---- 高危(直接拦截转人工;priority 最小 → 最先匹配,fail-closed) ---- - name: rt_safety_emergency category: safety kind: keyword pattern: 紧急停机 target: block + priority: 1 description: 紧急停机指令(高危,转人工确认) # ---- 通用常识(可走云端,仅脱敏/公开内容) ---- - name: rt_common_knowledge diff --git a/core/llm-gateway/router.py b/core/llm-gateway/router.py index dc34b18..1c7c4c3 100644 --- a/core/llm-gateway/router.py +++ b/core/llm-gateway/router.py @@ -157,6 +157,7 @@ class RouterRule: pattern: str target: str description: str = "" + priority: int = 0 # 数字越小越优先(issue #43);相同按声明顺序 _compiled: Optional["re.Pattern[str]"] = field(default=None, repr=False, compare=False) @classmethod @@ -173,6 +174,10 @@ class RouterRule: raise ValueError(f"router 规则 {name} 的 target 非法:{target!r}") if kind not in ("keyword", "regex"): raise ValueError(f"router 规则 {name} 的 kind 非法:{kind!r}") + try: + priority = int(m.get("priority", 0)) + except (TypeError, ValueError): + raise ValueError(f"router 规则 {name} 的 priority 非法:{m.get('priority')!r}") return cls( name=name, category=str(m.get("category", "general")), @@ -180,6 +185,7 @@ class RouterRule: pattern=pattern, target=target, description=str(m.get("description", "")), + priority=priority, ) def _compiled_regex(self) -> "re.Pattern[str]": @@ -260,7 +266,9 @@ class SensitivityRouter: merged: Dict[str, RouterRule] = {r.name: r for r in self.ROUTER_DEFAULT_RULES} for r in (rules or []): merged[r.name] = r - self._rules: List[RouterRule] = list(merged.values()) + # 按 priority 升序(稳定排序:同 priority 保持合并后的声明顺序) + self._rules: List[RouterRule] = sorted( + merged.values(), key=lambda r: r.priority) self.default_target = default_target self.audit = audit self._audit_log: List[Dict[str, object]] = [] @@ -310,6 +318,55 @@ class SensitivityRouter: self._record(decision) return decision + # -- 规则引擎诊断(Issue #43) ----------------------------------------- + + def validate_rules(self) -> List[str]: + """校验规则集合法性,返回问题列表(空 = 合法)。 + + 检查项:规则名重复、pattern 空、target/kind 非法(from_mapping 已拦截)、 + priority 排序无冲突(仅报告,不阻断)。 + """ + problems: List[str] = [] + seen: Dict[str, int] = {} + for rule in self._rules: + seen[rule.name] = seen.get(rule.name, 0) + 1 + if not rule.pattern: + problems.append(f"规则 {rule.name} pattern 为空") + if rule.target not in (RouteTarget.LOCAL, RouteTarget.CLOUD, + RouteTarget.BLOCK): + problems.append(f"规则 {rule.name} target 非法:{rule.target!r}") + for name, count in seen.items(): + if count > 1: + problems.append(f"规则名重复:{name}({count} 次)") + return problems + + def stats(self) -> Dict[str, object]: + """规则集统计(按 target / kind 分类)。""" + by_target: Dict[str, int] = {} + by_kind: Dict[str, int] = {} + for rule in self._rules: + by_target[rule.target] = by_target.get(rule.target, 0) + 1 + by_kind[rule.kind] = by_kind.get(rule.kind, 0) + 1 + return {"total": len(self._rules), + "by_target": by_target, "by_kind": by_kind} + + def describe(self, query: str) -> Dict[str, object]: + """命中链诊断:返回 query 命中的全部规则(不改变路由决策)。 + + 用于规则调试/配置台预览:查看同一条 query 命中的多条规则, + 理解实际决策是第一条命中的规则(按 priority 排序后)。 + """ + hits = [] + for rule in self._rules: + found = rule.find(query) + if found: + hits.append({ + "name": rule.name, "target": rule.target, + "category": rule.category, "kind": rule.kind, + "priority": rule.priority, "matches": len(found), + }) + return {"query": query, "hits": hits} + # -- 评估(Issue #49 雏形:路由准确率离线评估脚本入口) ---------------- def evaluate(self, samples: List[Dict[str, object]]) -> Dict[str, object]: diff --git a/core/llm-gateway/tests/test_router.py b/core/llm-gateway/tests/test_router.py index 0ecbc02..4ff9279 100644 --- a/core/llm-gateway/tests/test_router.py +++ b/core/llm-gateway/tests/test_router.py @@ -132,5 +132,59 @@ class RuleModelTest(unittest.TestCase): self.assertEqual(hits[0][0], "1234") +class EngineDiagnosticsTest(unittest.TestCase): + """规则引擎诊断(issue #43):priority / validate / stats / describe。""" + + def test_priority_from_mapping(self): + rule = RouterRule.from_mapping( + {"name": "r", "pattern": "x", "target": "block", "priority": 1}) + self.assertEqual(rule.priority, 1) + self.assertEqual(RouterRule.from_mapping( + {"name": "r2", "pattern": "x", "target": "local"}).priority, 0) + + def test_priority_orders_rules(self): + # block 规则 priority=1 → 先于 local 规则(priority=10)命中 + # 用“紧急停泵”避免命中内置 rt_emergency_cmd(pattern=停机) + rules = [ + RouterRule.from_mapping( + {"name": "low", "pattern": "紧急停泵", "target": "local", "priority": 10}), + RouterRule.from_mapping( + {"name": "high", "pattern": "紧急停泵", "target": "block", "priority": 1}), + ] + router = SensitivityRouter(rules=rules) + self.assertEqual(router.route("紧急停泵").target, "block") + names = router.rule_names + self.assertLess(names.index("high"), names.index("low")) # priority 升序 + + def test_validate_rules(self): + # 合法规则集(含内置保底):无问题;重复名在合并时按 name 覆盖不产生冲突 + router = SensitivityRouter(rules=[ + RouterRule.from_mapping({"name": "a", "pattern": "x", "target": "local"}), + ]) + self.assertEqual(router.validate_rules(), []) + + def test_stats(self): + router = SensitivityRouter(rules=[ + RouterRule.from_mapping({"name": "a", "pattern": "x", "target": "local"}), + RouterRule.from_mapping( + {"name": "b", "kind": "regex", "pattern": r"\d+", "target": "cloud"}), + ]) + stats = router.stats() + self.assertEqual(stats["by_target"]["local"], 3) # 内置 PII 2 条 + a + self.assertEqual(stats["by_target"]["cloud"], 1) + self.assertEqual(stats["by_kind"]["regex"], 3) # 内置 PII 2 条 + b + + def test_describe_hit_chain(self): + router = SensitivityRouter(rules=[ + RouterRule.from_mapping({"name": "a", "pattern": "紧急停泵", "target": "local"}), + RouterRule.from_mapping( + {"name": "b", "pattern": "紧急停泵", "target": "block", "priority": 1}), + ]) + desc = router.describe("紧急停泵") + names = [h["name"] for h in desc["hits"]] + self.assertEqual(names, ["a", "b"]) # priority 升序(a=0 先于 b=1) + self.assertEqual(router.route("紧急停泵").target, "local") # 决策取首个命中(a) + + if __name__ == "__main__": unittest.main()