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）

.gitignore

@@ -25,3 +25,9 @@ cached_results/
 Thumbs.db
 .idea/
 .vscode/
+
+# 安全扫描器工作目录（不入库）
+.mimosa/
+
+# 参考项目目录（不入库，仅作设计吸收来源）
+extra/

BRANCH.md

@@ -0,0 +1,29 @@
+# 分支：v1-model-routing — 本地多智能体协作模型路由（第一代）
+
+> **快照点**：`1e51167`（v1 MVP 完成时点，仓库首个提交）。
+> 此分支为**历史路标**，冻结不再演进；集成主线见 `master`。
+
+## 这一代是什么
+
+- **核心命题**：用"规则知识库 + 任务拆解 Planner + 专业执行器池 + 质量控制器(Judge) + 最后处理者"
+  在限定条件下替代单一通用大模型——**本地多智能体协作路由**
+- 架构：8 领域规则知识库 → 规则分类器（置信度 1−e^−s）→ Planner 任务拆解（DAG）
+  → 黑板/前向链 → 规则执行器（L0 零参数、确定性模板）→ RuleJudge 五维评分 → mock 兜底
+- 两级路由：`domain_groups`（tech/professional/lifestyle/general）→ 组内 RuleClassifier；三级子领域识别
+- 接口：FastAPI `/chat` `/health` `/metrics`；核心包 `router_system/` 零第三方依赖
+
+## 如何运行
+
+```powershell
+.venv\Scripts\python.exe scripts/demo.py --trace      # L0 演示（离线可跑）
+.venv\Scripts\python.exe scripts/eval.py              # 迷你评估
+.venv\Scripts\python.exe -m pytest tests -q           # 测试（初版 20 项）
+.venv\Scripts\python.exe scripts/serve.py --port 8000 # 网关
+```
+
+## 定位与结论
+
+- **方向结论**：规则路由在 9 域平衡集实测 74.4% 准确率、68.9% 升级率（`research/routerarena/`）——
+  假阳性过高，**被 v2 端云协作风取代**；本代代码在主线保留为 legacy（`POST /chat/legacy`）
+- 设计文档：`实现方案_多专业小模型+路由模型.md`、`可行性调研与落地实现路线报告.md`、
+  `research/2026_papers_survey.md`、`research/routerarena/01_results_and_gap_analysis.md`

gateway/api.py

@@ -1,94 +1,95 @@
-"""FastAPI 网关：对外提供 /chat /health /metrics 接口。
+"""FastAPI 网关：对外提供 /chat /health /metrics 接口。
-
+
-启动：
+启动：
-    uvicorn gateway.api:app --host 0.0.0.0 --port 8000
+    uvicorn gateway.api:app --host 0.0.0.0 --port 8000
-或：
+或：
-    python -m gateway.api
+    python -m gateway.api
-
+
-依赖：fastapi, uvicorn, pydantic（见 requirements.txt）
+依赖：fastapi, uvicorn, pydantic（见 requirements.txt）
-"""
+"""
-from __future__ import annotations
+from __future__ import annotations
-
+
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
-
+
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field
-
+
-from router_system.config import load_config
+from router_system.config import load_config
-from router_system.router import Router, build_router
+from router_system.router import Router, build_router
-
+
-# ---- 全局单例 ----
+# ---- 全局单例 ----
-_router: Optional[Router] = None
+_router: Optional[Router] = None
-
+
-
+
-def get_router() -> Router:
+def get_router() -> Router:
-    global _router
+    global _router
-    if _router is None:
+    if _router is None:
-        _router = build_router()
+        _router = build_router()
-    return _router
+    return _router
-
+
-
+
-# ---- 请求/响应模型 ----
+# ---- 请求/响应模型 ----
-class QueryRequest(BaseModel):
+class QueryRequest(BaseModel):
-    query: str = Field(..., min_length=1, max_length=8000, description="用户查询")
+    query: str = Field(..., min_length=1, max_length=8000, description="用户查询")
-
+
-
+
-class QueryResponse(BaseModel):
+class QueryResponse(BaseModel):
-    response: str
+    response: str
-    domain: str
+    domain: str
-    difficulty: str
+    difficulty: str
-    confidence: float
+    confidence: float
-    upgraded: bool
+    upgraded: bool
-    quality_score: float
+    quality_score: float
-    model_used: str
+    model_used: str
-    route: List[str]
+    route: List[str]
-    latency_ms: float
+    latency_ms: float
-    cache_hit: bool
+    cache_hit: bool
-    cache_level: Optional[str]
+    cache_level: Optional[str]
-    cost_est: float
+    cost_est: float
-    error: Optional[str] = None
+    error: Optional[str] = None
-
+    route_explanation: Optional[Dict[str, Any]] = None  # T-R1 可解释评分（缓存命中为 None）
-
+
-class HealthResponse(BaseModel):
+
-    status: str
+class HealthResponse(BaseModel):
-    domains: List[str]
+    status: str
-    classifier: str
+    domains: List[str]
-    judge: str
+    classifier: str
-    fallback: str
+    judge: str
-
+    fallback: str
-
+
-# ---- FastAPI 应用 ----
+
-try:
+# ---- FastAPI 应用 ----
-    from fastapi import FastAPI
+try:
-
+    from fastapi import FastAPI
-    app = FastAPI(
+
-        title="Multi-Expert Router API",
+    app = FastAPI(
-        description="多专业小模型 + 路由模型系统（MVP）",
+        title="Multi-Expert Router API",
-        version="0.1.0",
+        description="多专业小模型 + 路由模型系统（MVP）",
-    )
+        version="0.1.0",
-
+    )
-    @app.get("/health", response_model=HealthResponse, tags=["system"])
+
-    async def health():
+    @app.get("/health", response_model=HealthResponse, tags=["system"])
-        return get_router().health()
+    async def health():
-
+        return get_router().health()
-    @app.post("/chat", response_model=QueryResponse, tags=["chat"])
+
-    async def chat(req: QueryRequest):
+    @app.post("/chat", response_model=QueryResponse, tags=["chat"])
-        result = await get_router().route(req.query)
+    async def chat(req: QueryRequest):
-        return QueryResponse(**result.to_dict())
+        result = await get_router().route(req.query)
-
+        return QueryResponse(**result.to_dict())
-    @app.get("/metrics", tags=["system"])
+
-    async def metrics():
+    @app.get("/metrics", tags=["system"])
-        r = get_router()
+    async def metrics():
-        return {
+        r = get_router()
-            "router": r.stats.summary(),
+        return {
-            "cache": r.cache.stats(),
+            "router": r.stats.summary(),
-        }
+            "cache": r.cache.stats(),
-
+        }
-except ImportError:
+
-    # fastapi 未安装时，提供 CLI 入口提示
+except ImportError:
-    app = None
+    # fastapi 未安装时，提供 CLI 入口提示
-    print("[gateway] 未安装 fastapi，请执行: pip install -r requirements.txt")
+    app = None
-
+    print("[gateway] 未安装 fastapi，请执行: pip install -r requirements.txt")
-
+
-if __name__ == "__main__":
+
-    import uvicorn
+if __name__ == "__main__":
-    uvicorn.run("gateway.api:app", host="0.0.0.0", port=8000, reload=False)␍
+    import uvicorn
+    uvicorn.run("gateway.api:app", host="0.0.0.0", port=8000, reload=False)

router_system/cache.py

@@ -1,141 +1,172 @@
-"""两阶段路由缓存（对齐实现方案）：
+"""两阶段路由缓存（对齐实现方案）：
-- L1 精确缓存：完全相同的查询 -> 直接命中
+- L1 精确缓存：完全相同的查询 -> 直接命中
-- L2 语义缓存：字符 n-gram 余弦相似度（零依赖）-> 相似查询命中
+- L2 语义缓存：字符 n-gram 余弦相似度（零依赖）-> 相似查询命中
-- 命中 N 次（promote_frequency）后提升为精确缓存
+- 命中 N 次（promote_frequency）后提升为精确缓存
-
+
-说明：语义缓存中的"完全相同查询"（相似度=1.0）直接计为 exact 命中；
+说明：语义缓存中的"完全相同查询"（相似度=1.0）直接计为 exact 命中；
-高频语义命中会提升为 O(1) 的精确缓存条目。
+高频语义命中会提升为 O(1) 的精确缓存条目。
-
+
-只缓存"未升级"的结果（升级路径每次都走大模型，不缓存，避免陈旧）。
+只缓存"未升级"的结果（升级路径每次都走大模型，不缓存，避免陈旧）。
-"""
+
-from __future__ import annotations
+性能设计（2026-09 优化）：
-
+- 每条语义缓存条目在写入时预计算并缓存向量范数，查询时免重复计算（原来每对比较都重算）
-import re
+- 语义查找单遍完成：扫描即跟踪最优条目与命中计数，命中后不再二次线性查找
-from dataclasses import dataclass
+- 相似度达到 1.0（完全相同查询）时提前终止扫描（余弦相似度上界，不可能更优）
-from typing import Any, Dict, List, Optional, Tuple
+- 语义条目容器为 OrderedDict：提升/淘汰均为 O(1)（原 list pop 是 O(n) 搬移），
-
+  且按 query 天然去重（原实现同一 query 可重复追加，重复条目白白参与扫描）
-
+- n-gram 向量经 lru_cache 复用：同一次未命中请求里 get 与紧随的 put
-@dataclass
+  不再重复做两遍分词+计数
-class CacheEntry:
+"""
-    result: Dict[str, Any]
+from __future__ import annotations
-    hits: int = 1
+
-
+import re
-
+from collections import OrderedDict
-def _ngrams(text: str, n: int = 3) -> List[str]:
+from dataclasses import dataclass
-    """字符 n-gram（去空白、小写），用于轻量语义相似度。"""
+from functools import lru_cache
-    cleaned = re.sub(r"\s+", "", text.lower())
+from typing import Any, Dict, List, Optional, Tuple
-    if len(cleaned) < n:
+
-        return [cleaned]
+
-    return [cleaned[i:i + n] for i in range(len(cleaned) - n + 1)]
+@dataclass
-
+class CacheEntry:
-
+    result: Dict[str, Any]
-def _cosine(vec_a: Dict[str, float], vec_b: Dict[str, float]) -> float:
+    hits: int = 1
-    if not vec_a or not vec_b:
+
-        return 0.0
+
-    common = set(vec_a) & set(vec_b)
+def _ngrams(text: str, n: int = 3) -> List[str]:
-    dot = sum(vec_a[k] * vec_b[k] for k in common)
+    """字符 n-gram（去空白、小写），用于轻量语义相似度。"""
-    na = sum(v * v for v in vec_a.values()) ** 0.5
+    cleaned = re.sub(r"\s+", "", text.lower())
-    nb = sum(v * v for v in vec_b.values()) ** 0.5
+    if len(cleaned) < n:
-    if na == 0 or nb == 0:
+        return [cleaned]
-        return 0.0
+    return [cleaned[i:i + n] for i in range(len(cleaned) - n + 1)]
-    return dot / (na * nb)
+
-
+
-
+def _tf_vector(grams: List[str]) -> Dict[str, float]:
-def _tf_vector(grams: List[str]) -> Dict[str, float]:
+    vec: Dict[str, float] = {}
-    vec: Dict[str, float] = {}
+    for g in grams:
-    for g in grams:
+        vec[g] = vec.get(g, 0.0) + 1.0
-        vec[g] = vec.get(g, 0.0) + 1.0
+    return vec
-    return vec
+
-
+
-
+def _norm(vec: Dict[str, float]) -> float:
-class RouterCache:
+    return sum(v * v for v in vec.values()) ** 0.5
-    """L1 精确缓存 + L2 语义缓存。"""
+
-
+
-    def __init__(self, semantic_enabled: bool = True, similarity_threshold: float = 0.88,
+def _dot(vec_a: Dict[str, float], vec_b: Dict[str, float]) -> float:
-                 promote_frequency: int = 5, max_exact: int = 10000, max_semantic: int = 5000):
+    """点积：遍历较小的一方，另一侧用 get 兜底。"""
-        self.semantic_enabled = semantic_enabled
+    if len(vec_a) > len(vec_b):
-        self.similarity_threshold = similarity_threshold
+        vec_a, vec_b = vec_b, vec_a
-        self.promote_frequency = promote_frequency
+    return sum(v * vec_b.get(k, 0.0) for k, v in vec_a.items())
-        self.max_exact = max_exact
+
-        self.max_semantic = max_semantic
+
-        self._exact: Dict[str, CacheEntry] = {}
+@lru_cache(maxsize=1024)
-        self._semantic: List[Tuple[str, CacheEntry]] = []  # (query, entry)
+def _embed(text: str) -> Tuple[Dict[str, float], float]:
-        self._sem_vecs: Dict[str, Dict[str, float]] = {}
+    """n-gram 向量 + 范数（带缓存）。
-        self.hits = {"exact": 0, "semantic": 0}
+
-        self.misses = 0
+    纯函数且同一次未命中请求会在 get/put 各用一次，缓存可省一遍分词。
-
+    契约：返回的 dict 仅供只读（内部无任何修改点），调用方不得改写。
-    # ---- 查询 ----
+    """
-    def get(self, query: str) -> Optional[Tuple[Optional[str], Dict[str, Any]]]:
+    vec = _tf_vector(_ngrams(text))
-        """返回 (level, result)；未命中返回 None。level: 'exact' | 'semantic'"""
+    return vec, _norm(vec)
-        entry = self._exact.get(query)
+
-        if entry is not None:
+
-            self.hits["exact"] += 1
+class RouterCache:
-            return ("exact", entry.result)
+    """L1 精确缓存 + L2 语义缓存。"""
-
+
-        if self.semantic_enabled:
+    def __init__(self, semantic_enabled: bool = True, similarity_threshold: float = 0.88,
-            q_vec = _tf_vector(_ngrams(query))
+                 promote_frequency: int = 5, max_exact: int = 10000, max_semantic: int = 5000):
-            best_sim = 0.0
+        self.semantic_enabled = semantic_enabled
-            best_query: Optional[str] = None
+        self.similarity_threshold = similarity_threshold
-            best_result: Optional[Dict[str, Any]] = None
+        self.promote_frequency = promote_frequency
-            for q, e in self._semantic:
+        self.max_exact = max_exact
-                sim = _cosine(q_vec, self._sem_vecs.get(q, {}))
+        self.max_semantic = max_semantic
-                if sim > best_sim:
+        self._exact: Dict[str, CacheEntry] = {}
-                    best_sim = sim
+        # (query -> entry)，插入序即淘汰序；len 兼容原 list 断言
-                    best_query = q
+        self._semantic: "OrderedDict[str, CacheEntry]" = OrderedDict()
-                    best_result = e.result
+        self._sem_vecs: Dict[str, Dict[str, float]] = {}
-            if best_query is not None and best_sim >= self.similarity_threshold:
+        self._sem_norms: Dict[str, float] = {}             # 预计算范数，避免查询期重算
-                # 完全相同查询（相似度=1.0）计为 exact 命中
+        self.hits = {"exact": 0, "semantic": 0}
-                is_exact = best_sim >= 0.999
+        self.misses = 0
-                level = "exact" if is_exact else "semantic"
+
-                self.hits[level] += 1
+    # ---- 查询 ----
-                self._semantic_hit(best_query)
+    def get(self, query: str) -> Optional[Tuple[Optional[str], Dict[str, Any]]]:
-                return (level, best_result)
+        """返回 (level, result)；未命中返回 None。level: 'exact' | 'semantic'"""
-
+        entry = self._exact.get(query)
-        self.misses += 1
+        if entry is not None:
-        return None
+            self.hits["exact"] += 1
-
+            return ("exact", entry.result)
-    def _semantic_hit(self, query: str):
+
-        """语义命中：累计命中次数，达到阈值提升为精确缓存。"""
+        if self.semantic_enabled:
-        for i, (q, e) in enumerate(self._semantic):
+            q_vec, q_norm = _embed(query)
-            if q == query:
+            best_sim = 0.0
-                e.hits += 1
+            best_q = ""
-                if e.hits >= self.promote_frequency:
+            if q_norm > 0.0:
-                    self._exact[query] = e
+                # 单遍扫描（插入序）：同时跟踪最优条目，sim=1.0 提前终止
-                    self._semantic.pop(i)
+                for q, _e in self._semantic.items():
-                    self._sem_vecs.pop(query, None)
+                    n_q = self._sem_norms.get(q, 0.0)
-                break
+                    if n_q <= 0.0:
-
+                        continue
-    # ---- 写入 ----
+                    sim = _dot(q_vec, self._sem_vecs.get(q, {})) / (q_norm * n_q)
-    def put(self, query: str, result: Dict[str, Any]):
+                    if sim > best_sim:
-        if query in self._exact:
+                        best_sim = sim
-            return
+                        best_q = q
-        entry = CacheEntry(result=result)
+                        if sim >= 1.0:
-        if self.semantic_enabled:
+                            break  # 余弦相似度上界：完全相同查询，提前终止
-            if len(self._semantic) >= self.max_semantic:
+            if best_q and best_sim >= self.similarity_threshold:
-                old_q, _ = self._semantic.pop(0)
+                best_entry = self._semantic[best_q]
-                self._sem_vecs.pop(old_q, None)
+                # 完全相同查询（相似度=1.0）计为 exact 命中
-            self._semantic.append((query, entry))
+                is_exact = best_sim >= 0.999
-            self._sem_vecs[query] = _tf_vector(_ngrams(query))
+                level = "exact" if is_exact else "semantic"
-        else:
+                self.hits[level] += 1
-            self._exact[query] = entry
+                self._bump_semantic(best_q, best_entry)
-            if len(self._exact) > self.max_exact:
+                return (level, best_entry.result)
-                self._exact.pop(next(iter(self._exact)))
+
-
+        self.misses += 1
-    # ---- 统计 ----
+        return None
-    def stats(self) -> Dict[str, Any]:
+
-        total = self.hits["exact"] + self.hits["semantic"] + self.misses
+    def _bump_semantic(self, query: str, entry: CacheEntry):
-        return {
+        """语义命中：累计命中次数，达到阈值提升为精确缓存（OrderedDict 删除 O(1)）。"""
-            "exact_hits": self.hits["exact"],
+        entry.hits += 1
-            "semantic_hits": self.hits["semantic"],
+        if entry.hits >= self.promote_frequency:
-            "misses": self.misses,
+            self._exact[query] = entry
-            "hit_rate": round((self.hits["exact"] + self.hits["semantic"]) / total, 4) if total else 0.0,
+            del self._semantic[query]
-            "exact_size": len(self._exact),
+            self._sem_vecs.pop(query, None)
-            "semantic_size": len(self._semantic),
+            self._sem_norms.pop(query, None)
-        }
+
-
+    # ---- 写入 ----
-    def clear(self):
+    def put(self, query: str, result: Dict[str, Any]):
-        self._exact.clear()
+        if query in self._exact or query in self._semantic:
-        self._semantic.clear()
+            return  # 已缓存（语义区按 query 天然去重，避免重复条目参与扫描）
-        self._sem_vecs.clear()
+        entry = CacheEntry(result=result)
-        self.hits = {"exact": 0, "semantic": 0}
+        if self.semantic_enabled:
-        self.misses = 0␍
+            if len(self._semantic) >= self.max_semantic:
+                old_q, _old = self._semantic.popitem(last=False)  # O(1) 淘汰最旧
+                self._sem_vecs.pop(old_q, None)
+                self._sem_norms.pop(old_q, None)
+            self._semantic[query] = entry
+            vec, norm = _embed(query)  # 与 get miss 共享同一份缓存向量
+            self._sem_vecs[query] = vec
+            self._sem_norms[query] = norm
+        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

router_system/classifier.py

@@ -6,11 +6,18 @@
 置信度设计：每个领域有一组 (关键词, 权重)。命中权重求和得原始分 s，
 confidence = 1 - exp(-s)，保证 s=1 -> 0.63，s=2 -> 0.86，s=3 -> 0.95。
 无领域命中（或最高分领域为 general）时置信度低，触发 should_fallback。
+
+规则外置（T-R2，采纳 llmrouter「规则文档即配置」设计）：领域规则支持从
+config/routes.json（或 .yaml，pyyaml 可选）加载，文件按 domain 覆盖内置
+DOMAIN_RULES（与 v2 知识库"文件按 id 覆盖"同一惯例）；文件缺失/格式非法
+整体安全回退内置（llmrouter 失败安全思想）。
 """
 from __future__ import annotations
 
+import json
 import math
-from typing import Dict, List, Tuple
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
 
 from .difficulty import estimate_difficulty
 from .models import Classification
@@ -33,7 +40,7 @@ DOMAIN_RULES: Dict[str, List[Tuple[str, float]]] = {
         ("algorithm", 0.8), ("sort", 0.7), ("array", 0.7), ("regex", 0.8),
         ("import", 0.7), ("loop", 0.7), ("recursion", 0.8), ("refactor", 0.8),
         ("deploy", 0.8), ("docker", 0.8), ("kubernetes", 0.8),
-        ("async", 0.7), ("flask", 0.7), ("django", 0.7), ("api", 0.7),
+        ("async", 0.7), ("flask", 0.7), ("django", 0.7),
         ("索引", 0.8), ("优化", 0.7), ("查询", 0.6),
     ],
     "math": [
@@ -82,6 +89,50 @@ _STOPWORDS = {
     "or", "do", "does", "can", "could", "would", "should", "please", "me", "my",
 }
 
+# 外置规则默认路径（约定优于配置：文件存在即自动加载，T-R2）
+DEFAULT_RULES_FILE = Path(__file__).resolve().parent.parent / "config" / "routes.json"
+
+
+def load_domain_rules(path: Optional[str | Path] = None
+                      ) -> Optional[Dict[str, List[Tuple[str, float]]]]:
+    """加载外置分类规则文件，返回 {domain: [(关键词, 权重), ...]}。
+
+    文件缺失 / 格式非法 / 条目非法时返回 None（调用方安全回退内置规则，
+    不抛异常——规则文档可被人工编辑，编辑错误不应打垮路由）。
+    支持 .json（必有）与 .yaml（pyyaml 可选依赖）。
+    """
+    p = Path(path) if path else DEFAULT_RULES_FILE
+    if not p.exists():
+        return None
+    try:
+        if p.suffix.lower() in (".yaml", ".yml"):
+            try:
+                import yaml  # type: ignore
+            except ImportError:
+                return None
+            with open(p, "r", encoding="utf-8") as f:
+                data = yaml.safe_load(f)
+        else:
+            with open(p, "r", encoding="utf-8") as f:
+                data = json.load(f)
+    except (OSError, ValueError, Exception):  # noqa: BLE001  解析失败一律回退
+        return None
+    if not isinstance(data, dict):
+        return None
+    rules: Dict[str, List[Tuple[str, float]]] = {}
+    for domain, entries in data.items():
+        if not isinstance(domain, str) or not isinstance(entries, list):
+            return None
+        pairs: List[Tuple[str, float]] = []
+        for entry in entries:
+            if (not isinstance(entry, (list, tuple)) or len(entry) != 2
+                    or not isinstance(entry[0], str)
+                    or not isinstance(entry[1], (int, float))):
+                return None
+            pairs.append((entry[0], float(entry[1])))
+        rules[domain] = pairs
+    return rules
+
 
 class BaseClassifier:
     def classify(self, query: str) -> Classification:
@@ -98,24 +149,29 @@ class RuleClassifier(BaseClassifier):
         self.confidence_floor = confidence_floor
         self.rules = DOMAIN_RULES
 
-    def _score(self, query: str) -> Tuple[Dict[str, float], Dict[str, List[str]]]:
+    def _score(self, q: str) -> Dict[str, float]:
-        q = query.lower()
+        """对已 lowercase 的查询按领域规则打分，返回有命中的领域分数。
+
+        命中词列表只对最终胜出领域有意义（classify 的唯一消费点），
+        故不在各领域上白建 list——胜出后由 _matched_rules 单独收集。
+        """
         scores: Dict[str, float] = {}
-        matched: Dict[str, List[str]] = {}
         for domain, rules in self.rules.items():
             s = 0.0
-            hits = []
             for kw, w in rules:
                 if kw in q:
                     s += w
-                    hits.append(kw)
             if s > 0:
                 scores[domain] = s
-                matched[domain] = hits
+        return scores
-        return scores, matched
+
+    def _matched_rules(self, q: str, domain: str) -> List[str]:
+        """收集指定领域命中的关键词（保持登记顺序）。"""
+        return [kw for kw, _w in self.rules.get(domain, ()) if kw in q]
 
     def classify(self, query: str) -> Classification:
-        raw, matched = self._score(query)
+        q = query.lower()
+        raw = self._score(q)
         if not raw:
             # 完全无命中 -> general，低置信度
             diff, ds = estimate_difficulty(query)
@@ -128,7 +184,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 +195,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
 
@@ -149,7 +206,7 @@ class RuleClassifier(BaseClassifier):
             difficulty=diff,
             difficulty_score=ds,
             raw_scores={k: round(v, 3) for k, v in raw.items()},
-            matched_rules=matched.get(best_domain, []),
+            matched_rules=self._matched_rules(q, best_domain),
         )
 
 
@@ -157,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):
@@ -172,32 +231,58 @@ 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)
+            inputs = self.tokenizer(query, return_tensors="pt", truncation=True, max_length=256)
-        with torch.no_grad():
+            with torch.no_grad():
-            logits = self.model(**inputs).logits
+                logits = self.model(**inputs).logits
-        probs = torch.softmax(logits, dim=-1)[0]
+            probs = torch.softmax(logits, dim=-1)[0]
-        idx = int(probs.argmax())
+            idx = int(probs.argmax())
-        diff, ds = estimate_difficulty(query)
+            diff, ds = estimate_difficulty(query)
-        return Classification(
+            return Classification(
-            domain=self.labels[idx],
+                domain=self.labels[idx],
-            confidence=round(float(probs[idx]), 4),
+                confidence=round(float(probs[idx]), 4),
-            difficulty=diff,
+                difficulty=diff,
-            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:
-    """根据配置构建分类器。cfg 为 classifier 段配置。"""
+    """根据配置构建分类器。cfg 为 classifier 段配置。
+
+    T-R2：cfg.rules_file 指定外置规则文件；未指定时若约定路径
+    config/routes.json 存在则自动加载。文件按 domain 覆盖内置规则。
+    """
     ctype = cfg.get("type", "rule")
     floor = cfg.get("confidence_floor", 0.55)
     if ctype == "rule":
-        return RuleClassifier(confidence_floor=floor)
+        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
     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）")
 

router_system/difficulty.py

@@ -1,52 +1,83 @@
-"""查询难度估计器（启发式，纯标准库）。
+"""查询难度估计器（启发式，纯标准库）。
-
+
-依据：RouterArena 用 Bloom 分类法把问题分为 easy/medium/hard。
+依据：RouterArena 用 Bloom 分类法把问题分为 easy/medium/hard。
-这里用查询长度、指令动词、数学/推理标记做轻量估计。
+这里用查询长度、指令动词、数学/推理标记做轻量估计。
-"""
+"""
-from __future__ import annotations
+from __future__ import annotations
-
+
-import re
+import re
-from typing import Tuple
+from typing import Tuple
-
+
-# 触发 hard 的指令动词 / 推理标记
+# 触发 hard 的指令动词 / 推理标记
-_HARD_MARKERS = [
+_HARD_MARKERS = [
-    "证明", "推导", "为什么", "如何", "对比", "比较", "分析", "评估", "设计", "优化",
+    "证明", "推导", "为什么", "如何", "对比", "比较", "分析", "评估", "设计", "优化",
-    "复杂度", "时间复杂度", "空间复杂度", "原理", "机制", "优缺点", "区别", "推论", "定理",
+    "复杂度", "时间复杂度", "空间复杂度", "原理", "机制", "优缺点", "区别", "推论", "定理",
-    "proof", "prove", "derive", "explain why", "why", "how", "compare", "contrast",
+    "proof", "prove", "derive", "explain why", "why", "how", "compare", "contrast",
-    "analy", "evaluate", "design", "optimize", "refactor", "architect",
+    "analy", "evaluate", "design", "optimize", "refactor", "architect",
-    "implement", "debug", "review", "plan", "synthesize",
+    "implement", "debug", "review", "plan", "synthesize",
-    "int", "sum", "sqrt", "lim", "log", "derivative", "integral",
+    "int", "sum", "sqrt", "lim", "log", "derivative", "integral",
-]
+]
-# 触发 medium 的标记
+# 触发 medium 的标记
-_MEDIUM_MARKERS = [
+_MEDIUM_MARKERS = [
-    "用", "写", "计算", "求解", "生成", "翻译", "总结", "解释",
+    "用", "写", "计算", "求解", "生成", "翻译", "总结", "解释",
-    "注意", "建议", "是否", "实现", "步骤",
+    "注意", "建议", "是否", "实现", "步骤",
-    "write", "code", "function", "script", "calculate", "solve", "summarize",
+    "write", "code", "function", "script", "calculate", "solve", "summarize",
-    "translate", "fix", "explain", "describe", "list",
+    "translate", "fix", "explain", "describe", "list",
-]
+]
-
+
-
+
-def estimate_difficulty(query: str) -> Tuple[str, float]:
+def _compile_markers(markers):
-    """返回 (difficulty, score)，score 属于 [0,1]。"""
+    """标记表编译：英文标记用词边界正则，中文标记保持子串匹配。
-    q = query.lower()
+
-    hard_hits = sum(1 for m in _HARD_MARKERS if m in q)
+    T-R3（采纳 llmrouter 词边界思想，super_hard 先于 hard 的同理）：
-    medium_hits = sum(1 for m in _MEDIUM_MARKERS if m in q)
+    纯子串匹配会让 "int" 误命中 "print"/"point"、"log" 误命中 "logic"、
-    length = len(query)
+    "list" 误命中 "listen"——英文必须整词命中才计数。
-
+    """
-    score = 0.0
+    substr, words = [], []
-    score += min(0.30, length / 600.0)          # 长度贡献
+    for m in markers:
-    score += min(0.55, hard_hits * 0.25)        # 推理标记贡献
+        (words if m.isascii() else substr).append(m)
-    score += min(0.30, medium_hits * 0.08)      # 一般指令贡献
+    pattern = re.compile(r"\b(?:%s)\b" % "|".join(re.escape(w) for w in words)) \
-
+        if words else None
-    # 额外：代码 / 数学表达式（多步骤信号）
+    return substr, pattern
-    if "```" in query or re.search(r"\b(def|class|function|import)\b", q):
+
-        score += 0.15
+
-    if re.search(r"[0-9]+\s*[+\-*/^=]\s*[0-9xya-z]", q):
+_HARD_SUBSTR, _HARD_RE = _compile_markers(_HARD_MARKERS)
-        score += 0.15
+_MEDIUM_SUBSTR, _MEDIUM_RE = _compile_markers(_MEDIUM_MARKERS)
-
+
-    score = max(0.0, min(1.0, score))
+# 预编译正则（模块级一次，避免每次调用走 re 内部缓存查找）
-    if score >= 0.55:
+_RE_CODE_EXPR = re.compile(r"\b(def|class|function|import)\b")
-        return "hard", score
+_RE_ARITH_EXPR = re.compile(r"[0-9]+\s*[+\-*/^=]\s*[0-9xya-z]")
-    if score >= 0.25:
+
-        return "medium", score
+
-    return "easy", score␍
+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 = _hit_count(_HARD_SUBSTR, _HARD_RE, q)
+    medium_hits = _hit_count(_MEDIUM_SUBSTR, _MEDIUM_RE, q)
+    length = len(query)
+
+    score = 0.0
+    score += min(0.30, length / 600.0)          # 长度贡献
+    score += min(0.55, hard_hits * 0.25)        # 推理标记贡献
+    score += min(0.30, medium_hits * 0.08)      # 一般指令贡献
+
+    # 额外：代码 / 数学表达式（多步骤信号）
+    if "```" in query or _RE_CODE_EXPR.search(q):
+        score += 0.15
+    if _RE_ARITH_EXPR.search(q):
+        score += 0.15
+
+    score = max(0.0, min(1.0, score))
+    if score >= 0.55:
+        return "hard", score
+    if score >= 0.25:
+        return "medium", score
+    return "easy", score

router_system/experts.py

@@ -1,270 +1,268 @@
-"""专家模型池：统一 Expert 接口，支持三种后端。
+"""专家模型池：统一 Expert 接口，支持三种后端。
-
+
-- MockExpert   ：确定性模板输出（零依赖，离线可跑，便于测试与演示）
+- MockExpert   ：确定性模板输出（零依赖，离线可跑，便于测试与演示）
-- HFExpert     ：HuggingFace transformers 真实小模型（可选，需 ML 依赖）
+- HFExpert     ：HuggingFace transformers 真实小模型（可选，需 ML 依赖）
-- APIExpert    ：OpenAI 兼容 API（可选，需 API Key，如 DeepSeek）
+- APIExpert    ：OpenAI 兼容 API（可选，需 API Key，如 DeepSeek）
-
+
-成本估计：cost_est 按参数量粗估（美元/百万 token 的近似比例）。
+成本估计：cost_est 按参数量粗估（美元/百万 token 的近似比例）。
-"""
+"""
-from __future__ import annotations
+from __future__ import annotations
-
+
-import asyncio
+import asyncio
-import re
+import re
-from typing import Dict, List, Optional
+from functools import lru_cache
-
+from typing import Dict, List, Optional, Tuple
-from .models import ExpertResponse
+
-
+from .config import get_api_key
-# 按模型规模粗估的相对成本（$ / 1M output tokens，近似）
+from .llm_client import OpenAICompatClient
-MODEL_COST_EST = {
+from .models import ExpertResponse
-    "mock": 0.0,
+
-    "0.5b": 0.02,
+# 按模型规模粗估的相对成本（$ / 1M output tokens，近似）
-    "1b": 0.05,
+MODEL_COST_EST = {
-    "1.7b": 0.08,
+    "mock": 0.0,
-    "3b": 0.12,
+    "0.5b": 0.02,
-    "4b": 0.15,
+    "1b": 0.05,
-    "7b": 0.25,
+    "1.7b": 0.08,
-    "70b": 2.50,
+    "3b": 0.12,
-    "api": 1.00,
+    "4b": 0.15,
-}
+    "7b": 0.25,
-
+    "70b": 2.50,
-
+    "api": 1.00,
-def _cost_for(model_name: str, default: str = "1b") -> float:
+}
-    mn = model_name.lower()
+
-    for key in ("0.5b", "1.7b", "3b", "4b", "7b", "70b"):
+
-        if key in mn:
+def _cost_for(model_name: str, default: str = "1b") -> float:
-            return MODEL_COST_EST[key]
+    mn = model_name.lower()
-    if "api" in mn or mn in ("deepseek-chat", "gpt-4o-mini", "claude"):
+    for key in ("0.5b", "1.7b", "3b", "4b", "7b", "70b"):
-        return MODEL_COST_EST["api"]
+        if key in mn:
-    return MODEL_COST_EST.get(default, 0.1)
+            return MODEL_COST_EST[key]
-
+    if "api" in mn or mn in ("deepseek-chat", "gpt-4o-mini", "claude"):
-
+        return MODEL_COST_EST["api"]
-def extract_content_terms(query: str) -> List[str]:
+    return MODEL_COST_EST.get(default, 0.1)
-    """抽取查询中的"内容词"（中文词/英文单词），用于 Judge 覆盖度与 Mock 回显。"""
+
-    q = query.lower()
+
-    terms: List[str] = []
+@lru_cache(maxsize=2048)
-    # 英文单词（>=2 字符）
+def _extract_content_terms_cached(query: str) -> Tuple[str, ...]:
-    for w in re.findall(r"[a-z][a-z0-9_]{1,}", q):
+    q = query.lower()
-        if w not in _STOPWORDS_EN and w not in terms:
+    terms: List[str] = []
-            terms.append(w)
+    # 英文单词（>=2 字符）
-    # 中文：按 2-4 字窗口切分，保留含中文字符的片段
+    for w in re.findall(r"[a-z][a-z0-9_]{1,}", q):
-    cn = re.findall(r"[\u4e00-\u9fff]{2,8}", q)
+        if w not in _STOPWORDS_EN and w not in terms:
-    for c in cn:
+            terms.append(w)
-        terms.append(c)
+    # 中文：按 2-4 字窗口切分，保留含中文字符的片段
-    return terms
+    cn = re.findall(r"[\u4e00-\u9fff]{2,8}", q)
-
+    for c in cn:
-
+        terms.append(c)
-_STOPWORDS_EN = {
+    return tuple(terms)
-    "the", "a", "an", "is", "are", "to", "of", "in", "on", "for", "with", "and",
+
-    "or", "do", "does", "can", "could", "would", "should", "please", "me", "my",
+
-    "this", "that", "it", "be", "was", "were", "have", "has", "had", "will",
+def extract_content_terms(query: str) -> Tuple[str, ...]:
-    "not", "no", "yes", "i", "you", "he", "she", "we", "they",
+    """抽取查询中的"内容词"（中文词/英文单词），用于 Judge 覆盖度与 Mock 回显。
-}
+
-
+    纯函数，但 Router 主链路中专家生成与 Judge 评估会对同一查询各调一次，
-
+    故以 lru_cache 去重；返回不可变 tuple（调用方只读）。
-class Expert:
+    """
-    name: str = "expert"
+    return _extract_content_terms_cached(query)
-
+
-    async def generate(self, query: str, difficulty: str) -> ExpertResponse:
+
-        raise NotImplementedError
+_STOPWORDS_EN = {
-
+    "the", "a", "an", "is", "are", "to", "of", "in", "on", "for", "with", "and",
-
+    "or", "do", "does", "can", "could", "would", "should", "please", "me", "my",
-class MockExpert(Expert):
+    "this", "that", "it", "be", "was", "were", "have", "has", "had", "will",
-    """确定性模板专家：零依赖，离线可跑。
+    "not", "no", "yes", "i", "you", "he", "she", "we", "they",
-
+}
-    输出会回显查询中的内容词以提高 Judge 覆盖度，并带领域结构，
+
-    使端到端管线（分类 -> 专家 -> Judge -> 缓存）可被稳定测试与演示。
+
-    """
+class Expert:
-
+    name: str = "expert"
-    def __init__(self, name: str, domain: str, model: str = "mock"):
+    nominal_latency_ms: float = 100.0   # 标称延迟元数据（可解释评分 speed 维输入）
-        self.name = name
+
-        self.domain = domain
+    async def generate(self, query: str, difficulty: str) -> ExpertResponse:
-        self.model = model
+        raise NotImplementedError
-
+
-    async def generate(self, query: str, difficulty: str) -> ExpertResponse:
+
-        await asyncio.sleep(0.001)  # 模拟极短推理延迟
+class MockExpert(Expert):
-        terms = extract_content_terms(query)
+    """确定性模板专家：零依赖，离线可跑。
-        body = self._template(query, terms, difficulty)
+
-        # 预估 token 数：中文约 1.5 字符/token，英文约 4 字符/token
+    输出会回显查询中的内容词以提高 Judge 覆盖度，并带领域结构，
-        tokens = max(8, int(len(body) / 2.2))
+    使端到端管线（分类 -> 专家 -> Judge -> 缓存）可被稳定测试与演示。
-        return ExpertResponse(
+    """
-            text=body,
+
-            model_used=self.model,
+    nominal_latency_ms = 1.0
-            latency_ms=1.0,
+
-            tokens=tokens,
+    def __init__(self, name: str, domain: str, model: str = "mock"):
-            cost_est=_cost_for(self.model) * tokens / 1_000_000,
+        self.name = name
-        )
+        self.domain = domain
-
+        self.model = model
-    def _template(self, query: str, terms: List[str], difficulty: str) -> str:
+
-        kw = "、".join(terms[:6]) if terms else "该主题"
+    async def generate(self, query: str, difficulty: str) -> ExpertResponse:
-        if self.domain == "code":
+        await asyncio.sleep(0.001)  # 模拟极短推理延迟
-            return (
+        terms = extract_content_terms(query)
-                f"（mock 代码专家）针对「{query}」的实现思路如下：\n\n"
+        body = self._template(query, terms, difficulty)
-                f"```python\n"
+        # 预估 token 数：中文约 1.5 字符/token，英文约 4 字符/token
-                f"def solve() -> None:\n"
+        tokens = max(8, int(len(body) / 2.2))
-                f"    # 关键点：{kw}\n"
+        return ExpertResponse(
-                f"    # 1. 明确输入输出约束\n"
+            text=body,
-                f"    # 2. 选择合适数据结构\n"
+            model_used=self.model,
-                f"    # 3. 处理边界条件（空输入、极端值）\n"
+            latency_ms=1.0,
-                f"    # 4. 补充单元测试\n"
+            tokens=tokens,
-                f"    pass\n"
+            cost_est=_cost_for(self.model) * tokens / 1_000_000,
-                f"```\n\n"
+        )
-                f"复杂度：平均 O(n)。请按上述步骤补充具体实现。"
+
-            )
+    def _template(self, query: str, terms: List[str], difficulty: str) -> str:
-        if self.domain == "math":
+        kw = "、".join(terms[:6]) if terms else "该主题"
-            return (
+        if self.domain == "code":
-                f"（mock 数学专家）求解「{query}」的步骤：\n\n"
+            return (
-                f"1. 明确已知条件与目标：{kw}\n"
+                f"（mock 代码专家）针对「{query}」的实现思路如下：\n\n"
-                f"2. 选择合适的方法（代数变形 / 积分 / 归纳等）\n"
+                f"```python\n"
-                f"3. 逐步推导并验证中间结果\n"
+                f"def solve() -> None:\n"
-                f"4. 检查边界与特殊情况\n\n"
+                f"    # 关键点：{kw}\n"
-                f"结论：在标准假设下，结果可化简为闭合形式。完整推导见正式解答。"
+                f"    # 1. 明确输入输出约束\n"
-            )
+                f"    # 2. 选择合适数据结构\n"
-        if self.domain == "legal":
+                f"    # 3. 处理边界条件（空输入、极端值）\n"
-            return (
+                f"    # 4. 补充单元测试\n"
-                f"（mock 法律专家）关于「{query}」的初步法律分析：\n\n"
+                f"    pass\n"
-                f"相关要点：{kw}\n"
+                f"```\n\n"
-                f"1. 适用法规：请以现行有效法条为准（建议核对最新修订版）\n"
+                f"复杂度：平均 O(n)。请按上述步骤补充具体实现。"
-                f"2. 合同/合规风险点识别\n"
+            )
-                f"3. 责任划分与救济途径\n\n"
+        if self.domain == "math":
-                f"⚠️ 提示：以上为一般性分析，不构成正式法律意见，个案请咨询执业律师。"
+            return (
-            )
+                f"（mock 数学专家）求解「{query}」的步骤：\n\n"
-        if self.domain == "medical":
+                f"1. 明确已知条件与目标：{kw}\n"
-            return (
+                f"2. 选择合适的方法（代数变形 / 积分 / 归纳等）\n"
-                f"（mock 医学专家）关于「{query}」的科普性说明：\n\n"
+                f"3. 逐步推导并验证中间结果\n"
-                f"相关关键词：{kw}\n"
+                f"4. 检查边界与特殊情况\n\n"
-                f"1. 常见表现与可能原因\n"
+                f"结论：在标准假设下，结果可化简为闭合形式。完整推导见正式解答。"
-                f"2. 一般处理建议与注意事项\n"
+            )
-                f"3. 何时需要就医（警示信号）\n\n"
+        if self.domain == "legal":
-                f"⚠️ 提示：内容仅供健康科普，不能替代医生诊断；如有不适请及时就医。"
+            return (
-            )
+                f"（mock 法律专家）关于「{query}」的初步法律分析：\n\n"
-        return (
+                f"相关要点：{kw}\n"
-            f"（mock 通用专家）关于「{query}」的回答：\n\n"
+                f"1. 适用法规：请以现行有效法条为准（建议核对最新修订版）\n"
-            f"核心要点：{kw}\n"
+                f"2. 合同/合规风险点识别\n"
-            f"1. 背景与定义\n"
+                f"3. 责任划分与救济途径\n\n"
-            f"2. 主要分类/维度\n"
+                f"⚠️ 提示：以上为一般性分析，不构成正式法律意见，个案请咨询执业律师。"
-            f"3. 实际应用与注意事项\n\n"
+            )
-            f"如需更深入的分析，可以补充更多上下文。"
+        if self.domain == "medical":
-        )
+            return (
-
+                f"（mock 医学专家）关于「{query}」的科普性说明：\n\n"
-
+                f"相关关键词：{kw}\n"
-class HFExpert(Expert):
+                f"1. 常见表现与可能原因\n"
-    """可选：HuggingFace 真实小模型（需 requirements-ml.txt）。"""
+                f"2. 一般处理建议与注意事项\n"
-
+                f"3. 何时需要就医（警示信号）\n\n"
-    def __init__(self, name: str, domain: str, model: str):
+                f"⚠️ 提示：内容仅供健康科普，不能替代医生诊断；如有不适请及时就医。"
-        self.name = name
+            )
-        self.domain = domain
+        return (
-        self.model = model
+            f"（mock 通用专家）关于「{query}」的回答：\n\n"
-        self._loaded = False
+            f"核心要点：{kw}\n"
-        self._model = None
+            f"1. 背景与定义\n"
-        self._tokenizer = None
+            f"2. 主要分类/维度\n"
-
+            f"3. 实际应用与注意事项\n\n"
-    def _ensure_loaded(self):
+            f"如需更深入的分析，可以补充更多上下文。"
-        if self._loaded:
+        )
-            return
+
-        try:
+
-            from transformers import AutoModelForCausalLM, AutoTokenizer
+class HFExpert(Expert):
-        except ImportError as e:
+    """可选：HuggingFace 真实小模型（需 requirements-ml.txt）。"""
-            raise RuntimeError("HFExpert 需要安装 ML 依赖：pip install -r requirements-ml.txt") from e
+
-        self._tokenizer = AutoTokenizer.from_pretrained(self.model)
+    nominal_latency_ms = 300.0
-        self._model = AutoModelForCausalLM.from_pretrained(
+
-            self.model, device_map="auto", torch_dtype="auto"
+    def __init__(self, name: str, domain: str, model: str):
-        )
+        self.name = name
-        self._loaded = True
+        self.domain = domain
-
+        self.model = model
-    async def generate(self, query: str, difficulty: str) -> ExpertResponse:
+        self._loaded = False
-        self._ensure_loaded()
+        self._model = None
-        return await asyncio.to_thread(self._generate_sync, query)
+        self._tokenizer = None
-
+
-    def _generate_sync(self, query: str) -> ExpertResponse:
+    def _ensure_loaded(self):
-        messages = [{"role": "user", "content": query}]
+        if self._loaded:
-        text = self._tokenizer.apply_chat_template(messages, tokenize=False)
+            return
-        inputs = self._tokenizer(text, return_tensors="pt").to(self._model.device)
+        try:
-        outputs = self._model.generate(
+            from transformers import AutoModelForCausalLM, AutoTokenizer
-            **inputs,
+        except ImportError as e:
-            max_new_tokens=512,
+            raise RuntimeError("HFExpert 需要安装 ML 依赖：pip install -r requirements-ml.txt") from e
-            temperature=0.2,
+        self._tokenizer = AutoTokenizer.from_pretrained(self.model)
-            do_sample=True,
+        self._model = AutoModelForCausalLM.from_pretrained(
-        )
+            self.model, device_map="auto", torch_dtype="auto"
-        body = self._tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
+        )
-        return ExpertResponse(
+        self._loaded = True
-            text=body,
+
-            model_used=self.model,
+    async def generate(self, query: str, difficulty: str) -> ExpertResponse:
-            latency_ms=0.0,
+        self._ensure_loaded()
-            tokens=512,
+        return await asyncio.to_thread(self._generate_sync, query)
-            cost_est=_cost_for(self.model) * 512 / 1_000_000,
+
-        )
+    def _generate_sync(self, query: str) -> ExpertResponse:
-
+        messages = [{"role": "user", "content": query}]
-
+        text = self._tokenizer.apply_chat_template(messages, tokenize=False)
-class APIExpert(Expert):
+        inputs = self._tokenizer(text, return_tensors="pt").to(self._model.device)
-    """可选：OpenAI 兼容 Chat Completions（DeepSeek / OpenAI / 本地 vLLM）。"""
+        outputs = self._model.generate(
-
+            **inputs,
-    def __init__(self, name: str, domain: str, model: str, base_url: str, api_key: str):
+            max_new_tokens=512,
-        self.name = name
+            temperature=0.2,
-        self.domain = domain
+            do_sample=True,
-        self.model = model
+        )
-        self.base_url = base_url.rstrip("/")
+        body = self._tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
-        self.api_key = api_key
+        return ExpertResponse(
-        self._client = None
+            text=body,
-
+            model_used=self.model,
-    def _get_client(self):
+            latency_ms=0.0,
-        if self._client is None:
+            tokens=512,
-            import httpx
+            cost_est=_cost_for(self.model) * 512 / 1_000_000,
-            self._client = httpx.AsyncClient(timeout=60.0)
+        )
-        return self._client
+
-
+
-    async def generate(self, query: str, difficulty: str) -> ExpertResponse:
+class APIExpert(Expert):
-        client = self._get_client()
+    """可选：OpenAI 兼容 Chat Completions（DeepSeek / OpenAI / 本地 vLLM）。"""
-        resp = await client.post(
+
-            f"{self.base_url}/chat/completions",
+    nominal_latency_ms = 800.0
-            headers={"Authorization": f"Bearer {self.api_key}"},
+
-            json={
+    def __init__(self, name: str, domain: str, model: str, base_url: str, api_key: str):
-                "model": self.model,
+        self.name = name
-                "messages": [{"role": "user", "content": query}],
+        self.domain = domain
-                "temperature": 0.2,
+        self.model = model
-                "max_tokens": 1024,
+        self._client = OpenAICompatClient(base_url, api_key, timeout=60.0)
-            },
+
-        )
+    async def generate(self, query: str, difficulty: str) -> ExpertResponse:
-        resp.raise_for_status()
+        data = await self._client.chat(
-        data = resp.json()
+            self.model,
-        body = data["choices"][0]["message"]["content"]
+            [{"role": "user", "content": query}],
-        usage = data.get("usage", {})
+            temperature=0.2,
-        tokens = usage.get("completion_tokens", int(len(body) / 2.2))
+            max_tokens=1024,
-        return ExpertResponse(
+        )
-            text=body,
+        body = self._client.completion_text(data)
-            model_used=self.model,
+        tokens = self._client.completion_tokens(data, body)
-            latency_ms=0.0,
+        return ExpertResponse(
-            tokens=tokens,
+            text=body,
-            cost_est=_cost_for(self.model) * tokens / 1_000_000,
+            model_used=self.model,
-        )
+            latency_ms=0.0,
-
+            tokens=tokens,
-
+            cost_est=_cost_for(self.model) * tokens / 1_000_000,
-def build_expert(domain: str, cfg: Dict) -> Expert:
+        )
-    """根据配置构建领域专家。cfg 为 experts.<domain> 段配置。"""
+
-    etype = cfg.get("type", "mock")
+
-    model = cfg.get("model", "mock")
+def build_expert(domain: str, cfg: Dict) -> Expert:
-    name = f"expert-{domain}"
+    """根据配置构建领域专家。cfg 为 experts.<domain> 段配置。"""
-    if etype == "mock":
+    etype = cfg.get("type", "mock")
-        return MockExpert(name, domain, model)
+    model = cfg.get("model", "mock")
-    if etype == "hf":
+    name = f"expert-{domain}"
-        return HFExpert(name, domain, model)
+    if etype == "mock":
-    if etype == "api":
+        return MockExpert(name, domain, model)
-        base_url = cfg.get("base_url", "https://api.deepseek.com/v1")
+    if etype == "hf":
-        api_key = cfg.get("api_key") or _env(cfg.get("api_key_env", ""))
+        return HFExpert(name, domain, model)
-        if not api_key:
+    if etype == "api":
-            raise RuntimeError(f"APIExpert({domain}) 缺少 API Key（env: {cfg.get('api_key_env')}）")
+        base_url = cfg.get("base_url", "https://api.deepseek.com/v1")
-        return APIExpert(name, domain, model, base_url, api_key)
+        api_key = cfg.get("api_key") or get_api_key(cfg)
-    raise ValueError(f"未知专家后端类型: {etype}（支持 mock | hf | api）")
+        if not api_key:
-
+            raise RuntimeError(f"APIExpert({domain}) 缺少 API Key（env: {cfg.get('api_key_env')}）")
-
+        return APIExpert(name, domain, model, base_url, api_key)
-def _env(name: str) -> Optional[str]:
+    raise ValueError(f"未知专家后端类型: {etype}（支持 mock | hf | api）")
-    import os
+
-    return os.environ.get(name) if name else None
+
-
+def build_expert_pool(experts_cfg: Dict[str, Dict], domains: List[str]) -> Dict[str, Expert]:
-
+    """构建完整专家池。"""
-def build_expert_pool(experts_cfg: Dict[str, Dict], domains: List[str]) -> Dict[str, Expert]:
+    pool: Dict[str, Expert] = {}
-    """构建完整专家池。"""
+    for domain in domains:
-    pool: Dict[str, Expert] = {}
+        cfg = experts_cfg.get(domain, {"type": "mock", "model": "mock"})
-    for domain in domains:
+        pool[domain] = build_expert(domain, cfg)
-        cfg = experts_cfg.get(domain, {"type": "mock", "model": "mock"})
+    return pool
-        pool[domain] = build_expert(domain, cfg)
-    return pool␍

router_system/explain.py

@@ -0,0 +1,106 @@
+"""可解释路由评分层（T-R1，采纳 ai-model-router 五维评分 + 硬过滤带拒绝理由设计）。
+
+对分类产生的候选领域专家做透明评分与解释。路由胜负保持既有逻辑不变
+（本模块是纯解释层，不参与决策）：
+
+- 硬过滤先行：无规则命中/低置信度/缺专家的候选先淘汰，逐条记录人话拒绝理由
+- 五维加权评分（各维归一到 0-1，权重可配）：
+    capability   分类器该领域原始分归一（语义匹配强度）
+    cost_efficiency  1/(1+cost*100) 平滑映射，零成本本地模型恒 1.0
+    speed        1/(1+标称延迟/200)（200ms 参考延迟）
+    reliability  Judge 通过率（无数据时 0.9 中性先验）
+    quality      难度 × 领域纵深启发式
+- 置信度可推导：confidence = 0.8 + (最高分 - 次高分)，分差越大越自信（封顶 0.99）
+- 同分决胜按领域名字典序（与仓库既有确定性约定一致）
+
+输出 RoutingExplanation.to_dict() 直接进 /chat 响应的 route_explanation 字段。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+# 领域纵深启发式（该领域知识可承载的任务深度，0-1）
+_DOMAIN_DEPTH = {
+    "code": 0.90, "math": 0.90, "legal": 0.85, "medical": 0.85, "general": 0.60,
+}
+
+DEFAULT_WEIGHTS: Dict[str, float] = {
+    "capability": 0.35,
+    "cost_efficiency": 0.20,
+    "speed": 0.15,
+    "reliability": 0.20,
+    "quality": 0.10,
+}
+
+
+@dataclass
+class CandidateProfile:
+    """一个候选领域专家的评分输入（由 Router 从分类/专家池/统计装配）。"""
+    domain: str
+    model: str                          # 专家模型名
+    capability: float                   # 分类器该领域原始分归一 0-1
+    cost_per_mtok: float                # $ / 1M output tokens 估计
+    nominal_latency_ms: float           # 标称延迟（专家后端元数据）
+    reliability: float                  # Judge 通过率（无数据给中性先验）
+    difficulty_score: float             # 查询难度分 0-1（质量维输入）
+    hard_fail_reason: Optional[str] = None   # 非 None 即硬过滤淘汰（人话理由）
+
+
+@dataclass
+class RoutingExplanation:
+    """一次路由的完整解释：胜者 + 排名 + 拒绝名单 + 权重快照。"""
+    winner: str                         # 胜出领域；无存活候选时为 "fallback"
+    confidence: float                   # 0.8 + 分差推导（封顶 0.99）
+    ranked: List[Dict[str, Any]] = field(default_factory=list)
+    rejected: List[Dict[str, Any]] = field(default_factory=list)
+    weights: Dict[str, float] = field(default_factory=dict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "winner": self.winner,
+            "confidence": round(self.confidence, 4),
+            "ranked": self.ranked,
+            "rejected": self.rejected,
+            "weights": dict(self.weights),
+        }
+
+
+def score_candidate(c: CandidateProfile, weights: Dict[str, float]) -> Dict[str, Any]:
+    """单候选五维评分（全部绝对映射 0-1，跨候选可比、可单测）。"""
+    dims = {
+        "capability": max(0.0, min(1.0, c.capability)),
+        "cost_efficiency": 1.0 / (1.0 + max(0.0, c.cost_per_mtok) * 100.0),
+        "speed": 1.0 / (1.0 + max(0.0, c.nominal_latency_ms) / 200.0),
+        "reliability": max(0.0, min(1.0, c.reliability)),
+        "quality": max(0.0, min(1.0, c.difficulty_score)) * _DOMAIN_DEPTH.get(c.domain, 0.60),
+    }
+    total = sum(dims[k] * weights.get(k, 0.0) for k in dims)
+    return {"domain": c.domain, "model": c.model,
+            "total": round(total, 4), "dims": {k: round(v, 4) for k, v in dims.items()}}
+
+
+def explain_routing(candidates: List[CandidateProfile],
+                    weights: Optional[Dict[str, float]] = None) -> RoutingExplanation:
+    """硬过滤 -> 五维加权 -> 排名与置信度推导（确定性：同分按领域名字典序）。"""
+    w = dict(DEFAULT_WEIGHTS if weights is None else weights)
+    rejected: List[Dict[str, Any]] = []
+    alive: List[CandidateProfile] = []
+    for c in candidates:
+        if c.hard_fail_reason is not None:
+            rejected.append({"domain": c.domain, "reason": c.hard_fail_reason})
+        else:
+            alive.append(c)
+
+    scored = [score_candidate(c, w) for c in alive]
+    scored.sort(key=lambda s: (-s["total"], s["domain"]))
+
+    if not scored:
+        # 全部被硬过滤（含低置信度直连回退路径）：决策即走大模型回退
+        return RoutingExplanation(winner="fallback", confidence=0.80,
+                                  ranked=[], rejected=rejected, weights=w)
+
+    gap = scored[0]["total"] - (scored[1]["total"] if len(scored) > 1 else 0.0)
+    confidence = min(0.99, 0.80 + max(0.0, gap))
+    return RoutingExplanation(winner=scored[0]["domain"], confidence=confidence,
+                              ranked=scored, rejected=rejected, weights=w)

router_system/fallback.py

@@ -1,99 +1,84 @@
-"""大模型回退层：Mock 与 OpenAI 兼容 API 两种后端。"""
+"""大模型回退层：Mock 与 OpenAI 兼容 API 两种后端。"""
-from __future__ import annotations
+from __future__ import annotations
-
+
-import asyncio
+import asyncio
-from typing import Dict, Optional
+from typing import Dict
-
+
-from .models import ExpertResponse
+from .config import get_api_key
-
+from .llm_client import OpenAICompatClient
-
+from .models import ExpertResponse
-class FallbackProvider:
+
-    name: str = "fallback"
+
-
+class FallbackProvider:
-    async def generate(self, query: str) -> ExpertResponse:
+    name: str = "fallback"
-        raise NotImplementedError
+
-
+    async def generate(self, query: str) -> ExpertResponse:
-
+        raise NotImplementedError
-class MockFallback(FallbackProvider):
+
-    """确定性 mock 大模型：标识为 fallback，便于测试升级路径。"""
+
-
+class MockFallback(FallbackProvider):
-    def __init__(self, model: str = "mock-large"):
+    """确定性 mock 大模型：标识为 fallback，便于测试升级路径。"""
-        self.model = model
+
-        self.name = f"fallback-{model}"
+    def __init__(self, model: str = "mock-large"):
-
+        self.model = model
-    async def generate(self, query: str) -> ExpertResponse:
+        self.name = f"fallback-{model}"
-        await asyncio.sleep(0.002)
+
-        body = (
+    async def generate(self, query: str) -> ExpertResponse:
-            f"（大模型回退）「{query}」\n\n"
+        await asyncio.sleep(0.002)
-            "这是一条来自大模型回退路径的完整回答。\n"
+        body = (
-            "要点：\n"
+            f"（大模型回退）「{query}」\n\n"
-            "1. 对复杂/跨域任务给出综合推理\n"
+            "这是一条来自大模型回退路径的完整回答。\n"
-            "2. 补充领域专家未覆盖的上下文\n"
+            "要点：\n"
-            "3. 给出可执行的后续建议\n"
+            "1. 对复杂/跨域任务给出综合推理\n"
-        )
+            "2. 补充领域专家未覆盖的上下文\n"
-        return ExpertResponse(
+            "3. 给出可执行的后续建议\n"
-            text=body,
+        )
-            model_used=self.model,
+        return ExpertResponse(
-            latency_ms=2.0,
+            text=body,
-            tokens=120,
+            model_used=self.model,
-            cost_est=2.0 * 120 / 1_000_000,
+            latency_ms=2.0,
-        )
+            tokens=120,
-
+            cost_est=2.0 * 120 / 1_000_000,
-
+        )
-class APIFallback(FallbackProvider):
+
-    """OpenAI 兼容大模型 API（如 DeepSeek / OpenAI / 本地 vLLM）。"""
+
-
+class APIFallback(FallbackProvider):
-    def __init__(self, model: str, base_url: str, api_key: str):
+    """OpenAI 兼容大模型 API（如 DeepSeek / OpenAI / 本地 vLLM）。"""
-        self.model = model
+
-        self.base_url = base_url.rstrip("/")
+    def __init__(self, model: str, base_url: str, api_key: str):
-        self.api_key = api_key
+        self.model = model
-        self.name = f"fallback-{model}"
+        self.name = f"fallback-{model}"
-        self._client = None
+        self._client = OpenAICompatClient(base_url, api_key, timeout=90.0)
-
+
-    def _get_client(self):
+    async def generate(self, query: str) -> ExpertResponse:
-        if self._client is None:
+        data = await self._client.chat(
-            import httpx
+            self.model,
-            self._client = httpx.AsyncClient(timeout=90.0)
+            [{"role": "user", "content": query}],
-        return self._client
+            temperature=0.3,
-
+            max_tokens=2048,
-    async def generate(self, query: str) -> ExpertResponse:
+        )
-        client = self._get_client()
+        body = self._client.completion_text(data)
-        resp = await client.post(
+        tokens = self._client.completion_tokens(data, body)
-            f"{self.base_url}/chat/completions",
+        return ExpertResponse(
-            headers={"Authorization": f"Bearer {self.api_key}"},
+            text=body,
-            json={
+            model_used=self.model,
-                "model": self.model,
+            latency_ms=0.0,
-                "messages": [{"role": "user", "content": query}],
+            tokens=tokens,
-                "temperature": 0.3,
+            cost_est=2.0 * tokens / 1_000_000,
-                "max_tokens": 2048,
+        )
-            },
+
-        )
+
-        resp.raise_for_status()
+def build_fallback(cfg: Dict) -> FallbackProvider:
-        data = resp.json()
+    """cfg 为 fallback 段配置。"""
-        body = data["choices"][0]["message"]["content"]
+    ftype = cfg.get("type", "mock")
-        usage = data.get("usage", {})
+    model = cfg.get("model", "deepseek-chat")
-        tokens = usage.get("completion_tokens", int(len(body) / 2.2))
+    if ftype == "mock":
-        return ExpertResponse(
+        return MockFallback(model=model)
-            text=body,
+    if ftype == "api":
-            model_used=self.model,
+        api_key = cfg.get("api_key") or get_api_key(cfg)
-            latency_ms=0.0,
+        if not api_key:
-            tokens=tokens,
+            raise RuntimeError(
-            cost_est=2.0 * tokens / 1_000_000,
+                f"APIFallback 缺少 API Key：请设置环境变量 {cfg.get('api_key_env')} 或配置 api_key"
-        )
+            )
-
+        return APIFallback(model, cfg.get("base_url", "https://api.deepseek.com/v1"), api_key)
-
+    raise ValueError(f"未知 fallback 类型: {ftype}（支持 mock | api）")
-def build_fallback(cfg: Dict) -> FallbackProvider:
-    """cfg 为 fallback 段配置。"""
-    ftype = cfg.get("type", "mock")
-    model = cfg.get("model", "deepseek-chat")
-    if ftype == "mock":
-        return MockFallback(model=model)
-    if ftype == "api":
-        import os
-        api_key = cfg.get("api_key") or os.environ.get(cfg.get("api_key_env", "API_KEY"))
-        if not api_key:
-            raise RuntimeError(
-                f"APIFallback 缺少 API Key：请设置环境变量 {cfg.get('api_key_env')} 或配置 api_key"
-            )
-        return APIFallback(model, cfg.get("base_url", "https://api.deepseek.com/v1"), api_key)
-    raise ValueError(f"未知 fallback 类型: {ftype}（支持 mock | api）")␍

router_system/judge.py

@@ -1,174 +1,168 @@
-"""质量控制器（Judge）：评估专家输出，决定是否升级大模型。
+"""质量控制器（Judge）：评估专家输出，决定是否升级大模型。
-
+
-- RuleJudge：零依赖启发式（内容覆盖度 / 长度充分性 / 领域格式 / 安全提示），
+- RuleJudge：零依赖启发式（内容覆盖度 / 长度充分性 / 领域格式 / 安全提示），
-  稳定可测，适合 MVP 与离线演示。
+  稳定可测，适合 MVP 与离线演示。
-- LLMJudge：可选，基于 transformers 小模型或 API 的 LLM-as-Judge。
+- LLMJudge：可选，基于 transformers 小模型或 API 的 LLM-as-Judge。
-
+
-设计对齐实现方案：overall_score < judge_fallback_threshold -> 升级大模型。
+设计对齐实现方案：overall_score < judge_fallback_threshold -> 升级大模型。
-"""
+"""
-from __future__ import annotations
+from __future__ import annotations
-
+
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field
-from typing import Dict, List
+from typing import Dict, List
-
+
-from .experts import extract_content_terms
+from .experts import extract_content_terms
-
+
-# 各领域期望的响应长度范围（字符数）
+# 各领域期望的响应长度范围（字符数）
-_EXPECTED_LEN = {
+_EXPECTED_LEN = {
-    "code": (60, 2000),
+    "code": (60, 2000),
-    "math": (60, 2000),
+    "math": (60, 2000),
-    "legal": (80, 3000),
+    "legal": (80, 3000),
-    "medical": (80, 3000),
+    "medical": (80, 3000),
-    "general": (40, 2000),
+    "general": (40, 2000),
-}
+}
-
+
-# 领域格式检查：响应应包含的标记
+# 领域格式检查：响应应包含的标记
-_DOMAIN_FORMAT_HINTS = {
+_DOMAIN_FORMAT_HINTS = {
-    "code": ["```", "def ", "function", "class "],
+    "code": ["```", "def ", "function", "class "],
-    "math": ["步骤", "推导", "=", "解", "step"],
+    "math": ["步骤", "推导", "=", "解", "step"],
-    "legal": ["⚠", "法律", "意见", "合规", "contract", "law"],
+    "legal": ["⚠", "法律", "意见", "合规", "contract", "law"],
-    "medical": ["⚠", "就医", "医生", "症状", "诊断", "symptom"],
+    "medical": ["⚠", "就医", "医生", "症状", "诊断", "symptom"],
-    "general": [],
+    "general": [],
-}
+}
-
+
-
+
-@dataclass
+@dataclass
-class QualityEvaluation:
+class QualityEvaluation:
-    overall_score: float
+    overall_score: float
-    scores: Dict[str, float] = field(default_factory=dict)
+    scores: Dict[str, float] = field(default_factory=dict)
-    needs_fallback: bool = False
+    needs_fallback: bool = False
-    reasons: List[str] = field(default_factory=list)
+    reasons: List[str] = field(default_factory=list)
-
+
-    def to_dict(self) -> Dict:
+    def to_dict(self) -> Dict:
-        return {
+        return {
-            "overall_score": round(self.overall_score, 4),
+            "overall_score": round(self.overall_score, 4),
-            "scores": {k: round(v, 4) for k, v in self.scores.items()},
+            "scores": {k: round(v, 4) for k, v in self.scores.items()},
-            "needs_fallback": self.needs_fallback,
+            "needs_fallback": self.needs_fallback,
-            "reasons": self.reasons,
+            "reasons": self.reasons,
-        }
+        }
-
+
-
+
-class BaseJudge:
+class BaseJudge:
-    async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
+    async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
-        raise NotImplementedError
+        raise NotImplementedError
-
+
-
+
-class RuleJudge(BaseJudge):
+class RuleJudge(BaseJudge):
-    """启发式质量评估（零依赖）。"""
+    """启发式质量评估（零依赖）。"""
-
+
-    def __init__(self, fallback_threshold: float = 0.70):
+    def __init__(self, fallback_threshold: float = 0.70):
-        self.fallback_threshold = fallback_threshold
+        self.fallback_threshold = fallback_threshold
-
+
-    async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
+    async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
-        scores: Dict[str, float] = {}
+        scores: Dict[str, float] = {}
-        reasons: List[str] = []
+        reasons: List[str] = []
-
+
-        # 1) 内容覆盖度：查询中的内容词有多少出现在响应里
+        # 1) 内容覆盖度：查询中的内容词有多少出现在响应里
-        terms = extract_content_terms(query)
+        terms = extract_content_terms(query)
-        if terms:
+        if terms:
-            hit = sum(1 for t in terms if t in response.lower())
+            lowered = response.lower()  # lowercase 一次，避免逐词重复复制长响应
-            coverage = hit / len(terms)
+            hit = sum(1 for t in terms if t in lowered)
-            scores["coverage"] = coverage
+            coverage = hit / len(terms)
-            if coverage < 0.4:
+            scores["coverage"] = coverage
-                reasons.append(f"内容覆盖度低 ({coverage:.0%})")
+            if coverage < 0.4:
-        else:
+                reasons.append(f"内容覆盖度低 ({coverage:.0%})")
-            scores["coverage"] = 1.0
+        else:
-
+            scores["coverage"] = 1.0
-        # 2) 长度充分性
+
-        lo, hi = _EXPECTED_LEN.get(domain, (40, 2000))
+        # 2) 长度充分性
-        n = len(response)
+        lo, hi = _EXPECTED_LEN.get(domain, (40, 2000))
-        if n < lo:
+        n = len(response)
-            scores["length"] = max(0.0, n / lo)
+        if n < lo:
-            reasons.append(f"响应过短 ({n} 字符)")
+            scores["length"] = max(0.0, n / lo)
-        elif n > hi:
+            reasons.append(f"响应过短 ({n} 字符)")
-            scores["length"] = 0.8
+        elif n > hi:
-            reasons.append(f"响应过长 ({n} 字符)")
+            scores["length"] = 0.8
-        else:
+            reasons.append(f"响应过长 ({n} 字符)")
-            scores["length"] = 1.0
+        else:
-
+            scores["length"] = 1.0
-        # 3) 领域格式检查
+
-        hints = _DOMAIN_FORMAT_HINTS.get(domain, [])
+        # 3) 领域格式检查
-        if hints:
+        hints = _DOMAIN_FORMAT_HINTS.get(domain, [])
-            hit_hints = sum(1 for h in hints if h in response)
+        if hints:
-            scores["format"] = min(1.0, 0.4 + 0.2 * hit_hints)
+            hit_hints = sum(1 for h in hints if h in response)
-            if hit_hints == 0:
+            scores["format"] = min(1.0, 0.4 + 0.2 * hit_hints)
-                reasons.append("缺少领域格式特征")
+            if hit_hints == 0:
-        else:
+                reasons.append("缺少领域格式特征")
-            scores["format"] = 1.0
+        else:
-
+            scores["format"] = 1.0
-        # 4) 安全/免责提示（法律、医疗领域应有警示语）
+
-        if domain in ("legal", "medical") and ("⚠" not in response and "提示" not in response):
+        # 4) 安全/免责提示（法律、医疗领域应有警示语）
-            scores["safety"] = 0.6
+        if domain in ("legal", "medical") and ("⚠" not in response and "提示" not in response):
-            reasons.append("缺少免责提示")
+            scores["safety"] = 0.6
-        else:
+            reasons.append("缺少免责提示")
-            scores["safety"] = 1.0
+        else:
-
+            scores["safety"] = 1.0
-        weights = {"coverage": 0.4, "length": 0.2, "format": 0.2, "safety": 0.2}
+
-        overall = sum(scores.get(k, 0.0) * w for k, w in weights.items())
+        weights = {"coverage": 0.4, "length": 0.2, "format": 0.2, "safety": 0.2}
-        needs = overall < self.fallback_threshold
+        overall = sum(scores.get(k, 0.0) * w for k, w in weights.items())
-        if needs:
+        needs = overall < self.fallback_threshold
-            reasons.append("质量分低于阈值，建议升级大模型")
+        if needs:
-        return QualityEvaluation(
+            reasons.append("质量分低于阈值，建议升级大模型")
-            overall_score=round(overall, 4),
+        return QualityEvaluation(
-            scores=scores,
+            overall_score=round(overall, 4),
-            needs_fallback=needs,
+            scores=scores,
-            reasons=reasons,
+            needs_fallback=needs,
-        )
+            reasons=reasons,
-
+        )
-
+
-class LLMJudge(BaseJudge):
+
-    """可选：LLM-as-Judge（API 后端）。"""
+class LLMJudge(BaseJudge):
-
+    """可选：LLM-as-Judge（API 后端）。"""
-    def __init__(self, model: str, base_url: str, api_key: str, fallback_threshold: float = 0.70):
+
-        self.model = model
+    def __init__(self, model: str, base_url: str, api_key: str, fallback_threshold: float = 0.70):
-        self.base_url = base_url.rstrip("/")
+        from .llm_client import OpenAICompatClient
-        self.api_key = api_key
+        self.model = model
-        self.fallback_threshold = fallback_threshold
+        self.fallback_threshold = fallback_threshold
-        self._client = None
+        # 请求体与旧实现一致：不下发 temperature / max_tokens
-
+        self._client = OpenAICompatClient(base_url, api_key, timeout=60.0)
-    def _get_client(self):
+
-        if self._client is None:
+    async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
-            import httpx
+        prompt = (
-            self._client = httpx.AsyncClient(timeout=60.0)
+            f"你是质量评审员。评估以下回答对查询的满足程度，输出 0-1 分（相关性/正确性/完整性）。\n"
-        return self._client
+            f"查询: {query}\n领域: {domain}\n回答: {response[:2000]}\n"
-
+            f"只输出一个 0 到 1 之间的数字。"
-    async def evaluate(self, query: str, response: str, domain: str) -> QualityEvaluation:
+        )
-        client = self._get_client()
+        data = await self._client.chat(
-        prompt = (
+            self.model,
-            f"你是质量评审员。评估以下回答对查询的满足程度，输出 0-1 分（相关性/正确性/完整性）。\n"
+            [{"role": "user", "content": prompt}],
-            f"查询: {query}\n领域: {domain}\n回答: {response[:2000]}\n"
+            temperature=None,
-            f"只输出一个 0 到 1 之间的数字。"
+            max_tokens=None,
-        )
+        )
-        resp = await client.post(
+        try:
-            f"{self.base_url}/chat/completions",
+            score = float(self._client.completion_text(data).strip())
-            headers={"Authorization": f"Bearer {self.api_key}"},
+            score = max(0.0, min(1.0, score))
-            json={"model": self.model, "messages": [{"role": "user", "content": prompt}]},
+        except Exception:
-        )
+            score = 0.5
-        resp.raise_for_status()
+        return QualityEvaluation(
-        try:
+            overall_score=score,
-            score = float(resp.json()["choices"][0]["message"]["content"].strip())
+            scores={"llm_judge": score},
-            score = max(0.0, min(1.0, score))
+            needs_fallback=score < self.fallback_threshold,
-        except Exception:
+        )
-            score = 0.5
+
-        return QualityEvaluation(
+
-            overall_score=score,
+def build_judge(cfg: Dict, fallback_threshold: float = 0.70) -> BaseJudge:
-            scores={"llm_judge": score},
+    """cfg 为 judge 段配置。"""
-            needs_fallback=score < self.fallback_threshold,
+    jtype = cfg.get("type", "rule")
-        )
+    if jtype == "rule":
-
+        return RuleJudge(fallback_threshold=fallback_threshold)
-
+    if jtype == "llm":
-def build_judge(cfg: Dict, fallback_threshold: float = 0.70) -> BaseJudge:
+        from .config import get_api_key
-    """cfg 为 judge 段配置。"""
+        api_key = cfg.get("api_key") or get_api_key(cfg)
-    jtype = cfg.get("type", "rule")
+        return LLMJudge(
-    if jtype == "rule":
+            cfg.get("model", "deepseek-chat"),
-        return RuleJudge(fallback_threshold=fallback_threshold)
+            cfg.get("base_url", "https://api.deepseek.com/v1"),
-    if jtype == "llm":
+            api_key or "",
-        import os
+            fallback_threshold=fallback_threshold,
-        api_key = cfg.get("api_key") or os.environ.get(cfg.get("api_key_env", "API_KEY"))
+        )
-        return LLMJudge(
+    raise ValueError(f"未知 judge 类型: {jtype}（支持 rule | llm）")
-            cfg.get("model", "deepseek-chat"),
-            cfg.get("base_url", "https://api.deepseek.com/v1"),
-            api_key or "",
-            fallback_threshold=fallback_threshold,
-        )
-    raise ValueError(f"未知 judge 类型: {jtype}（支持 rule | llm）")␍

router_system/llm_client.py

@@ -0,0 +1,69 @@
+"""OpenAI 兼容 Chat Completions 共享客户端（架构去重）。
+
+此前 experts.APIExpert / judge.LLMJudge / fallback.APIFallback 各自复制一份
+"懒建 AsyncClient + POST /chat/completions + choices/usage 解析"的等价逻辑
+（超时、温度、token 上限各自维护）。本模块将其统一为单一组件：
+
+- 懒建 httpx.AsyncClient 并长期复用（与仓库内其他客户端一致）
+- temperature / max_tokens 传 None 时不出现在请求体（保持各后端原请求形状）
+- 密钥解析统一走 config.get_api_key（此前 experts 的默认环境名与另两处不一致）
+
+依赖说明：httpx 仅 api 类型后端需要，import 延迟到连接创建时，
+router_system 核心保持零第三方依赖约束不变。
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+
+class OpenAICompatClient:
+    """OpenAI 兼容 /chat/completions 客户端（懒建连接、跨调用复用）。"""
+
+    def __init__(self, base_url: str, api_key: str, timeout: float = 60.0):
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self._timeout = timeout
+        self._client = None
+
+    def _get_client(self):
+        if self._client is None:
+            import httpx
+            self._client = httpx.AsyncClient(timeout=self._timeout)
+        return self._client
+
+    async def chat(
+        self,
+        model: str,
+        messages: List[Dict[str, str]],
+        temperature: Optional[float] = 0.2,
+        max_tokens: Optional[int] = 1024,
+    ) -> Dict[str, Any]:
+        """调用 /chat/completions，返回完整响应 JSON（含 usage）。
+
+        HTTP 非 2xx 时抛 httpx.HTTPStatusError；temperature/max_tokens
+        为 None 时对应字段不下发（与旧实现中 LLMJudge 的请求体一致）。
+        """
+        payload: Dict[str, Any] = {"model": model, "messages": messages}
+        if temperature is not None:
+            payload["temperature"] = temperature
+        if max_tokens is not None:
+            payload["max_tokens"] = max_tokens
+        client = self._get_client()
+        resp = await client.post(
+            f"{self.base_url}/chat/completions",
+            headers={"Authorization": f"Bearer {self.api_key}"},
+            json=payload,
+        )
+        resp.raise_for_status()
+        return resp.json()
+
+    @staticmethod
+    def completion_text(data: Dict[str, Any]) -> str:
+        """提取首条回复文本。"""
+        return data["choices"][0]["message"]["content"]
+
+    @staticmethod
+    def completion_tokens(data: Dict[str, Any], text: str) -> int:
+        """提取 completion token 数；usage 缺失时按 ~2.2 字符/token 估算。"""
+        usage = data.get("usage", {})
+        return usage.get("completion_tokens", int(len(text) / 2.2))

router_system/models.py

@@ -1,68 +1,70 @@
-"""核心数据模型（纯标准库，无外部依赖）"""
+"""核心数据模型（纯标准库，无外部依赖）"""
-from __future__ import annotations
+from __future__ import annotations
-
+
-import time
+import time
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional
-
+
-
+
-@dataclass
+@dataclass
-class Classification:
+class Classification:
-    """分类器输出：领域 + 置信度 + 难度"""
+    """分类器输出：领域 + 置信度 + 难度"""
-    domain: str
+    domain: str
-    confidence: float
+    confidence: float
-    difficulty: str                      # easy | medium | hard
+    difficulty: str                      # easy | medium | hard
-    difficulty_score: float = 0.5
+    difficulty_score: float = 0.5
-    raw_scores: Dict[str, float] = field(default_factory=dict)
+    raw_scores: Dict[str, float] = field(default_factory=dict)
-    matched_rules: List[str] = field(default_factory=list)
+    matched_rules: List[str] = field(default_factory=list)
-
+
-
+
-@dataclass
+@dataclass
-class ExpertResponse:
+class ExpertResponse:
-    """专家模型输出"""
+    """专家模型输出"""
-    text: str
+    text: str
-    model_used: str
+    model_used: str
-    latency_ms: float = 0.0
+    latency_ms: float = 0.0
-    tokens: int = 0
+    tokens: int = 0
-    cost_est: float = 0.0                # 相对成本估计（美元，近似）
+    cost_est: float = 0.0                # 相对成本估计（美元，近似）
-
+
-
+
-@dataclass
+@dataclass
-class RouterResult:
+class RouterResult:
-    """一次路由的完整结果"""
+    """一次路由的完整结果"""
-    query: str
+    query: str
-    response: str
+    response: str
-    domain: str
+    domain: str
-    difficulty: str
+    difficulty: str
-    confidence: float
+    confidence: float
-    upgraded: bool                       # 是否升级到大模型
+    upgraded: bool                       # 是否升级到大模型
-    quality_score: float
+    quality_score: float
-    model_used: str
+    model_used: str
-    route: List[str] = field(default_factory=list)   # 路由决策轨迹
+    route: List[str] = field(default_factory=list)   # 路由决策轨迹
-    latency_ms: float = 0.0
+    latency_ms: float = 0.0
-    cache_hit: bool = False
+    cache_hit: bool = False
-    cache_level: Optional[str] = None    # exact | semantic
+    cache_level: Optional[str] = None    # exact | semantic
-    cost_est: float = 0.0
+    cost_est: float = 0.0
-    error: Optional[str] = None
+    error: Optional[str] = None
-
+    route_explanation: Optional[Dict[str, Any]] = None  # 可解释评分（T-R1，缓存命中为 None）
-    def to_dict(self) -> Dict[str, Any]:
+
-        return {
+    def to_dict(self) -> Dict[str, Any]:
-            "query": self.query,
+        return {
-            "response": self.response,
+            "query": self.query,
-            "domain": self.domain,
+            "response": self.response,
-            "difficulty": self.difficulty,
+            "domain": self.domain,
-            "confidence": round(self.confidence, 4),
+            "difficulty": self.difficulty,
-            "upgraded": self.upgraded,
+            "confidence": round(self.confidence, 4),
-            "quality_score": round(self.quality_score, 4),
+            "upgraded": self.upgraded,
-            "model_used": self.model_used,
+            "quality_score": round(self.quality_score, 4),
-            "route": self.route,
+            "model_used": self.model_used,
-            "latency_ms": round(self.latency_ms, 2),
+            "route": self.route,
-            "cache_hit": self.cache_hit,
+            "latency_ms": round(self.latency_ms, 2),
-            "cache_level": self.cache_level,
+            "cache_hit": self.cache_hit,
-            "cost_est": round(self.cost_est, 6),
+            "cache_level": self.cache_level,
-            "error": self.error,
+            "cost_est": round(self.cost_est, 6),
-        }
+            "error": self.error,
-
+            "route_explanation": self.route_explanation,
-
+        }
-def now_ms() -> float:
+
-    return time.perf_counter() * 1000.0␍
+
+def now_ms() -> float:
+    return time.perf_counter() * 1000.0

router_system/router.py

@@ -1,217 +1,257 @@
-"""主路由器：协调 缓存 -> 分类 -> 专家 -> Judge -> 大模型回退 的完整链路。
+"""主路由器：协调 缓存 -> 分类 -> 专家 -> Judge -> 大模型回退 的完整链路。
-
+
-流程（对齐实现方案）：
+流程（对齐实现方案）：
-1. 检查缓存（L1 精确 / L2 语义）
+1. 检查缓存（L1 精确 / L2 语义）
-2. 低置信度查询直接走大模型（should_fallback）
+2. 低置信度查询直接走大模型（should_fallback）
-3. 分类器输出领域 + 难度
+3. 分类器输出领域 + 难度
-4. 选择专家模型生成
+4. 选择专家模型生成
-5. Judge 评估质量
+5. Judge 评估质量
-6. 质量不达标 -> 升级大模型
+6. 质量不达标 -> 升级大模型
-7. 记录指标、写缓存、返回结果
+7. 记录指标、写缓存、返回结果
-"""
+"""
-from __future__ import annotations
+from __future__ import annotations
-
+
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional
-
+
-from .cache import RouterCache
+from .cache import RouterCache
-from .classifier import BaseClassifier, build_classifier
+from .classifier import BaseClassifier, build_classifier
-from .config import load_config
+from .config import load_config
-from .experts import Expert, build_expert_pool
+from .experts import Expert, build_expert_pool
-from .fallback import FallbackProvider, build_fallback
+from .fallback import FallbackProvider, build_fallback
-from .judge import BaseJudge, build_judge
+from .judge import BaseJudge, build_judge
-from .models import Classification, ExpertResponse, RouterResult, now_ms
+from .models import Classification, ExpertResponse, RouterResult, now_ms
-from .stats import Stats
+from .stats import Stats
-
+
-
+
-class Router:
+class Router:
-    def __init__(
+    def __init__(
-        self,
+        self,
-        classifier: BaseClassifier,
+        classifier: BaseClassifier,
-        experts: Dict[str, Expert],
+        experts: Dict[str, Expert],
-        judge: BaseJudge,
+        judge: BaseJudge,
-        fallback: FallbackProvider,
+        fallback: FallbackProvider,
-        cache: Optional[RouterCache] = None,
+        cache: Optional[RouterCache] = None,
-        stats: Optional[Stats] = None,
+        stats: Optional[Stats] = None,
-        config: Optional[Dict[str, Any]] = None,
+        config: Optional[Dict[str, Any]] = None,
-    ):
+    ):
-        self.classifier = classifier
+        self.classifier = classifier
-        self.experts = experts
+        self.experts = experts
-        self.judge = judge
+        self.judge = judge
-        self.fallback = fallback
+        self.fallback = fallback
-        self.cache = cache or RouterCache()
+        self.cache = cache or RouterCache()
-        self.stats = stats or Stats()
+        self.stats = stats or Stats()
-        cfg = config or {}
+        cfg = config or {}
-        rcfg = cfg.get("router", {})
+        rcfg = cfg.get("router", {})
-        self.low_confidence_threshold = rcfg.get("low_confidence_threshold", 0.60)
+        self.low_confidence_threshold = rcfg.get("low_confidence_threshold", 0.60)
-        self.judge_fallback_threshold = rcfg.get("judge_fallback_threshold", 0.70)
+        self.judge_fallback_threshold = rcfg.get("judge_fallback_threshold", 0.70)
-        self.cache_enabled = cfg.get("cache", {}).get("enabled", True)
+        self.cache_enabled = cfg.get("cache", {}).get("enabled", True)
-
+        # T-R1：可解释路由评分层（纯解释，不影响决策；默认开启）
-    # ---------------------------------------------------------------
+        self.explain_enabled = bool(rcfg.get("explain_enabled", True))
-    async def route(self, query: str) -> RouterResult:
+
-        start = now_ms()
+    # ---------------------------------------------------------------
-        route: list = []
+    async def route(self, query: str) -> RouterResult:
-
+        start = now_ms()
-        # ---- Step 1: 缓存 ----
+        route: list = []
-        if self.cache_enabled:
+
-            hit = self.cache.get(query)
+        # ---- Step 1: 缓存 ----
-            if hit is not None:
+        if self.cache_enabled:
-                level, cached = hit
+            hit = self.cache.get(query)
-                latency = now_ms() - start
+            if hit is not None:
-                result = RouterResult(
+                level, cached = hit
-                    query=query,
+                latency = now_ms() - start
-                    response=cached.get("response", ""),
+                result = RouterResult(
-                    domain=cached.get("domain", "general"),
+                    query=query,
-                    difficulty=cached.get("difficulty", "medium"),
+                    response=cached.get("response", ""),
-                    confidence=cached.get("confidence", 0.0),
+                    domain=cached.get("domain", "general"),
-                    upgraded=False,
+                    difficulty=cached.get("difficulty", "medium"),
-                    quality_score=cached.get("quality_score", 0.0),
+                    confidence=cached.get("confidence", 0.0),
-                    model_used=cached.get("model_used", ""),
+                    upgraded=False,
-                    route=["cache:" + level],
+                    quality_score=cached.get("quality_score", 0.0),
-                    latency_ms=latency,
+                    model_used=cached.get("model_used", ""),
-                    cache_hit=True,
+                    route=["cache:" + level],
-                    cache_level=level,
+                    latency_ms=latency,
-                    cost_est=0.0,
+                    cache_hit=True,
-                )
+                    cache_level=level,
-                self.stats.record(latency, result.domain, result.difficulty, False, True, level, 0.0, result.model_used)
+                    cost_est=0.0,
-                return result
+                )
-            route.append("cache:miss")
+                self.stats.record(latency, result.domain, result.difficulty, False, True, level, 0.0, result.model_used)
-
+                return result
-        # ---- Step 2: 分类 ----
+            route.append("cache:miss")
-        classification = self.classifier.classify(query)
+
-        route.append(f"classify:{classification.domain}@{classification.confidence:.2f}/{classification.difficulty}")
+        # ---- 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")
+        # T-R1：候选装配与可解释评分（纯解释层，不改变后续任何决策分支）
-            fb = await self._call_fallback(query)
+        explanation = self._explain(classification) if self.explain_enabled else None
-            latency = now_ms() - start
+        if explanation is not None:
-            result = self._finalize(query, classification, fb, quality_score=0.0,
+            route.append(f"explain:{explanation.winner}@{explanation.confidence:.2f}")
-                                    upgraded=True, route=route, latency_ms=latency,
+
-                                    model_used=fb.model_used, cost_est=fb.cost_est)
+        # 低置信度 -> 直接走大模型
-            self._record(result, latency)
+        if self.classifier.should_fallback(classification, self.low_confidence_threshold):
-            return result
+            route.append("direct_fallback")
-
+            return await self._fallback_result(query, classification, route, start,
-        # ---- Step 3: 选择专家 ----
+                                               explanation=explanation)
-        domain = classification.domain
+
-        expert = self.experts.get(domain)
+        # ---- Step 3: 选择专家 ----
-        if expert is None:
+        domain = classification.domain
-            expert = self.experts.get("general")
+        expert = self.experts.get(domain)
-            route.append("expert:fallback-to-general")
+        if expert is None:
-        else:
+            expert = self.experts.get("general")
-            route.append(f"expert:{expert.name}")
+            route.append("expert:fallback-to-general")
-
+        else:
-        # ---- Step 4: 生成 ----
+            route.append(f"expert:{expert.name}")
-        try:
+
-            expert_resp = await expert.generate(query, classification.difficulty)
+        # ---- Step 4: 生成 ----
-        except Exception as e:
+        try:
-            self.stats.record_error()
+            expert_resp = await expert.generate(query, classification.difficulty)
-            route.append(f"expert_error:{type(e).__name__}")
+        except Exception as e:
-            fb = await self._call_fallback(query)
+            self.stats.record_error()
-            latency = now_ms() - start
+            route.append(f"expert_error:{type(e).__name__}")
-            result = self._finalize(query, classification, fb, quality_score=0.0,
+            return await self._fallback_result(query, classification, route, start,
-                                    upgraded=True, route=route, latency_ms=latency,
+                                               error=str(e), explanation=explanation)
-                                    model_used=fb.model_used, cost_est=fb.cost_est,
+
-                                    error=str(e))
+        # ---- Step 5: Judge 评估 ----
-            self._record(result, latency)
+        try:
-            return result
+            evaluation = await self.judge.evaluate(query, expert_resp.text, domain)
-
+        except Exception:
-        # ---- Step 5: Judge 评估 ----
+            evaluation = None
-        try:
+            route.append("judge_error")
-            evaluation = await self.judge.evaluate(query, expert_resp.text, domain)
+
-        except Exception:
+        quality_score = evaluation.overall_score if evaluation else 0.0
-            evaluation = None
+        route.append(f"judge:{quality_score:.2f}")
-            route.append("judge_error")
+
-
+        upgraded = False
-        quality_score = evaluation.overall_score if evaluation else 0.0
+        final_resp = expert_resp
-        route.append(f"judge:{quality_score:.2f}")
+        if evaluation is not None and evaluation.needs_fallback:
-
+            route.append("upgrade")
-        upgraded = False
+            final_resp = await self._call_fallback(query)
-        final_resp = expert_resp
+            upgraded = True
-        if evaluation is not None and evaluation.needs_fallback:
+
-            route.append("upgrade")
+        latency = now_ms() - start
-            final_resp = await self._call_fallback(query)
+        result = self._finalize(query, classification, final_resp, quality_score=quality_score,
-            upgraded = True
+                                upgraded=upgraded, route=route, latency_ms=latency,
-
+                                model_used=final_resp.model_used, cost_est=final_resp.cost_est,
-        latency = now_ms() - start
+                                explanation=explanation)
-        result = self._finalize(query, classification, final_resp, quality_score=quality_score,
+        self._record(result, latency)
-                                upgraded=upgraded, route=route, latency_ms=latency,
+
-                                model_used=final_resp.model_used, cost_est=final_resp.cost_est)
+        # 未升级的结果写缓存（解释不随缓存复用——缓存命中路径 explanation=None）
-        self._record(result, latency)
+        if self.cache_enabled and not upgraded and result.response:
-
+            self.cache.put(query, result.to_dict())
-        # 未升级的结果写缓存
+
-        if self.cache_enabled and not upgraded and result.response:
+        return result
-            self.cache.put(query, result.to_dict())
+
-
+    # ---------------------------------------------------------------
-        return result
+    def _explain(self, classification: Classification):
-
+        """装配候选画像并生成可解释评分（T-R1；任何异常不影响主链路）。"""
-    # ---------------------------------------------------------------
+        try:
-    async def _call_fallback(self, query: str) -> ExpertResponse:
+            from .explain import CandidateProfile, explain_routing
-        try:
+            from .experts import _cost_for
-            return await self.fallback.generate(query)
+
-        except Exception as e:
+            raw = classification.raw_scores or {}
-            # 回退也失败：返回错误占位响应
+            raw_max = max(raw.values()) if raw else 0.0
-            return ExpertResponse(
+            candidates = []
-                text=f"[系统错误] 专家与大模型回退均失败：{type(e).__name__}: {e}",
+            for dom, expert in self.experts.items():
-                model_used=f"error:{self.fallback.name}",
+                cap = (raw.get(dom, 0.0) / raw_max) if raw_max > 0 else 0.0
-                cost_est=0.0,
+                reason = "无规则命中（capability=0）" if cap <= 0.0 else None
-            )
+                candidates.append(CandidateProfile(
-
+                    domain=dom, model=getattr(expert, "model", expert.name),
-    @staticmethod
+                    capability=cap,
-    def _finalize(query: str, classification: Classification, resp: ExpertResponse,
+                    cost_per_mtok=_cost_for(getattr(expert, "model", "mock")),
-                  quality_score: float, upgraded: bool, route: list,
+                    nominal_latency_ms=getattr(expert, "nominal_latency_ms", 100.0),
-                  latency_ms: float, model_used: str, cost_est: float,
+                    reliability=self.stats.domain_reliability(dom),
-                  error: Optional[str] = None) -> RouterResult:
+                    difficulty_score=classification.difficulty_score,
-        return RouterResult(
+                    hard_fail_reason=reason))
-            query=query,
+            return explain_routing(candidates)
-            response=resp.text,
+        except Exception:  # noqa: BLE001  解释层故障不影响路由（D-G4 同款纪律）
-            domain=classification.domain,
+            return None
-            difficulty=classification.difficulty,
+
-            confidence=classification.confidence,
+    # ---------------------------------------------------------------
-            upgraded=upgraded,
+    async def _fallback_result(self, query: str, classification: Classification,
-            quality_score=quality_score,
+                               route: list, start: float,
-            model_used=model_used,
+                               error: Optional[str] = None,
-            route=route,
+                               explanation=None) -> RouterResult:
-            latency_ms=latency_ms,
+        """兜底路径的公共收尾：调用大模型回退 -> finalize -> 记录指标。
-            cache_hit=False,
+
-            cost_est=cost_est,
+        低置信度直连、专家异常两条路径共用，避免收尾逻辑三处重复。
-            error=error,
+        """
-        )
+        fb = await self._call_fallback(query)
-
+        latency = now_ms() - start
-    def _record(self, result: RouterResult, latency_ms: float):
+        result = self._finalize(query, classification, fb, quality_score=0.0,
-        self.stats.record(
+                                upgraded=True, route=route, latency_ms=latency,
-            latency_ms,
+                                model_used=fb.model_used, cost_est=fb.cost_est,
-            result.domain,
+                                error=error, explanation=explanation)
-            result.difficulty,
+        self._record(result, latency)
-            result.upgraded,
+        return result
-            result.cache_hit,
+
-            result.cache_level,
+    async def _call_fallback(self, query: str) -> ExpertResponse:
-            result.cost_est,
+        try:
-            result.model_used,
+            return await self.fallback.generate(query)
-        )
+        except Exception as e:
-
+            # 回退也失败：返回错误占位响应
-    # ---------------------------------------------------------------
+            return ExpertResponse(
-    def health(self) -> Dict[str, Any]:
+                text=f"[系统错误] 专家与大模型回退均失败：{type(e).__name__}: {e}",
-        return {
+                model_used=f"error:{self.fallback.name}",
-            "status": "ok",
+                cost_est=0.0,
-            "domains": list(self.experts.keys()),
+            )
-            "classifier": type(self.classifier).__name__,
+
-            "judge": type(self.judge).__name__,
+    @staticmethod
-            "fallback": type(self.fallback).__name__,
+    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, explanation=None) -> RouterResult:
-def build_router(config_path: Optional[str] = None) -> Router:
+        return RouterResult(
-    """从配置构建完整 Router（默认 mock 全链路，零依赖可跑）。"""
+            query=query,
-    config = load_config(config_path)
+            response=resp.text,
-    classifier = build_classifier(config.get("classifier", {}))
+            domain=classification.domain,
-    experts = build_expert_pool(config.get("experts", {}), config.get("domains", []))
+            difficulty=classification.difficulty,
-    judge = build_judge(config.get("judge", {}), config.get("router", {}).get("judge_fallback_threshold", 0.70))
+            confidence=classification.confidence,
-    fallback = build_fallback(config.get("fallback", {}))
+            upgraded=upgraded,
-    cache_cfg = config.get("cache", {})
+            quality_score=quality_score,
-    cache = RouterCache(
+            model_used=model_used,
-        semantic_enabled=cache_cfg.get("semantic_enabled", True),
+            route=route,
-        similarity_threshold=cache_cfg.get("similarity_threshold", 0.88),
+            latency_ms=latency_ms,
-        promote_frequency=cache_cfg.get("promote_frequency", 5),
+            cache_hit=False,
-    )
+            cost_est=cost_est,
-    stats = Stats()
+            error=error,
-    return Router(classifier, experts, judge, fallback, cache, stats, config)␍
+            route_explanation=explanation.to_dict() if explanation is not None else None,
+        )
+
+    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)

router_system/stats.py

@@ -1,63 +1,74 @@
-"""运行指标收集（线程安全，零依赖）。"""
+"""运行指标收集（线程安全，零依赖）。"""
-from __future__ import annotations
+from __future__ import annotations
-
+
-import threading
+import threading
-from collections import Counter, deque
+from collections import Counter, deque
-from typing import Any, Deque, Dict
+from typing import Any, Deque, Dict
-
+
-
+
-class Stats:
+class Stats:
-    def __init__(self, window: int = 1000):
+    def __init__(self, window: int = 1000):
-        self._lock = threading.Lock()
+        self._lock = threading.Lock()
-        self.requests = 0
+        self.requests = 0
-        self.domain_counter: Counter = Counter()
+        self.domain_counter: Counter = Counter()
-        self.difficulty_counter: Counter = Counter()
+        self.difficulty_counter: Counter = Counter()
-        self.upgraded = 0
+        self.upgraded = 0
-        self.cache_hits = 0
+        self.upgraded_by_domain: Counter = Counter()   # 可解释评分 reliability 维输入
-        self.cache_levels: Counter = Counter()
+        self.cache_hits = 0
-        self.errors = 0
+        self.cache_levels: Counter = Counter()
-        self.latencies: Deque[float] = deque(maxlen=window)
+        self.errors = 0
-        self.cost_total = 0.0
+        self.latencies: Deque[float] = deque(maxlen=window)
-        self.model_usage: Counter = Counter()
+        self.cost_total = 0.0
-
+        self.model_usage: Counter = Counter()
-    def record(self, latency_ms: float, domain: str, difficulty: str,
+
-               upgraded: bool, cache_hit: bool, cache_level: str | None,
+    def record(self, latency_ms: float, domain: str, difficulty: str,
-               cost_est: float, model_used: str):
+               upgraded: bool, cache_hit: bool, cache_level: str | None,
-        with self._lock:
+               cost_est: float, model_used: str):
-            self.requests += 1
+        with self._lock:
-            self.domain_counter[domain] += 1
+            self.requests += 1
-            self.difficulty_counter[difficulty] += 1
+            self.domain_counter[domain] += 1
-            if upgraded:
+            self.difficulty_counter[difficulty] += 1
-                self.upgraded += 1
+            if upgraded:
-            if cache_hit:
+                self.upgraded += 1
-                self.cache_hits += 1
+                self.upgraded_by_domain[domain] += 1
-                if cache_level:
+            if cache_hit:
-                    self.cache_levels[cache_level] += 1
+                self.cache_hits += 1
-            self.latencies.append(latency_ms)
+                if cache_level:
-            self.cost_total += cost_est
+                    self.cache_levels[cache_level] += 1
-            self.model_usage[model_used] += 1
+            self.latencies.append(latency_ms)
-
+            self.cost_total += cost_est
-    def record_error(self):
+            self.model_usage[model_used] += 1
-        with self._lock:
+
-            self.errors += 1
+    def record_error(self):
-
+        with self._lock:
-    def summary(self) -> Dict[str, Any]:
+            self.errors += 1
-        with self._lock:
+
-            n = self.requests
+    def domain_reliability(self, domain: str, prior: float = 0.9) -> float:
-            lat = list(self.latencies)
+        """领域可靠性 = 1 - 升级率（无数据时返回中性先验，可解释评分用）。"""
-            avg_lat = sum(lat) / len(lat) if lat else 0.0
+        with self._lock:
-            p99 = sorted(lat)[int(len(lat) * 0.99) - 1] if len(lat) >= 100 else (max(lat) if lat else 0.0)
+            n = self.domain_counter.get(domain, 0)
-            return {
+            if n <= 0:
-                "total_requests": n,
+                return prior
-                "domain_distribution": dict(self.domain_counter),
+            up = self.upgraded_by_domain.get(domain, 0)
-                "difficulty_distribution": dict(self.difficulty_counter),
+            return max(0.0, min(1.0, 1.0 - up / n))
-                "fallback_rate": round(self.upgraded / n, 4) if n else 0.0,
+
-                "upgraded_requests": self.upgraded,
+    def summary(self) -> Dict[str, Any]:
-                "cache_hit_rate": round(self.cache_hits / n, 4) if n else 0.0,
+        with self._lock:
-                "cache_levels": dict(self.cache_levels),
+            n = self.requests
-                "avg_latency_ms": round(avg_lat, 3),
+            lat = list(self.latencies)
-                "p99_latency_ms": round(p99, 3),
+            avg_lat = sum(lat) / len(lat) if lat else 0.0
-                "total_cost_est_usd": round(self.cost_total, 6),
+            p99 = sorted(lat)[int(len(lat) * 0.99) - 1] if len(lat) >= 100 else (max(lat) if lat else 0.0)
-                "model_usage": dict(self.model_usage),
+            return {
-                "errors": self.errors,
+                "total_requests": n,
-            }␍
+                "domain_distribution": dict(self.domain_counter),
+                "difficulty_distribution": dict(self.difficulty_counter),
+                "fallback_rate": round(self.upgraded / n, 4) if n else 0.0,
+                "upgraded_requests": self.upgraded,
+                "cache_hit_rate": round(self.cache_hits / n, 4) if n else 0.0,
+                "cache_levels": dict(self.cache_levels),
+                "avg_latency_ms": round(avg_lat, 3),
+                "p99_latency_ms": round(p99, 3),
+                "total_cost_est_usd": round(self.cost_total, 6),
+                "model_usage": dict(self.model_usage),
+                "errors": self.errors,
+            }

tests/conftest.py

@@ -10,5 +10,5 @@ from router_system.router import build_router
 
 @pytest.fixture()
 def router():
-    """????????? mock ?????????????"""
+    """构建全 mock 链路的 Router 实例（离线可跑，供各测试复用）。"""
     return build_router()

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"}
@@ -13,9 +49,9 @@ def test_exact_hit():
 
 def test_semantic_hit():
     c = RouterCache(semantic_enabled=True, similarity_threshold=0.5)
-    c.put("?python?????", {"response": "code", "domain": "code"})
+    c.put("用 python 实现快速排序", {"response": "code", "domain": "code"})
-    # ?????????? L2
+    # 相似但不完全相同的查询应命中 L2 语义缓存
-    hit = c.get("?python????????")
+    hit = c.get("用 python 实现快速排序的算法")
     assert hit is not None
     assert hit[0] == "semantic"
 
@@ -24,7 +60,7 @@ def test_promote_to_exact():
     c = RouterCache(promote_frequency=3)
     result = {"response": "x", "domain": "general"}
     c.put("query", result)
-    # ?????? 3 ? ? ???????
+    # 语义命中累计 3 次后提升为精确缓存
     for _ in range(3):
         hit = c.get("query")
         assert hit is not None

tests/test_classifier.py

@@ -1,44 +1,63 @@
-"""分类器单元测试。"""
+"""分类器单元测试。"""
-from router_system.classifier import RuleClassifier
+from router_system.classifier import RuleClassifier
-
+
-
+
-def test_code_classification():
+def test_tie_break_is_deterministic():
-    clf = RuleClassifier()
+    """同分决胜：按领域名字典序，与规则表排列顺序无关。"""
-    r = clf.classify("用 Python 写一个快速排序函数")
+    clf = RuleClassifier()
-    assert r.domain == "code"
+    clf.rules = {"zeta": [("x", 1.0)], "alpha": [("x", 1.0)]}
-    assert r.confidence > 0.7
+    r = clf.classify("x")
-
+    assert r.domain == "alpha"
-
+
-def test_math_classification():
+
-    clf = RuleClassifier()
+def test_distinctiveness_penalty():
-    r = clf.classify("求解方程 x^2 - 5x + 6 = 0")
+    """次高分占比高（语义含混）时置信度被压低；单一领域命中不受影响。"""
-    assert r.domain == "math"
+    clf = RuleClassifier()
-    assert r.confidence > 0.7
+    clf.rules = {"a": [("kw", 1.0)], "b": [("kw", 0.9)]}
-
+    r_ambiguous = clf.classify("kw")
-
+    clf_clear = RuleClassifier()
-def test_legal_classification():
+    clf_clear.rules = {"a": [("kw", 1.0)], "b": [("other", 0.1)]}
-    clf = RuleClassifier()
+    r_clear = clf_clear.classify("kw")
-    r = clf.classify("劳动合同到期不续签需要支付经济补偿吗")
+    assert r_clear.confidence > r_ambiguous.confidence
-    assert r.domain == "legal"
+
-
+
-
+def test_code_classification():
-def test_medical_classification():
+    clf = RuleClassifier()
-    clf = RuleClassifier()
+    r = clf.classify("用 Python 写一个快速排序函数")
-    r = clf.classify("高血压患者日常饮食需要注意什么")
+    assert r.domain == "code"
-    assert r.domain == "medical"
+    assert r.confidence > 0.7
-
+
-
+
-def test_general_low_confidence():
+def test_math_classification():
-    clf = RuleClassifier()
+    clf = RuleClassifier()
-    r = clf.classify("今天天气怎么样")
+    r = clf.classify("求解方程 x^2 - 5x + 6 = 0")
-    # 未命中任何领域 -> 低置信度，触发 should_fallback
+    assert r.domain == "math"
-    assert r.domain == "general"
+    assert r.confidence > 0.7
-    assert clf.should_fallback(r, 0.6) is True
+
-
+
-
+def test_legal_classification():
-def test_difficulty_estimation():
+    clf = RuleClassifier()
-    clf = RuleClassifier()
+    r = clf.classify("劳动合同到期不续签需要支付经济补偿吗")
-    easy = clf.classify("1 + 1 = ?")
+    assert r.domain == "legal"
-    hard = clf.classify("证明费马大定理并推导其推论，给出详细步骤")
+
-    assert hard.difficulty in ("medium", "hard")
+
-    assert easy.difficulty == "easy"␍
+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"

tests/test_explain.py

@@ -0,0 +1,90 @@
+"""可解释路由评分层单元测试（T-R1，采纳 ai-model-router 设计）。"""
+import pytest
+
+from router_system.explain import CandidateProfile, explain_routing
+from router_system.router import build_router
+
+
+def _cand(domain, capability, cost=1.0, latency=100.0, reliability=0.9, fail=None):
+    return CandidateProfile(domain=domain, model=f"m-{domain}", capability=capability,
+                            cost_per_mtok=cost, nominal_latency_ms=latency,
+                            reliability=reliability, difficulty_score=0.5,
+                            hard_fail_reason=fail)
+
+
+def test_hard_filter_rejects_with_reason():
+    """硬过滤淘汰带人话理由，被拒者不进排名。"""
+    exp = explain_routing([
+        _cand("code", 1.0),
+        _cand("math", 0.0, fail="无规则命中（capability=0）"),
+    ])
+    assert exp.winner == "code"
+    assert any(r["domain"] == "math" and "无规则命中" in r["reason"] for r in exp.rejected)
+    assert all(s["domain"] != "math" for s in exp.ranked)
+
+
+def test_all_rejected_falls_back():
+    """全部候选被硬过滤：决策者是大模型回退，置信度取基准 0.8。"""
+    exp = explain_routing([_cand("code", 0.0, fail="x"), _cand("math", 0.0, fail="y")])
+    assert exp.winner == "fallback"
+    assert exp.confidence == 0.80
+    assert len(exp.rejected) == 2
+
+
+def test_zero_cost_local_beats_costly():
+    """成本效率维平滑映射：零成本恒 1.0，贵模型显著吃亏。"""
+    cheap = explain_routing([_cand("code", 1.0, cost=0.0, latency=1.0)]).ranked[0]
+    pricey = explain_routing([_cand("math", 1.0, cost=2.5, latency=800.0)]).ranked[0]
+    assert cheap["dims"]["cost_efficiency"] == 1.0
+    assert pricey["dims"]["cost_efficiency"] < 0.5
+    assert cheap["total"] > pricey["total"]
+
+
+def test_confidence_gap_derivation():
+    """置信度 = 0.8 + 分差：独苗封顶 0.99，双候选同分回落 0.8。"""
+    single = explain_routing([_cand("code", 1.0)])
+    tied = explain_routing([_cand("code", 1.0), _cand("math", 1.0)])
+    assert single.confidence == 0.99
+    assert tied.confidence == 0.80
+
+
+def test_tie_break_is_deterministic():
+    """同分决胜按领域名字典序（与仓库既有确定性约定一致）。"""
+    exp = explain_routing([_cand("zeta", 1.0), _cand("alpha", 1.0)])
+    assert exp.winner == "alpha"
+
+
+def test_weights_snapshot_and_rank_order():
+    """排名按 total 降序，权重快照随解释输出。"""
+    exp = explain_routing([
+        _cand("code", 1.0, cost=0.0, latency=1.0),
+        _cand("math", 0.5, cost=2.0, latency=900.0),
+    ])
+    totals = [s["total"] for s in exp.ranked]
+    assert totals == sorted(totals, reverse=True)
+    assert exp.weights["capability"] == 0.35
+
+
+@pytest.mark.asyncio
+async def test_route_attaches_explanation():
+    """主链路返回结构化 route_explanation；缓存命中路径为 None。"""
+    router = build_router()
+    r1 = await router.route("用 Python 写一个快速排序函数")
+    assert r1.route_explanation is not None
+    assert r1.route_explanation["winner"] == "code"
+    assert any("explain:" in step for step in r1.route)
+    assert any("无规则命中" in r["reason"]
+               for r in r1.route_explanation["rejected"])
+    r2 = await router.route("用 Python 写一个快速排序函数")
+    assert r2.cache_hit is True
+    assert r2.route_explanation is None
+
+
+@pytest.mark.asyncio
+async def test_low_confidence_explains_fallback():
+    """低置信度直连回退：解释层给出 fallback 胜者与拒绝名单。"""
+    router = build_router()
+    r = await router.route("今天天气怎么样")
+    assert r.upgraded is True
+    assert r.route_explanation is not None
+    assert r.route_explanation["winner"] == "fallback"

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"

tests/test_gateway.py

@@ -1,4 +1,4 @@
-"""???????? fastapi + httpx??"""
+"""网关集成测试（需 fastapi + httpx，链路全 mock）。"""
 import pytest
 
 pytest.importorskip("fastapi")
@@ -23,7 +23,7 @@ def test_health(client):
 
 
 def test_chat(client):
-    resp = client.post("/chat", json={"query": "? Python ???????"})
+    resp = client.post("/chat", json={"query": "用 Python 写一个快速排序函数"})
     assert resp.status_code == 200
     data = resp.json()
     assert data["response"]

tests/test_rules_external.py

@@ -0,0 +1,53 @@
+"""分类规则外置单元测试（T-R2，采纳 llmrouter「规则文档即配置」设计）。"""
+import json
+
+from router_system.classifier import (DOMAIN_RULES, RuleClassifier, build_classifier,
+                                      load_domain_rules)
+
+
+def test_load_missing_file_returns_none(tmp_path):
+    """文件缺失：返回 None（调用方回退内置），不抛异常。"""
+    assert load_domain_rules(tmp_path / "nope.json") is None
+
+
+def test_external_rules_override_domain(tmp_path):
+    """外置文件按 domain 覆盖内置：改写 code 规则即改变路由行为。"""
+    f = tmp_path / "routes.json"
+    f.write_text(json.dumps({"code": [["速排", 2.0]]}, ensure_ascii=False),
+                 encoding="utf-8")
+    rules = load_domain_rules(f)
+    assert rules is not None and rules["code"] == [("速排", 2.0)]
+
+    clf = RuleClassifier()
+    clf.rules = {**DOMAIN_RULES, **rules}
+    r = clf.classify("讲讲速排的思路")
+    assert r.domain == "code"
+
+
+def test_build_classifier_merges_over_builtin(tmp_path):
+    """build_classifier 合并语义：文件只写 legal，其余领域保持内置。"""
+    f = tmp_path / "routes.json"
+    f.write_text(json.dumps({"legal": [["劳动合同", 3.0]]}, ensure_ascii=False),
+                 encoding="utf-8")
+    clf = build_classifier({"type": "rule", "rules_file": str(f)})
+    assert clf.rules["legal"] == [("劳动合同", 3.0)]
+    assert clf.rules["code"] == DOMAIN_RULES["code"]  # 未覆盖领域保持内置
+    r = clf.classify("劳动合同到期不续签需要支付经济补偿吗")
+    assert r.domain == "legal"
+
+
+def test_malformed_file_falls_back_to_builtin(tmp_path):
+    """格式非法：整体回退内置（失败安全，编辑错误不打垮路由）。"""
+    f = tmp_path / "routes.json"
+    f.write_text('{"code": [["坏数据", "不是数字"]]}', encoding="utf-8")
+    assert load_domain_rules(f) is None
+    clf = build_classifier({"type": "rule", "rules_file": str(f)})
+    assert clf.rules == DOMAIN_RULES
+    r = clf.classify("用 Python 写一个快速排序函数")
+    assert r.domain == "code"
+
+
+def test_explicit_rules_file_missing_uses_builtin(tmp_path):
+    """显式路径不存在：静默回退内置，分类行为不变。"""
+    clf = build_classifier({"type": "rule", "rules_file": str(tmp_path / "no.json")})
+    assert clf.rules == DOMAIN_RULES