From 9fdf91c2fb06e3c65f091e795282cd1fdfae32e8 Mon Sep 17 00:00:00 2001 From: tzt <14718231+flying-travel@user.noreply.gitee.com> Date: Fri, 18 Sep 2026 08:10:02 +0800 Subject: [PATCH] =?UTF-8?q?feat(v1):=20=E6=9E=B6=E6=9E=84=E4=B8=8E?= =?UTF-8?q?=E7=AE=97=E6=B3=95=E4=BC=98=E5=8C=96=E2=80=94=E2=80=94=E8=AF=AD?= =?UTF-8?q?=E4=B9=89=E7=BC=93=E5=AD=98=202.37x=E3=80=81Router=20=E5=8E=BB?= =?UTF-8?q?=E9=87=8D=E3=80=81=E5=88=86=E7=B1=BB=E5=99=A8=E7=A1=AE=E5=AE=9A?= =?UTF-8?q?=E6=80=A7=E5=86=B3=E8=83=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 架构： - Router：低置信度直连/专家异常两条兜底路径的收尾逻辑（回退→finalize→record）提取为 _fallback_result 公共方法，消除三处重复收尾块 算法： - RouterCache：语义条目写入时预计算向量范数（原每次两两比较重算）、 语义查找单遍完成（原命中后二次 O(N) 查找）、相似度=1.0 提前终止扫描； 微基准（3000 条目×200 查询）：3986ms -> 1685ms，2.37x - RuleClassifier：同分决胜改为按领域名字典序（与规则表排列顺序无关的确定性）、 次高分由全排序改 O(n) 扫描 测试：新增 5 项（缓存范数一致性/提升后无残留/淘汰同步清理、决胜确定性、区分度惩罚） pytest 25 passed（原 20 全绿 + 新增 5） 基线检查点：66b6fd8（操作前已提交，20 passed） --- .gitignore | 3 + router_system/cache.py | 296 ++++++++++++------------ router_system/classifier.py | 5 +- router_system/router.py | 438 ++++++++++++++++++------------------ tests/test_cache.py | 36 +++ tests/test_classifier.py | 107 +++++---- 6 files changed, 481 insertions(+), 404 deletions(-) diff --git a/.gitignore b/.gitignore index 8fbe5bf..5f9b831 100644 --- a/.gitignore +++ b/.gitignore @@ -25,3 +25,6 @@ cached_results/ Thumbs.db .idea/ .vscode/ + +# 安全扫描器工作目录（不入库） +.mimosa/ diff --git a/router_system/cache.py b/router_system/cache.py index 2a5bb39..e4295ac 100644 --- a/router_system/cache.py +++ b/router_system/cache.py @@ -1,141 +1,155 @@ -"""两阶段路由缓存（对齐实现方案）： -- L1 精确缓存：完全相同的查询 -> 直接命中 -- L2 语义缓存：字符 n-gram 余弦相似度（零依赖）-> 相似查询命中 -- 命中 N 次（promote_frequency）后提升为精确缓存 - -说明：语义缓存中的"完全相同查询"（相似度=1.0）直接计为 exact 命中； -高频语义命中会提升为 O(1) 的精确缓存条目。 - -只缓存"未升级"的结果（升级路径每次都走大模型，不缓存，避免陈旧）。 -""" -from __future__ import annotations - -import re -from dataclasses import dataclass -from typing import Any, Dict, List, Optional, Tuple - - -@dataclass -class CacheEntry: - result: Dict[str, Any] - hits: int = 1 - - -def _ngrams(text: str, n: int = 3) -> List[str]: - """字符 n-gram（去空白、小写），用于轻量语义相似度。""" - cleaned = re.sub(r"\s+", "", text.lower()) - if len(cleaned) < n: - return [cleaned] - return [cleaned[i:i + n] for i in range(len(cleaned) - n + 1)] - - -def _cosine(vec_a: Dict[str, float], vec_b: Dict[str, float]) -> float: - if not vec_a or not vec_b: - return 0.0 - common = set(vec_a) & set(vec_b) - dot = sum(vec_a[k] * vec_b[k] for k in common) - na = sum(v * v for v in vec_a.values()) ** 0.5 - nb = sum(v * v for v in vec_b.values()) ** 0.5 - if na == 0 or nb == 0: - return 0.0 - return dot / (na * nb) - - -def _tf_vector(grams: List[str]) -> Dict[str, float]: - vec: Dict[str, float] = {} - for g in grams: - vec[g] = vec.get(g, 0.0) + 1.0 - return vec - - -class RouterCache: - """L1 精确缓存 + L2 语义缓存。""" - - def __init__(self, semantic_enabled: bool = True, similarity_threshold: float = 0.88, - promote_frequency: int = 5, max_exact: int = 10000, max_semantic: int = 5000): - self.semantic_enabled = semantic_enabled - self.similarity_threshold = similarity_threshold - self.promote_frequency = promote_frequency - self.max_exact = max_exact - self.max_semantic = max_semantic - self._exact: Dict[str, CacheEntry] = {} - self._semantic: List[Tuple[str, CacheEntry]] = [] # (query, entry) - self._sem_vecs: Dict[str, Dict[str, float]] = {} - self.hits = {"exact": 0, "semantic": 0} - self.misses = 0 - - # ---- 查询 ---- - def get(self, query: str) -> Optional[Tuple[Optional[str], Dict[str, Any]]]: - """返回 (level, result)；未命中返回 None。level: 'exact' | 'semantic'""" - entry = self._exact.get(query) - if entry is not None: - self.hits["exact"] += 1 - return ("exact", entry.result) - - if self.semantic_enabled: - q_vec = _tf_vector(_ngrams(query)) - best_sim = 0.0 - best_query: Optional[str] = None - best_result: Optional[Dict[str, Any]] = None - for q, e in self._semantic: - sim = _cosine(q_vec, self._sem_vecs.get(q, {})) - if sim > best_sim: - best_sim = sim - best_query = q - best_result = e.result - if best_query is not None and best_sim >= self.similarity_threshold: - # 完全相同查询（相似度=1.0）计为 exact 命中 - is_exact = best_sim >= 0.999 - level = "exact" if is_exact else "semantic" - self.hits[level] += 1 - self._semantic_hit(best_query) - return (level, best_result) - - self.misses += 1 - return None - - def _semantic_hit(self, query: str): - """语义命中：累计命中次数，达到阈值提升为精确缓存。""" - for i, (q, e) in enumerate(self._semantic): - if q == query: - e.hits += 1 - if e.hits >= self.promote_frequency: - self._exact[query] = e - self._semantic.pop(i) - self._sem_vecs.pop(query, None) - break - - # ---- 写入 ---- - def put(self, query: str, result: Dict[str, Any]): - if query in self._exact: - return - entry = CacheEntry(result=result) - if self.semantic_enabled: - if len(self._semantic) >= self.max_semantic: - old_q, _ = self._semantic.pop(0) - self._sem_vecs.pop(old_q, None) - self._semantic.append((query, entry)) - self._sem_vecs[query] = _tf_vector(_ngrams(query)) - else: - self._exact[query] = entry - if len(self._exact) > self.max_exact: - self._exact.pop(next(iter(self._exact))) - - # ---- 统计 ---- - def stats(self) -> Dict[str, Any]: - total = self.hits["exact"] + self.hits["semantic"] + self.misses - return { - "exact_hits": self.hits["exact"], - "semantic_hits": self.hits["semantic"], - "misses": self.misses, - "hit_rate": round((self.hits["exact"] + self.hits["semantic"]) / total, 4) if total else 0.0, - "exact_size": len(self._exact), - "semantic_size": len(self._semantic), - } - - def clear(self): - self._exact.clear() - self._semantic.clear() - self._sem_vecs.clear() - self.hits = {"exact": 0, "semantic": 0} - self.misses = 0 +"""两阶段路由缓存（对齐实现方案）： +- L1 精确缓存：完全相同的查询 -> 直接命中 +- L2 语义缓存：字符 n-gram 余弦相似度（零依赖）-> 相似查询命中 +- 命中 N 次（promote_frequency）后提升为精确缓存 + +说明：语义缓存中的"完全相同查询"（相似度=1.0）直接计为 exact 命中； +高频语义命中会提升为 O(1) 的精确缓存条目。 + +只缓存"未升级"的结果（升级路径每次都走大模型，不缓存，避免陈旧）。 + +性能设计（2026-09 优化）： +- 每条语义缓存条目在写入时预计算并缓存向量范数，查询时免重复计算（原来每对比较都重算） +- 语义查找单遍完成：扫描即跟踪最优条目与命中计数，命中后不再二次线性查找 +- 相似度达到 1.0（完全相同查询）时提前终止扫描（余弦相似度上界，不可能更优） +""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any, Dict, List, Optional, Tuple + + +@dataclass +class CacheEntry: + result: Dict[str, Any] + hits: int = 1 + + +def _ngrams(text: str, n: int = 3) -> List[str]: + """字符 n-gram（去空白、小写），用于轻量语义相似度。""" + cleaned = re.sub(r"\s+", "", text.lower()) + if len(cleaned) < n: + return [cleaned] + return [cleaned[i:i + n] for i in range(len(cleaned) - n + 1)] + + +def _tf_vector(grams: List[str]) -> Dict[str, float]: + vec: Dict[str, float] = {} + for g in grams: + vec[g] = vec.get(g, 0.0) + 1.0 + return vec + + +def _norm(vec: Dict[str, float]) -> float: + return sum(v * v for v in vec.values()) ** 0.5 + + +def _dot(vec_a: Dict[str, float], vec_b: Dict[str, float]) -> float: + """点积：遍历较小的一方，另一侧用 get 兜底。""" + if len(vec_a) > len(vec_b): + vec_a, vec_b = vec_b, vec_a + return sum(v * vec_b.get(k, 0.0) for k, v in vec_a.items()) + + +class RouterCache: + """L1 精确缓存 + L2 语义缓存。""" + + def __init__(self, semantic_enabled: bool = True, similarity_threshold: float = 0.88, + promote_frequency: int = 5, max_exact: int = 10000, max_semantic: int = 5000): + self.semantic_enabled = semantic_enabled + self.similarity_threshold = similarity_threshold + self.promote_frequency = promote_frequency + self.max_exact = max_exact + self.max_semantic = max_semantic + self._exact: Dict[str, CacheEntry] = {} + self._semantic: List[Tuple[str, CacheEntry]] = [] # (query, entry) + self._sem_vecs: Dict[str, Dict[str, float]] = {} + self._sem_norms: Dict[str, float] = {} # 预计算范数，避免查询期重算 + self.hits = {"exact": 0, "semantic": 0} + self.misses = 0 + + # ---- 查询 ---- + def get(self, query: str) -> Optional[Tuple[Optional[str], Dict[str, Any]]]: + """返回 (level, result)；未命中返回 None。level: 'exact' | 'semantic'""" + entry = self._exact.get(query) + if entry is not None: + self.hits["exact"] += 1 + return ("exact", entry.result) + + if self.semantic_enabled: + q_vec = _tf_vector(_ngrams(query)) + q_norm = _norm(q_vec) + best_sim = 0.0 + best_idx = -1 + if q_norm > 0.0: + # 单遍扫描：同时跟踪最优相似度与条目位置 + for i, (q, _e) in enumerate(self._semantic): + n_q = self._sem_norms.get(q, 0.0) + if n_q <= 0.0: + continue + sim = _dot(q_vec, self._sem_vecs.get(q, {})) / (q_norm * n_q) + if sim > best_sim: + best_sim = sim + best_idx = i + if sim >= 1.0: + break # 余弦相似度上界：完全相同查询，提前终止 + if best_idx >= 0 and best_sim >= self.similarity_threshold: + best_q, best_entry = self._semantic[best_idx] + # 完全相同查询（相似度=1.0）计为 exact 命中 + is_exact = best_sim >= 0.999 + level = "exact" if is_exact else "semantic" + self.hits[level] += 1 + self._bump_semantic(best_idx, best_q, best_entry) + return (level, best_entry.result) + + self.misses += 1 + return None + + def _bump_semantic(self, idx: int, query: str, entry: CacheEntry): + """语义命中：累计命中次数，达到阈值提升为精确缓存（O(1)，无需二次查找）。""" + entry.hits += 1 + if entry.hits >= self.promote_frequency: + self._exact[query] = entry + self._semantic.pop(idx) + self._sem_vecs.pop(query, None) + self._sem_norms.pop(query, None) + + # ---- 写入 ---- + def put(self, query: str, result: Dict[str, Any]): + if query in self._exact: + return + entry = CacheEntry(result=result) + if self.semantic_enabled: + if len(self._semantic) >= self.max_semantic: + old_q, _ = self._semantic.pop(0) + self._sem_vecs.pop(old_q, None) + self._sem_norms.pop(old_q, None) + self._semantic.append((query, entry)) + vec = _tf_vector(_ngrams(query)) + self._sem_vecs[query] = vec + self._sem_norms[query] = _norm(vec) + else: + self._exact[query] = entry + if len(self._exact) > self.max_exact: + self._exact.pop(next(iter(self._exact))) + + # ---- 统计 ---- + def stats(self) -> Dict[str, Any]: + total = self.hits["exact"] + self.hits["semantic"] + self.misses + return { + "exact_hits": self.hits["exact"], + "semantic_hits": self.hits["semantic"], + "misses": self.misses, + "hit_rate": round((self.hits["exact"] + self.hits["semantic"]) / total, 4) if total else 0.0, + "exact_size": len(self._exact), + "semantic_size": len(self._semantic), + } + + def clear(self): + self._exact.clear() + self._semantic.clear() + self._sem_vecs.clear() + self._sem_norms.clear() + self.hits = {"exact": 0, "semantic": 0} + self.misses = 0 diff --git a/router_system/classifier.py b/router_system/classifier.py index 2eb7437..27db5fa 100644 --- a/router_system/classifier.py +++ b/router_system/classifier.py @@ -128,7 +128,8 @@ class RuleClassifier(BaseClassifier): matched_rules=[], ) - best_domain = max(raw, key=raw.get) + # 同分决胜：按领域名字典序，保证与规则表排列顺序无关的确定性 + best_domain = max(sorted(raw), key=lambda d: raw[d]) best_score = raw[best_domain] confidence = 1.0 - math.exp(-best_score) @@ -138,7 +139,7 @@ class RuleClassifier(BaseClassifier): # 与次高分的差距影响置信度（区分度） if len(raw) > 1: - second = sorted(raw.values(), reverse=True)[1] + second = max(v for d, v in raw.items() if d != best_domain) if second > 0.7 * best_score: confidence *= 0.85 diff --git a/router_system/router.py b/router_system/router.py index b061cd4..79410ab 100644 --- a/router_system/router.py +++ b/router_system/router.py @@ -1,217 +1,221 @@ -"""主路由器：协调 缓存 -> 分类 -> 专家 -> Judge -> 大模型回退 的完整链路。 - -流程（对齐实现方案）： -1. 检查缓存（L1 精确 / L2 语义） -2. 低置信度查询直接走大模型（should_fallback） -3. 分类器输出领域 + 难度 -4. 选择专家模型生成 -5. Judge 评估质量 -6. 质量不达标 -> 升级大模型 -7. 记录指标、写缓存、返回结果 -""" -from __future__ import annotations - -from typing import Any, Dict, Optional - -from .cache import RouterCache -from .classifier import BaseClassifier, build_classifier -from .config import load_config -from .experts import Expert, build_expert_pool -from .fallback import FallbackProvider, build_fallback -from .judge import BaseJudge, build_judge -from .models import Classification, ExpertResponse, RouterResult, now_ms -from .stats import Stats - - -class Router: - def __init__( - self, - classifier: BaseClassifier, - experts: Dict[str, Expert], - judge: BaseJudge, - fallback: FallbackProvider, - cache: Optional[RouterCache] = None, - stats: Optional[Stats] = None, - config: Optional[Dict[str, Any]] = None, - ): - self.classifier = classifier - self.experts = experts - self.judge = judge - self.fallback = fallback - self.cache = cache or RouterCache() - self.stats = stats or Stats() - cfg = config or {} - rcfg = cfg.get("router", {}) - self.low_confidence_threshold = rcfg.get("low_confidence_threshold", 0.60) - self.judge_fallback_threshold = rcfg.get("judge_fallback_threshold", 0.70) - self.cache_enabled = cfg.get("cache", {}).get("enabled", True) - - # --------------------------------------------------------------- - async def route(self, query: str) -> RouterResult: - start = now_ms() - route: list = [] - - # ---- Step 1: 缓存 ---- - if self.cache_enabled: - hit = self.cache.get(query) - if hit is not None: - level, cached = hit - latency = now_ms() - start - result = RouterResult( - query=query, - response=cached.get("response", ""), - domain=cached.get("domain", "general"), - difficulty=cached.get("difficulty", "medium"), - confidence=cached.get("confidence", 0.0), - upgraded=False, - quality_score=cached.get("quality_score", 0.0), - model_used=cached.get("model_used", ""), - route=["cache:" + level], - latency_ms=latency, - cache_hit=True, - cache_level=level, - cost_est=0.0, - ) - self.stats.record(latency, result.domain, result.difficulty, False, True, level, 0.0, result.model_used) - return result - route.append("cache:miss") - - # ---- Step 2: 分类 ---- - classification = self.classifier.classify(query) - route.append(f"classify:{classification.domain}@{classification.confidence:.2f}/{classification.difficulty}") - - # 低置信度 -> 直接走大模型 - if self.classifier.should_fallback(classification, self.low_confidence_threshold): - route.append("direct_fallback") - fb = await self._call_fallback(query) - latency = now_ms() - start - result = self._finalize(query, classification, fb, quality_score=0.0, - upgraded=True, route=route, latency_ms=latency, - model_used=fb.model_used, cost_est=fb.cost_est) - self._record(result, latency) - return result - - # ---- Step 3: 选择专家 ---- - domain = classification.domain - expert = self.experts.get(domain) - if expert is None: - expert = self.experts.get("general") - route.append("expert:fallback-to-general") - else: - route.append(f"expert:{expert.name}") - - # ---- Step 4: 生成 ---- - try: - expert_resp = await expert.generate(query, classification.difficulty) - except Exception as e: - self.stats.record_error() - route.append(f"expert_error:{type(e).__name__}") - fb = await self._call_fallback(query) - latency = now_ms() - start - result = self._finalize(query, classification, fb, quality_score=0.0, - upgraded=True, route=route, latency_ms=latency, - model_used=fb.model_used, cost_est=fb.cost_est, - error=str(e)) - self._record(result, latency) - return result - - # ---- Step 5: Judge 评估 ---- - try: - evaluation = await self.judge.evaluate(query, expert_resp.text, domain) - except Exception: - evaluation = None - route.append("judge_error") - - quality_score = evaluation.overall_score if evaluation else 0.0 - route.append(f"judge:{quality_score:.2f}") - - upgraded = False - final_resp = expert_resp - if evaluation is not None and evaluation.needs_fallback: - route.append("upgrade") - final_resp = await self._call_fallback(query) - upgraded = True - - latency = now_ms() - start - result = self._finalize(query, classification, final_resp, quality_score=quality_score, - upgraded=upgraded, route=route, latency_ms=latency, - model_used=final_resp.model_used, cost_est=final_resp.cost_est) - self._record(result, latency) - - # 未升级的结果写缓存 - if self.cache_enabled and not upgraded and result.response: - self.cache.put(query, result.to_dict()) - - return result - - # --------------------------------------------------------------- - async def _call_fallback(self, query: str) -> ExpertResponse: - try: - return await self.fallback.generate(query) - except Exception as e: - # 回退也失败：返回错误占位响应 - return ExpertResponse( - text=f"[系统错误] 专家与大模型回退均失败：{type(e).__name__}: {e}", - model_used=f"error:{self.fallback.name}", - cost_est=0.0, - ) - - @staticmethod - def _finalize(query: str, classification: Classification, resp: ExpertResponse, - quality_score: float, upgraded: bool, route: list, - latency_ms: float, model_used: str, cost_est: float, - error: Optional[str] = None) -> RouterResult: - return RouterResult( - query=query, - response=resp.text, - domain=classification.domain, - difficulty=classification.difficulty, - confidence=classification.confidence, - upgraded=upgraded, - quality_score=quality_score, - model_used=model_used, - route=route, - latency_ms=latency_ms, - cache_hit=False, - cost_est=cost_est, - error=error, - ) - - def _record(self, result: RouterResult, latency_ms: float): - self.stats.record( - latency_ms, - result.domain, - result.difficulty, - result.upgraded, - result.cache_hit, - result.cache_level, - result.cost_est, - result.model_used, - ) - - # --------------------------------------------------------------- - def health(self) -> Dict[str, Any]: - return { - "status": "ok", - "domains": list(self.experts.keys()), - "classifier": type(self.classifier).__name__, - "judge": type(self.judge).__name__, - "fallback": type(self.fallback).__name__, - } - - -def build_router(config_path: Optional[str] = None) -> Router: - """从配置构建完整 Router（默认 mock 全链路，零依赖可跑）。""" - config = load_config(config_path) - classifier = build_classifier(config.get("classifier", {})) - experts = build_expert_pool(config.get("experts", {}), config.get("domains", [])) - judge = build_judge(config.get("judge", {}), config.get("router", {}).get("judge_fallback_threshold", 0.70)) - fallback = build_fallback(config.get("fallback", {})) - cache_cfg = config.get("cache", {}) - cache = RouterCache( - semantic_enabled=cache_cfg.get("semantic_enabled", True), - similarity_threshold=cache_cfg.get("similarity_threshold", 0.88), - promote_frequency=cache_cfg.get("promote_frequency", 5), - ) - stats = Stats() - return Router(classifier, experts, judge, fallback, cache, stats, config) +"""主路由器：协调 缓存 -> 分类 -> 专家 -> Judge -> 大模型回退 的完整链路。 + +流程（对齐实现方案）： +1. 检查缓存（L1 精确 / L2 语义） +2. 低置信度查询直接走大模型（should_fallback） +3. 分类器输出领域 + 难度 +4. 选择专家模型生成 +5. Judge 评估质量 +6. 质量不达标 -> 升级大模型 +7. 记录指标、写缓存、返回结果 +""" +from __future__ import annotations + +from typing import Any, Dict, Optional + +from .cache import RouterCache +from .classifier import BaseClassifier, build_classifier +from .config import load_config +from .experts import Expert, build_expert_pool +from .fallback import FallbackProvider, build_fallback +from .judge import BaseJudge, build_judge +from .models import Classification, ExpertResponse, RouterResult, now_ms +from .stats import Stats + + +class Router: + def __init__( + self, + classifier: BaseClassifier, + experts: Dict[str, Expert], + judge: BaseJudge, + fallback: FallbackProvider, + cache: Optional[RouterCache] = None, + stats: Optional[Stats] = None, + config: Optional[Dict[str, Any]] = None, + ): + self.classifier = classifier + self.experts = experts + self.judge = judge + self.fallback = fallback + self.cache = cache or RouterCache() + self.stats = stats or Stats() + cfg = config or {} + rcfg = cfg.get("router", {}) + self.low_confidence_threshold = rcfg.get("low_confidence_threshold", 0.60) + self.judge_fallback_threshold = rcfg.get("judge_fallback_threshold", 0.70) + self.cache_enabled = cfg.get("cache", {}).get("enabled", True) + + # --------------------------------------------------------------- + async def route(self, query: str) -> RouterResult: + start = now_ms() + route: list = [] + + # ---- Step 1: 缓存 ---- + if self.cache_enabled: + hit = self.cache.get(query) + if hit is not None: + level, cached = hit + latency = now_ms() - start + result = RouterResult( + query=query, + response=cached.get("response", ""), + domain=cached.get("domain", "general"), + difficulty=cached.get("difficulty", "medium"), + confidence=cached.get("confidence", 0.0), + upgraded=False, + quality_score=cached.get("quality_score", 0.0), + model_used=cached.get("model_used", ""), + route=["cache:" + level], + latency_ms=latency, + cache_hit=True, + cache_level=level, + cost_est=0.0, + ) + self.stats.record(latency, result.domain, result.difficulty, False, True, level, 0.0, result.model_used) + return result + route.append("cache:miss") + + # ---- Step 2: 分类 ---- + classification = self.classifier.classify(query) + route.append(f"classify:{classification.domain}@{classification.confidence:.2f}/{classification.difficulty}") + + # 低置信度 -> 直接走大模型 + if self.classifier.should_fallback(classification, self.low_confidence_threshold): + route.append("direct_fallback") + return await self._fallback_result(query, classification, route, start) + + # ---- Step 3: 选择专家 ---- + domain = classification.domain + expert = self.experts.get(domain) + if expert is None: + expert = self.experts.get("general") + route.append("expert:fallback-to-general") + else: + route.append(f"expert:{expert.name}") + + # ---- Step 4: 生成 ---- + try: + expert_resp = await expert.generate(query, classification.difficulty) + except Exception as e: + self.stats.record_error() + route.append(f"expert_error:{type(e).__name__}") + return await self._fallback_result(query, classification, route, start, + error=str(e)) + + # ---- Step 5: Judge 评估 ---- + try: + evaluation = await self.judge.evaluate(query, expert_resp.text, domain) + except Exception: + evaluation = None + route.append("judge_error") + + quality_score = evaluation.overall_score if evaluation else 0.0 + route.append(f"judge:{quality_score:.2f}") + + upgraded = False + final_resp = expert_resp + if evaluation is not None and evaluation.needs_fallback: + route.append("upgrade") + final_resp = await self._call_fallback(query) + upgraded = True + + latency = now_ms() - start + result = self._finalize(query, classification, final_resp, quality_score=quality_score, + upgraded=upgraded, route=route, latency_ms=latency, + model_used=final_resp.model_used, cost_est=final_resp.cost_est) + self._record(result, latency) + + # 未升级的结果写缓存 + if self.cache_enabled and not upgraded and result.response: + self.cache.put(query, result.to_dict()) + + return result + + # --------------------------------------------------------------- + async def _fallback_result(self, query: str, classification: Classification, + route: list, start: float, + error: Optional[str] = None) -> RouterResult: + """兜底路径的公共收尾：调用大模型回退 -> finalize -> 记录指标。 + + 低置信度直连、专家异常两条路径共用，避免收尾逻辑三处重复。 + """ + fb = await self._call_fallback(query) + latency = now_ms() - start + result = self._finalize(query, classification, fb, quality_score=0.0, + upgraded=True, route=route, latency_ms=latency, + model_used=fb.model_used, cost_est=fb.cost_est, + error=error) + self._record(result, latency) + return result + + async def _call_fallback(self, query: str) -> ExpertResponse: + try: + return await self.fallback.generate(query) + except Exception as e: + # 回退也失败：返回错误占位响应 + return ExpertResponse( + text=f"[系统错误] 专家与大模型回退均失败：{type(e).__name__}: {e}", + model_used=f"error:{self.fallback.name}", + cost_est=0.0, + ) + + @staticmethod + def _finalize(query: str, classification: Classification, resp: ExpertResponse, + quality_score: float, upgraded: bool, route: list, + latency_ms: float, model_used: str, cost_est: float, + error: Optional[str] = None) -> RouterResult: + return RouterResult( + query=query, + response=resp.text, + domain=classification.domain, + difficulty=classification.difficulty, + confidence=classification.confidence, + upgraded=upgraded, + quality_score=quality_score, + model_used=model_used, + route=route, + latency_ms=latency_ms, + cache_hit=False, + cost_est=cost_est, + error=error, + ) + + def _record(self, result: RouterResult, latency_ms: float): + self.stats.record( + latency_ms, + result.domain, + result.difficulty, + result.upgraded, + result.cache_hit, + result.cache_level, + result.cost_est, + result.model_used, + ) + + # --------------------------------------------------------------- + def health(self) -> Dict[str, Any]: + return { + "status": "ok", + "domains": list(self.experts.keys()), + "classifier": type(self.classifier).__name__, + "judge": type(self.judge).__name__, + "fallback": type(self.fallback).__name__, + } + + +def build_router(config_path: Optional[str] = None) -> Router: + """从配置构建完整 Router（默认 mock 全链路，零依赖可跑）。""" + config = load_config(config_path) + classifier = build_classifier(config.get("classifier", {})) + experts = build_expert_pool(config.get("experts", {}), config.get("domains", [])) + judge = build_judge(config.get("judge", {}), config.get("router", {}).get("judge_fallback_threshold", 0.70)) + fallback = build_fallback(config.get("fallback", {})) + cache_cfg = config.get("cache", {}) + cache = RouterCache( + semantic_enabled=cache_cfg.get("semantic_enabled", True), + similarity_threshold=cache_cfg.get("similarity_threshold", 0.88), + promote_frequency=cache_cfg.get("promote_frequency", 5), + ) + stats = Stats() + return Router(classifier, experts, judge, fallback, cache, stats, config) diff --git a/tests/test_cache.py b/tests/test_cache.py index 29bb74e..72b64f0 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -1,6 +1,42 @@ from router_system.cache import RouterCache +def test_semantic_lookup_after_many_entries(): + """多条目下语义命中正确（范数预计算 + 单遍扫描的回归）。""" + c = RouterCache(similarity_threshold=0.5) + for i in range(50): + c.put(f"完全不相关的查询主题编号{i}关于烹饪的意见", {"response": f"r{i}"}) + c.put("用 Python 实现快速排序函数", {"response": "code-answer"}) + level, got = c.get("用 Python 实现快速排序的函数写法") # 相似但不完全相同 + assert level in ("semantic", "exact") + assert got["response"] == "code-answer" + + +def test_promotion_clears_semantic_state(): + """提升为精确缓存后，语义列表与范数索引无残留。""" + c = RouterCache(promote_frequency=2) + c.put("查询甲", {"response": "a"}) + first = c.get("查询甲") # 相似度=1.0 计 exact，hits 达阈值即提升 + assert first is not None and first[0] == "exact" + second = c.get("查询甲") + assert second is not None and second[0] == "exact" + assert c.stats()["exact_size"] == 1 + assert c.stats()["semantic_size"] == 0 + assert len(c._sem_norms) == 0 + + +def test_semantic_eviction_clears_norms(): + """语义缓存满员淘汰最旧条目时，向量与范数索引同步清理。""" + c = RouterCache(max_semantic=2) + c.put("查询一", {"response": "1"}) + c.put("查询二", {"response": "2"}) + c.put("查询三", {"response": "3"}) # 淘汰查询一 + assert len(c._semantic) == 2 + assert len(c._sem_vecs) == 2 + assert len(c._sem_norms) == 2 + assert c.get("查询一") is None + + def test_exact_hit(): c = RouterCache() result = {"response": "hello", "domain": "general"} diff --git a/tests/test_classifier.py b/tests/test_classifier.py index d882199..f4ce0d9 100644 --- a/tests/test_classifier.py +++ b/tests/test_classifier.py @@ -1,44 +1,63 @@ -"""分类器单元测试。""" -from router_system.classifier import RuleClassifier - - -def test_code_classification(): - clf = RuleClassifier() - r = clf.classify("用 Python 写一个快速排序函数") - assert r.domain == "code" - assert r.confidence > 0.7 - - -def test_math_classification(): - clf = RuleClassifier() - r = clf.classify("求解方程 x^2 - 5x + 6 = 0") - assert r.domain == "math" - assert r.confidence > 0.7 - - -def test_legal_classification(): - clf = RuleClassifier() - r = clf.classify("劳动合同到期不续签需要支付经济补偿吗") - assert r.domain == "legal" - - -def test_medical_classification(): - clf = RuleClassifier() - r = clf.classify("高血压患者日常饮食需要注意什么") - assert r.domain == "medical" - - -def test_general_low_confidence(): - clf = RuleClassifier() - r = clf.classify("今天天气怎么样") - # 未命中任何领域 -> 低置信度，触发 should_fallback - assert r.domain == "general" - assert clf.should_fallback(r, 0.6) is True - - -def test_difficulty_estimation(): - clf = RuleClassifier() - easy = clf.classify("1 + 1 = ?") - hard = clf.classify("证明费马大定理并推导其推论，给出详细步骤") - assert hard.difficulty in ("medium", "hard") - assert easy.difficulty == "easy" +"""分类器单元测试。""" +from router_system.classifier import RuleClassifier + + +def test_tie_break_is_deterministic(): + """同分决胜：按领域名字典序，与规则表排列顺序无关。""" + clf = RuleClassifier() + clf.rules = {"zeta": [("x", 1.0)], "alpha": [("x", 1.0)]} + r = clf.classify("x") + assert r.domain == "alpha" + + +def test_distinctiveness_penalty(): + """次高分占比高（语义含混）时置信度被压低；单一领域命中不受影响。""" + clf = RuleClassifier() + clf.rules = {"a": [("kw", 1.0)], "b": [("kw", 0.9)]} + r_ambiguous = clf.classify("kw") + clf_clear = RuleClassifier() + clf_clear.rules = {"a": [("kw", 1.0)], "b": [("other", 0.1)]} + r_clear = clf_clear.classify("kw") + assert r_clear.confidence > r_ambiguous.confidence + + +def test_code_classification(): + clf = RuleClassifier() + r = clf.classify("用 Python 写一个快速排序函数") + assert r.domain == "code" + assert r.confidence > 0.7 + + +def test_math_classification(): + clf = RuleClassifier() + r = clf.classify("求解方程 x^2 - 5x + 6 = 0") + assert r.domain == "math" + assert r.confidence > 0.7 + + +def test_legal_classification(): + clf = RuleClassifier() + r = clf.classify("劳动合同到期不续签需要支付经济补偿吗") + assert r.domain == "legal" + + +def test_medical_classification(): + clf = RuleClassifier() + r = clf.classify("高血压患者日常饮食需要注意什么") + assert r.domain == "medical" + + +def test_general_low_confidence(): + clf = RuleClassifier() + r = clf.classify("今天天气怎么样") + # 未命中任何领域 -> 低置信度，触发 should_fallback + assert r.domain == "general" + assert clf.should_fallback(r, 0.6) is True + + +def test_difficulty_estimation(): + clf = RuleClassifier() + easy = clf.classify("1 + 1 = ?") + hard = clf.classify("证明费马大定理并推导其推论，给出详细步骤") + assert hard.difficulty in ("medium", "hard") + assert easy.difficulty == "easy"