feat(v1): T-R3 采纳 llmrouter 词边界与失败安全——难度英文标记整词命中 + HF 分类器两级回落
- difficulty:标记表编译拆分——英文标记改词边界正则(修真实误判:'int' 子串 命中 'print'/'point'、'log' 命中 'logic'、'list' 命中 'listen'),中文标记 保持子串语义;命中行为对合法用例不变(整词出现照常计数) - classifier:HuggingFaceClassifier 推理期异常回落内置 RuleClassifier(单次 推理异常不打垮路由);build_classifier 的 hf 分支构造失败(ML 依赖缺失/ 模型加载失败)打印提示并回落规则分类器(外置规则照常合并) - 新增 tests/test_failsafe.py 5 项;全量 43 passed(38+5)
This commit is contained in:
@@ -214,6 +214,8 @@ class HuggingFaceClassifier(BaseClassifier):
|
|||||||
"""可选:基于 transformers 的序列分类模型。
|
"""可选:基于 transformers 的序列分类模型。
|
||||||
|
|
||||||
仅当安装 torch+transformers 且模型可加载时可用;否则抛错提示。
|
仅当安装 torch+transformers 且模型可加载时可用;否则抛错提示。
|
||||||
|
失败安全(T-R3,采纳 llmrouter 分类失败静默降级思想):模型加载成功但
|
||||||
|
推理期异常时,自动回落内置规则分类器,不让单次推理异常打垮路由。
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, model_name: str, num_labels: int = 5, confidence_floor: float = 0.55):
|
def __init__(self, model_name: str, num_labels: int = 5, confidence_floor: float = 0.55):
|
||||||
@@ -229,8 +231,10 @@ class HuggingFaceClassifier(BaseClassifier):
|
|||||||
)
|
)
|
||||||
self.labels = ["code", "math", "legal", "medical", "general"]
|
self.labels = ["code", "math", "legal", "medical", "general"]
|
||||||
self.confidence_floor = confidence_floor
|
self.confidence_floor = confidence_floor
|
||||||
|
self._rule_fallback = RuleClassifier(confidence_floor=confidence_floor)
|
||||||
|
|
||||||
def classify(self, query: str) -> Classification:
|
def classify(self, query: str) -> Classification:
|
||||||
|
try:
|
||||||
import torch # type: ignore
|
import torch # type: ignore
|
||||||
|
|
||||||
inputs = self.tokenizer(query, return_tensors="pt", truncation=True, max_length=256)
|
inputs = self.tokenizer(query, return_tensors="pt", truncation=True, max_length=256)
|
||||||
@@ -246,6 +250,8 @@ class HuggingFaceClassifier(BaseClassifier):
|
|||||||
difficulty_score=ds,
|
difficulty_score=ds,
|
||||||
raw_scores={self.labels[i]: round(float(probs[i]), 3) for i in range(len(self.labels))},
|
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:
|
def build_classifier(cfg: Dict) -> BaseClassifier:
|
||||||
@@ -265,6 +271,18 @@ def build_classifier(cfg: Dict) -> BaseClassifier:
|
|||||||
clf.rules = {**DOMAIN_RULES, **external}
|
clf.rules = {**DOMAIN_RULES, **external}
|
||||||
return clf
|
return clf
|
||||||
if ctype == "hf":
|
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)")
|
raise ValueError(f"未知分类器类型: {ctype}(支持 rule | hf)")
|
||||||
|
|
||||||
|
|||||||
@@ -25,16 +25,43 @@ _MEDIUM_MARKERS = [
|
|||||||
"translate", "fix", "explain", "describe", "list",
|
"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 内部缓存查找)
|
||||||
_RE_CODE_EXPR = re.compile(r"\b(def|class|function|import)\b")
|
_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]")
|
_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]:
|
def estimate_difficulty(query: str) -> Tuple[str, float]:
|
||||||
"""返回 (difficulty, score),score 属于 [0,1]。"""
|
"""返回 (difficulty, score),score 属于 [0,1]。"""
|
||||||
q = query.lower()
|
q = query.lower()
|
||||||
hard_hits = sum(1 for m in _HARD_MARKERS if m in q)
|
hard_hits = _hit_count(_HARD_SUBSTR, _HARD_RE, q)
|
||||||
medium_hits = sum(1 for m in _MEDIUM_MARKERS if m in q)
|
medium_hits = _hit_count(_MEDIUM_SUBSTR, _MEDIUM_RE, q)
|
||||||
length = len(query)
|
length = len(query)
|
||||||
|
|
||||||
score = 0.0
|
score = 0.0
|
||||||
|
|||||||
@@ -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"
|
||||||
Reference in New Issue
Block a user