diff --git a/router_system/classifier.py b/router_system/classifier.py index 099aad3..c28fd4d 100644 --- a/router_system/classifier.py +++ b/router_system/classifier.py @@ -214,6 +214,8 @@ class HuggingFaceClassifier(BaseClassifier): """可选：基于 transformers 的序列分类模型。 仅当安装 torch+transformers 且模型可加载时可用；否则抛错提示。 + 失败安全（T-R3，采纳 llmrouter 分类失败静默降级思想）：模型加载成功但 + 推理期异常时，自动回落内置规则分类器，不让单次推理异常打垮路由。 """ def __init__(self, model_name: str, num_labels: int = 5, confidence_floor: float = 0.55): @@ -229,23 +231,27 @@ class HuggingFaceClassifier(BaseClassifier): ) self.labels = ["code", "math", "legal", "medical", "general"] self.confidence_floor = confidence_floor + self._rule_fallback = RuleClassifier(confidence_floor=confidence_floor) def classify(self, query: str) -> Classification: - import torch # type: ignore + try: + import torch # type: ignore - inputs = self.tokenizer(query, return_tensors="pt", truncation=True, max_length=256) - with torch.no_grad(): - logits = self.model(**inputs).logits - probs = torch.softmax(logits, dim=-1)[0] - idx = int(probs.argmax()) - diff, ds = estimate_difficulty(query) - return Classification( - domain=self.labels[idx], - confidence=round(float(probs[idx]), 4), - difficulty=diff, - difficulty_score=ds, - raw_scores={self.labels[i]: round(float(probs[i]), 3) for i in range(len(self.labels))}, - ) + inputs = self.tokenizer(query, return_tensors="pt", truncation=True, max_length=256) + with torch.no_grad(): + logits = self.model(**inputs).logits + probs = torch.softmax(logits, dim=-1)[0] + idx = int(probs.argmax()) + diff, ds = estimate_difficulty(query) + return Classification( + domain=self.labels[idx], + confidence=round(float(probs[idx]), 4), + difficulty=diff, + difficulty_score=ds, + raw_scores={self.labels[i]: round(float(probs[i]), 3) for i in range(len(self.labels))}, + ) + except Exception: # noqa: BLE001 推理失败回落规则分类器（失败安全） + return self._rule_fallback.classify(query) def build_classifier(cfg: Dict) -> BaseClassifier: @@ -265,6 +271,18 @@ def build_classifier(cfg: Dict) -> BaseClassifier: clf.rules = {**DOMAIN_RULES, **external} return clf if ctype == "hf": - return HuggingFaceClassifier(cfg.get("model", "Qwen/Qwen3-0.6B"), confidence_floor=floor) + # 失败安全（T-R3）：ML 依赖缺失 / 模型加载失败时回落规则分类器 + try: + return HuggingFaceClassifier( + cfg.get("model", "Qwen/Qwen3-0.6B"), confidence_floor=floor) + except Exception as e: # noqa: BLE001 + print(f"[classifier] HF 分类器不可用（{type(e).__name__}），回落规则分类器") + clf = RuleClassifier(confidence_floor=floor) + rules_file = cfg.get("rules_file") + external = load_domain_rules(rules_file) if rules_file \ + else (load_domain_rules() if DEFAULT_RULES_FILE.exists() else None) + if external: + clf.rules = {**DOMAIN_RULES, **external} + return clf raise ValueError(f"未知分类器类型: {ctype}（支持 rule | hf）") diff --git a/router_system/difficulty.py b/router_system/difficulty.py index 31fa5f9..efb2f5c 100644 --- a/router_system/difficulty.py +++ b/router_system/difficulty.py @@ -25,16 +25,43 @@ _MEDIUM_MARKERS = [ "translate", "fix", "explain", "describe", "list", ] + +def _compile_markers(markers): + """标记表编译：英文标记用词边界正则，中文标记保持子串匹配。 + + T-R3（采纳 llmrouter 词边界思想，super_hard 先于 hard 的同理）： + 纯子串匹配会让 "int" 误命中 "print"/"point"、"log" 误命中 "logic"、 + "list" 误命中 "listen"——英文必须整词命中才计数。 + """ + substr, words = [], [] + for m in markers: + (words if m.isascii() else substr).append(m) + pattern = re.compile(r"\b(?:%s)\b" % "|".join(re.escape(w) for w in words)) \ + if words else None + return substr, pattern + + +_HARD_SUBSTR, _HARD_RE = _compile_markers(_HARD_MARKERS) +_MEDIUM_SUBSTR, _MEDIUM_RE = _compile_markers(_MEDIUM_MARKERS) + # 预编译正则（模块级一次，避免每次调用走 re 内部缓存查找） _RE_CODE_EXPR = re.compile(r"\b(def|class|function|import)\b") _RE_ARITH_EXPR = re.compile(r"[0-9]+\s*[+\-*/^=]\s*[0-9xya-z]") +def _hit_count(substr, pattern, q: str) -> int: + """中文子串命中数 + 英文整词命中数。""" + n = sum(1 for m in substr if m in q) + if pattern is not None: + n += len(pattern.findall(q)) + return n + + def estimate_difficulty(query: str) -> Tuple[str, float]: """返回 (difficulty, score)，score 属于 [0,1]。""" q = query.lower() - hard_hits = sum(1 for m in _HARD_MARKERS if m in q) - medium_hits = sum(1 for m in _MEDIUM_MARKERS if m in q) + hard_hits = _hit_count(_HARD_SUBSTR, _HARD_RE, q) + medium_hits = _hit_count(_MEDIUM_SUBSTR, _MEDIUM_RE, q) length = len(query) score = 0.0 diff --git a/tests/test_failsafe.py b/tests/test_failsafe.py new file mode 100644 index 0000000..278489f --- /dev/null +++ b/tests/test_failsafe.py @@ -0,0 +1,48 @@ +"""失败安全与词边界单元测试（T-R3，采纳 llmrouter 设计）。""" +from router_system.classifier import RuleClassifier, build_classifier +from router_system.difficulty import estimate_difficulty + + +def test_word_boundary_stops_substring_false_positive(): + """英文标记整词命中：'int' 不再被 'print'/'point' 误触发 hard。""" + diff, score = estimate_difficulty("如何用 print 函数打印结果") + # 旧实现 'int' 误命中 print 叠加 如何 -> hard；现在应为 medium 以下 + assert diff in ("easy", "medium") + + diff2, _ = estimate_difficulty("请总结这段逻辑代码的思路") + # 'sum' 不被 'summary/总结' 类场景误判：'逻辑' 含 'log' 也不再整词误命中 + assert diff2 in ("easy", "medium") + + +def test_word_boundary_still_counts_whole_words(): + """整词出现照常计数：int/log 作为真词仍触发 hard 信号。""" + diff, _ = estimate_difficulty("explain how to implement int overflow and log parsing in depth") + assert diff in ("medium", "hard") + + +def test_chinese_markers_keep_substring_semantics(): + """中文标记保持子串语义：'证明' 命中 '证明费马大定理'。""" + diff, _ = estimate_difficulty("证明费马大定理并推导其推论，给出详细步骤") + assert diff in ("medium", "hard") + + +def test_build_hf_falls_back_when_unavailable(monkeypatch): + """HF 分类器构造失败（依赖缺失/模型加载失败）安全回落规则分类器。""" + def _boom(self, *a, **k): + raise RuntimeError("HuggingFaceClassifier 需要安装 ML 依赖") + import router_system.classifier as C + monkeypatch.setattr(C.HuggingFaceClassifier, "__init__", _boom) + clf = build_classifier({"type": "hf", "model": "whatever"}) + assert isinstance(clf, RuleClassifier) + assert clf.classify("用 Python 写一个快速排序函数").domain == "code" + + +def test_hf_runtime_failure_falls_back_to_rules(): + """推理期异常（torch 缺失/tokenizer 损坏）回落内置规则分类器。""" + from router_system.classifier import HuggingFaceClassifier + hf = HuggingFaceClassifier.__new__(HuggingFaceClassifier) + hf.labels = ["code", "math", "legal", "medical", "general"] + hf.confidence_floor = 0.55 + hf._rule_fallback = RuleClassifier(confidence_floor=0.55) + r = hf.classify("用 Python 写一个快速排序函数") # 本环境无 torch/tokenizer -> 必走回落 + assert r.domain == "code"